fix: serialize media publishing decisions
This commit is contained in:
+135
-60
@@ -5,11 +5,14 @@ import io
|
||||
import json
|
||||
import mimetypes
|
||||
import re
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from urllib.parse import urljoin, urlsplit
|
||||
|
||||
import fcntl
|
||||
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
from PIL import Image, ImageDraw, ImageFont, UnidentifiedImageError
|
||||
|
||||
@@ -27,6 +30,52 @@ MEDIA_ROLES_BY_ENTITY = {
|
||||
}
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _manifest_lock(path: Path):
|
||||
"""Serialize decisions made by API and CLI against one manifest."""
|
||||
lock_path = path.with_name(f"{path.name}.lock")
|
||||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with lock_path.open("w", encoding="utf-8") as lock:
|
||||
fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
|
||||
|
||||
|
||||
def _manifest_version(manifest: dict) -> int:
|
||||
try:
|
||||
return max(1, int(manifest.get("version", 1)))
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ValueError("manifest version is invalid") from exc
|
||||
|
||||
|
||||
def _asset_id(item: dict) -> str | None:
|
||||
value = item.get("sha256")
|
||||
return value if isinstance(value, str) and len(value) == 64 else None
|
||||
|
||||
|
||||
def _selected_upgrade_candidates(manifest: dict, asset_ids: list[str] | None) -> list[dict]:
|
||||
candidates = [item for item in manifest.get("assets", []) if item.get("status") == "upgrade_stored"]
|
||||
if asset_ids is None:
|
||||
return candidates
|
||||
selected = set(asset_ids)
|
||||
by_id = {_asset_id(item): item for item in candidates}
|
||||
if len(selected) != len(asset_ids):
|
||||
raise ValueError("asset_ids must be unique")
|
||||
missing = sorted(selected - set(by_id))
|
||||
if missing:
|
||||
raise ValueError(f"selected asset is not an upgrade_stored candidate: {missing[0]}")
|
||||
return [by_id[asset_id] for asset_id in asset_ids]
|
||||
|
||||
|
||||
def _check_expected_version(manifest: dict, expected_version: int | None) -> int:
|
||||
version = _manifest_version(manifest)
|
||||
if expected_version is not None and expected_version != version:
|
||||
raise ValueError(f"manifest version conflict: expected {expected_version}, current {version}")
|
||||
return version
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MediaCandidate:
|
||||
source_page: str
|
||||
@@ -197,70 +246,96 @@ def approve_stored_assets(path: Path, *, note: str) -> dict:
|
||||
return {"approved": approved}
|
||||
|
||||
|
||||
def publish_quality_upgrades(path: Path, *, note: str, minimum_dimension: int = 256) -> dict:
|
||||
"""Atomically promote every reviewed quality candidate and retain its fallback."""
|
||||
manifest = json.loads(path.read_text(encoding="utf-8"))
|
||||
assets = manifest.get("assets", [])
|
||||
by_url = {item.get("asset_url"): item for item in assets}
|
||||
candidates = [item for item in assets if item.get("status") == "upgrade_stored"]
|
||||
|
||||
# Validate the complete set before changing any public mapping.
|
||||
for candidate in candidates:
|
||||
fallback = by_url.get(candidate.get("duplicate_of"))
|
||||
if not fallback or fallback.get("status") != "approved":
|
||||
raise ValueError(f"{candidate.get('asset_url')}: approved fallback is missing")
|
||||
if not fallback.get("entity_key"):
|
||||
raise ValueError(f"{candidate.get('asset_url')}: fallback has no entity_key")
|
||||
if min(int(candidate.get("width") or 0), int(candidate.get("height") or 0)) < minimum_dimension:
|
||||
raise ValueError(f"{candidate.get('asset_url')}: candidate is below {minimum_dimension}px")
|
||||
if not candidate.get("sha256") or not candidate.get("local_path"):
|
||||
raise ValueError(f"{candidate.get('asset_url')}: stored candidate metadata is incomplete")
|
||||
|
||||
reviewed_at = datetime.now(timezone.utc).isoformat()
|
||||
for candidate in candidates:
|
||||
fallback = by_url[candidate["duplicate_of"]]
|
||||
candidate.update({
|
||||
"status": "approved",
|
||||
"entity_key": fallback["entity_key"],
|
||||
"supersedes": fallback["asset_url"],
|
||||
"reviewed_at": reviewed_at,
|
||||
"review_note": note,
|
||||
})
|
||||
fallback.update({
|
||||
"status": "superseded",
|
||||
"replaced_by": candidate["asset_url"],
|
||||
"reviewed_at": reviewed_at,
|
||||
"review_note": note,
|
||||
})
|
||||
|
||||
manifest["updated_at"] = reviewed_at
|
||||
temporary = path.with_suffix(path.suffix + ".tmp")
|
||||
temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
temporary.replace(path)
|
||||
return {"published": len(candidates), "retained_fallbacks": len(candidates)}
|
||||
def preview_quality_upgrades(path: Path, *, asset_ids: list[str], expected_version: int | None = None) -> dict:
|
||||
"""Return the exact selected set and version used for a later decision."""
|
||||
with _manifest_lock(path):
|
||||
manifest = json.loads(path.read_text(encoding="utf-8"))
|
||||
version = _check_expected_version(manifest, expected_version)
|
||||
candidates = _selected_upgrade_candidates(manifest, asset_ids)
|
||||
return {
|
||||
"manifest_version": version,
|
||||
"assets": [{
|
||||
"id": _asset_id(item), "asset_url": item.get("asset_url"),
|
||||
"duplicate_of": item.get("duplicate_of"), "label": item.get("label"),
|
||||
"width": item.get("width"), "height": item.get("height"),
|
||||
} for item in candidates],
|
||||
}
|
||||
|
||||
|
||||
def rollback_quality_upgrade(path: Path, *, asset_url: str, note: str) -> dict:
|
||||
def publish_quality_upgrades(
|
||||
path: Path, *, note: str, asset_ids: list[str] | None = None,
|
||||
expected_version: int | None = None, minimum_dimension: int = 256,
|
||||
) -> dict:
|
||||
"""Atomically promote an explicit reviewed set and retain its fallbacks."""
|
||||
with _manifest_lock(path):
|
||||
manifest = json.loads(path.read_text(encoding="utf-8"))
|
||||
previous_version = _check_expected_version(manifest, expected_version)
|
||||
assets = manifest.get("assets", [])
|
||||
by_url = {item.get("asset_url"): item for item in assets}
|
||||
candidates = _selected_upgrade_candidates(manifest, asset_ids)
|
||||
|
||||
for candidate in candidates:
|
||||
fallback = by_url.get(candidate.get("duplicate_of"))
|
||||
if not fallback or fallback.get("status") != "approved":
|
||||
raise ValueError(f"{candidate.get('asset_url')}: approved fallback is missing")
|
||||
if not fallback.get("entity_key"):
|
||||
raise ValueError(f"{candidate.get('asset_url')}: fallback has no entity_key")
|
||||
if min(int(candidate.get("width") or 0), int(candidate.get("height") or 0)) < minimum_dimension:
|
||||
raise ValueError(f"{candidate.get('asset_url')}: candidate is below {minimum_dimension}px")
|
||||
if not candidate.get("sha256") or not candidate.get("local_path"):
|
||||
raise ValueError(f"{candidate.get('asset_url')}: stored candidate metadata is incomplete")
|
||||
|
||||
reviewed_at = datetime.now(timezone.utc).isoformat()
|
||||
for candidate in candidates:
|
||||
fallback = by_url[candidate["duplicate_of"]]
|
||||
candidate.update({
|
||||
"status": "approved", "entity_key": fallback["entity_key"],
|
||||
"supersedes": fallback["asset_url"], "reviewed_at": reviewed_at,
|
||||
"review_note": note,
|
||||
})
|
||||
fallback.update({
|
||||
"status": "superseded", "replaced_by": candidate["asset_url"],
|
||||
"reviewed_at": reviewed_at, "review_note": note,
|
||||
})
|
||||
|
||||
manifest["version"] = previous_version + 1
|
||||
manifest["updated_at"] = reviewed_at
|
||||
temporary = path.with_suffix(path.suffix + ".tmp")
|
||||
temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
temporary.replace(path)
|
||||
return {"published": len(candidates), "retained_fallbacks": len(candidates), "manifest_version": manifest["version"]}
|
||||
|
||||
|
||||
def rollback_quality_upgrade(
|
||||
path: Path, *, asset_url: str | None = None, asset_id: str | None = None,
|
||||
note: str, expected_version: int | None = None,
|
||||
) -> dict:
|
||||
"""Restore the superseded fallback for one explicitly selected upgrade."""
|
||||
manifest = json.loads(path.read_text(encoding="utf-8"))
|
||||
assets = manifest.get("assets", [])
|
||||
candidate = next((item for item in assets if item.get("asset_url") == asset_url), None)
|
||||
if not candidate or candidate.get("status") != "approved" or not candidate.get("supersedes"):
|
||||
raise ValueError("rollback requires an approved upgrade with a supersedes link")
|
||||
fallback = next((item for item in assets if item.get("asset_url") == candidate["supersedes"]), None)
|
||||
if not fallback or fallback.get("status") != "superseded" or fallback.get("replaced_by") != asset_url:
|
||||
raise ValueError("rollback requires the matching superseded fallback")
|
||||
if not asset_id and not asset_url:
|
||||
raise ValueError("rollback requires an asset id")
|
||||
with _manifest_lock(path):
|
||||
manifest = json.loads(path.read_text(encoding="utf-8"))
|
||||
previous_version = _check_expected_version(manifest, expected_version)
|
||||
assets = manifest.get("assets", [])
|
||||
candidate = next((item for item in assets if (asset_id and _asset_id(item) == asset_id) or (asset_url and item.get("asset_url") == asset_url)), None)
|
||||
if not candidate or candidate.get("status") != "approved" or not candidate.get("supersedes"):
|
||||
raise ValueError("rollback requires an approved upgrade with a supersedes link")
|
||||
selected_url = candidate.get("asset_url")
|
||||
fallback = next((item for item in assets if item.get("asset_url") == candidate["supersedes"]), None)
|
||||
if not fallback or fallback.get("status") != "superseded" or fallback.get("replaced_by") != selected_url:
|
||||
raise ValueError("rollback requires the matching superseded fallback")
|
||||
|
||||
reviewed_at = datetime.now(timezone.utc).isoformat()
|
||||
candidate.update({"status": "upgrade_stored", "reviewed_at": reviewed_at, "review_note": note})
|
||||
candidate.pop("supersedes", None)
|
||||
fallback.update({"status": "approved", "reviewed_at": reviewed_at, "review_note": note})
|
||||
fallback.pop("replaced_by", None)
|
||||
manifest["updated_at"] = reviewed_at
|
||||
temporary = path.with_suffix(path.suffix + ".tmp")
|
||||
temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
temporary.replace(path)
|
||||
return {"rolled_back": asset_url, "restored": fallback["asset_url"]}
|
||||
reviewed_at = datetime.now(timezone.utc).isoformat()
|
||||
candidate.update({"status": "upgrade_stored", "reviewed_at": reviewed_at, "review_note": note})
|
||||
candidate.pop("supersedes", None)
|
||||
fallback.update({"status": "approved", "reviewed_at": reviewed_at, "review_note": note})
|
||||
fallback.pop("replaced_by", None)
|
||||
manifest["version"] = previous_version + 1
|
||||
manifest["updated_at"] = reviewed_at
|
||||
temporary = path.with_suffix(path.suffix + ".tmp")
|
||||
temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
temporary.replace(path)
|
||||
return {"rolled_back": selected_url, "restored": fallback["asset_url"], "manifest_version": manifest["version"]}
|
||||
|
||||
|
||||
def reclassify_manifest(path: Path) -> dict:
|
||||
|
||||
Reference in New Issue
Block a user