from __future__ import annotations

from dataclasses import asdict, is_dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Callable
import re
import math
import hashlib
import json
import webbrowser

from . import (
    book_program_adapters,
    cad_section_export,
    case_management,
    calc_book,
    core,
    dependency_graph,
    dem_import,
    engineering_output,
    flood_mainline,
    project_model,
    river_special,
    section_cad_import,
    workflow_guide,
)
from .auth_client import (
    AuthClientConfig,
    AuthClientError,
    OnlineAuthClient,
    clear_auth_state,
    is_offline_usable,
    is_state_active,
    load_auth_state,
    save_auth_state,
)
from .export import export_calculation_book_docx, export_calculation_book_xlsx
from .program_registry import get_program, list_modules, list_programs, registry_summary
from .versioning import load_version_info
from .workspace_store import WorkspaceStore


def _mask_phone(phone: str) -> str:
    digits = "".join(ch for ch in str(phone or "") if ch.isdigit())
    if len(digits) == 11:
        return f"{digits[:3]}****{digits[-4:]}"
    return str(phone or "")


def _is_wechat_placeholder(value: str) -> bool:
    return bool(re.fullmatch(r"wx_[0-9a-f]{16}", str(value or "").strip(), flags=re.IGNORECASE))


def _bound_phone(state: dict[str, str] | None) -> str:
    phone = str((state or {}).get("phone") or "").strip()
    return "" if _is_wechat_placeholder(phone) else phone


def _account_identity(state: dict[str, str] | None, fallback: str) -> str:
    state = state or {}
    phone = _bound_phone(state)
    if phone:
        return _mask_phone(phone)
    nickname = str(state.get("nickname") or "").strip()
    if nickname:
        return nickname
    for key, prefix in (("unionid", "微信用户"), ("openid", "微信用户")):
        value = str(state.get(key) or "").strip()
        if value:
            return f"{prefix}·{value[-6:]}"
    pseudo_account = str(state.get("phone") or "").strip()
    if pseudo_account:
        return pseudo_account
    return fallback


class HydrologyBridgeApi:
    # @hook UI2 20260804
    # Exposes pywebview JS bridge methods for auth, registry, calculation, export, and dashboard data.
    def __init__(self, request_login_restart: Callable[[], None] | None = None) -> None:
        self.version_info = load_version_info()
        self._dashboard_data = self._empty_dashboard_data()
        self._project = project_model.new_project(name="当前工程", project_type="桥梁")
        self._request_login_restart = request_login_restart
        self._workspace = WorkspaceStore()

    def workspace_state(self) -> dict[str, Any]:
        return self._workspace.state()

    def workspace_create(self, name: str, description: str = "") -> dict[str, Any]:
        try:
            return {"ok": True, "project": self._workspace.create(name, description)}
        except (ValueError, OSError) as exc:
            return {"ok": False, "message": str(exc)}

    def workspace_export(self, project_id: str) -> dict[str, Any]:
        try:
            return {"ok": True, "project": self._workspace.read(project_id)}
        except (ValueError, OSError, KeyError) as exc:
            return {"ok": False, "message": str(exc)}

    def workspace_open(self, project_id: str) -> dict[str, Any]:
        try:
            project = self._workspace.read(project_id)
            self._project = project_model.new_project(project_id=project_id, name=project["name"])
            for record in project.get("records", []):
                project_model.add_result(self._project, "proposed", 1.0, record["program_id"], record["params"], record["result"])
            return {"ok": True, "project": project}
        except (ValueError, OSError, KeyError) as exc:
            return {"ok": False, "message": str(exc)}

    def workspace_update(self, project_id: str, change: dict[str, Any]) -> dict[str, Any]:
        try:
            project = self._workspace.update(project_id, change)
            if self._project.id == project_id:
                self._project.name = project["name"]
            return {"ok": True, "project": project}
        except (ValueError, OSError) as exc:
            return {"ok": False, "message": str(exc)}

    def workspace_save_draft(self, project_id: str, program_id: str, draft: dict[str, Any]) -> dict[str, Any]:
        try:
            get_program(program_id)
            return self._workspace.save_draft(project_id, program_id, draft)
        except (ValueError, OSError, KeyError) as exc:
            return {"ok": False, "message": str(exc)}

    def workspace_add_record(self, project_id: str, program_id: str, params: dict[str, Any], result: dict[str, Any]) -> dict[str, Any]:
        try:
            get_program(program_id)
            return self._workspace.add_record(project_id, program_id, params, result)
        except (ValueError, OSError, KeyError) as exc:
            return {"ok": False, "message": str(exc)}

    def workspace_preferences(self, change: dict[str, Any]) -> dict[str, Any]:
        return self._workspace.preferences(change)

    def workspace_import(self, payload: dict[str, Any]) -> dict[str, Any]:
        try:
            return {"ok": True, "project": self._workspace.import_project(payload)}
        except (ValueError, OSError, KeyError) as exc:
            return {"ok": False, "message": str(exc)}

    def get_version(self) -> dict[str, Any]:
        info = load_version_info()
        self.version_info = info
        return {
            "ok": True,
            "app_id": info.app_id,
            "name": info.name,
            "version": info.version,
            "channel": info.channel,
            "auth_required": info.auth_required,
            "update_enabled": info.update_enabled,
            "ui": "html",
            "bridge": "ui2",
        }

    def auth_heartbeat(self) -> dict[str, Any]:
        info = load_version_info()
        self.version_info = info
        if not info.auth_required:
            return {
                "ok": True,
                "authorized": True,
                "source": "auth-disabled",
                "requires_membership": False,
                "message": "auth not required",
            }

        state = load_auth_state()
        if not state:
            return self._unauthorized("未检测到本机登录授权，请微信扫码登录并开通会员后使用。")

        client = OnlineAuthClient(AuthClientConfig(info.auth_server_url, timeout_seconds=5))
        try:
            new_state = client.heartbeat(state, info.version)
        except AuthClientError as exc:
            if is_offline_usable(state, info.offline_grace_days):
                return self._authorized(state, "offline", f"在线授权服务器暂不可用，已进入离线宽限期：{exc}")
            if is_state_active(state):
                return self._unauthorized(f"授权心跳失败：{exc}", state)
            return self._unauthorized(f"当前账号未开通会员或会员已过期，请开通会员后使用。服务器返回：{exc}", state)

        save_auth_state(new_state)
        # @hook wechat-membership-payment 20260814
        # Logged-in WeChat accounts without an active license must enter payment flow, not a QR-login loop.
        if not is_state_active(new_state):
            return self._unauthorized("当前账号未开通会员或会员已过期，请开通会员后使用。", new_state)
        return self._authorized(new_state, "online", "授权心跳正常。")

    def logout_auth(self, restart_login: bool = False) -> dict[str, Any]:
        # @hook AUTH-logout 20260805
        # Clears the local membership token; HTML UI owns the next login screen to avoid legacy tkinter fallback.
        clear_auth_state()
        if restart_login and self._request_login_restart is not None:
            self._request_login_restart()
        return {
            "ok": True,
            "status": "logged_out",
            "message": "本机登录状态已清除，请重新微信扫码登录。",
            "restart_login": bool(restart_login and self._request_login_restart is not None),
        }

    def request_bind_phone_code(self, phone: str) -> dict[str, Any]:
        normalized = re.sub(r"\D", "", str(phone or ""))
        if not re.fullmatch(r"1\d{10}", normalized):
            return {"ok": False, "message": "请输入正确的 11 位手机号。"}
        state = load_auth_state()
        if not state:
            return {"ok": False, "message": "请先微信扫码登录，再绑定手机号。"}
        info = load_version_info()
        self.version_info = info
        client = OnlineAuthClient(AuthClientConfig(info.auth_server_url, timeout_seconds=10))
        try:
            data = client.request_bind_phone_code(state, normalized)
        except AuthClientError as exc:
            return {"ok": False, "message": str(exc)}
        result: dict[str, Any] = {
            "ok": True,
            "phone": normalized,
            "message": data.get("message") or "验证码已发送。",
        }
        if data.get("debug_code"):
            result["debug_code"] = data.get("debug_code")
        return result

    def bind_account_phone(self, phone: str, code: str) -> dict[str, Any]:
        normalized = re.sub(r"\D", "", str(phone or ""))
        code_text = re.sub(r"\D", "", str(code or ""))
        if not re.fullmatch(r"1\d{10}", normalized):
            return {"ok": False, "message": "请输入正确的 11 位手机号。"}
        if not re.fullmatch(r"\d{4,8}", code_text):
            return {"ok": False, "message": "请输入短信验证码。"}
        state = load_auth_state()
        if not state:
            return {"ok": False, "message": "请先微信扫码登录，再绑定手机号。"}
        info = load_version_info()
        self.version_info = info
        client = OnlineAuthClient(AuthClientConfig(info.auth_server_url, timeout_seconds=10))
        try:
            new_state = client.bind_account_phone(state, normalized, code_text)
        except AuthClientError as exc:
            return {"ok": False, "message": str(exc)}
        save_auth_state(new_state)
        if is_state_active(new_state):
            auth = self._authorized(new_state, "online", "手机号绑定成功，会员状态已同步。")
        else:
            auth = self._unauthorized("手机号绑定成功，当前账号未开通会员或会员已过期。", new_state)
        return {
            "ok": True,
            "phone": normalized,
            "auth": auth,
            "message": "手机号绑定成功，会员状态已同步。",
        }

    def open_external_url(self, url: str) -> dict[str, Any]:
        if not str(url).lower().startswith(("http://", "https://")):
            return {"ok": False, "message": "只允许打开 http/https 链接。"}
        try:
            webbrowser.open(str(url))
        except Exception as exc:  # noqa: BLE001
            return {"ok": False, "message": f"打开浏览器失败：{exc}"}
        return {"ok": True, "url": str(url)}

    def start_wechat_login(self) -> dict[str, Any]:
        # @hook HTML-wechat-login 20260805
        # Starts WeChat QR login directly from the HTML UI instead of using the legacy tkinter dialog.
        info = load_version_info()
        self.version_info = info
        client = OnlineAuthClient(AuthClientConfig(info.auth_server_url, timeout_seconds=10))
        try:
            data = client.start_wechat_login(info.version)
        except AuthClientError as exc:
            return {"ok": False, "message": str(exc)}
        return {
            "ok": True,
            "state": data.get("state"),
            "auth_url": data.get("auth_url") or data.get("url") or data.get("login_url"),
            "expires_in": data.get("expires_in") or data.get("expire_seconds") or 600,
            "message": data.get("message") or "请在打开的微信页面完成扫码确认。",
        }

    def get_wechat_login_status(self, state: str) -> dict[str, Any]:
        info = load_version_info()
        self.version_info = info
        if not state:
            return {"ok": False, "status": "missing_state", "message": "缺少微信登录状态码，请重新发起扫码。"}
        client = OnlineAuthClient(AuthClientConfig(info.auth_server_url, timeout_seconds=10))
        try:
            data = client.get_wechat_login_status(str(state))
        except AuthClientError as exc:
            return {"ok": False, "message": str(exc)}

        status = str(data.get("status") or "").lower()
        authenticated = bool(data.get("authenticated")) or status in {"authenticated", "success", "confirmed", "logged_in"}
        if authenticated:
            try:
                new_state = client.auth_state_from_response(data)
            except AuthClientError as exc:
                return {"ok": False, "status": "invalid_auth_response", "message": str(exc)}
            save_auth_state(new_state)
            auth = (
                self._authorized(new_state, "online", "微信扫码登录成功。")
                if is_state_active(new_state)
                else self._unauthorized("微信扫码登录成功，当前账号未开通会员或会员已过期，请开通会员后使用。", new_state)
            )
            return {
                "ok": True,
                "status": "authenticated",
                "message": "微信扫码登录成功。",
                "auth": auth,
            }

        if status == "device_conflict" or data.get("requires_device_replace"):
            return {
                "ok": True,
                "status": "device_conflict",
                "message": data.get("message") or "当前账号设备席位已满，请确认是否替换旧设备。",
                "requires_device_replace": True,
                "device_conflict": data.get("device_conflict") or {},
            }

        return {
            "ok": True,
            "status": status or "pending",
            "message": data.get("message") or "等待微信扫码确认。",
            "requires_membership": bool(data.get("requires_membership")),
        }

    def confirm_wechat_device_replace(self, state: str) -> dict[str, Any]:
        # @hook AUTH-device-seat-confirm-client 20260814
        # Confirms a recoverable device-seat conflict from the HTML login modal.
        info = load_version_info()
        self.version_info = info
        if not state:
            return {"ok": False, "status": "missing_state", "message": "缺少微信登录状态码，请重新扫码。"}
        client = OnlineAuthClient(AuthClientConfig(info.auth_server_url, timeout_seconds=10))
        try:
            data = client.confirm_wechat_device_replace(str(state))
        except AuthClientError as exc:
            return {"ok": False, "message": str(exc)}
        status = str(data.get("status") or "").lower()
        if status not in {"authenticated", "success", "confirmed", "logged_in"} and not data.get("access_token"):
            return {
                "ok": True,
                "status": status or "device_conflict",
                "message": data.get("message") or "设备替换尚未完成，请稍后重试。",
                "requires_device_replace": bool(data.get("requires_device_replace")),
                "device_conflict": data.get("device_conflict") or {},
            }
        try:
            new_state = client.auth_state_from_response(data)
        except AuthClientError as exc:
            return {"ok": False, "status": "invalid_auth_response", "message": str(exc)}
        save_auth_state(new_state)
        auth = (
            self._authorized(new_state, "online", "已替换旧设备并完成微信登录。")
            if is_state_active(new_state)
            else self._unauthorized("已完成微信登录，当前账号未开通会员或会员已过期，请开通会员后使用。", new_state)
        )
        return {
            "ok": True,
            "status": "authenticated",
            "message": data.get("message") or "已替换旧设备并完成微信登录。",
            "auth": auth,
        }

    def get_membership_plans(self) -> dict[str, Any]:
        state = load_auth_state() or {}
        account = str(state.get("phone") or "").strip()
        if not account:
            return {"ok": False, "message": "请先微信扫码登录，再开通会员。", "plans": []}
        info = load_version_info()
        client = OnlineAuthClient(AuthClientConfig(info.auth_server_url, timeout_seconds=10))
        try:
            payload = client.get_membership_plans(account)
        except AuthClientError as exc:
            return {"ok": False, "message": str(exc), "plans": []}
        return {
            "ok": True,
            "account": _account_identity(state, account),
            "plans": payload.get("plans", []),
            "rules": payload.get("rules", {}),
        }

    def create_payment_order(self, plan_id: str = "year", provider: str = "wechat") -> dict[str, Any]:
        state = load_auth_state() or {}
        account = str(state.get("phone") or "").strip()
        if not account:
            return {"ok": False, "message": "请先微信扫码登录，再开通会员。"}
        info = load_version_info()
        client = OnlineAuthClient(AuthClientConfig(info.auth_server_url, timeout_seconds=10))
        try:
            payload = client.create_payment_order(account, str(plan_id or "year"), str(provider or "wechat"))
        except AuthClientError as exc:
            return {"ok": False, "message": str(exc)}
        payload["ok"] = True
        return payload

    def get_payment_order(self, order_no: str) -> dict[str, Any]:
        if not order_no:
            return {"ok": False, "message": "缺少支付订单号。"}
        info = load_version_info()
        client = OnlineAuthClient(AuthClientConfig(info.auth_server_url, timeout_seconds=10))
        try:
            payload = client.get_payment_order(str(order_no))
        except AuthClientError as exc:
            return {"ok": False, "message": str(exc)}
        result = {"ok": True, **payload}
        if str(payload.get("status") or "").lower() == "paid":
            result["auth"] = self.auth_heartbeat()
        return result

    def get_modules(self) -> dict[str, Any]:
        return {
            "ok": True,
            "summary": registry_summary(),
            "modules": list_modules(),
        }

    def list_programs(self, module_key: str | None = None) -> dict[str, Any]:
        return {
            "ok": True,
            "module_key": module_key or "all",
            "programs": list_programs(module_key),
        }

    def get_module_workbench(
        self,
        module_key: str,
        case_id: str = "proposed",
        frequency: float | str = 1.0,
    ) -> dict[str, Any]:
        # Catalog pages are independent entry points.  Dependency information is
        # retained in each calculation's validation, but never blocks users from
        # opening a program or entering project data.
        modules = {row["key"]: row for row in list_modules()}
        if module_key not in modules:
            return {"ok": False, "status": "unknown_module", "message": f"未知模块：{module_key}"}

        completed = project_model.get_completed_programs(self._project, case_id, frequency)
        rows: list[dict[str, Any]] = []
        for program in list_programs(module_key):
            dependency = dependency_graph.check_dependencies(
                str(program["program_id"]),
                self._project,
                case_id=case_id,
                frequency=frequency,
            )
            is_completed = str(program["program_id"]) in completed
            if is_completed:
                status_kind = "done"
                status_label = "已完成"
                status_message = "已有计算成果"
            else:
                status_kind = "ready"
                status_label = "可算"
                status_message = "可独立录入参数；引用其他成果时在本程序内核对来源。"
            row = dict(program)
            row.update(
                {
                    "statusKind": status_kind,
                    "statusLabel": status_label,
                    "statusMessage": status_message,
                    "dependency": dependency,
                    "href": f"{str(program['program_id']).lower()}.html",
                }
            )
            rows.append(row)

        module = dict(modules[module_key])
        module["program_count"] = len(rows)
        module["status_done"] = sum(1 for row in rows if row.get("status") == "已接入")
        module["completed_count"] = sum(1 for row in rows if row["statusKind"] == "done")
        module["ready_count"] = sum(1 for row in rows if row["statusKind"] == "ready")
        module["blocked_count"] = 0
        return {"ok": True, "module": module, "programs": rows}

    def calculate(self, program_id: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        params = dict(params or {})
        metadata = params.pop('_parameter_metadata', {})
        metadata = {str(key): value for key, value in metadata.items() if isinstance(value, dict)} if isinstance(metadata, dict) else {}
        try:
            program = get_program(program_id)
        except KeyError:
            return {"ok": False, "status": "unknown_program", "message": f"未知程序ID：{program_id}"}

        auth = self.auth_heartbeat()
        if not auth.get("authorized"):
            return {
                "ok": False,
                "status": "unauthorized",
                "requires_membership": True,
                "message": auth.get("message") or "未授权，不能计算。",
            }

        adapter = _CALCULATORS.get(program_id)
        if adapter is None:
            return {
                "ok": False,
                "status": "needs_adapter",
                "program": program.to_dict(),
                "message": "程序已登记并可从 Web UI 调用；该程序的 Web 参数适配将在 B3/B4 接入。",
                "required_inputs": list(program.inputs),
                "outputs": list(program.outputs),
            }

        try:
            engineering_output.validate_finite(params)
            result = adapter(params)
            engineering_output.validate_finite(result, '计算结果')
        except Exception as exc:  # noqa: BLE001 - bridge must return structured errors to JS.
            return {
                "ok": False,
                "status": "calculation_error",
                "program": program.to_dict(),
                "message": str(exc),
            }

        payload = _to_plain(result)
        payload = _with_calculation_contract(program_id, params, payload, metadata)
        payload['calculated_at'] = datetime.now().isoformat(timespec='seconds')
        payload['parameter_hash'] = hashlib.sha256(json.dumps(params, sort_keys=True, ensure_ascii=False).encode('utf-8')).hexdigest()[:16]
        payload['parameter_snapshot'] = [
            {'name': metadata.get(key, {}).get('label') or engineering_output.display_label(key), 'value': value,
             'unit': metadata.get(key, {}).get('unit', ''),
             'note': metadata.get(key, {}).get('source', '本次提交值；资料来源待核定')}
            for key, value in params.items()
        ]
        self._update_dashboard_from_result(payload)
        try:
            project_model.add_result(self._project, "proposed", 1.0, program_id, params, payload)
        except Exception:
            pass
        return {
            "ok": True,
            "status": "calculated",
            "program": program.to_dict(),
            "result": payload,
        }

    def _project_name(self) -> str:
        value = getattr(self._project, "name", "") or ""
        return str(value or "当前工程")

    def _normalize_calc_book_content(
        self,
        program: Any,
        result: dict[str, Any] | str,
        content: str,
    ) -> str:
        """Ensure Web UI exports always contain the four required calculation-book sections."""
        required = ["公式来源", "参数代入值", "中间过程", "最终结果"]
        if content and all(name in content for name in required):
            return content

        data = result if isinstance(result, dict) else {}
        source = str(data.get("source") or data.get("formula_source") or getattr(program, "source", "") or "见程序登记表与对应规范条文")
        inputs = data.get("input_rows") or data.get("inputs") or []
        metrics = data.get("metrics") or data.get("result_rows") or data.get("outputs") or []
        formula = str(data.get("formula") or data.get("formula_text") or "")

        def table_lines(title: str, rows: Any) -> list[str]:
            lines = [title, "| 项目 | 数值 | 单位 | 说明 |", "|---|---:|---|---|"]
            if isinstance(rows, dict):
                iterable = [{"name": key, "value": value, "unit": "", "note": ""} for key, value in rows.items()]
            elif isinstance(rows, list):
                iterable = rows
            else:
                iterable = []
            for item in iterable:
                if isinstance(item, dict):
                    lines.append(
                        "| {name} | {value} | {unit} | {note} |".format(
                            name=item.get("name") or item.get("label") or item.get("key") or "-",
                            value=item.get("value", "-"),
                            unit=item.get("unit", ""),
                            note=item.get("note", ""),
                        )
                    )
                else:
                    lines.append(f"| {item} | - |  |  |")
            if len(lines) == 3:
                lines.append("| 由当前计算表单或上游成果提供 | 见原始计算内容 |  | 自动归档 |")
            return lines

        normalized = [
            str(getattr(program, "name", "") or "计算书"),
            "",
            "一、公式来源",
            source,
            "",
            "二、参数代入值",
            *table_lines("", inputs)[1:],
            "",
            "三、中间过程",
            formula or "按上述输入参数代入程序登记表中的公式与规范方法，逐步形成中间量；原始程序过程如下：",
            content or "当前程序未返回独立过程文本，已按结构化输入和结果生成归档计算书。",
            "",
            "四、最终结果",
            *table_lines("", metrics)[1:],
        ]
        return "\n".join(normalized)

    def export_calc_book(self, program_id: str, result: dict[str, Any] | str) -> dict[str, Any]:
        try:
            program = get_program(program_id)
        except KeyError:
            return {"ok": False, "status": "unknown_program", "message": f"未知程序ID：{program_id}"}

        if isinstance(result, str):
            content = result
            target_path = ""
            watermark = "送审稿"
        else:
            content = str(result.get("calculation_book_text") or result.get("text") or result.get("content") or "")
            target_path = str(result.get("target_path") or result.get("path") or "")
            watermark = str(result.get("watermark") or result.get("document_stage") or result.get("stage") or "送审稿")

        if not content:
            return {"ok": False, "status": "missing_content", "message": "缺少计算书文本内容。"}
        if not target_path:
            return {"ok": False, "status": "missing_path", "message": "缺少导出目标路径。"}

        path = Path(target_path)
        structured = isinstance(result, dict) and bool(result.get('metrics') or result.get('parameter_snapshot'))
        content = engineering_output.snapshot_text(program, result) if structured else self._normalize_calc_book_content(program, result, content)
        if structured:
            content = f"项目：{self._project_name()}；软件版本：{self.version_info.version}；文件阶段：{watermark}\n\n" + content
        figures = [(s['title'], engineering_output.svg_chart(s)) for s in engineering_output.charts(result)] if structured else []
        project_name = self._project_name()
        version = f"V{self.version_info.version}"
        try:
            if path.suffix.lower() == '.xlsx' and structured:
                engineering_output.export_snapshot_xlsx(content, path, result, program.name)
            elif path.suffix.lower() == ".xlsx":
                export_calculation_book_xlsx(
                    content,
                    path,
                    title=program.name,
                    project_name=project_name,
                    version=version,
                    watermark=watermark,
                )
            elif path.suffix.lower() == ".docx":
                export_calculation_book_docx(
                    content,
                    path,
                    title=program.name,
                    project_name=project_name,
                    version=version,
                    watermark=watermark,
                    figures=figures,
                )
            else:
                return {"ok": False, "status": "unsupported_format", "message": "仅支持 .xlsx 或 .docx。"}
        except Exception as exc:  # noqa: BLE001
            return {"ok": False, "status": "export_error", "message": str(exc)}

        return {"ok": True, "status": "exported", "path": str(path)}

    def export_section_cad(self, program_id: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
        # @hook CAD-SECTION-EXPORT 20260817
        # Exports water-surface section results to DXF with water level and thalweg annotations.
        payload = payload or {}
        result = payload.get("result")
        params = payload.get("params")
        target_path = payload.get("target_path") or payload.get("path")
        if not isinstance(result, dict):
            result = {}
        if not isinstance(params, dict):
            params = {}
        return cad_section_export.export_section_dxf(
            program_id,
            result=result,
            params=params,
            target_path=target_path,
        )

    def generate_calc_book(
        self,
        cb_id: str,
        project_id: str = "",
        options: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        # @hook B9 20260804
        # Generates CB-01 through CB-14 seven-chapter calculation books through the HTML bridge.
        options = options or {}
        if options.get('example') is not True:
            return {'ok':False, 'status':'example_only', 'message':'旧计算书生成器仅提供教学例题，不能作为当前项目成果。请先在计算页面计算，再导出计算书或Excel图表。'}
        target_path = str(options.get("target_path") or options.get("path") or "")
        watermark = "教学例题·非项目成果"
        project_name = str(options.get("project_name") or self._project_name())
        version = f"V{self.version_info.version}"
        try:
            return calc_book.generate_calc_book_document(
                cb_id,
                project_id=project_id,
                target_path=target_path or None,
                project_name=project_name,
                version=version,
                watermark=watermark,
            )
        except Exception as exc:  # noqa: BLE001 - bridge must return structured errors to JS.
            return {"ok": False, "status": "calc_book_error", "message": str(exc), "cb_id": cb_id}

    def get_dashboard_data(self) -> dict[str, Any]:
        data = dict(self._dashboard_data)
        data["updated_at"] = datetime.now().strftime("%H:%M:%S")
        return {"ok": True, **data}

    def get_section_cad_import_modes(self) -> dict[str, Any]:
        # @hook SECTION-CAD-MODES 20260817
        # Exposes non-mandatory CAD layer recognition modes for section import.
        return section_cad_import.available_import_modes()

    def analyze_section_cad_layers(self, layers: Any = None, mode: str = "smart") -> dict[str, Any]:
        return section_cad_import.analyze_layers(layers, mode=mode)

    def select_section_cad_file(self) -> dict[str, Any]:
        # @hook SECTION-CAD-IMPORT-FILE 20260817
        # Opens the native file picker so users can import local DXF files directly.
        def _tk_file_dialog() -> str:
            import tkinter as tk
            from tkinter import filedialog

            root = tk.Tk()
            root.withdraw()
            root.attributes("-topmost", True)
            try:
                return filedialog.askopenfilename(
                    title="选择本地 CAD/DXF 文件",
                    filetypes=[
                        ("CAD/DXF 文件", "*.dxf *.dwg"),
                        ("DXF 文件", "*.dxf"),
                        ("DWG 文件", "*.dwg"),
                        ("所有文件", "*.*"),
                    ],
                )
            finally:
                root.destroy()

        selected: Any = None
        primary_error = ""
        try:
            import webview  # type: ignore

            window = webview.windows[0] if getattr(webview, "windows", None) else None
            if window is not None:
                selected = window.create_file_dialog(
                    webview.OPEN_DIALOG,
                    allow_multiple=False,
                    file_types=("CAD/DXF 文件 (*.dxf;*.dwg)", "DXF 文件 (*.dxf)", "DWG 文件 (*.dwg)", "所有文件 (*.*)"),
                )
        except Exception as exc:  # noqa: BLE001
            primary_error = str(exc)

        if not selected:
            try:
                selected = _tk_file_dialog()
            except Exception as exc:  # noqa: BLE001
                detail = f"；pywebview 错误：{primary_error}" if primary_error else ""
                return {"ok": False, "message": f"打开本地文件选择器失败：{exc}{detail}"}

        if not selected:
            return {"ok": False, "cancelled": True, "message": "已取消选择文件。"}
        path = selected[0] if isinstance(selected, (list, tuple)) else selected
        return {"ok": True, "path": str(path), "message": "已选择本地 CAD/DXF 文件。"}

    def analyze_section_cad_file(self, path: str = "", mode: str = "smart") -> dict[str, Any]:
        # @hook SECTION-CAD-IMPORT-FILE 20260817
        # Reads local DXF layers and suggests section role mappings.
        return section_cad_import.analyze_cad_file(path, mode=mode)

    def import_section_cad_profile(
        self,
        path: str = "",
        mapping: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        mapping = mapping or {}
        ground_layer = mapping.get("ground_line") or mapping.get("groundLayer")
        return section_cad_import.extract_section_profile(path, ground_layer)

    def validate_section_cad_mapping(
        self,
        layers: Any = None,
        mapping: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        return section_cad_import.validate_manual_mapping(layers, mapping)

    def select_dem_file(self) -> dict[str, Any]:
        def _tk_file_dialog() -> str:
            import tkinter as tk
            from tkinter import filedialog
            root = tk.Tk()
            root.withdraw()
            root.attributes("-topmost", True)
            try:
                return filedialog.askopenfilename(title="选择 DEM 地形文件", filetypes=[("DEM 文件", "*.tif *.tiff *.asc *.grd"), ("所有文件", "*.*")])
            finally:
                root.destroy()
        selected: Any = None
        try:
            import webview  # type: ignore
            window = webview.windows[0] if getattr(webview, "windows", None) else None
            if window is not None:
                selected = window.create_file_dialog(webview.OPEN_DIALOG, allow_multiple=False, file_types=("DEM 文件 (*.tif;*.tiff;*.asc;*.grd)", "所有文件 (*.*)"))
        except Exception:
            selected = None
        if not selected:
            selected = _tk_file_dialog()
        if not selected:
            return {"ok": False, "cancelled": True, "message": "已取消选择 DEM 文件。"}
        path = selected[0] if isinstance(selected, (list, tuple)) else selected
        return {"ok": True, "path": str(path), "message": "已选择 DEM 文件。"}

    def read_dem_metadata(self, path: str = "") -> dict[str, Any]:
        try:
            return dem_import.read_metadata(path)
        except dem_import.DemImportError as exc:
            return {"ok": False, "message": str(exc)}

    def get_dem_runtime_status(self) -> dict[str, Any]:
        return dem_import.runtime_status()

    def install_dem_runtime(self) -> dict[str, Any]:
        try:
            return dem_import.install_runtime()
        except dem_import.DemImportError as exc:
            return {"ok": False, "message": str(exc)}

    def sample_dem_points(self, path: str = "", points: Any = None, input_crs: str = "") -> dict[str, Any]:
        try:
            return dem_import.sample_points(path, points or [], input_crs)
        except dem_import.DemImportError as exc:
            return {"ok": False, "message": str(exc)}

    def sample_dem_profile(self, path: str = "", start_x: Any = "", start_y: Any = "", end_x: Any = "", end_y: Any = "", count: Any = 51, input_crs: str = "") -> dict[str, Any]:
        try:
            return dem_import.sample_profile(path, start_x, start_y, end_x, end_y, count, input_crs)
        except dem_import.DemImportError as exc:
            return {"ok": False, "message": str(exc)}

    def check_dependencies(
        self,
        program_id: str,
        case_id: str = "proposed",
        frequency: float | str = 1.0,
    ) -> dict[str, Any]:
        return dependency_graph.check_dependencies(
            program_id,
            self._project,
            case_id=case_id,
            frequency=frequency,
        )

    def get_program_input_schema(self, program_id: str) -> dict[str, Any]:
        """Expose exact adapter input fields to the Web UI."""
        from . import calculation_forms

        schema = calculation_forms.program_input_schema(program_id)
        if schema is None:
            schema = book_program_adapters.program_input_schema(program_id)
        return {"ok": True, "schema": schema}

    def list_stale_results(self) -> dict[str, Any]:
        return {"ok": True, "results": project_model.list_stale_results(self._project)}

    def run_case_compare(
        self,
        program_id: str,
        case_pair: Any = None,
    ) -> dict[str, Any]:
        pair = _normalize_case_pair(case_pair)
        export_path = Path(r"D:\workspace") / f"hydrology_case_compare_{program_id.lower().replace('-', '_')}.xlsx"
        compare = case_management.run_case_compare(
            self._project,
            program_id,
            pair,
            export_path=export_path,
        )
        return {"ok": True, "status": "exported", **compare}

    def get_workflow(self, project_type: str = "桥梁") -> dict[str, Any]:
        workflow = workflow_guide.get_workflow(project_type, self._project)
        return {"ok": True, **workflow}

    def _authorized(self, state: dict[str, str], source: str, message: str) -> dict[str, Any]:
        phone = state.get("phone", "")
        display_account = _account_identity(state, "微信已登录 · 待绑定手机号")
        return {
            "ok": True,
            "authorized": True,
            "logged_in": True,
            "needs_payment": False,
            "action": "use",
            "license_active": True,
            "source": source,
            "requires_membership": False,
            "message": message,
            "phone": phone,
            "bound_phone": _bound_phone(state),
            "phone_is_bound": bool(_bound_phone(state)),
            "nickname": state.get("nickname", ""),
            "openid": state.get("openid", ""),
            "unionid": state.get("unionid", ""),
            "display_account": display_account,
            "device_name": state.get("device_name", ""),
            "device_id": state.get("device_id", ""),
            "plan": state.get("plan", ""),
            "expire_at": state.get("expire_at", ""),
            "trial_remaining": state.get("trial_remaining", ""),
        }

    def _unauthorized(self, message: str, state: dict[str, str] | None = None) -> dict[str, Any]:
        state = state or {}
        phone = state.get("phone", "")
        logged_in = bool(phone or state.get("access_token") or state.get("openid") or state.get("unionid"))
        display_account = _account_identity(state, "未登录")
        return {
            "ok": False,
            "authorized": False,
            "logged_in": logged_in,
            "needs_payment": logged_in,
            "action": "pay" if logged_in else "login",
            "license_active": False,
            "requires_membership": True,
            "message": message,
            "phone": phone,
            "bound_phone": _bound_phone(state),
            "phone_is_bound": bool(_bound_phone(state)),
            "nickname": state.get("nickname", ""),
            "openid": state.get("openid", ""),
            "unionid": state.get("unionid", ""),
            "display_account": display_account,
            "device_name": state.get("device_name", ""),
            "device_id": state.get("device_id", ""),
            "plan": state.get("plan", ""),
            "expire_at": state.get("expire_at", ""),
        }

    def _empty_dashboard_data(self) -> dict[str, Any]:
        return {
            "source": "project-results",
            "has_data": False,
            "peak_flow": {"value": "—", "unit": "m³/s", "status": "待计算"},
            "water_level": {"value": "—", "unit": "m", "status": "待计算"},
            "rain_type": {"value": "—", "unit": "", "status": "待计算"},
            "updated_at": "",
        }

    def _update_dashboard_from_result(self, payload: dict[str, Any]) -> None:
        peak = _first_value(payload, ("peak_flow", "peak_q", "Qp", "Q", "flood_peak"))
        water = _first_value(payload, ("water_level", "stage", "Z", "design_water_level"))
        rain = _first_value(payload, ("rain_type", "rainfall_pattern", "storm_pattern"))
        if peak is None and water is None and rain is None:
            return

        data = self._empty_dashboard_data()
        data["has_data"] = True
        if peak is not None:
            data["peak_flow"] = {"value": _format_metric(peak), "unit": "m³/s", "status": "最近计算"}
        if water is not None:
            data["water_level"] = {"value": _format_metric(water), "unit": "m", "status": "最近计算"}
        if rain is not None:
            data["rain_type"] = {"value": str(rain), "unit": "", "status": "最近计算"}
        self._dashboard_data = data


def _to_plain(value: Any) -> Any:
    if is_dataclass(value):
        return _to_plain(asdict(value))
    if isinstance(value, dict):
        return {str(key): _to_plain(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_to_plain(item) for item in value]
    if hasattr(value, "__dict__"):
        return _to_plain(vars(value))
    return value


def _contract_number(value: Any) -> float | None:
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return float(value) if math.isfinite(value) else None
    try:
        text = str(value).strip()
        if not text or text in {"-", "—"}:
            return None
        number = float(text)
        return number if math.isfinite(number) else None
    except (TypeError, ValueError):
        return None


def _contract_scalar(value: Any) -> bool:
    return value is None or isinstance(value, (str, int, float, bool))


def _append_unique_diag(diag: list[str], message: str) -> None:
    if message and message not in diag:
        diag.append(message)


def _normalize_diag(value: Any) -> list[str]:
    if isinstance(value, str):
        return [value] if value.strip() else []
    if isinstance(value, (list, tuple)):
        return [str(item) for item in value if str(item).strip()]
    return []


def _series_point_value(point: dict[str, Any], keys: tuple[str, ...]) -> Any:
    for key in keys:
        if key in point and point.get(key) not in (None, ""):
            return point.get(key)
    return None


def _infer_result_series(payload: dict[str, Any]) -> list[dict[str, Any]]:
    # @hook RESULT-CONTRACT 20260824
    # Converts existing process-line outputs into the command-book result_series contract without changing formulas.
    existing = payload.get("result_series")
    if isinstance(existing, list):
        return existing

    series_items = payload.get("series")
    if isinstance(series_items, list):
        for series in series_items:
            if not isinstance(series, dict):
                continue
            points = series.get("points")
            if not isinstance(points, list):
                continue
            x_key = str(series.get("xKey") or "time_h")
            y_key = str(series.get("yKey") or "discharge_m3s")
            title = str(series.get("title") or "成果序列")
            converted: list[dict[str, Any]] = []
            for index, point in enumerate(points):
                if not isinstance(point, dict):
                    continue
                x_raw = _series_point_value(point, (x_key, "time_h", "time", "t", "x"))
                y_raw = _series_point_value(point, (y_key, "discharge_m3s", "flow_m3s", "q", "Q", "value", "y"))
                x = _contract_number(x_raw)
                y = _contract_number(y_raw)
                converted.append(
                    {
                        "time_h": x if x is not None else float(index),
                        "value": y,
                        "label": title,
                    }
                )
            if any(item.get("value") is not None for item in converted):
                return converted

    curve_points = payload.get("curve_points")
    if isinstance(curve_points, list):
        converted = []
        for index, point in enumerate(curve_points):
            if not isinstance(point, dict):
                continue
            x = _contract_number(_series_point_value(point, ("time_h", "time", "t", "x")))
            y = _contract_number(_series_point_value(point, ("flow_m3s", "discharge_m3s", "q", "Q", "value", "y")))
            converted.append({"time_h": x if x is not None else float(index), "value": y, "label": "过程成果"})
        if any(item.get("value") is not None for item in converted):
            return converted

    return []


def _with_calculation_contract(program_id: str, params: dict[str, Any], payload: Any, parameter_metadata: dict | None = None) -> Any:
    if not isinstance(payload, dict):
        return payload

    normalized = dict(payload)
    diag = _normalize_diag(normalized.get("diag") or normalized.get("diagnostics"))
    _append_unique_diag(diag, "结果仅对应本次输入；工程采用前请复核参数依据、适用条件与量级。")

    result_series = _infer_result_series(normalized)
    if result_series:
        normalized["result_series"] = result_series
        _append_unique_diag(diag, "已保存完整成果序列，可用于曲线绘制和计算书归档。")
    else:
        _append_unique_diag(diag, "本程序未形成逐时成果序列，表格、断面或关系曲线按专用成果字段输出。")

    if normalized.get("rating_curve") or normalized.get("profile_curve") or normalized.get("section_profile") or normalized.get("section_profiles"):
        _append_unique_diag(diag, "已输出水位流量关系曲线、沿程水面线/能量线或断面水位标注图。")
    tables = list(normalized.get("tables") or [])
    input_rows = normalized.get("input_rows") if isinstance(normalized.get("input_rows"), list) else []
    result_rows = normalized.get("result_rows") if isinstance(normalized.get("result_rows"), list) else []
    metric_rows = normalized.get("metrics") if isinstance(normalized.get("metrics"), list) else []

    # Keep the full calculation trail for every registered program, including
    # older calculators that predate the Web table renderer. Specialised
    # tables remain intact; common audit tables are appended only when absent.
    titles = {str(table.get("title", "")) for table in tables if isinstance(table, dict)}
    if "参数代入与公式核对表" not in titles:
        metadata = parameter_metadata or params.get("_parameter_metadata") or {}
        if not isinstance(metadata, dict):
            metadata = {}
        scalar_rows = [
            {"name": str(metadata.get(key, {}).get("label") or engineering_output.display_label(key)), "value": str(value), "unit": str(metadata.get(key, {}).get("unit") or ""), "note": "本次提交参数"}
            for key, value in params.items()
            if _contract_scalar(value) and not str(key).startswith("_")
        ]
        formula = str(normalized.get("formula") or "").strip()
        source = str(normalized.get("source") or "").strip()
        if input_rows or scalar_rows or formula or source:
            tables.insert(0, {
                "title": "参数代入与公式核对表",
                "rows": [
                    *input_rows,
                    *scalar_rows,
                    *([{"name": "公式来源", "value": source, "unit": "", "note": ""}] if source else []),
                    *([{"name": "采用公式", "value": formula, "unit": "", "note": ""}] if formula else []),
                ],
            })
            titles.add("参数代入与公式核对表")
    if "计算过程与成果汇总" not in titles and (result_rows or metric_rows):
        tables.append({"title": "计算过程与成果汇总", "rows": result_rows or metric_rows})

    normalized["tables"] = sorted(tables, key=lambda t: t.get('title') in {'参数代入与公式核对表', '计算过程与成果汇总'})
    engineering_output.label_result_tables(normalized, parameter_metadata)
    if tables:
        _append_unique_diag(diag, "已输出工程计算表，空缺值使用空值或横线，不以 0 代替缺测。")

    status_code = str(normalized.get("status_code") or "ok")
    design_params = {str(key): _to_plain(value) for key, value in params.items() if _contract_scalar(value)}
    contract = {
        "program_id": program_id,
        "mode": "design",
        "status_code": status_code,
        "design_params": design_params,
        "obs_series": None,
        "fcst_series": None,
        "t0": None,
        "result_series": result_series or None,
        "ci_low": normalized.get("ci_low") if isinstance(normalized.get("ci_low"), list) else None,
        "ci_high": normalized.get("ci_high") if isinstance(normalized.get("ci_high"), list) else None,
        "diag": diag,
    }
    normalized["status_code"] = status_code
    normalized["diag"] = diag
    normalized["calculation_contract"] = contract
    return normalized


def _first_value(payload: dict[str, Any], keys: tuple[str, ...]) -> Any:
    for key in keys:
        if key in payload and payload[key] not in (None, ""):
            return payload[key]
    for value in payload.values():
        if isinstance(value, dict):
            nested = _first_value(value, keys)
            if nested is not None:
                return nested
    return None


def _format_metric(value: Any) -> str:
    if isinstance(value, (int, float)):
        return f"{value:.2f}".rstrip("0").rstrip(".")
    return str(value)


def _normalize_case_pair(case_pair: Any) -> tuple[str, str]:
    if isinstance(case_pair, dict):
        left = str(case_pair.get("existing") or case_pair.get("left") or "existing")
        right = str(case_pair.get("proposed") or case_pair.get("right") or "proposed")
        return left, right
    if isinstance(case_pair, (list, tuple)) and len(case_pair) >= 2:
        return str(case_pair[0]), str(case_pair[1])
    return "existing", "proposed"


def _bf(params: dict[str, Any], key: str, default: float) -> float:
    value = params.get(key, default)
    if value in (None, ""):
        value = default
    return float(value)


def _bs(params: dict[str, Any], key: str, default: str) -> str:
    value = params.get(key, default)
    return str(value if value not in (None, "") else default)


def _brow(name: str, value: Any, unit: str = "", note: str = "") -> dict[str, str]:
    if isinstance(value, float):
        text = f"{value:.3f}".rstrip("0").rstrip(".")
    else:
        text = str(value)
    return {"name": name, "value": text, "unit": unit, "note": note}


def _btable(title: str, rows: list[dict[str, str]]) -> dict[str, Any]:
    return {"title": title, "rows": rows}


def _bbook(title: str, source: str, inputs: list[dict[str, str]], results: list[dict[str, str]]) -> str:
    lines = [title, "", "一、公式来源", source, "", "二、参数代入值", "| 项目 | 数值 | 单位 | 说明 |", "|---|---:|---|---|"]
    lines.extend(f"| {r['name']} | {r['value']} | {r.get('unit', '')} | {r.get('note', '')} |" for r in inputs)
    lines.extend(["", "三、中间过程", "按本程序登记的规范公式和参数表逐项代入，过程量随结构化成果归档。"])
    lines.extend(["", "四、最终结果", "| 项目 | 数值 | 单位 | 说明 |", "|---|---:|---|---|"])
    lines.extend(f"| {r['name']} | {r['value']} | {r.get('unit', '')} | {r.get('note', '')} |" for r in results)
    return "\n".join(lines)


def _frequency_label(freq: str, mode: str = "return_period") -> str:
    text = str(freq)
    if mode == "percent":
        return f"P={text}" if text.endswith("%") else f"P={text}%"
    if text.endswith("%"):
        return f"P={text}"
    return f"{text}年一遇"


def _flows_to_payload(
    program_id: str,
    title: str,
    source: str,
    result: Any,
    flow_map: dict[str, float | None],
    inputs: list[dict[str, str]],
    frequency_mode: str = "return_period",
) -> dict[str, Any]:
    rows = [
        _brow(_frequency_label(freq, frequency_mode), value if value is not None else "-", "m³/s")
        for freq, value in flow_map.items()
    ]
    numeric = [float(value) for value in flow_map.values() if value is not None]
    payload = {
        "program_id": program_id,
        "title": title,
        "source": source,
        "metrics": rows,
        "tables": [_btable("频率洪峰成果", rows)],
        "calculation_book_text": _bbook(f"{title}计算书", source, inputs, rows),
    }
    if numeric:
        payload["peak_flow"] = max(numeric)
    return payload


def _rainfall_dict(params: dict[str, Any], prefix: str, default: dict[str, float]) -> dict[str, float]:
    return {freq: _bf(params, f"{prefix}_{freq}", value) for freq, value in default.items()}


def _calc_tieyuan(params: dict[str, Any]) -> dict[str, Any]:
    from .calculation_forms import TIEYUAN_RESULT_COLUMNS

    def number(key: str, default: Any = None) -> float:
        value = params.get(key, default)
        if value in (None, ""):
            value = default
        return core._require_positive(key, value)

    def optional(key: str) -> float | None:
        return None if params.get(key) in (None, "") else number(key)

    def flag(key: str, default: bool = False) -> bool:
        value = str(params.get(key, default)).strip().lower()
        if value in {"1", "true", "yes", "是"}:
            return True
        if value in {"", "0", "false", "no", "否", "none"}:
            return False
        raise core.HydrologyError(f"{key} 必须明确选择是或否")

    soil = _bs(params, "soil_class", "III").strip().upper()
    moisture = _bs(params, "moisture_condition", "medium").strip().lower()
    if soil not in core.TIE_SOIL_CLASS_PARAMS or moisture not in core.TIE_SOIL_CLASS_PARAMS[soil]:
        raise core.HydrologyError("请选择有效土类 II～VI 和前期水分 wet/medium/dry")
    soil_values = core.TIE_SOIL_CLASS_PARAMS[soil][moisture]
    soil_mode = _bs(params, "soil_param_mode", "legacy")
    if soil_mode not in {"legacy", "table", "manual"}:
        raise core.HydrologyError("不支持的产流损失参数模式")
    r_sources = {}
    r_values = {}
    for key, table_key in (("r_param", "r"), ("r1_param", "r1")):
        manual = soil_mode == "manual" or (soil_mode == "legacy" and params.get(key) not in (None, ""))
        r_values[key] = number(key) if manual else soil_values[table_key]
        r_sources[key] = "人工输入（保留原参数键）" if manual else f"core.TIE_SOIL_CLASS_PARAMS[{soil}][{moisture}]"

    has_direct_rain = any(params.get(f"h24_{freq}") not in (None, "") for freq in core.TIE_FREQUENCIES)
    inferred_rain_mode = "statistics" if not has_direct_rain and params.get("h24_mean") not in (None, "") else "direct"
    rain_mode = _bs(params, "rainfall_mode", inferred_rain_mode)
    kp_values, point_values = {}, {}
    rain_defaults = []
    if rain_mode == "statistics":
        mean, cv, cs_ratio = number("h24_mean"), number("h24_cv"), number("h24_cs_ratio", 3.5)
        eta, phi = number("areal_reduction", 1.0), number("shape_correction", 1.0)
        kp_values, point_values, rainfall = core.tieyuan_design_rainfall_values(mean, cv, eta, core.TIE_FREQUENCIES, cs_ratio, phi)
    elif rain_mode == "direct":
        rainfall = {}
        legacy_defaults = {"100": 180.0, "50": 160.0, "20": 135.0, "10": 118.0}
        for freq in core.TIE_FREQUENCIES:
            key = f"h24_{freq}"
            if params.get(key) not in (None, ""):
                rainfall[freq] = number(key)
            elif not params.get("rainfall_mode") and key not in params and freq in legacy_defaults:
                rainfall[freq] = legacy_defaults[freq]
                rain_defaults.append(freq)
        if not rainfall:
            raise core.HydrologyError("直接雨量模式至少需要一项24h面雨量；未填频率不自动补造")
        eta = phi = None
    else:
        raise core.HydrologyError("24h暴雨输入方式必须为 direct 或 statistics")

    auto_c = flag("runoff_coefficient_auto")
    manual_c = None if auto_c else number("runoff_coefficient", 0.86)
    if manual_c is not None and manual_c > 1:
        raise core.HydrologyError("径流系数 C 必须大于0且不大于1")
    data = core.TieyuanInput(
        basin_name=_bs(params, "basin_name", "示例流域"),
        area=number("area", params.get("area_km2") or 47.02),
        river_length_l1=number("river_length_l1", 13.8),
        river_slope_j1=number("river_slope_j1", 0.095724),
        slope_length_l2=optional("slope_length_l2"),
        slope_slope_i2=optional("slope_slope_i2"),
        a1=number("a1", 0.341), a2=number("a2", 0.025), n=number("n", 0.65),
        rainfall_24h=rainfall,
        rainfall_24h_is_areal=True,
        soil_class=soil,
        moisture_condition=moisture,
        runoff_coefficient=manual_c,
        runoff_coefficient_auto=auto_c,
        c_landform=_bs(params, "c_landform", "高低山"),
        r_param=r_values["r_param"], r1_param=r_values["r1_param"],
        beta1=number("beta1", 0.65),
    )
    result = core.calculate_tieyuan(data)
    source = "历史 TieyuanTab 与 core.calculate_tieyuan；R/r1使用已有土类水分表，自动C使用已有表3-3。A1/A2未新增规范参数表；历史Excel仅作19列成果结构参考，不采用其例值替换核心公式。"
    warnings = list(result.warnings)
    if rain_defaults:
        warnings.append("未设雨量模式的旧接口沿用历史示例面雨量：" + "、".join(rain_defaults) + "年；正式工程须核定。")
    warnings.append("A1/A2来自人工输入或历史软件默认值，不能视为已核定的规范查表值；β1仅兼容归档，当前核心洪峰公式未使用。")

    def input_row(label: str, value: Any, unit: str, origin: str) -> dict[str, Any]:
        return {"参数": label, "采用值": value, "单位": unit, "取值来源": origin}

    input_rows = [
        input_row("流域名称", data.basin_name, "", "工程输入"),
        input_row("流域面积 F", data.area, "km²", "工程输入 / 旧接口默认值"),
        input_row("主河道长度 L1", data.river_length_l1, "km", "工程输入 / 旧接口默认值"),
        input_row("主河道比降 J1", data.river_slope_j1, "小数", "原桥接数值口径；不自动换算旧草稿"),
        input_row("土类", soil, "", "下拉选择"),
        input_row("前期土壤水分", {"wet": "湿润", "medium": "中等", "dry": "干燥"}[moisture], "", "下拉选择"),
        input_row("地貌", data.c_landform, "", "下拉选择"),
        input_row("R", data.r_param, "", r_sources["r_param"]),
        input_row("r1", data.r1_param, "", r_sources["r1_param"]),
        input_row("C取值方式", "表3-3自动查算" if auto_c else "人工输入", "", "逐频率采用值见过程表"),
        input_row("人工C", "" if auto_c else manual_c, "", "自动模式不采用隐藏人工值" if auto_c else "人工输入 / 历史默认0.86"),
        input_row("24h暴雨来源", "均值频率计算" if rain_mode == "statistics" else "直接面雨量", "", "直接面雨量不重复折减"),
        input_row("β1", data.beta1, "", "历史兼容键；当前洪峰公式未使用"),
    ]
    for key in ("a1", "a2", "n"):
        origin = "人工输入；采用依据需核定" if params.get(key) not in (None, "") else "core.TieyuanInput历史默认值；不是规范查表值"
        input_rows.append(input_row(key.upper() if key != "n" else key, getattr(data, key), "", origin))
    if rain_mode == "statistics":
        for label, value, unit in (("24h点雨量均值", mean, "mm"), ("Cv", cv, ""), ("Cs/Cv", cs_ratio, ""), ("点面折减系数η", eta, ""), ("形状修正系数φ", phi, "")):
            input_rows.append(input_row(label, value, unit, "人工输入；core.tieyuan_design_rainfall_values"))

    # Reconstruct unrounded common operands; core's public result rounds them to 3 decimals.
    l2 = data.slope_length_l2 if data.slope_length_l2 is not None else data.area / (2.7 * data.river_length_l1)
    j2 = data.slope_slope_i2 if data.slope_slope_i2 is not None else data.river_slope_j1 * 0.6
    k1 = 0.278 * data.river_length_l1 / (data.a1 * data.river_slope_j1 ** 0.35)
    k2 = 0.278 * l2 ** 0.5 * data.area ** 0.5 / (data.a2 * j2 ** 0.33)
    x = k1 + 0.95 * k2
    common_rows = [
        input_row("坡面长度 L2", l2, "km", "人工输入" if data.slope_length_l2 is not None else "历史估算 L2=F/(2.7L1)"),
        input_row("山坡平均坡度 J2", j2, "小数", "人工输入" if data.slope_slope_i2 is not None else "历史估算 J2=0.6J1"),
        input_row("主河槽汇流因子 K1", k1, "", "0.278L1/(A1×J1^0.35)"),
        input_row("山坡汇流因子 K2", k2, "", "0.278×L2^0.5×F^0.5/(A2×J2^0.33)；保持现有core"),
        input_row("综合汇流因子 X", x, "", "K1+0.95K2"),
        input_row("汇流指数 y", result.y, "", f"core.TIE_Y_TABLE；K1/K2={k1/k2:.12g}"),
        input_row("γ", result.gamma, "", f"core.TIE_GAMMA_TABLE；K1+K2={k1+k2:.12g}"),
    ]
    fixed_rows, rain_rows, detail_rows = [], [], []
    for freq in core.TIE_FREQUENCIES:
        probability = core.TIE_FREQUENCY_PROBABILITIES[freq]
        label = f"P={probability:.6g}%（{freq}年一遇）"
        item = result.frequency_params.get(freq)
        values = [label] + [""] * 18
        if item is not None:
            storm_s = rainfall[freq] * 24 ** (data.n - 1)
            production = 0.278 * storm_s * data.area
            values = [label, eta if eta is not None else "", storm_s, production, item.loss_factor_k2,
                      k1, k2, x, k1/k2, result.y, k1+k2, result.gamma, item.runoff_coefficient,
                      data.n, item.n_prime, item.cn, item.p1, item.p, item.q]
            rain_rows.append({"重现期（年）": freq, "设计频率（%）": probability, "Kp": kp_values.get(freq, ""),
                              "24h点雨量（mm）": point_values.get(freq, ""), "点面折减系数η": eta if eta is not None else "",
                              "形状修正系数φ": phi if phi is not None else "", "24h面雨量（mm）": rainfall[freq],
                              "暴雨参数S": storm_s, "来源": "历史默认算例（待核定）" if freq in rain_defaults else ("均值×Kp×η×φ" if rain_mode == "statistics" else f"人工面雨量 h24_{freq}（不重复折减）")})
            detail_rows.append({"设计频率": label, "面雨量H24（mm）": item.rainfall_24h, "S": item.storm_parameter_s,
                                "C": item.runoff_coefficient, "表3-3时段t（h）": item.c_table_t if item.c_table_t is not None else "",
                                "损失因子k2": item.loss_factor_k2, "Cn": item.cn, "n′": item.n_prime, "P1": item.p1, "P": item.p, "洪峰流量（m³/s）": item.q,
                                "参数代入": f"Q=[({production:.12g}×{item.runoff_coefficient:g}×[1-(1-{item.p1:g})^{result.gamma:g}])/({x:.12g}×{item.p1:g})^{data.n:g}]^[1/(1-{data.n:g}×{result.y:g})]={item.q:g}",
                                "C来源": "core表3-3（逐频率查算）" if auto_c else "人工输入"})
            if auto_c and not core.TIE_C_S_VALUES[0] <= storm_s <= core.TIE_C_S_VALUES[-1]:
                warnings.append(f"{label}：S={storm_s:.6g}超出表3-3的S列范围，现有core按端点取C，正式采用前须复核。")
        fixed_rows.append(dict(zip(TIEYUAN_RESULT_COLUMNS, values)))

    tables = [
        {"title": "铁一院两所法固定成果表", "columns": list(TIEYUAN_RESULT_COLUMNS), "rows": fixed_rows},
        {"title": "输入参数及取值来源", "columns": list(input_rows[0]), "rows": input_rows},
        {"title": "24h设计暴雨计算表", "columns": list(rain_rows[0]), "rows": rain_rows},
        {"title": "汇流及折减参数表", "columns": list(common_rows[0]), "rows": common_rows},
        {"title": "逐频率产汇流详细过程", "columns": list(detail_rows[0]), "rows": detail_rows},
    ]
    payload = _flows_to_payload("1-02", "铁一院两所法设计洪水", source, result, result.flows, [])
    payload.update(tables=tables, warnings=warnings, flows=result.flows,
                   input_modes={"rainfall_mode": rain_mode, "soil_param_mode": soil_mode, "runoff_coefficient_auto": auto_c})
    lines = [payload["title"] + "计算书", source,
             "S=H24×24^(n-1)；产流因子k1=0.278SF；损失因子k2=R×S^(r1-1)。",
             "Cn按core.TIE_CN_TABLE插值；n′=Cn×n；P1按core.TIE_P1_TABLE插值；P=1-(1-P1)^γ。",
             "Q=[k1×C×P/(X×P1)^n]^[1/(1-n×y)]。查表与舍入沿用core；固定成果表与详细表使用同一计算结果。"]
    for table in tables:
        columns = table["columns"]
        lines.extend(["", table["title"], "| " + " | ".join(columns) + " |", "| " + " | ".join("---" for _ in columns) + " |"])
        lines.extend("| " + " | ".join(str(row.get(column, "")) for column in columns) + " |" for row in table["rows"])
    lines.extend(["", "复核提示", *warnings])
    payload["calculation_book_text"] = "\n".join(lines)
    return payload


def _calc_regional(params: dict[str, Any]) -> dict[str, Any]:
    from decimal import Decimal, InvalidOperation

    raw_region = params.get("region_id")
    try:
        parsed_region = Decimal(str(raw_region).strip())
        if not parsed_region.is_finite() or parsed_region != parsed_region.to_integral_value():
            raise ValueError
        if parsed_region not in core.REGION_BY_ID:
            raise ValueError
        region_id = int(parsed_region)
    except (InvalidOperation, ValueError, TypeError):
        raise core.HydrologyError("请从中文下拉中明确选择有效地区经验分区；不能省略、截断小数或猜测分区") from None
    flag = str(params.get("allow_out_of_range", False)).strip().lower()
    if flag not in {"true", "false", "1", "0", "yes", "no", "是", "否", "", "none"}:
        raise core.HydrologyError("是否允许外推必须明确选择")
    data = core.RegionalFormulaInput(
        basin_name=_bs(params, "basin_name", "示例流域"),
        area=_bf(params, "area", _bf(params, "area_km2", 47.02)),
        region_id=region_id,
        allow_out_of_range=flag in {"true", "1", "yes", "是"},
    )
    result = core.calculate_regional_formula(data)
    region = core.REGION_BY_ID[region_id]
    warnings = list(result.warnings)
    try:
        region.validate_area(data.area)
    except core.HydrologyError as exc:
        warnings.append(f"已明确允许超出分区面积范围：{exc}；本次为外推结果，正式采用前必须复核。")
    low, high = region.area_range
    area_range = (f"{low:g} < F < {high:g}" if low is not None and high is not None else
                  f"F > {low:g}" if low is not None else f"F < {high:g}" if high is not None else "源表未注明面积界限，仍须核对水系与适用条件")
    source = "core.REGION_PARAMS既有67分区参数；Q=K×F^(1-n)。分区63的0.1%沿用已核定K=0.00222，不修改历史资料。"
    inputs = [_brow("流域名称", data.basin_name), _brow("分区", result.region_name), _brow("流域面积", data.area, "km²"),
              _brow("适用面积范围", area_range, "km²"), _brow("允许外推", "是" if data.allow_out_of_range else "否"), _brow("指数 n", result.n)]
    payload = _flows_to_payload(
        "1-03",
        "地区经验公式洪峰流量",
        source,
        result,
        result.flows,
        inputs,
        frequency_mode="percent",
    )
    parameter_rows, flow_rows = [], []
    for freq, n, k in core.regional_formula_parameter_rows(region_id):
        parameter_rows.append({"设计频率": f"P={freq}", "指数n": n, "系数K": k, "计算式": "Q=K×F^(1-n)", "参数来源": f"core.REGION_PARAMS 分区{region_id}：{region.name}"})
        flow_rows.append({"设计频率": f"P={freq}", "流域面积F（km²）": data.area, "指数n": n, "系数K": k,
                          "面积项F^(1-n)": data.area ** (1-n), "洪峰流量Q（m³/s）": result.flows[freq],
                          "参数代入": f"{k:.12g}×{data.area:g}^(1-({n:g}))={result.flows[freq]:g}"})
    tables = [
        {"title": "地区经验公式洪峰成果表", "columns": list(flow_rows[0]), "rows": flow_rows},
        {"title": "地区经验公式参数表", "columns": list(parameter_rows[0]), "rows": parameter_rows},
        _btable("分区与适用范围核对", inputs),
    ]
    lines = [payload["calculation_book_text"]]
    for table in tables[:2]:
        columns = table["columns"]
        lines.extend(["", table["title"], "| " + " | ".join(columns) + " |", "| " + " | ".join("---" for _ in columns) + " |"])
        lines.extend("| " + " | ".join(str(row[column]) for column in columns) + " |" for row in table["rows"])
    lines.extend(["", "复核提示", *warnings])
    payload.update(tables=tables, warnings=warnings, flows=result.flows, region_id=region_id,
                   region_name=result.region_name, allow_out_of_range=data.allow_out_of_range,
                   calculation_book_text="\n".join(lines))
    return payload


def _calc_flood_synthesis(params: dict[str, Any]) -> dict[str, Any]:
    if 'components_json' not in params:
        raise core.HydrologyError('请输入实际子流域过程线；不再用预设三角形过程代替项目数据。')
    if 'components_json' in params:
        raw = json.loads(str(params['components_json']))
        if not isinstance(raw, list) or not raw:
            raise core.HydrologyError('请至少输入一条子流域过程线。')
        components = [core.FloodSynthesisComponent(str(item['name']), float(item.get('offset_h', 0)), [float(v) for v in item['flows']]) for item in raw]
        if any(len(c.flows) < 2 for c in components):
            raise core.HydrologyError('每条过程线至少需要两个流量值。')
    result = core.synthesize_flood_hydrographs(components, _bf(params, "delta_t_h", 1.0))
    rows = [_brow("合成洪峰", result.peak_flow_m3s, "m³/s"), _brow("峰现时间", result.peak_time_h, "h"), _brow("洪水总量", result.flood_volume_10000m3, "万m³")]
    points = [{"time_h": p.time_h, "discharge_m3s": p.total_flow} for p in result.points]
    process_rows = [{'时间（h）': p.time_h, **{f'{name}（m³/s）': flow for name, flow in p.component_flows.items()}, '合成流量（m³/s）': p.total_flow} for p in result.points]
    series = [{"title": "合成洪水过程线", "points": points, "xKey": "time_h", "yKey": "discharge_m3s", 'xLabel':'时间（h）', 'yLabel':'流量（m³/s）'}]
    for component in result.components:
        series.append({'title':component.name, 'xKey':'time_h', 'yKey':'discharge_m3s', 'xLabel':'时间（h）', 'yLabel':'流量（m³/s）', 'points':[{'time_h':p.time_h, 'discharge_m3s':p.component_flows[component.name]} for p in result.points]})
    return {
        "program_id": "1-06",
        "title": "洪水过程线合成",
        "source": "按统一时间步长及各子流域时移叠加实际输入的流量序列。同一流域不同计算方法的成果应对比，不能相加。",
        "metrics": rows,
        "tables": [_btable("合成成果", rows), {'title':'逐时段子流域叠加过程', 'columns':list(process_rows[0]), 'rows':process_rows}],
        "series": series,
        "peak_flow": result.peak_flow_m3s,
        "calculation_book_text": _bbook("洪水过程线合成计算书", "Q合(t)=ΣQi(t−Δti)；非整时段时移采用线性插值。洪量W=Σ[(Qj+Qj+1)/2×Δt×3600]。各分量必须来自不同汇水单元。", [_brow("时间步长", _bf(params, "delta_t_h", 1.0), "h")], rows),
    }


def _calc_reservoir_flood_volume(params: dict[str, Any]) -> dict[str, Any]:
    from .reservoir_storage import ReservoirFloodInput, calculate_reservoir_flood
    data = ReservoirFloodInput(_bf(params,'catchment_area_km2',47.02), _bf(params,'rainfall_decay_exponent',.67), _bf(params,'peak_flow_coefficient',.278), _bf(params,'design_24h_rainfall_mm',95), _bf(params,'net_rainfall_mm',55))
    return book_program_adapters._generic_payload('1-08','水库洪峰流量和洪水总量',data,calculate_reservoir_flood,'《水利大算盘》第五章第1节')


def _calc_small_river(params: dict[str, Any]) -> dict[str, Any]:
    rainfall_is_mean = str(params.get("rainfall_is_mean", "false")).lower() in {"1", "true", "yes", "是", "均值", "mean"}
    if rainfall_is_mean:
        mean_h = _bf(params, "rainfall_mean_mm", 70.0)
        rainfall = {freq: mean_h for freq in core.SMALL_RIVER_FREQUENCIES}
    else:
        rainfall = {
            "10": _bf(params, "rainfall_10", 56.0),
            "5": _bf(params, "rainfall_5", _bf(params, "rainfall_20", 68.0)),
            "2": _bf(params, "rainfall_2", _bf(params, "rainfall_50", 82.0)),
            "1": _bf(params, "rainfall_1", _bf(params, "rainfall_100", 95.0)),
        }
    data = core.SmallRiverInput(
        basin_name=_bs(params, "basin_name", "示例流域"),
        region=int(_bf(params, "region", 1)),
        area=_bf(params, "area", _bf(params, "area_km2", 47.02)),
        river_length=_bf(params, "river_length", 13.8),
        slope=_bf(params, "slope", 0.095724),
        cv=_bf(params, "cv", 0.45),
        rainfall=rainfall,
        rainfall_is_mean=rainfall_is_mean,
    )
    result = core.calculate_small_river(data)
    rainfall_mode = "流域平均雨量均值 H₀×Kp" if data.rainfall_is_mean else "各频率流域平均设计 H"
    inputs = [
        _brow("流域面积", data.area, "km²"),
        _brow("面积对应历时", result.duration, ""),
        _brow("河长", data.river_length, "km"),
        _brow("比降", data.slope, "小数"),
        _brow("H输入模式", rainfall_mode),
    ]
    source = "复用 core.calculate_small_river；H 为面积对应历时的流域平均雨量。直接设计 H 模式不再乘 Kp；均值 H₀ 模式按 Cv 与 Cs=3.5Cv 计算 Kp 后代入。"
    payload = _flows_to_payload(
        "1-09",
        "中小河流经验公式法",
        source,
        result,
        result.flows,
        inputs,
        frequency_mode="percent",
    )
    rainfall_rows = []
    for freq in core.SMALL_RIVER_FREQUENCIES:
        kp = result.kp_values.get(freq)
        original_h = rainfall.get(freq)
        design_h = result.design_rainfall.get(freq)
        note = f"Kp={kp:.3f}；输入H₀={original_h:g}" if kp is not None else "直接输入流域平均设计H"
        rainfall_rows.append(_brow(_frequency_label(freq, "percent"), design_h, "mm", note))
    payload["tables"].append(_btable("H输入口径与代入雨量", rainfall_rows))
    payload["calculation_book_text"] = _bbook("中小河流经验公式法计算书", source, inputs + rainfall_rows, payload["metrics"])
    payload["duration"] = result.duration
    return payload


def _calc_flood_compare(params: dict[str, Any]) -> dict[str, Any]:
    values = {
        "铁一两所法": _bf(params, "tieyuan_q_m3s", 250.0),
        "地区经验公式": _bf(params, "regional_q_m3s", 230.0),
        "瞬时单位线法": _bf(params, "unit_q_m3s", 255.88),
        "推理公式法": _bf(params, "rational_q_m3s", 240.0),
    }
    adopted = sum(values.values()) / len(values)
    rows = [_brow(name, value, "m³/s", f"偏差={(value - adopted) / adopted * 100:.1f}%") for name, value in values.items()]
    rows.append(_brow("建议采用值", adopted, "m³/s", "按多方法均值给出，正式采用需结合地区经验校核。"))
    return {"program_id": "1-10", "title": "设计洪水成果对比", "source": "对已计算多方法洪峰成果进行同表对比和偏差复核。", "metrics": rows, "tables": [_btable("多方法成果对比", rows)], "peak_flow": adopted, "calculation_book_text": _bbook("设计洪水成果对比计算书", "多方法成果对比。", [], rows)}


def _section_points(params: dict[str, Any], key: str = "points") -> tuple[tuple[float, float], ...]:
    from .manning_output import parse_section_points

    return parse_section_points(params.get(key, "0,1200;10,1198;20,1198;30,1200"))


def _fmt_num(value: Any, digits: int = 2) -> str:
    try:
        number = float(value)
    except (TypeError, ValueError):
        return "" if value in (None, "") else str(value)
    return f"{number:.{digits}f}"


def _engineering_table(title: str, columns: list[str], rows: list[dict[str, Any]]) -> dict[str, Any]:
    return {"title": title, "columns": columns, "rows": rows}


def _rating_curve_for_section(
    points: tuple[tuple[float, float], ...],
    roughness_n: float,
    slope: float,
    design_level: float | None,
    level_step_m: float | None = None,
    maximum_depth_m: float | None = None,
    target_levels: tuple[float, ...] = (),
) -> tuple[list[dict[str, str]], list[dict[str, float]]]:
    from .manning_output import rating_curve_for_section

    return rating_curve_for_section(points, roughness_n, slope, design_level,
                                    level_step_m, maximum_depth_m, target_levels)


def _calc_manning(params: dict[str, Any]) -> dict[str, Any]:
    from .manning_output import build_manning_output

    return build_manning_output(params, _section_points(params), _rating_curve_for_section)


def _calc_hec_standard_step(params: dict[str, Any]) -> dict[str, Any]:
    payload = dict(flood_mainline.calculate_wsl_01(params))
    payload["program_id"] = "4-03"
    payload["title"] = "多断面能量方程标准步法"
    payload["source"] = (
        "按多断面能量方程标准步法沿程推求水面线，输出水位计算表、"
        "水位流量关系曲线、沿程水面线/能量线和断面水位标注图。"
    )
    if payload.get("calculation_book_text"):
        payload["calculation_book_text"] = str(payload["calculation_book_text"]).replace(
            "水面线推求计算书",
            "多断面能量方程标准步法计算书",
            1,
        )
    return payload


def _calc_area_analogy(params: dict[str, Any]) -> dict[str, Any]:
    # 1-11 水文面积比拟法: Q设 = Q参 × (F设/F参)^n
    from .core import StationFlow, TIE_FREQUENCIES
    reference_area = _require_positive_helper("参证站面积F参", _bf(params, "reference_area_km2", _bf(params, "F参", 0.0)))
    design_area = _require_positive_helper("设计流域面积F设", _bf(params, "design_area_km2", _bf(params, "F设", 0.0)))
    reference_flows: dict[str, float] = {}
    for freq in TIE_FREQUENCIES:
        v = _bf(params, f"reference_q_{freq}", _bf(params, f"Q参_{freq}", 0.0))
        if v and v > 0:
            reference_flows[freq] = v
    n_values: dict[str, float] = {}
    for freq in TIE_FREQUENCIES:
        v = _bf(params, f"n_{freq}", 0.0)
        if v and v > 0:
            n_values[freq] = v
    n_unified = _bf(params, "unified_n", 0.667)
    # n值线性回归（多于2个站时启用）
    stations: list[core.StationFlow] = []
    count_value = _bf(params, 'stations_count', 0)
    if not count_value.is_integer() or count_value < 0 or count_value == 1:
        raise core.HydrologyError('回归参证站数量须为0（不回归）或至少2个站。')
    stations_count = int(count_value)
    if params.get('stations_json'):
        raw_stations = json.loads(str(params['stations_json']))
        if not isinstance(raw_stations, list) or len(raw_stations) < 2:
            raise core.HydrologyError('多站资料至少提供两个站，每站包含name、area和flows。')
        for i,item in enumerate(raw_stations,1):
            station_area = _require_positive_helper(f'站{i}面积', float(item['area']))
            flow_map = {str(f):_require_positive_helper(f'站{i}流量',float(q)) for f,q in item['flows'].items()}
            if not flow_map or any(f not in TIE_FREQUENCIES for f in flow_map):
                raise core.HydrologyError('参证站flows的键须为支持的重现期，例如100、50、20。')
            stations.append(core.StationFlow(str(item.get('name',f'站{i}')),station_area,flow_map))
        stations_count = 0
    for i in range(1, max(stations_count, 0) + 1):
        try:
            s_area = _bf(params, f"station_{i}_area", 0.0)
            s_q_ref = _bf(params, f"station_{i}_q_ref", 0.0)
            s_freq = _bs(params, f"station_{i}_freq", "")
            if s_area > 0 and s_q_ref > 0 and s_freq:
                stations.append(core.StationFlow(
                    name=_bs(params, f"station_{i}_name", f"站{i}"),
                    area=s_area,
                    flows={s_freq: s_q_ref},
                ))
            else:
                raise core.HydrologyError(f'回归参证站{i}缺少有效面积、频率或洪峰；两站以上请使用多站JSON输入。')
        except (TypeError, ValueError) as exc:
            raise core.HydrologyError(f'回归参证站{i}资料无效：{exc}') from exc
    if len(stations) >= 2 and not n_values:
        # 对每个频率分别回归
        all_freqs = set()
        for s in stations:
            all_freqs.update(s.flows.keys())
        for freq in all_freqs:
            try:
                idx_n = core.calculate_area_index(stations, frequencies=[freq])
                if idx_n.get(freq) is not None:
                    n_values[freq] = idx_n[freq]
                else:
                    raise core.HydrologyError(f'{freq}年回归至少需要两个不同面积且有同频率洪峰的站。')
            except (ValueError, core.HydrologyError) as exc:
                raise core.HydrologyError(f'{freq}年回归未完成：{exc}') from exc
    flows = core.calculate_area_analogy(
        reference_area=reference_area,
        design_area=design_area,
        reference_flows=reference_flows,
        n_values=n_values if n_values else None,
        unified_n=n_unified,
    )
    rows = []
    for freq, q in flows.items():
        if q is None:
            rows.append(_brow(f"{freq}年", "—", "m³/s", "缺参证站流量"))
        else:
            n_used = n_values.get(freq, n_unified)
            ratio = design_area / reference_area
            extra = f"n={n_used:.3f}; 倍比={ratio:.3f}^{n_used:.3f}"
            if ratio > 3 or ratio < 0.333:
                extra += " ⚠过度外推"
            rows.append(_brow(f"{freq}年", q, "m³/s", extra))
    if n_values:
        idx_text = "n值: " + ", ".join(f"{f}={v:.3f}" for f, v in n_values.items() if v)
    else:
        idx_text = f"统一n={n_unified:.3f}"
    inputs = [
        _brow("参证站面积", reference_area, "km²"),
        _brow("设计流域面积", design_area, "km²"),
        _brow("面积比F设/F参", design_area / reference_area, ""),
        _brow(idx_text.split(": ")[0], idx_text, ""),
    ]
    return {"program_id": "1-11", "title": "水文面积比拟法", "source": "复用 core.calculate_area_analogy; Q设 = Q参×(F设/F参)^n。可选 n 值线性回归。", "metrics": rows, "tables": [_btable("各频率设计流域洪峰", rows)], "peak_flow": max((q for q in flows.values() if q is not None), default=None), "calculation_book_text": _bbook("水文面积比拟法计算书", "复用 core.calculate_area_analogy; Q设 = Q参×(F设/F参)^n。", inputs, rows), "formula": "Q设 = Q参 × (F设/F参)^n"}


def _require_positive_helper(name: str, value: float) -> float:
    if value is None or not math.isfinite(float(value)) or float(value) <= 0:
        raise core.HydrologyError(f"{name} 必须大于 0")
    return float(value)


def _calc_double_log(params: dict[str, Any]) -> dict[str, Any]:
    # 1-12 双对数内插法: 在 log(F)-log(Q) 坐标上按两个参证站线性内插
    from .core import StationFlow, TIE_FREQUENCIES
    station_a = core.StationFlow(
        name=_bs(params, "station_a_name", "参证站A"),
        area=_require_positive_helper("参证站A面积", _bf(params, "station_a_area", 0.0)),
        flows={freq: _bf(params, f"station_a_q_{freq}", 0.0) for freq in TIE_FREQUENCIES if _bf(params, f"station_a_q_{freq}", 0.0) > 0},
    )
    station_b = core.StationFlow(
        name=_bs(params, "station_b_name", "参证站B"),
        area=_require_positive_helper("参证站B面积", _bf(params, "station_b_area", 0.0)),
        flows={freq: _bf(params, f"station_b_q_{freq}", 0.0) for freq in TIE_FREQUENCIES if _bf(params, f"station_b_q_{freq}", 0.0) > 0},
    )
    design_area = _require_positive_helper("设计流域面积F设", _bf(params, "design_area_km2", _bf(params, "design_area", 0.0)))
    result = core.double_log_interpolate_flows(station_a, station_b, design_area)
    rows = []
    for freq, data in result.items():
        if data.get("q_design") is None:
            rows.append(_brow(f"{freq}年", "—", "m³/s", "缺流量"))
        else:
            rows.append(_brow(
                f"{freq}年",
                data["q_design"],
                "m³/s",
                f"n={data['n']:.3f}; 位置={data['position']}; 权重={data['weight']:.3f}",
            ))
    inputs = [
        _brow(f"{station_a.name}面积", station_a.area, "km²"),
        _brow(f"{station_b.name}面积", station_b.area, "km²"),
        _brow("设计流域面积", design_area, "km²"),
    ]
    return {"program_id": "1-12", "title": "双对数内插法", "source": "复用 core.double_log_interpolate_flows; 在 log(F)-log(Q) 坐标上逐频率线性内插。", "metrics": rows, "tables": [_btable("各频率设计流域洪峰", rows)], "peak_flow": max((data["q_design"] for data in result.values() if data.get("q_design") is not None), default=None), "calculation_book_text": _bbook("双对数内插法计算书", "复用 core.double_log_interpolate_flows。", inputs, rows)}


def _calc_frequency_curve(params: dict[str, Any]) -> dict[str, Any]:
    # 1-13 水文频率适线计算（基于 SL 44-2006 附录 A P-III 型曲线）
    # 完全独立实现，参数估计用矩法，适线准则支持绝对/相对离差平方和最小
    samples_raw = _bs(params, "samples", "")
    samples_text = samples_raw.strip()
    if not samples_text:
        raise core.HydrologyError("请输入洪水样本系列（如 1246,890,650,420,...）")
    # 解析样本：支持逗号/分号/换行/空格分隔
    sample_strs = re.split(r"[,，;；\s]+", samples_text)
    samples = []
    for s in sample_strs:
        s = s.strip()
        if s:
            try:
                v = float(s)
                if math.isfinite(v) and v > 0:
                    samples.append(v)
                else:
                    raise core.HydrologyError("样本值必须大于 0")
            except ValueError:
                raise core.HydrologyError(f"样本中包含无法解析的值：{s}")
    n = len(samples)
    if n < 3:
        raise core.HydrologyError(f"频率适线至少需要 3 个样本，当前 {n} 个")
    # 矩法估计
    mean_value = sum(samples) / n
    variance = sum((x - mean_value) ** 2 for x in samples) / (n - 1)
    std = variance ** 0.5
    cv_estimated = std / mean_value if mean_value > 0 else 0
    if cv_estimated <= 0:
        raise core.HydrologyError("样本全部相同，无法估计频率曲线；请核对样本系列。")
    cs_method = _bs(params, "cs_method", "ratio")
    cs_ratio = _bf(params, "cs_cv_ratio", _bf(params, "cs_ratio", 2.5))
    if cs_method == "ratio":
        cs_estimated = cs_ratio * cv_estimated
    else:
        cs_estimated = _bf(params, "cs_fixed", 1.0)
    # 经验频率（数学期望公式 P_m = m/(n+1)）
    sorted_samples = sorted(samples, reverse=True)  # 降序
    empirical_freqs = []
    rank_data = [(v, (i + 1) / (n + 1)) for i, v in enumerate(sorted_samples)]
    for sample_val, freq_p in rank_data:
        empirical_freqs.append((sample_val, freq_p))
    # 适线准则
    criterion = _bs(params, "criterion", "absolute")  # 'absolute' or 'relative'
    # 适线调整：用 Cs/Cv 比值（用户可调）微调 Cs
    cs_cv_ratio_adjust = _bf(params, "fit_cs_cv_ratio", cs_ratio)
    if cs_cv_ratio_adjust != cs_ratio and cs_method == "ratio":
        cs_adjusted = cs_cv_ratio_adjust * cv_estimated
    else:
        cs_adjusted = cs_estimated
    if cs_method not in ("ratio", "fixed") or criterion not in ("absolute", "relative"):
        raise core.HydrologyError("请选择有效的 Cs 方法和偏差评价方式。")
    if not math.isfinite(cs_adjusted) or cs_adjusted <= 0:
        raise core.HydrologyError("当前 P-III 实现要求 Cs 大于 0；零偏和负偏样本需另行选用适用方法。")
    effective_ratio = cs_adjusted / cv_estimated
    # 生成理论频率曲线（使用 GAMMAINV）
    design_freqs_str = _bs(params, "design_freqs", "0.1,0.2,0.5,1,2,5,10,20,50")
    design_freqs = []
    for s in re.split(r"[,，;；\s]+", design_freqs_str):
        s = s.strip()
        if s:
            try:
                probability = float(s)
                if not math.isfinite(probability) or not 0 < probability < 100:
                    raise ValueError()
                design_freqs.append(probability)
            except ValueError:
                raise core.HydrologyError(f"设计频率必须是 0 至 100 之间的有效百分数：{s}")
    if not design_freqs:
        raise core.HydrologyError("请至少输入一个设计频率。")
    design_freqs = sorted(set(design_freqs))
    theoretical_curve = []
    design_values = {}
    for df_p in design_freqs:
        if df_p <= 0 or df_p >= 100:
            continue
        kp = core.pearson_iii_kp(cv_estimated, effective_ratio, df_p)
        qp = mean_value * kp
        theoretical_curve.append((df_p, qp))
        design_values[f"{df_p}%"] = round(qp, 3)
    # 适线拟合度计算（绝对离差平方和）
    abs_dev_sq = 0
    rel_dev_sq = 0
    for sample_val, freq_p in empirical_freqs:
        kp_calc = core.pearson_iii_kp(cv_estimated, effective_ratio, freq_p * 100)
        q_theory = mean_value * kp_calc
        abs_dev_sq += (sample_val - q_theory) ** 2
        if q_theory > 0:
            rel_dev_sq += ((sample_val - q_theory) / q_theory) ** 2
    fit_quality_abs = 1 - abs_dev_sq / sum((x - mean_value) ** 2 for x in samples)
    fit_quality_rel = 1 - rel_dev_sq / n if n > 0 else 1.0
    fit_quality = fit_quality_abs if criterion == "absolute" else fit_quality_rel
    # 输出 rows
    rows = [
        _brow("样本数", n, "个"),
        _brow("样本均值Ex", mean_value, ""),
        _brow("变差系数Cv", cv_estimated, ""),
        _brow("偏态系数Cs", cs_adjusted, ""),
        _brow("Cs/Cv", cs_cv_ratio_adjust if cs_method == "ratio" else cs_adjusted / cv_estimated if cv_estimated > 0 else 0, ""),
        _brow("偏差评价方式", {"absolute": "绝对离差平方和", "relative": "相对离差平方和"}.get(criterion, criterion), "", "矩法估计与手动调参；未执行自动最优化。"),
        _brow("适线拟合度", fit_quality, "", f"1 - Σ偏差² / Σ(x-Ex)² (绝对)" if criterion == "absolute" else f"1 - Σ相对偏差²/n (相对)"),
    ]
    empirical_rows = []
    for i, (sample_val, freq_p) in enumerate(empirical_freqs):
        empirical_rows.append(_brow(f"#{i+1}", sample_val, "", f"P={freq_p*100:.2f}% (m={i+1}, n={n})"))
    theoretical_rows = [_brow(f"P={p:g}%", q, "m³/s", "") for p, q in theoretical_curve]
    tables = [
        _btable("统计参数与适线准则", rows),
        _btable("经验频率样本", empirical_rows),
        _btable("理论频率曲线", theoretical_rows),
    ]
    peak_flow = max(q for _, q in theoretical_curve) if theoretical_curve else None
    return {
        "program_id": "1-13",
        "title": "水文频率适线计算",
        "source": "复用 core.pearson_iii_kp (SL 44-2006 附录 A P-III 型曲线); 经验频率 = m/(n+1); 矩法估计 Ex/Cv, Cs/Cv 由用户输入或默认 2.5。",
        "metrics": rows,
        "tables": tables,
        "peak_flow": peak_flow,
        "design_values": design_values,
        "fit_quality": round(fit_quality, 4),
        "series": [
            {"title": "P-III 理论频率曲线", "xKey": "probability_percent", "yKey": "discharge_m3s", "xLabel": "超越概率 P（%）", "yLabel": "洪峰流量 Q（m³/s）", "points": [{"probability_percent": p, "discharge_m3s": q} for p, q in theoretical_curve]},
            {"title": "样本经验频率点", "xKey": "probability_percent", "yKey": "discharge_m3s", "xLabel": "超越概率 P（%）", "yLabel": "洪峰流量 Q（m³/s）", "points": [{"probability_percent": p * 100, "discharge_m3s": q} for q, p in empirical_freqs]},
        ],
        "calculation_book_text": _bbook(
            "水文频率适线计算书",
            "基于 SL 44-2006 附录 A 皮尔逊Ⅲ型曲线; 完全独立实现, 不调用任何商业软件。",
            [_brow("样本数", n, ""), _brow("均值Ex", mean_value, ""), _brow("Cv", cv_estimated, "")],
            rows + theoretical_rows,
        ),
    }


_CALCULATORS = {
    "1-01": flood_mainline.calculate_fld_01,
    "1-02": _calc_tieyuan,
    "1-03": _calc_regional,
    "1-04": flood_mainline.calculate_fld_04,
    "1-05": flood_mainline.calculate_fld_05,
    "1-06": _calc_flood_synthesis,
    "1-08": _calc_reservoir_flood_volume,
    "1-09": _calc_small_river,
    "1-10": _calc_flood_compare,
    "1-11": _calc_area_analogy,
    "1-12": _calc_double_log,
    "1-13": _calc_frequency_curve,
    "4-01": flood_mainline.calculate_wsl_01,
    "4-02": _calc_manning,
    "4-03": _calc_hec_standard_step,
    "3-15": river_special.calculate_str_01,
    "3-16": river_special.calculate_str_02,
    "3-13": river_special.calculate_str_03,
    "3-17": river_special.calculate_str_04,
    "3-12": river_special.calculate_str_05,
    "3-14": river_special.calculate_str_06,
    "3-23": river_special.calculate_str_07,
    "3-24": river_special.calculate_str_08,
    "3-25": river_special.calculate_str_09,
    "4-04": river_special.calculate_str_10,
    "4-05": river_special.calculate_str_11,
    "2-01": book_program_adapters.calculate_irr_01,
    "2-02": book_program_adapters.calculate_irr_02,
    "2-03": book_program_adapters.calculate_irr_03,
    "2-04": book_program_adapters.calculate_irr_04,
    "2-05": book_program_adapters.calculate_irr_05,
    "2-06": book_program_adapters.calculate_irr_06,
    "2-07": book_program_adapters.calculate_irr_07,
    "2-08": book_program_adapters.calculate_irr_08,
    "2-09": book_program_adapters.calculate_irr_09,
    "3-01": book_program_adapters.calculate_chn_01,
    "3-02": book_program_adapters.calculate_chn_02,
    "3-03": book_program_adapters.calculate_chn_03,
    "3-04": book_program_adapters.calculate_chn_04,
    "3-05": book_program_adapters.calculate_chn_05,
    "3-06": book_program_adapters.calculate_chn_06,
    "3-07": book_program_adapters.calculate_chn_07,
    "3-08": book_program_adapters.calculate_chn_08,
    "3-18": book_program_adapters.calculate_chn_18,
    "3-19": book_program_adapters.calculate_chn_19,
    "3-20": book_program_adapters.calculate_chn_20,
    "3-21": book_program_adapters.calculate_chn_21,
    "3-22": book_program_adapters.calculate_chn_22,
    "3-09": book_program_adapters.calculate_riv_01,
    "3-10": book_program_adapters.calculate_riv_02,
    "3-11": book_program_adapters.calculate_riv_03,
    "5-01": book_program_adapters.calculate_rsr_01,
    "5-02": book_program_adapters.calculate_rsr_02,
    "5-03": book_program_adapters.calculate_rsr_03,
    "5-04": book_program_adapters.calculate_rsr_04,
    "5-05": book_program_adapters.calculate_rsr_05,
    "5-06": book_program_adapters.calculate_rsr_06,
    "5-07": book_program_adapters.calculate_rsr_07,
    "5-08": book_program_adapters.calculate_rsr_08,
    "5-09": book_program_adapters.calculate_rsr_09,
    "6-01": book_program_adapters.calculate_hyd_01,
    "6-02": book_program_adapters.calculate_hyd_02,
    "6-03": book_program_adapters.calculate_hyd_03,
    "6-04": book_program_adapters.calculate_hyd_04,
    "6-05": book_program_adapters.calculate_hyd_05,
    "6-06": book_program_adapters.calculate_hyd_06,
    "6-07": book_program_adapters.calculate_hyd_07,
    "6-08": book_program_adapters.calculate_hyd_08,
    "6-09": book_program_adapters.calculate_hyd_09,
    "6-10": book_program_adapters.calculate_hyd_10,
    "6-11": book_program_adapters.calculate_hyd_11,
    "7-01": book_program_adapters.calculate_hst_01,
    "7-02": book_program_adapters.calculate_hst_02,
    "7-03": book_program_adapters.calculate_hst_03,
    "7-04": book_program_adapters.calculate_hst_04,
    "7-05": book_program_adapters.calculate_hst_05,
    "7-06": book_program_adapters.calculate_hst_06,
    "7-07": book_program_adapters.calculate_hst_07,
    "7-08": book_program_adapters.calculate_hst_08,
    "7-09": book_program_adapters.calculate_hst_09,
    "7-10": book_program_adapters.calculate_hst_10,
    "7-11": book_program_adapters.calculate_hst_11,
    "7-12": book_program_adapters.calculate_hst_12,
}

