from __future__ import annotations

import hashlib
import importlib.util
import json
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path
from unittest import mock


PROJECT_ROOT = Path(__file__).resolve().parents[3]
PROJECT_DEV = Path(__file__).resolve().parents[1]
TMP_ROOT = PROJECT_ROOT / "dev" / "tmp"
MODULE_PATH = PROJECT_DEV / "bili_article_image_collector.py"
SPEC = importlib.util.spec_from_file_location("bili_article_image_collector", MODULE_PATH)
assert SPEC and SPEC.loader
collector = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = collector
SPEC.loader.exec_module(collector)

HOST_PATH = PROJECT_DEV / "bili_article_image_native_host.py"
HOST_SPEC = importlib.util.spec_from_file_location("bili_article_image_native_host", HOST_PATH)
assert HOST_SPEC and HOST_SPEC.loader
native_host = importlib.util.module_from_spec(HOST_SPEC)
sys.modules[HOST_SPEC.name] = native_host
HOST_SPEC.loader.exec_module(native_host)

VALIDATOR_PATH = PROJECT_DEV / "bili_article_image_source_validator.py"
VALIDATOR_SPEC = importlib.util.spec_from_file_location("bili_article_image_source_validator", VALIDATOR_PATH)
assert VALIDATOR_SPEC and VALIDATOR_SPEC.loader
source_validator = importlib.util.module_from_spec(VALIDATOR_SPEC)
sys.modules[VALIDATOR_SPEC.name] = source_validator
VALIDATOR_SPEC.loader.exec_module(source_validator)


def canonical_json(value: object) -> bytes:
    return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")


def accepted_snapshot_sha256(items: list[dict[str, object]]) -> str:
    canonical_items = []
    for item in items:
        published = datetime.fromisoformat(str(item["published_at"]))
        canonical_items.append({
            "body_complete": True,
            "body_text": str(item["body_text"]).replace("\r\n", "\n").replace("\r", "\n").rstrip("\n"),
            "image_count": len(item["images"]),
            "item_type": item["item_type"],
            "published_at_epoch_ms": int(published.timestamp() * 1000),
            "source_url": item["source_url"],
            "stable_id": item["stable_id"],
            "title": str(item["title"]).strip(),
        })
    canonical_items.sort(key=lambda item: str(item["stable_id"]))
    payload = json.dumps({"items": canonical_items, "schema_version": 1}, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(payload).hexdigest().upper()


class GenericArticleImageCollectorTests(unittest.TestCase):
    def setUp(self) -> None:
        TMP_ROOT.mkdir(parents=True, exist_ok=True)
        self.temp = tempfile.TemporaryDirectory(dir=TMP_ROOT)
        self.root = Path(self.temp.name)
        self.intake = self.root / "intake"
        self.intake.mkdir()

    def tearDown(self) -> None:
        self.temp.cleanup()

    def write_json(self, path: Path, value: object) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(canonical_json(value))

    def config(self, uid: str, name: str, *, policy: str = "verify_or_append", output: Path | None = None) -> tuple[Path, Path]:
        output_root = output or (self.root / f"out-{uid}")
        path = self.root / f"config-{uid}.json"
        self.write_json(
            path,
            {
                "schema_version": 1,
                "creator": {"uid": uid, "name": name},
                "page": {
                    "dynamic_url": f"https://space.bilibili.com/{uid}/dynamic",
                    "profile_url": f"https://space.bilibili.com/{uid}",
                },
                "output": {"root": str(output_root), "intake_root": str(self.intake), "manifest_name": "manifest.jsonl"},
                "selection": {
                    "date_start": "2026-08-01T00:00:00+08:00",
                    "date_end": "2026-08-31T23:59:59+08:00",
                    "window_days": None,
                    "timezone": "Asia/Shanghai",
                    "include_types": ["article", "text", "image"],
                },
                "readiness": {"deadline_seconds": 30, "observation_interval_ms": 100, "stable_observations": 3},
                "rerun": {"policy": policy},
                "verification": {"summary_path": None},
                "limits": {"max_items": 20, "max_body_bytes": 1048576, "max_images_per_item": 8, "max_image_bytes": 1048576},
            },
        )
        return path, output_root

    def image(self, name: str, extension: str) -> dict[str, object]:
        if extension == ".png":
            payload = b"\x89PNG\r\n\x1a\n" + b"p" * 24
        elif extension == ".webp":
            payload = b"RIFF" + b"\x10\x00\x00\x00" + b"WEBP" + b"w" * 20
        else:
            payload = b"\xff\xd8\xff" + b"j" * 29
        path = self.intake / f"{name}{extension}"
        path.write_bytes(payload)
        return {"path": path.name, "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest().upper(), "extension": extension}

    @staticmethod
    def observations(items: list[dict[str, object]], *states: str, ready_sha: str | None = None) -> list[dict[str, object]]:
        result = []
        fingerprint = ready_sha or accepted_snapshot_sha256(items)
        for index, state in enumerate(states):
            result.append(
                {
                    "elapsed_ms": index * 100 + 1,
                    "state": state,
                    "reason": state,
                    "snapshot_sha256": fingerprint if state == "READY" else None,
                }
            )
        return result

    def capture(
        self,
        uid: str,
        name: str,
        items: list[dict[str, object]],
        states: tuple[str, ...] = ("METADATA_NOT_READY", "READY", "READY", "READY"),
        *,
        ready_sha: str | None = None,
    ) -> Path:
        path = self.root / f"capture-{uid}-{len(list(self.root.glob('capture-*.json')))}.json"
        self.write_json(
            path,
            {
                "schema_version": 1,
                "creator_uid": uid,
                "creator_name": name,
                "dynamic_url": f"https://space.bilibili.com/{uid}/dynamic",
                "profile_url": f"https://space.bilibili.com/{uid}",
                "observations": self.observations(items, *states, ready_sha=ready_sha),
                "items": items,
            },
        )
        return path

    @staticmethod
    def item(stable_id: str, item_type: str, title: str, body: str, images: list[dict[str, object]] | None = None) -> dict[str, object]:
        return {
            "stable_id": stable_id,
            "item_type": item_type,
            "title": title,
            "source_url": f"https://www.bilibili.com/opus/{stable_id}",
            "published_at": "2026-08-12T10:00:00+08:00",
            "body_text": body,
            "body_complete": True,
            "images": images or [],
        }

    def test_two_configured_creators_full_body_images_and_safe_rerun(self) -> None:
        config_a, output_a = self.config("10001", "创作者甲")
        long_body = "正文段落\n" * 300
        capture_a = self.capture(
            "10001",
            "创作者甲",
            [
                self.item("opusA1", "article", "完整文章", long_body, [self.image("a1", ".png")]),
                self.item("opusA2", "text", "文字动态", "完整文字动态"),
                self.item("opusA3", "image", "图片动态", "图片说明", [self.image("a3", ".jpg")]),
            ],
        )
        code, result = collector.run(["--config", str(config_a), "collect", "--capture", str(capture_a)])
        self.assertEqual(0, code, result)
        self.assertEqual((3, 5), (result["new_items"], result["artifact_count"]))
        self.assertIn(long_body.encode("utf-8"), next(output_a.glob("*opusA1.txt")).read_bytes())
        code, verified = collector.run(["--config", str(config_a), "verify"])
        self.assertEqual(0, code, verified)
        self.assertEqual((3, 1, 2, 2), (verified["item_count"], verified["article_count"], verified["text_image_dynamic_count"], verified["original_image_count"]))
        code, rerun = collector.run(["--config", str(config_a), "collect", "--capture", str(capture_a)])
        self.assertEqual(0, code, rerun)
        self.assertEqual(("NO_NEW_ITEMS", 0), (rerun["status"], rerun["mutation_count"]))

        config_b, output_b = self.config("20002", "Creator-B")
        capture_b = self.capture("20002", "Creator-B", [self.item("opusB1", "text", "Second creator", "Independent corpus")])
        code, result_b = collector.run(["--config", str(config_b), "collect", "--capture", str(capture_b)])
        self.assertEqual(0, code, result_b)
        self.assertEqual(1, len(list(output_b.glob("*.txt"))))
        self.assertNotEqual(output_a, output_b)

    def test_pending_intermission_requires_three_consecutive_ready_samples(self) -> None:
        config_path, _ = self.config("30003", "稳定性样例")
        item = self.item("stable1", "text", "稳定", "正文")
        valid = self.capture("30003", "稳定性样例", [item], ("OWNER_PENDING", "READY", "READY", "DIMENSIONS_PENDING", "READY", "READY", "READY"))
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(valid)])
        self.assertEqual(0, code, result)
        self.assertEqual(7, result["readiness_attempts"])
        invalid = self.capture("30003", "稳定性样例", [item], ("READY", "READY", "METADATA_NOT_READY", "READY", "READY"))
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(invalid)])
        self.assertEqual((3, "E_READINESS_TIMEOUT", 0), (code, result["error_code"], result["mutation_count"]))

    def test_ready_tail_binds_exact_canonical_items_and_rejects_trailing_evidence(self) -> None:
        config_path, output_root = self.config("31003", "快照绑定样例")
        shared_vector = {
            "stable_id": "snap_1",
            "item_type": "article",
            "title": "Shared",
            "source_url": "https://www.bilibili.com/opus/snap_1",
            "published_at": datetime.fromisoformat("2026-08-12T10:00:00+08:00"),
            "body": "正文\nline\n".encode("utf-8"),
            "images": [{}],
        }
        self.assertEqual("4F0616496B08F8537A1C9F17B48D53BC2F93A26EB15AB5977C6E891ABA411454", collector._accepted_snapshot_sha256([shared_vector]))
        first = self.item("snapshot1", "article", "快照一", "BODY-A")
        second = self.item("snapshot2", "text", "快照二", "BODY-B")
        unrelated = hashlib.sha256(b"unrelated").hexdigest().upper()
        digest_mismatch = self.capture("31003", "快照绑定样例", [first], ready_sha=unrelated)
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(digest_mismatch)])
        self.assertEqual((3, "E_READINESS_DIGEST", 0), (code, result["error_code"], result["mutation_count"]))

        original_digest = accepted_snapshot_sha256([first])
        changed_body = dict(first)
        changed_body["body_text"] = "BODY-B-NOT-OBSERVED"
        body_drift = self.capture("31003", "快照绑定样例", [changed_body], ready_sha=original_digest)
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(body_drift)])
        self.assertEqual((3, "E_READINESS_DIGEST", 0), (code, result["error_code"], result["mutation_count"]))

        multi_not_shared = self.capture("31003", "快照绑定样例", [first, second], ready_sha=original_digest)
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(multi_not_shared)])
        self.assertEqual((3, "E_READINESS_DIGEST", 0), (code, result["error_code"], result["mutation_count"]))

        trailing_access = self.capture("31003", "快照绑定样例", [first], ("READY", "READY", "READY", "ACCESS_BLOCKED"))
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(trailing_access)])
        self.assertEqual((3, "E_ACCESS_CONTROL", 0), (code, result["error_code"], result["mutation_count"]))

        trailing_pending = self.capture("31003", "快照绑定样例", [first], ("READY", "READY", "READY", "METADATA_NOT_READY"))
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(trailing_pending)])
        self.assertEqual((3, "E_READINESS_TIMEOUT", 0), (code, result["error_code"], result["mutation_count"]))
        self.assertFalse(output_root.exists())

    def test_post_validation_image_drift_fails_before_pending_or_formal_mutation(self) -> None:
        config_path, output_root = self.config("32003", "图片冻结样例")
        image = self.image("drift", ".png")
        source_path = self.intake / str(image["path"])
        capture_path = self.capture("32003", "图片冻结样例", [self.item("drift1", "image", "图片漂移", "完整正文", [image])])
        original_validate = collector.validate_capture

        def validate_then_drift(config: collector.CollectorConfig, path: Path) -> dict[str, object]:
            value = original_validate(config, path)
            source_path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"q" * 24)
            return value

        with mock.patch.object(collector, "validate_capture", side_effect=validate_then_drift):
            code, result = collector.run(["--config", str(config_path), "collect", "--capture", str(capture_path)])
        self.assertEqual((3, "E_ARTIFACT_DRIFT", 0), (code, result["error_code"], result["mutation_count"]))
        self.assertFalse(output_root.exists())
        self.assertFalse((output_root / "manifest.jsonl").exists())
        self.assertEqual([], list(output_root.glob(".bili-article-image.pending.*.json")))
        self.assertEqual([], list(output_root.glob("*.txt")))
        self.assertEqual([], list(output_root.glob("*.png")))

        config_frozen, output_frozen = self.config("32004", "冻结载荷样例")
        frozen_image = self.image("frozen", ".png")
        frozen_source = self.intake / str(frozen_image["path"])
        expected_payload = frozen_source.read_bytes()
        frozen_capture = self.capture("32004", "冻结载荷样例", [self.item("frozen1", "image", "冻结载荷", "完整正文", [frozen_image])])
        original_create_new = collector._create_new
        mutated_after_freeze = False

        def create_then_mutate_intake(path: Path, payload: bytes) -> None:
            nonlocal mutated_after_freeze
            original_create_new(path, payload)
            if collector.OWNED_PENDING.fullmatch(path.name) and not mutated_after_freeze:
                frozen_source.write_bytes(b"\x89PNG\r\n\x1a\n" + b"r" * 24)
                mutated_after_freeze = True

        with mock.patch.object(collector, "_create_new", side_effect=create_then_mutate_intake):
            code, result = collector.run(["--config", str(config_frozen), "collect", "--capture", str(frozen_capture)])
        self.assertEqual((0, "CONTENT_SAVED", True), (code, result["status"], mutated_after_freeze))
        published = next(output_frozen.glob("*.png"))
        self.assertEqual(expected_payload, published.read_bytes())
        manifest_row = json.loads((output_frozen / "manifest.jsonl").read_text(encoding="utf-8"))
        self.assertEqual(
            (len(expected_payload), hashlib.sha256(expected_payload).hexdigest().upper()),
            (manifest_row["images"][0]["bytes"], manifest_row["images"][0]["sha256"]),
        )

    def test_identity_access_secret_partial_and_path_escape_fail_closed(self) -> None:
        config_path, output_root = self.config("40004", "安全样例")
        wrong = self.capture("40005", "安全样例", [self.item("bad1", "text", "bad", "body")])
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(wrong)])
        self.assertEqual((3, "E_CREATOR_IDENTITY"), (code, result["error_code"]))
        blocked = self.capture("40004", "安全样例", [self.item("bad2", "text", "bad", "body")], ("ACCESS_BLOCKED",))
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(blocked)])
        self.assertEqual((3, "E_ACCESS_CONTROL"), (code, result["error_code"]))
        partial = self.capture("40004", "安全样例", [self.item("bad3", "text", "bad", " ")])
        code, result = collector.run(["--config", str(config_path), "validate-capture", "--capture", str(partial)])
        self.assertEqual((3, "E_CONTENT_INCOMPLETE"), (code, result["error_code"]))
        config_value = json.loads(config_path.read_text(encoding="utf-8"))
        config_value["cookie"] = "synthetic-marker"
        self.write_json(config_path, config_value)
        code, result = collector.run(["--config", str(config_path), "verify"])
        self.assertEqual((3, "E_SECRET_FIELD"), (code, result["error_code"]))
        self.assertFalse(output_root.exists())
        self.config("40004", "安全样例", output=PROJECT_ROOT.parent / "escape")
        code, result = collector.run(["--config", str(self.root / "config-40004.json"), "verify"])
        self.assertEqual((3, "E_PATH_ESCAPE"), (code, result["error_code"]))

        config_path, _ = self.config("40004", "安全样例")
        outside_terminal = PROJECT_ROOT.parent / f"bili-article-image-outside-terminal-{os.getpid()}.json"
        self.assertFalse(outside_terminal.exists())
        code, result = collector.run(["--config", str(config_path), "verify", "--terminal", str(outside_terminal)])
        self.assertEqual((3, "E_PATH_ESCAPE", 0), (code, result["error_code"], result["mutation_count"]))
        self.assertFalse(outside_terminal.exists())

    def test_collision_precommit_recovery_and_reparse_are_mutation_zero(self) -> None:
        config_path, output_root = self.config("50005", "恢复样例")
        capture_path = self.capture("50005", "恢复样例", [self.item("recover1", "article", "恢复", "完整正文")])
        config = collector.load_config(config_path)
        empty = hashlib.sha256(b"").hexdigest().upper()
        with mock.patch.object(collector, "_manifest_snapshot", side_effect=[(0, empty), (1, "A" * 64), (0, empty)]):
            with self.assertRaises(collector.CollectorError) as raised:
                collector.collect(config, capture_path, None)
        self.assertEqual("E_PRECOMMIT_DRIFT", raised.exception.code)
        self.assertEqual([], list(output_root.glob("*.txt")))
        self.assertFalse((output_root / "manifest.jsonl").exists())
        self.assertEqual([], list(output_root.glob(".bili-article-image.pending.*.json")))

        capture = collector.validate_capture(config, capture_path)
        item = capture["items"][0]
        local = item["published_at"].astimezone(config.tz)
        stem = f"{local:%Y%m%d-%H%M%S}_{item['item_type']}_{collector._safe_component(item['title'], max_length=48)}_{item['stable_id']}"
        output_root.mkdir(parents=True, exist_ok=True)
        (output_root / f"{stem}.txt").write_bytes(b"collision")
        code, result = collector.run(["--config", str(config_path), "collect", "--capture", str(capture_path)])
        self.assertEqual((3, "E_TARGET_EXISTS", 0), (code, result["error_code"], result["mutation_count"]))

        probe = output_root / "reparse-probe"
        probe.mkdir()
        with mock.patch.object(collector, "_is_reparse", side_effect=lambda path: path == probe):
            with self.assertRaises(collector.CollectorError) as reparse:
                collector._safe_existing_chain(probe / "child")
        self.assertEqual("E_PATH_REPARSE", reparse.exception.code)

        pending = output_root / f".bili-article-image.pending.{'a' * 32}.json"
        pending.write_bytes(canonical_json({"schema_version": 1, "status": "PUBLISH_PENDING"}))
        before = pending.read_bytes()
        code, result = collector.run(["--config", str(config_path), "verify"])
        self.assertEqual((3, "E_RECOVERY_REQUIRED", 0), (code, result["error_code"], result["mutation_count"]))
        self.assertEqual(before, pending.read_bytes())

    def test_native_host_strict_boundary_and_real_corpus_readback(self) -> None:
        real_config = PROJECT_ROOT / "dev" / "tmp" / "bili-article-image-generic-real-validation-config-20260825.json"
        code, result = native_host.run_request(
            {"schema_version": 1, "action": "verify", "config_path": str(real_config), "capture_path": None, "terminal_path": None}
        )
        self.assertEqual(0, code, result)
        self.assertEqual((85, 22, 63, 16), (result["item_count"], result["article_count"], result["text_image_dynamic_count"], result["original_image_count"]))
        self.assertEqual("4CF0BB9936431C24278A1209C0E4CB1BD3645EF8E14BBC4B32F9B22629211EE3", result["manifest_sha256"])
        code, blocked = native_host.run_request(
            {"schema_version": 1, "action": "verify", "config_path": str(real_config), "capture_path": None, "terminal_path": None, "session_token": "synthetic"}
        )
        self.assertEqual((3, "E_HOST_SCHEMA", 0), (code, blocked["error_code"], blocked["mutation_count"]))

    def test_runtime_sources_have_no_current_creator_constant_or_secret_api(self) -> None:
        runtime_paths = [
            MODULE_PATH,
            PROJECT_DEV / "bili_article_image_capture.js",
            HOST_PATH,
            PROJECT_DEV / "bili_dynamic_collector.py",
            PROJECT_DEV / "bili_dynamic_refresh_extension" / "service_worker.js",
            PROJECT_DEV / "bili_dynamic_refresh_extension" / "page_extract.js",
            PROJECT_DEV / "bili_dynamic_refresh_native_host" / "constants.py",
            PROJECT_DEV / "bili_dynamic_refresh_native_host" / "protocol.py",
            PROJECT_DEV / "bili_article_image_collector.example.json",
        ]
        text = "\n".join(path.read_text(encoding="utf-8") for path in runtime_paths)
        for forbidden in ["1420210197", "青枫浦上Q", "document.cookie", "localStorage", "Profile", "--cookies", "Cookie:"]:
            self.assertNotIn(forbidden, text)

    def test_browser_capture_two_creator_and_readiness_contract(self) -> None:
        completed = subprocess.run(
            ["node", str(Path(__file__).with_name("test_bili_article_image_capture.mjs")), str(PROJECT_DEV / "bili_article_image_capture.js")],
            check=True,
            capture_output=True,
            text=True,
            timeout=20,
        )
        result = json.loads(completed.stdout)
        self.assertEqual({"status": "PASS", "creators": 2, "readiness_observations": 7, "network_requests": 0}, result)

    def test_exact_source_manifest(self) -> None:
        result = source_validator.validate()
        self.assertEqual(("SOURCE_VALID", 28, 0), (result["status"], result["file_count"], result["mutation_count"]))


if __name__ == "__main__":
    unittest.main()