vent_agent_2.0/mock_gateway/main_fan.py

"""JSON 走网关，文件用平台凭证直连；与真网关相同的内部接口契约。"""
import asyncio
import copy
from dataclasses import dataclass, field
import hashlib
import json
from pathlib import Path
import uuid

from fastapi import FastAPI, Header, HTTPException, Query, Request, UploadFile, File
from fastapi.responses import FileResponse, StreamingResponse

ROOT = Path(__file__).resolve().parents[1]
INTERNAL_TOKEN = "mock-internal-token"
PLATFORM_TOKEN = "mock-platform-file-token"
MANAGE_TOKEN = "mock-platform-manage-token"
CLIENT_ID = "mock-file-client"


def load_sample():
    return json.loads((ROOT / "examples/main_fan/reviewed_sample.json").read_text(encoding="utf-8"))


def task_input(scenario="original"):
    if scenario == "original":
        return {"mine_code": "mock-ci", "mine_name": "神木市瓷窑塔矿业有限公司",
                "fan_id": "00000541", "fan_name": "地面主通一号", "use_gateway": True,
                "evaluation_time": "2026-10-14T12:00:00+08:00"}
    if scenario == "report":
        return {"mine_code": "mock-zy", "mine_name": "陕西众源煤业集团有限公司",
                "fan_id": "mock-report-fan-1", "fan_name": "1号通风机", "use_gateway": True,
                "evaluation_time": "2026-10-14T12:00:00+08:00"}
    if scenario != "synthetic":
        raise ValueError("场景只能为 original、report 或 synthetic")
    # 额外计算分支验证；以下是合成数据，不宣称陕西局提供了这些量。
    base = {"mine_code": "mock-synthetic", "mine_name": "合成测试煤矿",
            "fan_id": "synthetic-fan-1", "fan_name": "1号风机",
            "evaluation_time": "2026-10-14T12:00:00+08:00", "points": [], "confirmed_bindings": []}
    for code, name, metric, value, unit in [("synthetic-q", "1号风机风量", "flow", 60, "m3/s"),
            ("synthetic-p", "1号风机静压增量", "static_pressure", 500, "Pa"),
            ("synthetic-w", "1号风机轴功率", "shaft_power", 50, "kW")]:
        base["points"].append({"point_code": code, "name": name, "mine_code": base["mine_code"],
            "mine_name": base["mine_name"], "metric": metric, "value": value, "unit": unit,
            "status": "normal", "observed_at": "2026-10-14T12:00:00+08:00",
            "evidence": {"source": "synthetic-test", "method": "synthetic", "excerpt": "合成测试数值"}})
        base["confirmed_bindings"].append({"point_code": code, "fan_id": base["fan_id"],
                                        "confirmed_by": "合成测试", "reason": "仅模拟已确认现场拓扑"})
    return base


@dataclass
class MockState:
    data_dir: Path
    sample: dict = field(default_factory=load_sample)
    registered: dict = field(default_factory=dict)
    enabled: dict = field(default_factory=dict)
    queues: dict = field(default_factory=dict)
    receipts: dict = field(default_factory=dict)
    uploads: dict = field(default_factory=dict)
    calls: list = field(default_factory=list)
    raw_files: bool = False
    operating_data: dict = field(default_factory=dict)
    data_agents: set[str] = field(default_factory=lambda: {"mock-main-fan"})

    @classmethod
    def from_raw_files(cls, data_dir, *, operating_data=None):
        """网页实测只建立原件索引，不加载人工转录的离线夹具。"""
        data_dir = Path(data_dir).resolve()
        assets = {}
        for index, source in enumerate(sorted(data_dir.rglob('*'))):
            if not source.is_file() or source.suffix.lower() not in {".pdf", ".png", ".jpg", ".jpeg", ".webp"}:
                continue
            if not source.resolve().is_relative_to(data_dir) or source.is_symlink():
                continue
            digest = hashlib.sha256(source.read_bytes()).hexdigest()
            key = "performance_pdf" if source.name == "通风机性能测定报告.pdf" else f"source-{digest[:16]}-{index}"
            assets[key] = {"relative_path": source.relative_to(data_dir).as_posix(), "sha256": digest}
        if not assets:
            raise ValueError("测试目录没有 PDF 或图片原件")
        return cls(data_dir, sample={"assets": assets, "devices": [], "points": []}, raw_files=True,
                   operating_data=operating_data or {})

    def asset(self, key):
        expected = self.sample["assets"][key]
        source = (self.data_dir / expected["relative_path"]).resolve()
        if not source.is_relative_to(self.data_dir.resolve()) or not source.is_file():
            raise ValueError("测试样例文件缺失或超出指定目录")
        if hashlib.sha256(source.read_bytes()).hexdigest() != expected["sha256"]:
            if self.raw_files:
                raise ValueError("原始测试文件发生变化，请重启模拟网关刷新文件索引")
            raise ValueError("测试样例内容已变化，请重新核验转录，禁止命中旧缓存")
        return source

    def file_index(self):
        if self.raw_files:
            return [{"file_id": key, "version": value["sha256"][:16], "filename": self.asset(key).name,
                     "path": f"/mock-platform/files/{key}", "params": {}}
                    for key, value in self.sample["assets"].items()]
        self.asset("performance_pdf")
        return [{"file_id": "performance_pdf", "version": self.sample["assets"]["performance_pdf"]["sha256"][:16],
                 "filename": "通风机性能测定报告.pdf", "path": "/mock-platform/files/performance_pdf", "params": {}}]

    def seed_reviewed_cache(self):
        """显式离线演示才调用，生产不自动加载样例缓存。"""
        from db import main_fan_store
        from algorithms.main_fan.models import DocumentFacts
        self.asset("performance_pdf")
        facts = DocumentFacts.model_validate(self.sample["document_facts"])
        main_fan_store.save_document(self.sample["assets"]["performance_pdf"]["sha256"],
                                     facts.model_dump(mode="json"), "manual_transcription")


def create_app(state: MockState):
    app = FastAPI(title="陕西局网关本地模拟（测试数据）")
    app.state.mock = state

    def internal(authorization, agent_code, *, active=True, data_access=False):
        if authorization != f"Bearer {INTERNAL_TOKEN}":
            raise HTTPException(401, "模拟网关凭证无效")
        if not agent_code:
            raise HTTPException(400, "缺少 X-Agent-Code")
        preauthorized = data_access and agent_code in state.data_agents
        if active and agent_code not in state.registered and not preauthorized:
            raise HTTPException(404, "未注册")
        if active and not state.enabled.get(agent_code, preauthorized):
            raise HTTPException(403, "智能体已停止")

    def success(data):
        return {"code": 200, "success": True, "data": data}

    @app.post("/internal/agents/register")
    async def register(payload: dict, authorization: str = Header(""), x_agent_code: str = Header("")):
        internal(authorization, x_agent_code, active=False)
        if payload.get("agentCode") != x_agent_code or not payload.get("checkTypeCode"):
            raise HTTPException(422, "注册身份不匹配或缺少研判类型")
        state.registered[x_agent_code] = payload
        state.enabled.setdefault(x_agent_code, True)
        state.queues.setdefault(x_agent_code, asyncio.Queue())
        state.calls.append({"operation": "register", "agent": x_agent_code})
        return success(payload)

    @app.get("/internal/monitor/token")
    async def token(authorization: str = Header(""), x_agent_code: str = Header("")):
        internal(authorization, x_agent_code, data_access=True)
        state.calls.append({"operation": "token", "agent": x_agent_code})
        return success({"access_token": PLATFORM_TOKEN, "client_id": CLIENT_ID})

    @app.post("/internal/monitor/proxy")
    async def proxy(payload: dict, authorization: str = Header(""), x_agent_code: str = Header("")):
        internal(authorization, x_agent_code, data_access=True)
        path = payload.get("path")
        state.calls.append({"operation": "proxy", "agent": x_agent_code, "path": path})
        mine = (payload.get("params") or {}).get("mineCode")
        if payload.get("method") != "GET":
            return success({"status_code": 405, "body": {"code": 405}})
        kind = {"/mock-platform/main-fan/devices": "devices", "/mock-platform/main-fan/points": "points",
                "/mock-platform/main-fan/reports": "files", "/mock-platform/main-fan/series":"operating_series",
                "/mock-platform/main-fan/states":"fan_state_records", "/mock-platform/main-fan/switches":"switch_records"}.get(path)
        if kind is None:
            return success({"status_code": 404, "body": {"code": 404}})
        rows = state.file_index() if kind == "files" else [r for r in
            [*state.sample.get(kind,[]),*state.operating_data.get(kind,[])] if not mine or r["mine_code"] == mine]
        return success({"status_code": 200, "body": {"code": 200, "data": {"items": copy.deepcopy(rows)}}})

    @app.get('/mock/main-fan/curves/{kind}')
    async def operating_curve(kind: str):
        raw = state.operating_data.get('raw_curves',{}).get(kind)
        if raw is None:
            raise HTTPException(404,'未配置该模拟曲线')
        return copy.deepcopy(raw)

    @app.get("/mock-platform/files/{asset_key}")
    async def download(asset_key: str, request: Request, authorization: str = Header(""), clientid: str = Header("")):
        # 直连文件平台只接收下发的平台鉴权，而不是网关内部凭证。
        if authorization != f"Bearer {PLATFORM_TOKEN}" or clientid != CLIENT_ID:
            raise HTTPException(401, "文件系统凭证无效")
        if asset_key not in state.sample["assets"]:
            raise HTTPException(404, "文件不存在")
        state.calls.append({"operation": "direct_file", "asset": asset_key,
                            "gateway_identity_header_present": bool(request.headers.get("X-Agent-Code"))})
        source = state.asset(asset_key)
        return FileResponse(source, filename=source.name)

    @app.post("/internal/monitor/upload")
    async def upload(file: UploadFile = File(...), path: str = Query("mock-evidence"),
                     authorization: str = Header(""), x_agent_code: str = Header("")):
        internal(authorization, x_agent_code)
        content = await file.read(50 * 1024 * 1024 + 1)
        if len(content) > 50 * 1024 * 1024:
            raise HTTPException(413, "测试文件过大")
        stored_path = "mock-evidence/" + hashlib.sha256(content).hexdigest() + "/" + Path(file.filename or "file").name
        state.uploads[stored_path] = {"sha256": hashlib.sha256(content).hexdigest(), "size": len(content)}
        state.calls.append({"operation": "upload", "agent": x_agent_code})
        return success({"data": {"status": "successed", "path": stored_path}})

    @app.post("/internal/push/ai_platform")
    async def push(payload: dict, authorization: str = Header(""), x_agent_code: str = Header("")):
        internal(authorization, x_agent_code)
        if payload.get("agentCode") != x_agent_code or len(payload.get("taskId", "")) > 32:
            raise HTTPException(422, "推送身份/任务标识错误")
        if any(not payload.get(k) for k in ("taskId", "mineCode", "mineName", "result", "checkTypeCode")):
            raise HTTPException(422, "推送缺少公共字段")
        key = (x_agent_code, payload["taskId"])
        prior = state.receipts.get(key)
        if prior and prior["payload"] != payload:
            raise HTTPException(409, "同一任务载荷变化")
        if not prior:
            receipt = {"checkId": "mock-check-" + uuid.uuid4().hex,
                       "clueIds": ["mock-clue-" + uuid.uuid4().hex for _ in payload.get("clues", [])]}
            state.receipts[key] = {"payload": payload, "receipt": receipt}
        state.calls.append({"operation": "push", "agent": x_agent_code, "task_id": payload["taskId"]})
        return success(state.receipts[key]["receipt"])

    @app.get("/internal/events/stream")
    async def events(request: Request, agentCode: str, authorization: str = Header(""), x_agent_code: str = Header("")):
        internal(authorization, x_agent_code)
        if agentCode != x_agent_code:
            raise HTTPException(403, "订阅身份不匹配")
        async def stream():
            yield ": connected\n\n"
            while state.enabled[agentCode] and not await request.is_disconnected():
                try:
                    item = await asyncio.wait_for(state.queues[agentCode].get(), .25)
                    yield "event: main_fan.judgment\nid: " + item["id"] + "\ndata: " + json.dumps({"data": item["data"]}, ensure_ascii=False) + "\n\n"
                except asyncio.TimeoutError:
                    yield ": heartbeat\n\n"
        return StreamingResponse(stream(), media_type="text/event-stream")

    @app.post("/mock/tasks")
    async def publish_task(payload: dict, authorization: str = Header(""), x_agent_code: str = Header("")):
        internal(authorization, x_agent_code)
        scenario = payload.get("scenario", "original")
        task_id = payload.get("taskId", "mock-" + uuid.uuid4().hex[:20])
        data = {"taskId": task_id, "input": task_input(scenario), "files": []}
        await state.queues[x_agent_code].put({"id": task_id, "data": data})
        return success({"taskId": task_id})

    @app.get("/openapi/agent/manage")
    async def manage(agentCode: str, action: str, token: str):
        if token != MANAGE_TOKEN:
            raise HTTPException(401, "模拟管理凭证无效")
        if agentCode not in state.registered:
            raise HTTPException(404, "未注册")
        if action in {"start", "restart"}:
            state.enabled[agentCode] = True
        elif action == "stop":
            state.enabled[agentCode] = False
        elif action not in {"status", "logs"}:
            raise HTTPException(422, "管理动作无效")
        data = {"agentCode": agentCode, "onlineStatus": "online" if state.enabled[agentCode] else "offline",
                "judgmentCount": sum(code == agentCode for code, _ in state.receipts)}
        if action == "logs":
            data["rows"] = [r for r in state.calls if r.get("agent") == agentCode]
        return {"code": 200, "message": "模拟管理操作成功", "data": data}
    return app