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()
|
