from __future__ import annotations
|
|
import csv
|
from datetime import datetime, timezone
|
import hashlib
|
import io
|
import json
|
import os
|
from pathlib import Path
|
from typing import Any, Callable, Mapping, Sequence
|
|
from .archive import create_exclusive_bytes, sha256_file
|
from .models import ContractError, ErrorCode
|
|
|
MANIFEST_COLUMNS = (
|
"task_id", "requested_by", "review_owner", "source_url", "source_site", "title",
|
"publisher", "report_date", "downloaded_at", "http_status", "content_type", "file_name",
|
"relative_path", "bytes", "sha256", "download_status", "error_or_note",
|
"source_cache_path", "source_file_name", "android_package", "extension_added",
|
"pdf_magic_valid", "remote_sha256", "local_sha256", "openability", "page_count",
|
"encryption_status", "schema_version", "row_id", "handoff_id", "run_id", "item_id",
|
"slot_id", "query", "candidate_id", "report_identity", "analysts", "selection_reason",
|
"remote_bytes", "local_bytes", "quota_reservation_id", "quota_terminal_event_id",
|
"quota_artifact_event_id", "status", "stop_code", "reused_without_new_trigger",
|
"external_evidence_hash", "manifested_at_utc",
|
)
|
|
|
def canonical_json_bytes(value: Any) -> bytes:
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=False).encode("utf-8")
|
|
|
def validate_manifest_row(row: Mapping[str, Any]) -> dict[str, str]:
|
if tuple(row.keys()) != MANIFEST_COLUMNS:
|
raise ContractError(ErrorCode.MANIFEST_INVALID, "keys", "48-key order required")
|
result = {k: "" if row[k] is None else str(row[k]) for k in MANIFEST_COLUMNS}
|
if result["source_site"] != "慧博 APP（安卓模拟器本地缓存）":
|
raise ContractError(ErrorCode.MANIFEST_INVALID, "source_site", "literal mismatch")
|
if result["android_package"] != "cn.com.hibor":
|
raise ContractError(ErrorCode.MANIFEST_INVALID, "android_package", "literal mismatch")
|
if result["download_status"] not in {"SUCCESS", "DUPLICATE", "FAILED", "STOPPED"}:
|
raise ContractError(ErrorCode.MANIFEST_INVALID, "download_status", "enum")
|
if result["status"] != result["download_status"]:
|
raise ContractError(ErrorCode.MANIFEST_INVALID, "status", "must equal download_status")
|
if result["content_type"] == "application/pdf" and result["pdf_magic_valid"].lower() != "true":
|
raise ContractError(ErrorCode.MANIFEST_INVALID, "content_type", "PDF claim without magic")
|
if result["reused_without_new_trigger"].lower() not in {"true", "false"}:
|
raise ContractError(ErrorCode.MANIFEST_INVALID, "reused", "BOOL required")
|
return result
|
|
|
def _create_or_verify(path: Path, data: bytes,
|
checkpoint: Callable[[], None] | None = None) -> None:
|
try:
|
create_exclusive_bytes(path, data, checkpoint=checkpoint)
|
except FileExistsError as exc:
|
if not path.is_file() or path.is_symlink():
|
raise ContractError(ErrorCode.PERSIST_LATE, "path", "existing non-file") from exc
|
try:
|
existing = path.read_bytes()
|
except OSError as read_exc:
|
raise ContractError(ErrorCode.RECOVERY_UNKNOWN, "path", "existing unreadable") from read_exc
|
if existing != data:
|
raise ContractError(ErrorCode.PERSIST_LATE, "path", "existing bytes differ") from exc
|
if checkpoint:
|
checkpoint()
|
|
|
def write_manifest_create_new(path: Path, rows: Sequence[Mapping[str, Any]], *,
|
checkpoint: Callable[[], None] | None = None) -> tuple[int, str]:
|
buffer = io.StringIO(newline="")
|
writer = csv.DictWriter(buffer, fieldnames=MANIFEST_COLUMNS, lineterminator="\r\n")
|
writer.writeheader()
|
for row in rows:
|
writer.writerow(validate_manifest_row(row))
|
data = buffer.getvalue().encode("utf-8")
|
_create_or_verify(path, data, checkpoint=checkpoint)
|
try:
|
parsed_rows = list(csv.DictReader(io.StringIO(path.read_text(encoding="utf-8"), newline="")))
|
except (OSError, UnicodeError, csv.Error) as exc:
|
raise ContractError(ErrorCode.MANIFEST_INVALID, "manifest", "reopen failed") from exc
|
if (parsed_rows and tuple(parsed_rows[0].keys()) != MANIFEST_COLUMNS) or (not parsed_rows and rows):
|
raise ContractError(ErrorCode.MANIFEST_INVALID, "manifest", "reopen schema mismatch")
|
if path.read_bytes() != data:
|
raise ContractError(ErrorCode.PERSIST_LATE, "manifest", "reopen bytes mismatch")
|
return len(data), hashlib.sha256(data).hexdigest()
|
|
|
def write_json_create_new(path: Path, value: Any, *,
|
checkpoint: Callable[[], None] | None = None) -> tuple[int, str]:
|
data = canonical_json_bytes(value)
|
_create_or_verify(path, data, checkpoint=checkpoint)
|
parsed = json.loads(path.read_text(encoding="utf-8"))
|
if canonical_json_bytes(parsed) != data:
|
raise ContractError(ErrorCode.PERSIST_LATE, "json", "canonical roundtrip mismatch")
|
return len(data), hashlib.sha256(data).hexdigest()
|
|
|
def write_bytes_create_new(path: Path, data: bytes, *,
|
checkpoint: Callable[[], None] | None = None) -> tuple[int, str]:
|
"""Create a closure artifact or accept only an exact durable replay."""
|
_create_or_verify(path, data, checkpoint=checkpoint)
|
try:
|
reopened = path.read_bytes()
|
except OSError as exc:
|
raise ContractError(ErrorCode.RECOVERY_UNKNOWN, "bytes", "reopen failed") from exc
|
if reopened != data:
|
raise ContractError(ErrorCode.PERSIST_LATE, "bytes", "reopen mismatch")
|
return len(data), hashlib.sha256(data).hexdigest()
|
|
|
def utc_now() -> str:
|
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
