vent_ai_platform/app/services/evaluation_service.py

from __future__ import annotations

from datetime import datetime, timezone

from sqlalchemy.orm import Session

from app.db.base import new_id
from app.db.models.evaluation import EvaluationRun
from app.db.repositories import diagnosis as diagnosis_repo
from app.db.repositories import evaluation as evaluation_repo
from app.db.repositories import monitoring as monitoring_repo
from app.db.repositories import work_order as work_order_repo
from app.schemas.evaluation import EvaluationRunRequest
from app.services.mine_service import require_mine


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


def _grade(score: float) -> str:
    if score >= 90:
        return "A"
    if score >= 80:
        return "B"
    if score >= 70:
        return "C"
    return "D"


def run_evaluation(
    db: Session,
    payload: EvaluationRunRequest,
) -> EvaluationRun:
    mine = require_mine(db, payload.mine_id)
    monitoring = monitoring_repo.monitoring_overview(db, mine.id)
    defects = diagnosis_repo.result_summary(db, mine.id)
    work_orders = work_order_repo.status_summary(db, mine.id)

    sensor_total = max(monitoring["sensor_count"], 1)
    online_ratio = monitoring["online_sensor_count"] / sensor_total
    high_defects = defects.get("high", 0)
    medium_defects = defects.get("medium", 0)
    open_orders = work_orders.get("pending", 0) + work_orders.get("processing", 0)
    total_orders = max(sum(work_orders.values()), 1)

    dimension_scores = {
        "安全性": max(50, 100 - high_defects * 8 - medium_defects * 3),
        "可靠性": round(70 + online_ratio * 30, 2),
        "经济性": 85.0,
        "管理合规": max(60, 100 - open_orders / total_orders * 30),
        "数据质量": round(80 + online_ratio * 15, 2),
    }
    default_weights = {
        "安全性": 0.3,
        "可靠性": 0.2,
        "经济性": 0.15,
        "管理合规": 0.2,
        "数据质量": 0.15,
    }
    total_score = round(
        sum(dimension_scores[key] * default_weights[key] for key in dimension_scores),
        2,
    )
    grade = _grade(total_score)
    recommendations = []
    if high_defects:
        recommendations.append("优先处置高风险缺陷并完成复核闭环。")
    if online_ratio < 0.95:
        recommendations.append("核查离线传感器通讯与供电状态，提高数据连续性。")
    if open_orders:
        recommendations.append("压降未闭环工单数量，重点关注逾期和超时任务。")
    if not recommendations:
        recommendations.append("维持当前通防业务管控状态，继续按周期复评。")

    run = evaluation_repo.create_run(
        db,
        {
            "mine_id": mine.id,
            "run_no": _run_no(),
            "period_start": payload.period_start,
            "period_end": payload.period_end,
            "status": "completed",
            "total_score": total_score,
            "grade": grade,
            "summary": (
                f"{mine.name}通防业务综合得分 {total_score}，等级 {grade}。"
                f"当前高风险缺陷 {high_defects} 项，传感器在线率 {online_ratio:.1%}。"
            ),
            "recommendations": recommendations,
            "weight_snapshot": default_weights,
            "input_snapshot": payload.context,
            "model_output": {
                "dimension_scores": dimension_scores,
                "monitoring": monitoring,
                "defects": defects,
                "work_orders": work_orders,
            },
            "started_at": datetime.now(timezone.utc),
            "finished_at": datetime.now(timezone.utc),
            "created_by": payload.requested_by,
        },
    )

    indicators = evaluation_repo.list_indicators(db, mine_id=mine.id)
    if indicators:
        for indicator in indicators:
            score = dimension_scores.get(indicator.dimension, 80)
            weight = indicator.weight if indicator.weight else default_weights.get(indicator.dimension, 0)
            evaluation_repo.create_score(
                db,
                {
                    "run_id": run.id,
                    "indicator_id": indicator.id,
                    "raw_value": score,
                    "score": score,
                    "weighted_score": round(score * weight, 2),
                    "weight": weight,
                    "level": _grade(score),
                    "evidence": "根据监测在线率、缺陷数量、工单闭环和业务基础数据综合计算。",
                    "metadata_json": {},
                },
            )
    return run