"""SDK 自测：清单校验 / 打包往返 / 装载策略 / 本地测试台（无需 worker）。

有设备的用例标 `@pytest.mark.device`（无 sophon.sail 时自动跳过）：

    uv run pytest              # 静态用例
    uv run pytest -m device    # 设备上跑推理
"""
from __future__ import annotations

import json
import subprocess
import sys
import tempfile
from pathlib import Path

import pytest

from algpkg_sdk import (AlgoBuilder, AlgoError, AlgoPackage, AlgoPackageError, Manifest,
                        ManifestError, install_sdk_alias, sha256_file)
from algpkg_sdk import testkit
from algpkg_sdk.cli import main as cli_main

EXAMPLES = Path(__file__).resolve().parents[1] / "examples"


def _manifest(**overrides) -> dict:
    payload = {
        "manifest_format": 1, "id": "t.demo", "name": "测试包", "version": "1.0.0",
        "api_version": 1, "kind": "detector", "entry": "Algorithm",
        "code_package": "alg_t_demo",
        "models": [{"name": "default", "file": "models/t.bmodel", "precision": "fp16",
                    "resolution": [640, 640], "batch": 1, "default": True}],
        "params_schema": [{"key": "conf", "type": "float", "default": 0.25,
                           "min": 0.01, "max": 0.99, "label": "置信度"}],
        "defaults": {"conf": 0.25},
    }
    payload.update(overrides)
    return payload


# ---------------------------------------------------------------------- 清单

def test_manifest_required_fields() -> None:
    Manifest.parse(_manifest())
    for missing in ("id", "version", "models"):
        bad = _manifest()
        bad.pop(missing)
        with pytest.raises(ManifestError):
            Manifest.parse(bad)


def test_manifest_rejects_bad_values() -> None:
    with pytest.raises(ManifestError):                       # id 形态
        Manifest.parse(_manifest(id="Bad Id!"))
    with pytest.raises(ManifestError):                       # 模型路径越界
        Manifest.parse(_manifest(models=[{"name": "d", "file": "../x.bmodel"}]))
    with pytest.raises(ManifestError):                       # api_version 过高
        Manifest.parse(_manifest(api_version=99))
    with pytest.raises(ManifestError):                       # params_schema 重复键
        Manifest.parse(_manifest(params_schema=[
            {"key": "conf", "type": "float"}, {"key": "conf", "type": "float"}]))


def test_manifest_default_model_pick() -> None:
    manifest = Manifest.parse(_manifest(models=[
        {"name": "fp16", "file": "models/a.bmodel"},
        {"name": "int8", "file": "models/b.bmodel", "default": True}]))
    assert manifest.default_model().name == "int8"
    assert manifest.model("fp16").file == "models/a.bmodel"
    with pytest.raises(ManifestError):
        manifest.model("nope")


def test_manifest_check_params_semantics() -> None:
    manifest = Manifest.parse(_manifest())
    merged = manifest.check_params({"conf": 0.5})
    assert merged["conf"] == 0.5
    assert manifest.check_params({})["conf"] == 0.25            # 补默认
    with pytest.raises(ManifestError):                          # 越界拒绝（不静默截断）
        manifest.check_params({"conf": 2.0})
    with pytest.raises(ManifestError):                          # 类型不符
        manifest.check_params({"conf": "abc"})


def test_manifest_legacy_and_entry_helpers() -> None:
    manifest = Manifest.parse(_manifest(entry="sub:Alg", legacy={"algorithm": "yolo"}))
    assert manifest.entry_module == "alg_t_demo.sub"
    assert manifest.entry_class == "Alg"
    assert manifest.legacy["algorithm"] == "yolo"


# ---------------------------------------------------------------------- 打包往返

def _make_source(root: Path, *, model_bytes: bytes = b"FAKE-BMODEL" * 512) -> Path:
    src = root / "t.demo"
    (src / "alg_t_demo").mkdir(parents=True)
    (src / "alg_t_demo" / "__init__.py").write_text(
        "from video_algpkg_api import Algorithm as Base, AlgoResult\n"
        "class Algorithm(Base):\n"
        "    def load(self):\n"
        "        self._data = self.ctx.model_bytes()\n"
        "        self.ctx.logger.info('loaded %s bytes', len(self._data))\n"
        "    def infer(self, frame):\n"
        "        return AlgoResult(boxes=None, value=len(self._data))\n",
        encoding="utf-8",
    )
    (src / "models").mkdir()
    (src / "models" / "t.bmodel").write_bytes(model_bytes)
    (src / "assets").mkdir()
    (src / "assets" / "labels.txt").write_text("person\ncar\n", encoding="utf-8")
    (src / "manifest.json").write_text(json.dumps(_manifest()), encoding="utf-8")
    return src


def test_build_roundtrip_and_bytes_fidelity() -> None:
    install_sdk_alias()
    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
        src = _make_source(root)
        out = root / "t.demo-1.0.0.mbank"
        summary = AlgoBuilder(src, verbose=False).build(out)

        assert out.is_file() and summary["algo_id"] == "t.demo"
        assert summary["sha256"] == sha256_file(out)

        pkg = AlgoPackage(out)
        pkg.open()
        manifest = pkg.require_manifest()
        assert manifest.version == "1.0.0"
        data, meta = pkg.read_model("default")
        assert data == (src / "models" / "t.bmodel").read_bytes()      # 包内模型与源文件逐字节相同
        assert meta["precision"] == "fp16"
        assert pkg.read_asset("assets/labels.txt").decode().splitlines() == ["person", "car"]
        files = {item["path"] for item in pkg.list_files()}
        assert {"manifest.json", "assets/labels.txt", "models/t.bmodel"} <= files
        pkg.close()


def test_build_creates_version_snapshot_and_signature() -> None:
    from sqlite_vfs.core import SQLiteVFS

    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
        src = _make_source(root)
        out = root / "signed.mbank"
        AlgoBuilder(src, verbose=False).build(out, sign_key="KEY-1", sign_key_id="ci")
        vfs = SQLiteVFS(str(out), readonly=True, check_same_thread=False)
        try:
            assert [s["name"] for s in vfs.list_snapshots()] == ["1.0.0"]
            report = vfs.verify_signature("KEY-1")
            assert report["ok"] and report["key_id"] == "ci"
            assert not vfs.verify_signature("KEY-WRONG")["ok"]
        finally:
            vfs.close()


def test_build_reports_missing_model_and_code() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
        src = _make_source(root)
        (src / "models" / "t.bmodel").unlink()
        with pytest.raises(Exception) as exc:
            AlgoBuilder(src, verbose=False).build(root / "x.mbank")
        assert "模型" in str(exc.value)

        src2 = _make_source(root / "other")
        import shutil
        shutil.rmtree(src2 / "alg_t_demo")
        with pytest.raises(Exception) as exc2:
            AlgoBuilder(src2, verbose=False).build(root / "y.mbank")
        assert "代码目录" in str(exc2.value)


def test_encrypt_removes_plain_source() -> None:
    from sqlite_vfs.core import SQLiteVFS

    key = "uA9mYq0T2wGZ0dX1vJ8kL3pR5sN7bC4eF6hK2mP0qT8="
    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
        src = _make_source(root)
        out = root / "enc.mbank"
        AlgoBuilder(src, key=key, verbose=False).build(out)
        vfs = SQLiteVFS(str(out), readonly=True, check_same_thread=False)
        try:
            files = [e["path"] for e in vfs.all_entries() if not e["is_directory"]]
        finally:
            vfs.close()
        assert not any(f.endswith(".py") for f in files)         # 明文源码不得留在包里
        # 不带 key 打开结构没问题；装载（import）会因缺密钥失败
        pkg = AlgoPackage(out)
        pkg.open()
        with pytest.raises(AlgoPackageError) as exc:
            pkg.mount()
        assert "密钥" in str(exc.value)
        pkg.close()


def test_bad_file_rejected() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        bad = Path(tmp) / "bad.mbank"
        bad.write_bytes(b"not a database at all")
        with pytest.raises(AlgoPackageError):
            AlgoPackage(bad).open()


# ---------------------------------------------------------------------- 多模型包

def _make_multi_model_source(root: Path) -> Path:
    """一个包带两份模型（检测 + 二次判定）：代码按名字取第二份（多模型包的标准写法）。"""
    src = root / "t.multi"
    (src / "alg_t_multi").mkdir(parents=True)
    (src / "alg_t_multi" / "__init__.py").write_text(
        "from video_algpkg_api import Algorithm as Base, AlgoResult\n"
        "class Algorithm(Base):\n"
        "    def load(self):\n"
        "        self._det = self.ctx.model_bytes()              # 选中模型（默认检测器）\n"
        "        self._cls = self.ctx.model_bytes('fp_filter')   # 按名字取第二份\n"
        "    def infer(self, frame):\n"
        "        return AlgoResult(boxes=None, value=len(self._det) + len(self._cls))\n",
        encoding="utf-8",
    )
    (src / "models").mkdir()
    (src / "models" / "det.bmodel").write_bytes(b"DET" * 256)
    (src / "models" / "fp.bmodel").write_bytes(b"FPFILTER" * 128)
    (src / "manifest.json").write_text(json.dumps(_manifest(
        id="t.multi", name="多模型包", code_package="alg_t_multi",
        models=[{"name": "detector", "file": "models/det.bmodel", "default": True},
                {"name": "fp_filter", "file": "models/fp.bmodel", "precision": "int8"}])),
        encoding="utf-8")
    return src


def test_multi_model_package_context_access() -> None:
    """`ctx.model_bytes(name)` 按 manifest.models[] 取包内任意模型；不给名 = 选中模型（旧写法兼容）。"""
    install_sdk_alias()
    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
        src = _make_multi_model_source(root)
        out = root / "multi.mbank"
        AlgoBuilder(src, verbose=False).build(out)

        pkg = AlgoPackage(out)
        pkg.open()
        pkg.mount()
        try:
            instance = pkg.new_instance("detector", {})
            ctx = instance.ctx
            det = (src / "models" / "det.bmodel").read_bytes()
            fp = (src / "models" / "fp.bmodel").read_bytes()
            assert ctx.model_name == "detector"
            assert ctx.model_bytes() == det                           # 选中模型（向后兼容）
            assert ctx.model_bytes("fp_filter") == fp                 # 按 name 取第二份
            assert ctx.model_bytes("models/fp.bmodel") == fp          # 按 file 路径也认
            assert [m["name"] for m in ctx.models] == ["detector", "fp_filter"]
            assert ctx.model_spec("fp_filter")["precision"] == "int8"
            assert ctx.model_spec()["name"] == "detector"
            with pytest.raises(AlgoError) as exc:
                ctx.model_bytes("nope")
            assert "fp_filter" in str(exc.value)                      # 报错带可选清单
            assert instance.infer(None).value == len(det) + len(fp)   # 包内代码真读到两份不同字节
        finally:
            pkg.close()


# ---------------------------------------------------------------------- 契约名（video_algpkg_api）

def test_contract_module_importable_without_prior_registration() -> None:
    """`video_algpkg_api` 是随 SDK 分发的**真实模块**，不只是运行时注册的别名。

    新解释器里**先** import 它也要成功（本地开发/IDE/类型检查的常见姿势），
    且与 `algpkg_sdk.api` 是同一批对象（重导出，非副本）——否则 worker 侧
    `isinstance(instance, Algorithm)` 会出现两个基类（"本地过、设备挂"）。
    """
    script = (
        "import video_algpkg_api as contract\n"
        "assert contract.__file__.endswith('video_algpkg_api/__init__.py'), contract.__file__\n"
        "import algpkg_sdk.api as api\n"
        "for name in contract.__all__:\n"
        "    assert getattr(contract, name) is getattr(api, name), name\n"
        "print('contract-ok')\n"
    )
    with tempfile.TemporaryDirectory() as tmp:
        proc = subprocess.run([sys.executable, "-c", script], capture_output=True,
                              text=True, timeout=30, cwd=tmp)
    assert proc.returncode == 0, proc.stderr
    assert "contract-ok" in proc.stdout


def test_contract_alias_still_registered_to_api_module() -> None:
    """worker 侧注册语义不变：导入 `algpkg_sdk` 后，`video_algpkg_api` 解析到 api 模块本身。

    （`install_sdk_alias` 用 setdefault：设备上 worker 先注册，真实模块被惰性遮蔽，
    两条路径拿到的是同一批对象。）
    """
    import video_algpkg_api as resolved

    from algpkg_sdk import api

    assert resolved is api


# ---------------------------------------------------------------------- CLI

def test_cli_init_build_inspect_check(tmp_path: Path) -> None:
    src = tmp_path / "cli.demo"
    assert cli_main(["init", str(src), "--id", "cli.demo", "--name", "CLI 示例"]) == 0
    assert (src / "manifest.json").is_file()
    code_pkg = next((src.glob("alg_cli_demo")), None)
    assert code_pkg is not None and (code_pkg / "__init__.py").is_file()

    model = tmp_path / "fake.bmodel"
    model.write_bytes(b"FAKE" * 1024)
    out = tmp_path / "cli.demo-1.0.0.mbank"
    assert cli_main(["build", str(src), "--out", str(out), "--model", f"default={model}",
                     "--quiet"]) == 0
    assert cli_main(["inspect", str(out), "--json"]) == 0
    assert cli_main(["verify", str(out), "--json"]) == 0
    assert cli_main(["list", str(tmp_path)]) == 0


def test_cli_check_reports_static_results(tmp_path: Path) -> None:
    """check 在无设备时也要给出"清单/结构 OK、试加载跳过"的诚实结论。"""
    src = tmp_path / "chk.demo"
    cli_main(["init", str(src), "--id", "chk.demo", "--name", "检查示例"])
    model = tmp_path / "fake.bmodel"
    model.write_bytes(b"FAKE" * 1024)
    out = tmp_path / "chk.mbank"
    cli_main(["build", str(src), "--out", str(out), "--model", f"default={model}", "--quiet"])
    report = testkit.check(out)
    names = {item["name"]: item for item in report["checks"]}
    assert names["manifest 合法"]["ok"]
    assert names["模型文件在包内"]["ok"]
    if not testkit.has_device():
        assert not names["代码挂载（试加载）"]["ok"] and "跳过" in names["代码挂载（试加载）"]["detail"]


def test_module_entry_point_help() -> None:
    proc = subprocess.run([sys.executable, "-m", "algpkg_sdk.cli", "--help"],
                          capture_output=True, text=True, timeout=30)
    assert proc.returncode == 0
    assert "check" in proc.stdout and "run" in proc.stdout and "compare" in proc.stdout


# ---------------------------------------------------------------------- 示例包（真实结构）

def test_example_person_count_manifest() -> None:
    example = EXAMPLES / "person-count"
    manifest = Manifest.parse(json.loads((example / "manifest.json").read_text()))
    assert manifest.id == "demo.person-count" and manifest.kind == "counter"
    assert manifest.output["type"] == "counter"
    assert (example / manifest.code_package).is_dir()
    # 模型文件由打包时 --model 指定（示例目录不带 bmodel），这里只校验声明形态
    assert manifest.default_model().file.startswith("models/")


# ---------------------------------------------------------------------- 设备用例

@pytest.mark.device
def test_check_deep_on_device() -> None:
    """设备：入口类可实例化（模型真加载）。

    默认用**自包含**示例包（examples/person-count 打出来的包）；内置包（依赖 worker 内部模块）
    属"特权包"，需额外传 `extra_path=[worker 工程目录]`。
    """
    import os

    pkg_path = os.environ.get("ALGPKG_DEVICE_PACKAGE")
    if not pkg_path or not Path(pkg_path).is_file():
        pytest.skip("未提供 ALGPKG_DEVICE_PACKAGE（指向含真实模型的 .mbank）")
    extra = [p for p in (os.environ.get("ALGPKG_EXTRA_PATH") or "").split(":") if p]
    report = testkit.check(pkg_path, deep=True, extra_path=extra)
    assert report["ok"], report["checks"]


@pytest.mark.device
def test_run_on_device() -> None:
    """设备：跑一段视频，拿到逐帧检出与单帧耗时。"""
    import os

    pkg_path = os.environ.get("ALGPKG_DEVICE_PACKAGE")
    clip = os.environ.get("ALGPKG_DEVICE_CLIP")
    if not pkg_path or not clip:
        pytest.skip("未提供 ALGPKG_DEVICE_PACKAGE / ALGPKG_DEVICE_CLIP")
    extra = [p for p in (os.environ.get("ALGPKG_EXTRA_PATH") or "").split(":") if p]
    report = testkit.run(pkg_path, source=clip, frames=3, warmup=1, extra_path=extra)
    summary = report.summary()
    assert summary["frames"] >= 1
    assert summary["infer_ms"]["avg"] > 0