shanxia2.0/backend/app/services/pl_cert_service.py

"""运输计划 - 证书 CRUD 与到期扫描"""
from __future__ import annotations

from datetime import date
from typing import Optional

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.pl_models import PlCert, PlDocumentReminder, PlPersonnel, PlVehicle
from app.schemas.pl.base_data import CertRead, CertRow


def list_owner_certs(
    db: Session, tenant_no: str, owner_type: str, owner_id: str
) -> list[PlCert]:
    return (
        db.query(PlCert)
        .filter(
            PlCert.tenant_no == tenant_no,
            PlCert.owner_type == owner_type,
            PlCert.owner_id == owner_id,
            PlCert.del_flag == 0,
        )
        .order_by(PlCert.order_num.asc(), PlCert.create_time.asc())
        .all()
    )


def list_owner_certs_batch(
    db: Session, tenant_no: str, owner_type: str, owner_ids: list[str]
) -> dict[str, list[PlCert]]:
    if not owner_ids:
        return {}
    rows = (
        db.query(PlCert)
        .filter(
            PlCert.tenant_no == tenant_no,
            PlCert.owner_type == owner_type,
            PlCert.owner_id.in_(owner_ids),
            PlCert.del_flag == 0,
        )
        .order_by(PlCert.order_num.asc(), PlCert.create_time.asc())
        .all()
    )
    grouped: dict[str, list[PlCert]] = {}
    for cert in rows:
        grouped.setdefault(cert.owner_id, []).append(cert)
    return grouped


def certs_to_read(certs: list[PlCert]) -> list[CertRead]:
    return [
        CertRead(
            id=c.id,
            cert_id=c.id,
            vehicle_part=c.vehicle_part,
            cert_name=c.cert_name,
            cert_no=c.cert_no,
            expire_date=c.expire_date,
            remind_days=c.remind_days,
            attachment_urls=c.attachment_urls,
            owner_type=c.owner_type,
            owner_id=c.owner_id,
            status=c.status,
            tenant_no=c.tenant_no,
            create_time=c.create_time,
            update_time=c.update_time,
        )
        for c in certs
    ]


def _normalize_cert_row(row: CertRow | dict) -> CertRow:
    if isinstance(row, CertRow):
        return row
    return CertRow.model_validate(row)


def save_owner_certs(
    db: Session,
    owner_type: str,
    owner_id: str,
    cert_rows: list[CertRow | dict],
    tenant_no: str,
    *,
    user_id: str,
    dept_id: str,
) -> None:
    """按 cert_id upsert；未出现在列表中的旧证软删。"""
    existing = list_owner_certs(db, tenant_no, owner_type, owner_id)
    existing_map = {c.id: c for c in existing}
    keep_ids: set[str] = set()

    for idx, row in enumerate(cert_rows):
        row = _normalize_cert_row(row)
        cert_id = row.cert_id
        if cert_id and cert_id in existing_map:
            cert = existing_map[cert_id]
            cert.vehicle_part = row.vehicle_part
            cert.cert_name = row.cert_name
            cert.cert_no = row.cert_no
            cert.expire_date = row.expire_date
            cert.remind_days = row.remind_days
            cert.attachment_urls = row.attachment_urls
            cert.order_num = idx
            cert.update_by = user_id
            keep_ids.add(cert.id)
        else:
            cert = PlCert(
                id=generate_uuid(),
                tenant_no=tenant_no,
                owner_type=owner_type,
                owner_id=owner_id,
                vehicle_part=row.vehicle_part,
                cert_name=row.cert_name,
                cert_no=row.cert_no,
                expire_date=row.expire_date,
                remind_days=row.remind_days,
                attachment_urls=row.attachment_urls,
                status="valid",
                order_num=idx,
                create_by=user_id,
                create_dept_id=dept_id,
            )
            db.add(cert)
            db.flush()
            keep_ids.add(cert.id)

    for cert in existing:
        if cert.id not in keep_ids:
            cert.del_flag = 1
            cert.update_by = user_id

    scan_cert_expiry(db, tenant_no)


def nearest_expire_date(
    db: Session, tenant_no: str, owner_type: str, owner_id: str
) -> Optional[date]:
    row = (
        db.query(PlCert.expire_date)
        .filter(
            PlCert.tenant_no == tenant_no,
            PlCert.owner_type == owner_type,
            PlCert.owner_id == owner_id,
            PlCert.del_flag == 0,
            PlCert.expire_date.isnot(None),
        )
        .order_by(PlCert.expire_date.asc())
        .first()
    )
    return row[0] if row else None


def _resolve_owner_label(
    db: Session, tenant_no: str, owner_type: str, owner_id: str
) -> str:
    if owner_type == "vehicle":
        vehicle = (
            db.query(PlVehicle)
            .filter(
                PlVehicle.id == owner_id,
                PlVehicle.tenant_no == tenant_no,
                PlVehicle.del_flag == 0,
            )
            .first()
        )
        return vehicle.license_plate if vehicle else owner_id

    personnel = (
        db.query(PlPersonnel)
        .filter(
            PlPersonnel.id == owner_id,
            PlPersonnel.tenant_no == tenant_no,
            PlPersonnel.del_flag == 0,
        )
        .first()
    )
    return personnel.personnel_name if personnel else owner_id


def scan_cert_expiry(db: Session, tenant_no: Optional[str] = None) -> int:
    """扫描 pl_cert 生成到期提醒，返回新建条数。"""
    today = date.today()
    query = db.query(PlCert).filter(
        PlCert.del_flag == 0,
        PlCert.expire_date.isnot(None),
        PlCert.status == "valid",
    )
    if tenant_no:
        query = query.filter(PlCert.tenant_no == tenant_no)

    created_count = 0
    for cert in query.all():
        days_remaining = (cert.expire_date - today).days
        # 仅跳过尚未进入提醒窗口的证书；已过期（days_remaining < 0）仍需提醒
        if days_remaining > cert.remind_days:
            continue

        owner_label = _resolve_owner_label(
            db, cert.tenant_no, cert.owner_type, cert.owner_id
        )
        driver_id = cert.owner_id if cert.owner_type == "personnel" else None
        driver_name = owner_label if cert.owner_type == "personnel" else None

        existing = (
            db.query(PlDocumentReminder)
            .filter(
                PlDocumentReminder.cert_id == cert.id,
                PlDocumentReminder.status == "pending",
                PlDocumentReminder.del_flag == 0,
            )
            .first()
        )
        if existing:
            existing.owner_type = cert.owner_type
            existing.owner_id = cert.owner_id
            existing.owner_label = owner_label
            existing.driver_id = driver_id
            existing.driver_name = driver_name
            existing.document_type = cert.cert_name
            existing.document_no = cert.cert_no
            existing.expire_date = cert.expire_date
            existing.days_remaining = days_remaining
            continue

        reminder = PlDocumentReminder(
            id=generate_uuid(),
            tenant_no=cert.tenant_no,
            cert_id=cert.id,
            owner_type=cert.owner_type,
            owner_id=cert.owner_id,
            owner_label=owner_label,
            driver_id=driver_id,
            driver_name=driver_name,
            document_type=cert.cert_name,
            document_no=cert.cert_no,
            expire_date=cert.expire_date,
            days_remaining=days_remaining,
            status="pending",
            create_by=cert.create_by or "system",
            create_dept_id=cert.create_dept_id or "",
        )
        db.add(reminder)
        created_count += 1

    return created_count