vent_agent_2.0/extensions/vent_airflow/tools.py

"""风量研判对外工具（需求 §2.3 的 8 个外部工具，其中工具 3 拆为预处理与校验两段）。

每个工具都是薄封装：参数解析、调用 algorithms/airflow_review 的确定性引擎、
把大对象落盘、返回紧凑摘要与证据来源。工具自身不做任何风量计算。
"""
from __future__ import annotations

from pathlib import Path
from typing import Any, Mapping

import config

from algorithms.airflow_review.engine.compare import compare_targets
from algorithms.airflow_review.engine.dedupe import dedupe_findings
from algorithms.airflow_review.engine.findings import FindingFactory, summarize
from algorithms.airflow_review.engine.reconcile import reconcile_sources
from algorithms.airflow_review.engine.trace import trace_evidence
from algorithms.airflow_review.engine.verify import verify_plan
from algorithms.airflow_review.parsers.measurement_report import parse_measurement_report
from algorithms.airflow_review.parsers.plan_extraction import validate_plan_extraction
from algorithms.airflow_review.parsers.plan_preprocess import preprocess_plan_document
from algorithms.airflow_review.report.render import render_report
from algorithms.airflow_review.rules.loader import load_rules
from extensions import define_tool
from extensions.vent_airflow import monitoring, workspace_paths
from extensions.vent_airflow.tool_support import (
    _readable_errors,
    _build_targets,
    _match_summary,
    _manual_summary,
    _collect_evidence,
    _evidence_ids_for_finding,
    _attach_evidence_chains,
    _sources,
    _save,
    _cleaned_text,
    _load_argument,
    _error,
)
from obs import get_logger
from tools.registry import source, sourced

logger = get_logger(__name__)

_GROUP = "风量研判"
_INLINE = config.AIRFLOW_MAX_INLINE_ITEMS


@define_tool(group=_GROUP, label="解析测风报表",
             summary="解析测风报表或人工填报表格，提取总进风、总回风、主要通风机、主要巷道与各用风地点的实测风量，构建研判数据集")
@_readable_errors
def parse_airflow_measurement_report(
    file_path: str,
    mine_id: str = "",
    report_period: str = "",
    report_type: str = "测风报表",
    scope: str = "",
) -> dict[str, Any]:
    """解析测风报表；scope 用逗号分隔，可取 mine_total_intake、mine_total_return、main_fans、main_roadways、workplaces。"""
    try:
        real = workspace_paths.resolve_input(file_path)
    except workspace_paths.WorkspacePathError as exc:
        return _error("parse_airflow_measurement_report", str(exc))
    scopes = [item.strip() for item in str(scope or "").replace("，", ",").split(",") if item.strip()]
    dataset = parse_measurement_report(real, mine_id=mine_id, report_type=report_type,
                                       report_period=report_period, scope=scopes or None)
    if dataset.get("status") == "error":
        return _error("parse_airflow_measurement_report", str(dataset.get("error")), dataset=dataset)
    path = _save(dataset, f"measurement-dataset-{mine_id or 'mine'}", suffix=".json")
    quality = dataset.get("data_quality") or {}
    payload = {
        "dataset_id": dataset["dataset_id"],
        "dataset_path": path,
        "source_file": dataset["source_file"],
        "sheet": dataset.get("sheet"),
        "mine_id": mine_id or dataset.get("mine_id"),
        "report_period": report_period,
        "mine_total_intake_m3_min": (dataset.get("mine_total_intake") or {}).get("airflow_m3_min"),
        "mine_total_return_m3_min": (dataset.get("mine_total_return") or {}).get("airflow_m3_min"),
        "return_systems": dataset.get("return_systems") or [],
        "main_fans": dataset.get("main_fans") or [],
        "effective_airflow_m3_min": (dataset.get("effective_airflow") or {}).get("value"),
        "effective_airflow_rate_percent": (dataset.get("effective_airflow_rate") or {}).get("value"),
        "internal_leakage_rate_percent": (dataset.get("internal_leakage_rate") or {}).get("value"),
        "location_count": len(dataset.get("locations") or []),
        "location_sample": (dataset.get("locations") or [])[:_INLINE],
        "data_quality": quality,
        "pending_review": dataset.get("pending_review") or [],
        "parse_warnings": dataset.get("parse_warnings") or [],
        "next_step": "若存在 pending_review，先与用户确认地点映射与单位，再进入实测比对",
    }
    warnings = [item.get("message") for item in payload["parse_warnings"]]
    if quality.get("unmapped_location_count"):
        warnings.append(f"{quality['unmapped_location_count']} 个测点未判定地点类型，已转人工复核")
    return sourced(ok=True, data=payload, sources=_sources(dataset.get("evidence"), dataset.get("evidence_ids")),
                   warnings=warnings)


# ---- 工具 2：安全监控查询 -------------------------------------------------

@define_tool(group=_GROUP, label="查询监控风量",
             summary="查询安全监控系统的风量、风速与传感器状态序列，按点/小时/班次/日/月聚合；无数据源时明确返回数据不足并转人工复核")
@_readable_errors
def query_monitoring_airflow(
    mine_id: str,
    target_ids: str = "",
    sensor_ids: str = "",
    start_time: str = "",
    end_time: str = "",
    aggregation: str = "day",
) -> dict[str, Any]:
    """查询监控风量；target_ids 与 sensor_ids 用逗号分隔，aggregation 取 point/hour/shift/day/month。"""
    result = monitoring.get_source().query(
        mine_id=mine_id,
        target_ids=[item.strip() for item in str(target_ids or "").replace("，", ",").split(",") if item.strip()],
        sensor_ids=[item.strip() for item in str(sensor_ids or "").replace("，", ",").split(",") if item.strip()],
        start_time=start_time, end_time=end_time, aggregation=aggregation or "day",
    )
    series = result.get("series") or []
    payload = {
        "status": result.get("status"),
        "mine_id": mine_id,
        "window": result.get("window"),
        "series_count": len(series),
        "series_sample": series[:_INLINE],
        "data_quality": result.get("data_quality") or {},
        "reason": result.get("reason", ""),
        "note": result.get("note", ""),
        "manual_review_required": bool(result.get("manual_review_required")),
        "next_step": ("监控数据可用，可用于交叉校核与时间窗补充"
                      if result.get("status") == "ok" else
                      "监控数据不可用：相关对象的连续监测结论必须转人工复核，不得用测风报表替代"),
    }
    if result.get("status") == "error":
        return _error("query_monitoring_airflow", str(result.get("error")))
    return sourced(ok=True, data=payload, sources=[],
                   warnings=[payload["reason"]] if payload["reason"] else [])


# ---- 工具 3a：配风计划预处理 ----------------------------------------------

@define_tool(group=_GROUP, label="预处理配风计划",
             summary="确定性预处理配风计划文档：抽干净文本、去页眉页脚、按坐标重建行序、识别章节/表格/公式，并给出提取模板与清洗后文本路径（不提取任何数值）")
@_readable_errors
def preprocess_air_plan_document(file_path: str, mine_id: str = "", month: str = "") -> dict[str, Any]:
    """预处理配风计划；返回提取模板、章节与表格线索，以及可被 read_file 分段读取的清洗文本路径。"""
    try:
        real = workspace_paths.resolve_input(file_path)
    except workspace_paths.WorkspacePathError as exc:
        return _error("preprocess_air_plan_document", str(exc))
    document = preprocess_plan_document(real, mine_id=mine_id, month=month)
    if document.get("status") == "error":
        return _error("preprocess_air_plan_document", str(document.get("error")))
    text_path = workspace_paths.write_artifact(_cleaned_text(document), f"plan-cleaned-{mine_id or 'mine'}",
                                               suffix=".txt")
    structure_path = _save(document, f"plan-preprocess-{mine_id or 'mine'}")
    payload = {
        "source_file": document["source_file"],
        "doc_type": document["doc_type"],
        "page_count": document["page_count"],
        "cleaned_text_path": text_path,
        "structure_path": structure_path,
        "extraction_template": document["extraction_template"],
        "section_count": len(document.get("sections") or []),
        "sections": (document.get("sections") or [])[:40],
        "formula_count": len(document.get("formulas") or []),
        "tables": [{"table_id": table["table_id"], "page_start": table["page_start"],
                    "page_end": table["page_end"], "row_count": table["row_count"],
                    "value_sum": round(sum(row["value"] for row in table["rows"]), 2),
                    "sample_rows": table["rows"][:5]} for table in (document.get("tables") or [])],
        "noise_removed": document.get("noise_removed"),
        "warnings": document.get("warnings") or [],
        "next_step": (f"用 read_file 分段读取 {text_path}（每次 200~400 行），按 extraction_template "
                      f"全量提取，每个数值带页码与原文片段；提取完调用 validate_plan_extraction 校验"),
    }
    return sourced(ok=True, data=payload, sources=[source(title=document["source_file"], url=f"file://{real}",
                                                snippet=f"共 {document['page_count']} 页")],
                   warnings=[item.get("message", "") for item in payload["warnings"]])


# ---- 工具 3b：提取结果校验 ------------------------------------------------

@define_tool(group=_GROUP, label="校验计划提取", name="validate_plan_extraction",
             summary="校验按模板提取出来的配风计划 JSON：字段完整性、单位、公式映射、层级归属；通过后写出规范化计划文件供复算核验使用")
@_readable_errors
def validate_plan_extraction_tool(extraction: str, mine_id: str = "", rule_version: str = "") -> dict[str, Any]:
    """校验提取结果；extraction 可直接传 JSON 字符串，也可传已写好的 JSON 文件路径。"""
    payload, error = _load_argument(extraction)
    if error:
        return _error("validate_plan_extraction", error)
    result = validate_plan_extraction(payload, mine_id=mine_id, rule_version=rule_version)
    # 只有"连输入都读不出来"才算工具失败。校验发现的阻断性问题属于**正常产出**：
    # 必须把 validation 明细完整交给模型，它才知道按什么去修；早先把这种情况也当失败，
    # 模型只拿到一句空错误，只能反复瞎试。
    if result.get("status") == "error" and result.get("error"):
        return _error("validate_plan_extraction", str(result["error"]))
    if result.get("plan") is None:
        return _error("validate_plan_extraction", "提取结果为空，未解析出任何计划对象")
    plan = result["plan"]
    path = _save(plan, f"plan-normalized-{mine_id or 'mine'}")
    validation = result["validation"]
    body = {
        "status": result["status"],
        "plan_path": path,
        "plan_id": plan["plan_id"],
        "plan_version": plan["plan_version"],
        "item_count": len(plan["items"]),
        "effective_window": plan["effective_window"],
        "validation": {**validation, "issues": validation["issues"][:_INLINE]},
        "pending_review": (result.get("pending_review") or [])[:_INLINE],
        "next_step": result["next_step"],
    }
    return sourced(ok=True, data=body, sources=[],
                   warnings=[issue["message"] for issue in validation["issues"] if issue["severity"] == "major"][:10])


# ---- 工具 4：配风计划核验 -------------------------------------------------

@define_tool(group=_GROUP, label="核验配风计划",
             summary="按规则库自下而上复算各用风地点、各回风系统与矿井总需风量，校验层级汇总关系，提取核验通过的权威需风量并阻断计算错误的对象")
@_readable_errors
def verify_and_extract_plan_airflow(
    plan_path: str,
    mine_id: str = "",
    rule_version: str = "",
    tolerance_relative_percent: float = 0.0,
    tolerance_absolute_m3_min: float = 0.0,
) -> dict[str, Any]:
    """核验配风计划；plan_path 为 validate_plan_extraction 返回的规范化计划文件路径。"""
    plan, error = _load_argument(plan_path)
    if error:
        return _error("verify_and_extract_plan_airflow", error)
    if not isinstance(plan, Mapping) or not plan.get("items"):
        return _error("verify_and_extract_plan_airflow",
                      "计划文件缺少 items；请先调用 validate_plan_extraction 生成规范化计划")
    rules = load_rules(rule_version or None)
    tolerance = {}
    if tolerance_relative_percent:
        tolerance["relative_percent"] = float(tolerance_relative_percent)
    if tolerance_absolute_m3_min:
        tolerance["absolute_m3_min"] = float(tolerance_absolute_m3_min)
    result = verify_plan(plan, rules, tolerance=tolerance, finding_factory=FindingFactory())
    path = _save(result, f"plan-verification-{mine_id or plan.get('mine_id') or 'mine'}")
    verified = result["verified_items"]
    findings = result["findings"]
    hierarchy = result["hierarchy"]
    body = {
        "plan_path": path,
        "plan_id": result["plan_id"],
        "plan_version": result["plan_version"],
        "plan_calculation_status": result["plan_calculation_status"],
        "plan_authoritative": result["plan_authoritative"],
        "rule_version": result["rule_version"],
        "tolerance": result["tolerance"],
        "verified_item_count": len(verified),
        "verified_items": verified,
        "blocked_items": result["blocked_items"],
        "manual_review_item_count": len(result["manual_review_items"]),
        "manual_review_summary": _manual_summary(result.get("manual_review_details") or []),
        "hierarchy": {
            "level_counts": {key: len(value) for key, value in (hierarchy.get("levels") or {}).items()},
            "issues": (hierarchy.get("issues") or [])[:_INLINE],
            "single_return_system": hierarchy.get("single_return_system"),
            "single_return_equivalence": hierarchy.get("single_return_equivalence"),
        },
        "finding_count": len(findings),
        "finding_summary": summarize(findings),
        "findings": findings[:_INLINE],
        "next_step": ("计划核验通过：把 verified_items 作为权威需风量进入实测比对"
                      if result["plan_authoritative"] else
                      "计划存在计算错误：相关对象不得进入正式实测比对，先按 findings 说明问题"),
    }
    return sourced(ok=True, data=body, sources=[],
                   warnings=[finding["conclusion"] for finding in findings
                             if finding.get("severity") == "major"][:5])


# ---- 工具 5：多源交叉校核 -------------------------------------------------

@define_tool(group=_GROUP, label="多源交叉校核", name="reconcile_airflow_sources",
             summary="对同一对象、同一时间窗的测风报表与安全监控数据做时间对齐、偏差计算与可信度判定；偏差超限或时间不对齐时转人工复核")
@_readable_errors
def reconcile_airflow_sources_tool(targets: str) -> dict[str, Any]:
    """交叉校核；targets 为 JSON 或 JSON 文件路径，元素含 target_id、target_type、comparison_window、sources[]。"""
    items, error = _load_argument(targets)
    if error:
        return _error("reconcile_airflow_sources", error)
    if isinstance(items, Mapping):
        items = [items]
    if not isinstance(items, list) or not items:
        return _error("reconcile_airflow_sources", "targets 必须是非空数组")
    rules = load_rules()
    factory = FindingFactory()
    results = [reconcile_sources(item, rules, factory) for item in items]
    findings = [result["finding"] for result in results if result.get("finding")]
    body = {
        "rule_version": rules.rule_version,
        "target_count": len(results),
        "results": [{key: value for key, value in result.items() if key != "finding"} for result in results][:_INLINE],
        "findings": findings,
        "finding_summary": summarize(findings),
        "conflict_count": sum(1 for result in results if result.get("consistency") == "conflict"),
        "manual_review_count": sum(1 for result in results if result.get("requires_manual_review")),
        "next_step": "存在冲突或时间不对齐的对象必须人工复核确认采信值，不得用平均值掩盖",
    }
    return sourced(ok=True, data=body, sources=[], warnings=[finding["conclusion"] for finding in findings][:5])


# ---- 工具 6：实际 vs 需风量比对 -------------------------------------------

@define_tool(group=_GROUP, label="实际与需风量比对",
             summary="把实测风量与核验通过的需风量按对象类型比对：矿井总进风/总回风与回风系统按 1.0，采掘工作面按 75% 红线；时间对不上或缺数据只输出待复核")
@_readable_errors
def compare_airflow_actual_vs_required(
    targets: str = "",
    dataset_path: str = "",
    verification_path: str = "",
    mine_id: str = "",
) -> dict[str, Any]:
    """比对：targets 直接给比对对象数组；也可给 dataset_path + verification_path 由工具自动匹配地点后组装。"""
    rules = load_rules()
    factory = FindingFactory()
    built_from: list[str] = []
    if targets:
        items, error = _load_argument(targets)
        if error:
            return _error("compare_airflow_actual_vs_required", error)
        if isinstance(items, Mapping):
            items = [items]
        if not isinstance(items, list) or not items:
            return _error("compare_airflow_actual_vs_required", "targets 必须是非空数组")
    else:
        if not dataset_path or not verification_path:
            return _error("compare_airflow_actual_vs_required",
                          "需要 targets，或同时给 dataset_path 与 verification_path")
        dataset, error = _load_argument(dataset_path)
        if error:
            return _error("compare_airflow_actual_vs_required", f"数据集读取失败：{error}")
        verification, error = _load_argument(verification_path)
        if error:
            return _error("compare_airflow_actual_vs_required", f"核验结果读取失败：{error}")
        items, match_report = _build_targets(dataset, verification, mine_id=mine_id)
        built_from = ["dataset_path", "verification_path"]
        if not items:
            return _error("compare_airflow_actual_vs_required",
                          "没有可比对的对象：实测地点与核验通过的需风量没有匹配上，请先解决地点映射",
                          match_report=match_report)
    result = compare_targets(items, rules, mine_id=mine_id, factory=factory)
    deduped = dedupe_findings(result["findings"])
    path = _save({**result, "findings": deduped["findings"], "dedupe": deduped},
                 f"comparison-{mine_id or 'mine'}")
    comparisons = result["comparisons"]
    body = {
        "result_path": path,
        "built_from": built_from,
        "rule_version": rules.rule_version,
        "summary": result["summary"],
        "comparisons": [{key: value for key, value in record.items() if key != "finding"}
                        for record in comparisons][:_INLINE],
        "finding_count": len(deduped["findings"]),
        "finding_summary": summarize(deduped["findings"]),
        "findings": [{key: value for key, value in finding.items() if key != "calculation_trace"}
                     for finding in deduped["findings"]][:_INLINE],
        "dedupe": {"merged_count": deduped["merged_count"], "reuse_groups": deduped["reuse_groups"]},
    }
    if built_from:
        body["match_report"] = _match_summary(match_report)
        if match_report.get("unmatched_dataset"):
            body["unmatched_locations"] = match_report["unmatched_dataset"][:_INLINE]
    return sourced(ok=True, data=body, sources=[],
                   warnings=[finding["conclusion"] for finding in deduped["findings"]
                             if finding.get("severity") == "major"][:5])


# ---- 工具 7：证据链追溯 ---------------------------------------------------

@define_tool(group=_GROUP, label="追溯证据链",
             summary="把 evidence_id 解析成完整证据链：原始文件、页码或单元格、数值、计算过程与规则版本；证据缺失时问题状态必须为人工复核")
@_readable_errors
def trace_airflow_evidence(
    evidence_path: str,
    finding_id: str = "",
    evidence_ids: str = "",
    trace_level: str = "full",
) -> dict[str, Any]:
    """追溯证据链；evidence_path 为任一工具产物文件（内含 evidence 映射），evidence_ids 用逗号分隔。"""
    payload, error = _load_argument(evidence_path)
    if error:
        return _error("trace_airflow_evidence", error)
    store = _collect_evidence(payload)
    if not store:
        return _error("trace_airflow_evidence",
                      "该文件里没有 evidence 映射；请传入测风报表解析结果或计划预处理/核验结果文件")
    wanted = [item.strip() for item in str(evidence_ids or "").replace("，", ",").split(",") if item.strip()]
    if not wanted and finding_id:
        wanted = _evidence_ids_for_finding(payload, finding_id)
    if not wanted:
        wanted = list(store)
    result = trace_evidence(evidence=store, evidence_ids=wanted, finding_id=finding_id,
                            trace_level=trace_level)
    return sourced(ok=True, data=result, sources=[],
                   warnings=[] if result["is_complete"] else [result["conclusion"]])


# ---- 工具 8：报告渲染 -----------------------------------------------------

@define_tool(group=_GROUP, label="渲染审查报告",
             summary="生成风量研判审查报告（md 与 docx），固定包含研判范围、数据来源、核查方法、问题清单、计算过程、证据索引、数据质量说明与待复核事项")
@_readable_errors
def render_airflow_review_report(
    findings: str,
    mine_id: str = "",
    period: str = "",
    data_quality_summary: str = "",
    review_scope: str = "",
    report_format: str = "both",
    comparisons: str = "",
    verified_items: str = "",
    evidence_path: str = "",
) -> dict[str, Any]:
    """渲染报告；findings/comparisons/verified_items/data_quality_summary 均为 JSON 或 JSON 文件路径，report_format 取 md/docx/both。

    evidence_path 指向任一含 evidence 映射的工具产物（如测风报表解析结果）时，报告会把
    每个问题的证据链展开并据此计算证据完整率；不给则在报告中注明证据链缺失。
    """
    parsed_findings, error = _load_argument(findings)
    if error:
        return _error("render_airflow_review_report", f"findings 读取失败：{error}")
    if isinstance(parsed_findings, Mapping):
        parsed_findings = parsed_findings.get("findings") or []
    if not isinstance(parsed_findings, list):
        return _error("render_airflow_review_report", "findings 必须是数组或含 findings 的对象")

    quality, _ = _load_argument(data_quality_summary, default={})
    comparison_list, _ = _load_argument(comparisons, default=[])
    verified, _ = _load_argument(verified_items, default=[])
    if isinstance(comparison_list, Mapping):
        comparison_list = comparison_list.get("comparisons") or []
    if isinstance(verified, Mapping):
        verified = verified.get("verified_items") or []
    scope = [item.strip() for item in str(review_scope or "").replace("，", ",").split(",") if item.strip()]

    deduped = dedupe_findings(parsed_findings) if parsed_findings else {"findings": []}
    enriched, trace_warnings = _attach_evidence_chains(
        deduped["findings"], evidence_path, verified if isinstance(verified, list) else [])
    try:
        real_output, virtual_output = workspace_paths.output_path(f"AIR-REVIEW-{mine_id or 'mine'}-{period or ''}")
    except workspace_paths.WorkspacePathError:
        real_output, virtual_output = None, None
    result = render_report(
        mine_id=mine_id, period=period, findings=enriched,
        data_quality_summary=quality or {}, review_scope=scope or None,
        report_format=report_format or "both", comparisons=comparison_list,
        verified_items=verified if isinstance(verified, list) else [],
        output_dir=real_output.parent if real_output else None,
    )
    body = dict(result)
    if virtual_output:
        # 报告文件名由 render_report 按 report_id 决定，因此以它**实际写出的文件**为准，
        # 不能拿预先分配的产物路径改名（两者只共享目录，文件名不同）。
        files = result.get("files") or {}
        parent = virtual_output.rsplit("/", 1)[0] if "/" in virtual_output             else str(Path(virtual_output).parent).replace("\\", "/")
        body["files"] = {kind: Path(path).name for kind, path in files.items()}
        body["file_paths"] = {kind: f"{parent}/{Path(path).name}" for kind, path in files.items()}
        body["file_url"] = body["file_paths"].get("md") or body["file_paths"].get("docx") or ""
    body["dedupe_merged_count"] = deduped.get("merged_count", 0)
    body["evidence_chain_attached"] = bool(evidence_path)
    return sourced(ok=True, data=body, sources=[],
                   warnings=[*(result.get("warnings") or []), *trace_warnings])


__all__ = [
    "parse_airflow_measurement_report",
    "query_monitoring_airflow",
    "preprocess_air_plan_document",
    "validate_plan_extraction_tool",
    "verify_and_extract_plan_airflow",
    "reconcile_airflow_sources_tool",
    "compare_airflow_actual_vs_required",
    "trace_airflow_evidence",
    "render_airflow_review_report",
]