"""按内容推进生成与 wheel 构建；失败时不激活本轮中间结果。

缓存可随时删除。输入只来自源码、登记资源与构建配置，不扫描作品或运行数据。
"""

from __future__ import annotations

import argparse
import fcntl
import hashlib
import importlib.metadata
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any

根 = Path(__file__).resolve().parents[1]
缓存 = 根 / "构建缓存"
接口 = "docs/接口契约/生成/openapi.generated.json"
客户端 = "web/src/接口/生成/客户端.ts"
类型 = "web/src/接口/生成/类型.ts"


def 文件摘要(路径: Path) -> str:
    with 路径.open("rb") as 文件:
        return hashlib.file_digest(文件, "sha256").hexdigest()


def 路径摘要(目录: Path, 路径们: list[str]) -> dict[str, str]:
    结果 = {}
    for 相对 in sorted(set(路径们)):
        路径 = 目录 / 相对
        文件们 = sorted(路径.rglob("*")) if 路径.is_dir() else [路径]
        for 文件 in 文件们:
            if 文件.is_file() and "__pycache__" not in 文件.parts:
                结果[文件.relative_to(目录).as_posix()] = 文件摘要(文件)
    return 结果


def 摘要(值: Any) -> str:
    return hashlib.sha256(json.dumps(值, sort_keys=True).encode()).hexdigest()


def 原子写(路径: Path, 值: Any) -> None:
    临时 = 路径.with_suffix(".tmp")
    临时.write_text(json.dumps(值, ensure_ascii=False, indent=2) + "\n")
    临时.replace(路径)


def 生产前端输入() -> list[str]:
    """当前 Vite HTML 入口及本地导入决定生产输入，tsc 测试检查由前端检查承担。

    src 入口目录整体保守纳入，目录外的本地导入递归跟随。自定义解析、非字面量
    动态导入或 glob 无法静态封闭时退回全 web 输入，不能漏算后悄悄缓存。
    """
    web = 根 / "web"
    所有 = {
        p
        for p in web.rglob("*")
        if p.is_file()
        and not {"node_modules", "dist", "test-results", "playwright-report"}.intersection(
            p.relative_to(web).parts
        )
        and not p.name.endswith(".tsbuildinfo")
    }

    def 保守输入() -> list[str]:
        return sorted(p.relative_to(根).as_posix() for p in 所有)

    配置 = web / "vite.config.ts"
    配置文本 = 配置.read_text()
    # 这些配置可以改变入口/解析/复制边界；未知配置仍走保守闭包。
    if re.search(r"\b(root|publicDir|rollupOptions|resolve|assetsInclude|lib)\s*:", 配置文本):
        return 保守输入()
    配置导入 = re.findall(r"(?:from\s*|import\s*)[\"']([^\"']+)[\"']", 配置文本)
    if any(v not in {"vite", "@vitejs/plugin-react"} for v in 配置导入):
        return 保守输入()
    待查 = {
        web / n
        for n in ("index.html", "vite.config.ts", "tsconfig.json", "package.json", "pnpm-lock.yaml")
    }
    html = (web / "index.html").read_text()
    for 入口 in re.findall(r"(?:src|href)=[\"']([^\"']+)[\"']", html):
        if ":" in 入口 or 入口.startswith("#"):
            continue
        文件 = web / 入口.lstrip("/")
        if not 文件.is_file():
            return 保守输入()
        # 入口所属源码目录含声明文件/隐式CSS依赖，整体保守纳入。
        入口目录 = web / 文件.relative_to(web).parts[0]
        待查.update(p for p in 所有 if p.is_relative_to(入口目录))
        待查.add(文件)
    待查.update(p for p in 所有 if p.is_relative_to(web / "public") or p.name.startswith(".env"))
    已查: set[Path] = set()
    导入式 = (
        r"(?:from\s*|import\s*(?:\(\s*)?|require\s*\(\s*|@import\s*|url\(\s*)[\"']([^\"']+)[\"']"
    )
    后缀 = (
        "",
        ".ts",
        ".tsx",
        ".js",
        ".jsx",
        ".json",
        ".css",
        "/index.ts",
        "/index.tsx",
        "/index.js",
    )
    while 待查:
        文件 = 待查.pop()
        if 文件 in 已查 or not 文件.is_file():
            continue
        已查.add(文件)
        if 文件.suffix not in {".ts", ".tsx", ".js", ".jsx", ".css"}:
            continue
        文本 = 文件.read_text()
        if re.search(r"import\.meta\.glob|(?:import|require)\s*\(\s*(?![\s\"'])", 文本):
            return 保守输入()
        for 名 in re.findall(导入式, 文本):
            if not 名.startswith((".", "/")):
                continue
            路径 = web / 名.lstrip("/") if 名.startswith("/") else 文件.parent / 名
            路径 = 路径.resolve()
            if not 路径.is_relative_to(web):
                raise ValueError("生产前端导入越出 web；请先显式登记跨目录构建输入")
            候选 = [Path(str(路径).split("?", 1)[0] + 后) for 后 in 后缀]
            实际 = next((p for p in 候选 if p.is_file()), None)
            if 实际 is None:
                if 根 / 类型 in 候选 or 根 / 客户端 in 候选:
                    continue
                return 保守输入()
            if 实际 not in 已查:
                待查.add(实际)
    return sorted(p.relative_to(根).as_posix() for p in 已查)


def 源清单() -> list[str]:
    登记 = json.loads((根 / "工具/资源登记.json").read_text())
    列表 = ["pyproject.toml", "uv.lock", "README.md"]
    列表 += [
        "工具/生成接口.py",
        "工具/生成输入.json",
        "工具/构建资源包.py",
        "工具/效果依赖.py",
        "工具/资源登记.json",
        "工具/构建编排.py",
    ]
    列表 += [
        p.relative_to(根).as_posix()
        for p in (根 / "src/muse").rglob("*")
        if p.is_file()
        and not {"资源", "__pycache__"}.intersection(p.relative_to(根 / "src/muse").parts)
    ]
    列表 += 生产前端输入()
    列表 += [v["源路径"] for v in 登记["资源"]] + 登记.get("能力文件", [])
    列表 += [p.relative_to(根).as_posix() for p in (根 / 登记["迁移目录"]).glob("*.sql")]
    for 相对 in 列表:
        路径 = 根 / 相对
        if not 路径.resolve().is_relative_to(根) or not 路径.is_file():
            raise ValueError(f"构建输入必须是仓内文件：{相对}")
    return sorted(set(列表))


def 构建(轮子: bool) -> dict:
    清单 = 源清单()
    原始 = 路径摘要(根, 清单)
    try:
        旧缓存 = json.loads((缓存 / "状态.json").read_text())
        if not isinstance(旧缓存, dict):
            旧缓存 = {}
    except (OSError, ValueError):
        旧缓存 = {}
    版本 = {"python": sys.version, "编排": 文件摘要(Path(__file__))}
    for 名 in ("fastapi", "pydantic", "pyyaml"):
        版本[名] = importlib.metadata.version(名)
    for 名 in ("pnpm", "node", "uv"):
        路径 = shutil.which(名)
        if 路径 is None:
            raise RuntimeError(f"缺少已安装工具：{名}")
        版本[名] = 文件摘要(Path(路径).resolve())
    for 名 in ("typescript", "vite", "openapi-typescript"):
        文件 = 根 / "web/node_modules" / 名 / "package.json"
        版本[名] = json.loads(文件.read_text())["version"]
    新缓存: dict[str, Any] = {}
    事件: list[dict] = []
    待激活: list[str] = []
    with tempfile.TemporaryDirectory(prefix="构建-", dir=缓存) as 临时:
        暂存 = Path(临时)

        # 同批全部步骤在暂存树运行；任何失败均不触及已有发布输出。
        def 复制(相对: str) -> None:
            源 = 根 / 相对
            目标 = 暂存 / 相对
            if not 源.exists():
                return
            目标.parent.mkdir(parents=True, exist_ok=True)
            if 源.is_dir():
                if 源.is_symlink() or any(p.is_symlink() for p in 源.rglob("*")):
                    raise ValueError(f"构建产物目录不得含软链：{相对}")
                shutil.copytree(源, 目标, dirs_exist_ok=True)
            else:
                shutil.copyfile(源, 目标)

        for 相对 in 清单 + [接口, "web/dist", "src/muse/资源"]:
            复制(相对)
        (暂存 / "web/node_modules").symlink_to((根 / "web/node_modules").resolve())
        环境 = {
            **os.environ,
            "PYTHONPATH": str(暂存 / "src"),
            "PYTHONDONTWRITEBYTECODE": "1",
            "UV_OFFLINE": "true",
            "UV_NO_SYNC": "true",
        }
        环境["SOURCE_DATE_EPOCH"] = "1767225600"

        def 执行(名称: str, 输入: list[str], 输出: list[str], 命令: list[list[str]]) -> None:
            指纹 = 摘要({"输入": 路径摘要(暂存, 输入), "版本": 版本})
            旧 = 旧缓存.get(名称, {})
            if not isinstance(旧, dict):
                旧 = {}
            当前 = 路径摘要(根, 输出)
            命中 = (
                bool(当前)
                and all((根 / p).exists() for p in 输出)
                and 旧.get("输入") == 指纹
                and 旧.get("输出") == 当前
            )
            if not 命中:
                for 参数 in 命令:
                    subprocess.run(参数, cwd=暂存, env=环境, check=True)
                if not all((暂存 / p).exists() for p in 输出):
                    raise RuntimeError(f"{名称} 未产生约定输出")
                待激活.extend(输出)
                当前 = 路径摘要(暂存, 输出)
            新缓存[名称] = {"输入": 指纹, "输出": 当前}
            事件.append({"步骤": 名称, "执行": not 命中})

        py = [p for p in 清单 if p.startswith("src/") or p in {"pyproject.toml", "uv.lock"}]
        执行(
            "OpenAPI",
            py + ["工具/生成接口.py", "工具/生成输入.json"],
            [接口, 客户端],
            [[sys.executable, "工具/生成接口.py"]],
        )
        执行(
            "TypeScript",
            [接口, "web/package.json", "web/pnpm-lock.yaml"],
            [类型],
            [
                [
                    "pnpm",
                    "--dir",
                    "web",
                    "exec",
                    "openapi-typescript",
                    "../" + 接口,
                    "-o",
                    "src/接口/生成/类型.ts",
                ]
            ],
        )
        web = [p for p in 清单 if p.startswith("web/")]
        # 发布只检查生产源码；完整 src/tests 类型门仍由 make 前端检查承担。
        (暂存 / "web/tsconfig.生产.json").write_text(
            json.dumps({"extends": "./tsconfig.json", "include": ["src"]}) + "\n"
        )
        执行(
            "web",
            web + [类型, 客户端],
            ["web/dist"],
            [
                [
                    "pnpm",
                    "--dir",
                    "web",
                    "exec",
                    "tsc",
                    "--project",
                    "tsconfig.生产.json",
                    "--noEmit",
                ],
                ["pnpm", "--dir", "web", "exec", "vite", "build"],
            ],
        )
        资源 = [p for p in 清单 if not p.startswith("web/")]
        执行(
            "资源", 资源 + ["web/dist"], ["src/muse/资源"], [[sys.executable, "工具/构建资源包.py"]]
        )
        wheel = None
        if 轮子:
            import tomllib

            配置 = tomllib.loads((暂存 / "pyproject.toml").read_text())
            包名 = 配置["project"]["name"].replace("-", "_")
            wheel = f"dist/{包名}-{配置['project']['version']}-py3-none-any.whl"
            执行(
                "wheel",
                py + ["src/muse/资源", "README.md"],
                [wheel],
                [["uv", "build", "--offline", "--wheel", "--out-dir", "dist"]],
            )
        elif "wheel" in 旧缓存:
            新缓存["wheel"] = 旧缓存["wheel"]
        # 构建期间输入有变化则拒绝发布，避免不同时间点的代码和清单混装。
        if 源清单() != 清单 or 路径摘要(根, 清单) != 原始:
            raise RuntimeError("构建期间源码发生变化，拒绝发布本轮产物")
        备份 = 暂存 / "原输出"
        已移动: list[str] = []
        try:
            for 相对 in dict.fromkeys(待激活):
                目标, 保存 = 根 / 相对, 备份 / 相对
                目标.parent.mkdir(parents=True, exist_ok=True)
                保存.parent.mkdir(parents=True, exist_ok=True)
                if 目标.exists():
                    目标.rename(保存)
                已移动.append(相对)
                (暂存 / 相对).rename(目标)
            原子写(缓存 / "状态.json", 新缓存)
        except BaseException:
            for 相对 in reversed(已移动):
                目标, 保存 = 根 / 相对, 备份 / 相对
                if 目标.is_dir():
                    shutil.rmtree(目标)
                elif 目标.exists():
                    目标.unlink()
                if 保存.exists():
                    保存.rename(目标)
            raise
    return {
        "步骤": 事件,
        "执行数": sum(e["执行"] for e in 事件),
        "wheel": str(根 / wheel) if wheel else None,
        "工具身份": 版本,
    }


def 执行构建(轮子: bool = False) -> dict:
    """共享聚合入口；进程锁覆盖构建和激活，返回本批实际 wheel 路径。"""
    缓存.mkdir(exist_ok=True)
    with (缓存 / "构建.lock").open("w") as 锁:
        fcntl.flock(锁, fcntl.LOCK_EX)
        return 构建(轮子)


def 主() -> int:
    解析器 = argparse.ArgumentParser(description=__doc__)
    解析器.add_argument("--wheel", action="store_true")
    解析器.add_argument("--报告", type=Path)
    参数 = 解析器.parse_args()
    # 生成会改写 src/muse/资源；验证在跑时改写会让本轮结论作废。
    from 验证锁 import 正在验证

    说明 = 正在验证()
    if 说明:
        print(f"拒绝生成：验证正在运行（{说明}）；等它结束再改源码与资源", file=sys.stderr)
        return 1
    报告 = 执行构建(参数.wheel)
    if 参数.报告:
        参数.报告.parent.mkdir(parents=True, exist_ok=True)
        原子写(参数.报告, 报告)
    print(json.dumps(报告, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(主())