MB-X Bilibili Pipeline
6 days ago 4a782f28b96ce51579e28cdbf9ecc223ff705ee8
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
#!/usr/bin/env python3
"""Validate the mechanically bound source set for the generic article/image collector."""
 
from __future__ import annotations
 
import hashlib
import json
import os
import stat
import sys
from pathlib import Path
from typing import Any
 
 
PROJECT_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_MANIFEST = Path(__file__).with_name("bili_article_image_source_manifest.json")
SHA256 = __import__("re").compile(r"[0-9A-F]{64}")
 
 
def _reparse(path: Path) -> bool:
    info = path.lstat()
    return stat.S_ISLNK(info.st_mode) or bool(getattr(info, "st_file_attributes", 0) & 0x400)
 
 
def _strict_json(path: Path) -> tuple[Any, bytes]:
    raw = path.read_bytes()
    if raw.startswith(b"\xef\xbb\xbf") or b"\r" in raw or not raw.endswith(b"\n") or raw.endswith(b"\n\n"):
        raise ValueError("manifest encoding differs")
    return json.loads(raw[:-1].decode("utf-8", errors="strict")), raw
 
 
def validate(manifest_path: Path = DEFAULT_MANIFEST) -> dict[str, Any]:
    manifest_path = Path(os.path.abspath(manifest_path))
    value, raw = _strict_json(manifest_path)
    if not isinstance(value, dict) or set(value) != {"schema", "task_id", "scope", "root", "files", "tree_sha256"}:
        raise ValueError("manifest schema differs")
    if value["schema"] != 1 or value["task_id"] != "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001" or value["scope"] != "generic-bilibili-article-image-collector-source":
        raise ValueError("manifest identity differs")
    root = (PROJECT_ROOT / value["root"]).resolve(strict=True)
    if root != PROJECT_ROOT / "dev" / "project-dev" or _reparse(root):
        raise ValueError("manifest root differs")
    files = value["files"]
    if not isinstance(files, list) or not files:
        raise ValueError("manifest files differ")
    seen: set[str] = set()
    actual: list[dict[str, Any]] = []
    for entry in files:
        if not isinstance(entry, dict) or set(entry) != {"path", "bytes", "sha256"}:
            raise ValueError("manifest entry schema differs")
        relative = entry["path"]
        if not isinstance(relative, str) or not relative or "\\" in relative or relative.startswith("/") or ".." in Path(relative).parts or relative in seen:
            raise ValueError("manifest entry path differs")
        seen.add(relative)
        candidate = (root / Path(*relative.split("/"))).resolve(strict=True)
        if candidate.parent == root and candidate == manifest_path:
            raise ValueError("manifest cannot self-bind")
        if not candidate.is_file() or _reparse(candidate):
            raise ValueError("source file is missing or unsafe")
        payload = candidate.read_bytes()
        fact = {"path": relative, "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest().upper()}
        if entry != fact or SHA256.fullmatch(str(entry["sha256"])) is None:
            raise ValueError("source file identity differs")
        actual.append(fact)
    material = bytearray()
    for entry in sorted(actual, key=lambda item: item["path"]):
        material.extend(entry["path"].encode("utf-8"))
        material.extend(b"\0")
        material.extend(str(entry["bytes"]).encode("ascii"))
        material.extend(b"\0")
        material.extend(entry["sha256"].encode("ascii"))
        material.extend(b"\n")
    tree = hashlib.sha256(material).hexdigest().upper()
    if value["tree_sha256"] != tree:
        raise ValueError("source tree identity differs")
    return {
        "schema_version": 1,
        "status": "SOURCE_VALID",
        "file_count": len(actual),
        "tree_sha256": tree,
        "manifest_bytes": len(raw),
        "manifest_sha256": hashlib.sha256(raw).hexdigest().upper(),
        "mutation_count": 0,
    }
 
 
def main() -> int:
    try:
        result = validate()
        code = 0
    except Exception:
        result = {"schema_version": 1, "status": "SOURCE_INVALID", "error_code": "E_SOURCE_IDENTITY", "mutation_count": 0}
        code = 3
    print(json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":")))
    return code
 
 
if __name__ == "__main__":
    raise SystemExit(main())