vent_ai_platform/app/services/work_order_service.py

from __future__ import annotations

from datetime import datetime, timezone

from sqlalchemy.orm import Session

from app.core.errors import APIError
from app.db.base import new_id
from app.db.models.work_order import WorkOrder
from app.db.repositories import work_order as work_order_repo
from app.schemas.work_order import WorkOrderCreate, WorkOrderEventCreate, WorkOrderUpdate


def _order_no() -> str:
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
    return f"WO-{stamp}-{new_id()[:6].upper()}"


def create_work_order(db: Session, payload: WorkOrderCreate) -> WorkOrder:
    data = payload.model_dump(exclude={"metadata"})
    data["order_no"] = _order_no()
    data["metadata_json"] = payload.metadata
    work_order = work_order_repo.create_work_order(db, data)
    work_order_repo.create_event(
        db,
        {
            "work_order_id": work_order.id,
            "event_type": "created",
            "to_status": work_order.status,
            "operator_id": "",
            "operator_name": "",
            "comment": "创建整改任务",
            "attachments": [],
            "created_at": datetime.now(timezone.utc),
        },
    )
    return work_order


def update_work_order(
    db: Session,
    work_order_id: str,
    payload: WorkOrderUpdate,
    *,
    operator_name: str = "",
) -> WorkOrder:
    work_order = work_order_repo.get_work_order(db, work_order_id)
    if work_order is None:
        raise APIError(404, "work_order_not_found", "整改任务不存在")
    old_status = work_order.status
    updates = payload.model_dump(exclude_unset=True)
    if "metadata" in updates:
        updates["metadata_json"] = updates.pop("metadata")
    work_order = work_order_repo.update_work_order(db, work_order, updates)
    if payload.status and payload.status != old_status:
        work_order_repo.create_event(
            db,
            {
                "work_order_id": work_order.id,
                "event_type": "status_changed",
                "from_status": old_status,
                "to_status": payload.status,
                "operator_id": "",
                "operator_name": operator_name,
                "comment": "",
                "attachments": [],
                "created_at": datetime.now(timezone.utc),
            },
        )
    return work_order


def add_event(
    db: Session,
    work_order_id: str,
    payload: WorkOrderEventCreate,
):
    if work_order_repo.get_work_order(db, work_order_id) is None:
        raise APIError(404, "work_order_not_found", "整改任务不存在")
    data = payload.model_dump()
    data["work_order_id"] = work_order_id
    data["created_at"] = datetime.now(timezone.utc)
    return work_order_repo.create_event(db, data)