vent_agent_3/service/data_service.py

from typing import Any, Dict, Optional

import requests

from config.field_mappings import TOP_LEVEL_MAP, TYPE_MAP
from config.settings import BASE_URL, COAL_WORKING_API, GAS_IDENTIFY_API, VENT_REPORT_API
from service.auth import AuthService


class VentDataService:
    """外部数据获取服务：瓦斯等级鉴定报告、工作面作业规程"""

    def __init__(self) -> None:
        self.base_url = BASE_URL

    # ── private HTTP helpers ──────────────────────────────────────────

    def _post_json(self, url: str, payload: dict) -> dict:
        """发送 JSON POST 请求并返回解析后的响应 dict（不处理业务字段）。"""
        headers: Dict[str, str] = {
            "X-Access-Token": AuthService.get_token(),
            "Content-Type": "application/json",
        }
        resp = requests.post(url, json=payload, headers=headers, timeout=15)
        resp.raise_for_status()
        return resp.json()

    def _get_json(self, url: str, params: dict | None = None) -> dict:
        """发送 GET 请求并返回解析后的响应 dict（不处理业务字段）。"""
        resp = requests.get(url, params=params or {})
        return resp.json() if resp.status_code == 200 else {}

    # ── data fetching ─────────────────────────────────────────────────

    def get_gas_identify(self, mine_name: str) -> dict:
        """调用接口获取瓦斯等级鉴定数据"""
        try:
            return self._get_json(GAS_IDENTIFY_API, {"mineName": mine_name})
        except Exception as e:
            print(f"[data_service] get_gas_identify 异常: {e}")
            return {}

    def get_coal_working_rule(self, mine_name: str) -> list:
        """调用接口获取工作面作业规程数据"""
        try:
            return self._get_json(COAL_WORKING_API, {"mineName": mine_name})
        except Exception as e:
            print(f"[data_service] get_coal_working_rule 异常: {e}")
            return []

    def get_vent_report_data(self, mine_name: str, date_month: str) -> dict:
        """调用接口获取工作面作业规程数据"""
        try:
            return self._get_json(
                VENT_REPORT_API,
                {"mineName": mine_name, "dateMonth": date_month},
            )
        except Exception as e:
            print(f"[data_service] get_vent_report_data 异常: {e}")
            return {}

    def _get_default_model_id(self) -> Optional[str]:
        """获取默认的模型版本 ID"""
        url = f"{self.base_url}/Vmodel/modelParamPub/list"
        headers = {"X-Access-Token": AuthService.get_token()}
        try:
            resp = requests.get(url, headers=headers, timeout=10)
            resp.raise_for_status()
            data = resp.json()
            records = data.get("result", {}).get("param", {}).get("records", [])
            if records:
                default_id = records[0].get("defaultmodelid")
                return str(default_id) if default_id is not None else None
            return None
        except Exception as e:
            print(f"⚠️ 获取默认模型ID失败: {e}，将不过滤数据")
            return None

    def fetch_all_data(self) -> Dict[str, Any]:
        """获取全量数据，并根据默认模型ID过滤，返回原始 JSON 的 'obj' 部分"""
        url = f"{self.base_url}/ventanaly-jingtaifengliang/xufengController/getNeedqAllData"
        data = self._post_json(url, {})

        if not data.get("success"):
            raise Exception(f"数据接口错误: {data.get('msg')}")

        obj = data["obj"]

        default_model_id = self._get_default_model_id()
        if default_model_id is not None:
            for list_key in (
                "retWorkFaceList",
                "headingFaceList",
                "roomList",
                "otherList",
                "ventilationSystemList",
            ):
                if list_key in obj and isinstance(obj[list_key], list):
                    obj[list_key] = [
                        item
                        for item in obj[list_key]
                        if str(item.get("nmodelID") or "") == default_model_id
                    ]
        return obj

    def get_summary(self) -> Dict[str, float]:
        obj = self.fetch_all_data()

        ret_work_faces: list = obj.get("retWorkFaceList") or []
        heading_faces: list = obj.get("headingFaceList") or []
        rooms: list = obj.get("roomList") or []
        others: list = obj.get("otherList") or []
        vent_systems: list = obj.get("ventilationSystemList") or []

        return {
            "q_cf": sum(x.get("fqneed", 0) for x in ret_work_faces),
            "q_hf": sum(x.get("fqneed", 0) for x in heading_faces),
            "q_room": sum(x.get("fqneed", 0) for x in rooms),
            "q_other": sum(x.get("fqneed", 0) for x in others),
            "q_system_total": sum(x.get("qra", 0) for x in vent_systems),
        }

    def get_translated_data(self) -> Dict[str, Any]:
        """返回完整翻译后的数据，所有 key 均为中文"""
        raw_data = self.fetch_all_data()
        return self._translate_obj(raw_data)

    # ── translation helpers ───────────────────────────────────────────

    # 列表键 → TYPE_MAP 键的查找表
    _LIST_KEY_TO_TYPE_MAP: Dict[str, str] = {
        "retWorkFaceList": "采煤工作面",
        "headingFaceList": "掘进工作面",
        "roomList": "硐室",
        "otherList": "其他用风地点",
        "ventilationSystemList": "回风系统",
    }

    def _translate_obj(self, obj: Dict[str, Any]) -> Dict[str, Any]:
        """将原始 obj 中列表字段的 key 翻译为中文。"""
        result: Dict[str, Any] = {}
        for key, value in obj.items():
            cn_key = TOP_LEVEL_MAP.get(key, key)
            if isinstance(value, list):
                list_type_name = self._LIST_KEY_TO_TYPE_MAP.get(key, "")
                field_map: Dict[str, str] = TYPE_MAP.get(list_type_name, {})
                translated_list = []
                for item in value:
                    translated_item: Dict[str, Any] = {}
                    for fk, fv in item.items():
                        cn_fk = field_map.get(fk, fk)
                        translated_item[cn_fk] = fv
                    translated_list.append(translated_item)
                result[cn_key] = translated_list
            else:
                result[cn_key] = value
        return result

    # ── ventilation calculators ───────────────────────────────────────

    def _post_calc(self, url_path: str, params: dict) -> str:
        """通用计算请求：POST → 校验 success → 返回 obj 的字符串形式。"""
        url = f"{self.base_url}/{url_path}"
        try:
            data = self._post_json(url, params)
        except Exception as e:
            return f"❌ 计算接口网络请求失败：{e}"

        if not data.get("success"):
            return f"❌ 计算接口调用失败：{data.get('msg') or data.get('message')}"
        return str(data["obj"])

    def calc_heading_face(self, params: dict) -> str:
        return self._post_calc(
            "ventanaly-jingtaifengliang/xufengController/simulateNeedqHeadingFace",
            params,
        )

    def calc_room(self, params: dict) -> str:
        return self._post_calc(
            "ventanaly-jingtaifengliang/xufengController/getNeedqRoom",
            params,
        )

    def calculate_working_face_ventilation(self, params: dict) -> str:
        """计算采煤/备用工作面需风量。原名 clac_ret_work_face。"""
        return self._post_calc(
            "ventanaly-jingtaifengliang/xufengController/simulateNeedqRetWorkFace",
            params,
        )

    def calculate_other_point_ventilation(self, params: dict) -> str:
        """计算其他用风地点需风量。原名 clac_other。"""
        return self._post_calc(
            "ventanaly-jingtaifengliang/xufengController/simulateNeedqOther",
            params,
        )

    def calculate_system_ventilation(self, params: dict) -> str:
        """计算回风系统总需风量。原名 clac_system。"""
        kaq = params.get("kaq")
        if not kaq:
            return "矿井通风需风系数不能为空"
        qcf = params.get("qcf", 0)
        qsc = params.get("qsc", 0)
        qhf = params.get("qhf", 0)
        qur = params.get("qur", 0)
        qrl = params.get("qrl", 0)
        qra = (qcf + qsc + qhf + qur + qrl) * kaq
        result = params.copy()
        result["qra"] = qra
        return str(result)

    # ── backward-compatible aliases for external callers ──────────────

    def clac_ret_work_face(self, params: dict) -> str:
        """[deprecated] 请使用 calculate_working_face_ventilation。"""
        return self.calculate_working_face_ventilation(params)

    def clac_other(self, params: dict) -> str:
        """[deprecated] 请使用 calculate_other_point_ventilation。"""
        return self.calculate_other_point_ventilation(params)

    def clac_system(self, params: dict) -> str:
        """[deprecated] 请使用 calculate_system_ventilation。"""
        return self.calculate_system_ventilation(params)