"""Bridge AutoShorts review packages to TrueTraceShorts website companion pages. The bridge is intentionally approval/hand-off oriented: it writes a website candidate JSON and can optionally invoke the website repo's local generator. It does not publish videos or call any platform APIs. """ from __future__ import annotations import hashlib import json import re import subprocess from dataclasses import dataclass, field from pathlib import Path from typing import Any REQUIRED_SPEC_FIELDS = ( "slug", "title", "shortTitle", "hook", "category", "riskLevel", "visibleScreen", "redFlag", "whyItWorks", "saferMove", "ifAlreadyClicked", "checklist", "toolRelevance", "seoTitle", "metaDescription", "socialTitle", ) @dataclass(frozen=True) class CompanionExportPlan: candidate_id: str slug: str candidate_json_path: Path video_path: Path | None generator_command: tuple[str, ...] | None def to_dict(self) -> dict[str, Any]: return { "candidate_id": self.candidate_id, "slug": self.slug, "candidate_json_path": str(self.candidate_json_path), "video_path": str(self.video_path) if self.video_path else None, "generator_command": list(self.generator_command) if self.generator_command else None, } @dataclass(frozen=True) class WebsiteCompanionPackage: """Approval-bound handoff artifact from AutoShortsBot to the website repo.""" candidate_id: str version: str | None page_slug: str website_path: Path page_payload_json: Path video_url: str | None thumbnail_path: Path | None redflag_markdown_path: Path | None safety_summary: str seo_summary: str tool_relevance: str update_allowed: bool = False video_sha256: str | None = None posting_pack_sha256: str | None = None website_payload_sha256: str | None = None review_package_path: Path | None = None generator_command: tuple[str, ...] | None = None notes: tuple[str, ...] = field(default_factory=tuple) def to_dict(self) -> dict[str, Any]: return { "candidate_id": self.candidate_id, "version": self.version, "page_slug": self.page_slug, "website_path": str(self.website_path), "page_payload_json": str(self.page_payload_json), "video_url": self.video_url, "thumbnail_path": str(self.thumbnail_path) if self.thumbnail_path else None, "redflag_markdown_path": str(self.redflag_markdown_path) if self.redflag_markdown_path else None, "safety_summary": self.safety_summary, "seo_summary": self.seo_summary, "tool_relevance": self.tool_relevance, "update_allowed": self.update_allowed, "video_sha256": self.video_sha256, "posting_pack_sha256": self.posting_pack_sha256, "website_payload_sha256": self.website_payload_sha256, "review_package_path": str(self.review_package_path) if self.review_package_path else None, "generator_command": list(self.generator_command) if self.generator_command else None, "notes": list(self.notes), } @dataclass(frozen=True) class WebsiteCompanionUpdateGate: candidate_id: str version: str | None video_sha256: str | None posting_pack_sha256: str | None website_payload_sha256: str | None no_raw_comment_text: bool no_external_user_text: bool safety_gate_valid: bool no_real_scam_data: bool @property def allowed(self) -> bool: return all( [ bool(self.candidate_id), bool(self.version), bool(self.video_sha256), bool(self.posting_pack_sha256), bool(self.website_payload_sha256), self.no_raw_comment_text, self.no_external_user_text, self.safety_gate_valid, self.no_real_scam_data, ] ) def to_dict(self) -> dict[str, Any]: return { "candidate_id": self.candidate_id, "version": self.version, "video_sha256": self.video_sha256, "posting_pack_sha256": self.posting_pack_sha256, "website_payload_sha256": self.website_payload_sha256, "no_raw_comment_text": self.no_raw_comment_text, "no_external_user_text": self.no_external_user_text, "safety_gate_valid": self.safety_gate_valid, "no_real_scam_data": self.no_real_scam_data, "allowed": self.allowed, } def load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def write_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def sha256_json(payload: dict[str, Any]) -> str: data = json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8") return hashlib.sha256(data).hexdigest() def slugify(value: str) -> str: return re.sub(r"^-+|-+$", "", re.sub(r"[^a-z0-9]+", "-", value.lower())).strip("-") def validate_companion_spec(spec: dict[str, Any]) -> None: missing = [field for field in REQUIRED_SPEC_FIELDS if field not in spec] if missing: raise ValueError(f"Companion spec missing required fields: {', '.join(missing)}") for field in ("ifAlreadyClicked", "checklist"): value = spec[field] if not isinstance(value, list) or not value or not all(isinstance(item, str) and item.strip() for item in value): raise ValueError(f"Companion spec field {field!r} must be a non-empty list of strings") if spec["riskLevel"] not in {"low", "medium", "high"}: raise ValueError("Companion spec riskLevel must be one of: low, medium, high") def companion_payload_has_unsafe_example_text(payload: dict[str, Any]) -> bool: """Detect obvious real-world scam data in website copy fields. Legitimate platform links live in videoUrl and source metadata, so exclude them from the scan. """ scrubbed = {key: value for key, value in payload.items() if key not in {"videoUrl", "source", "lastUpdated"}} joined = json.dumps(scrubbed, ensure_ascii=False) patterns = ( re.compile(r"https?://", re.I), re.compile(r"\bwww\.", re.I), re.compile(r"\b\+?\d[\d\s().-]{7,}\d\b"), re.compile(r"\b\d{4}[ -]?\d{4}[ -]?\d{4}[ -]?\d{4}\b"), re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b"), re.compile(r"\b[a-z0-9-]+\.(?:com|net|org|ch|de|co|io|app|shop|info)\b", re.I), re.compile(r"\b(?:visa|mastercard|paypal|western union|moneygram|amazon|apple|microsoft|google|dhl|fedex|ups)\b", re.I), ) return any(pattern.search(joined) for pattern in patterns) def build_update_gate(review: dict[str, Any], candidate: dict[str, Any], video_path: Path | None = None) -> WebsiteCompanionUpdateGate: website_payload_sha256 = sha256_json(candidate) video_sha256 = None if video_path and video_path.exists(): video_sha256 = sha256_file(video_path) else: video_sha256 = review.get("video_sha256") or review.get("sha256") return WebsiteCompanionUpdateGate( candidate_id=str(review.get("candidate_id") or candidate.get("id") or ""), version=review.get("version"), video_sha256=video_sha256, posting_pack_sha256=review.get("posting_pack_sha256"), website_payload_sha256=website_payload_sha256, no_raw_comment_text=not bool(candidate.get("raw_comments") or candidate.get("comment_text")), no_external_user_text=not bool(candidate.get("external_user_text") or candidate.get("raw_user_text")), safety_gate_valid=True, no_real_scam_data=not companion_payload_has_unsafe_example_text(candidate), ) def resolve_review_video_path(review: dict[str, Any], prefer_media_cache: bool = True) -> Path | None: candidates: list[str | None] = [] if prefer_media_cache: candidates.extend([review.get("media_cache_path"), review.get("output")]) else: candidates.extend([review.get("output"), review.get("media_cache_path")]) for candidate in candidates: if candidate: path = Path(candidate) if path.exists(): return path return None def build_website_candidate( review: dict[str, Any], spec: dict[str, Any], review_package_path: Path | None = None, ) -> dict[str, Any]: validate_companion_spec(spec) candidate_id = str(review.get("candidate_id") or spec.get("id") or spec["slug"]).strip() if not candidate_id: raise ValueError("Could not derive candidate_id") slug = slugify(str(spec["slug"])) if not slug: raise ValueError("Could not derive slug") related = spec.get("related", []) if related is None: related = [] if not isinstance(related, list) or not all(isinstance(item, str) for item in related): raise ValueError("Companion spec related must be a list of strings when provided") payload: dict[str, Any] = { "id": spec.get("id") or slug, "slug": slug, "title": spec["title"], "shortTitle": spec["shortTitle"], "hook": spec["hook"], "category": spec["category"], "riskLevel": spec["riskLevel"], "visibleScreen": spec["visibleScreen"], "redFlag": spec["redFlag"], "whyItWorks": spec["whyItWorks"], "saferMove": spec["saferMove"], "ifAlreadyClicked": spec["ifAlreadyClicked"], "checklist": spec["checklist"], "checklistDetails": spec.get("checklistDetails", []), "faq": spec.get("faq", []), "videoUrl": spec.get("videoUrl"), "thumbnail": spec.get("thumbnail"), "screenImage": spec.get("screenImage"), "screenImageAlt": spec.get("screenImageAlt") or spec["visibleScreen"], "visualBrief": spec.get("visualBrief"), "related": related, "toolRelevance": spec["toolRelevance"], "affiliateCategory": spec.get("affiliateCategory"), "seoTitle": spec["seoTitle"], "metaDescription": spec["metaDescription"], "socialTitle": spec["socialTitle"], "body": spec.get("body") or default_body(spec), "source": { "candidate_id": candidate_id, "version": review.get("version"), "review_package_sha256": review.get("sha256"), "posting_pack_sha256": review.get("posting_pack_sha256"), "approval_command": review.get("approval_command"), "review_package_path": str(review_package_path) if review_package_path else None, "youtube_private_upload_package_path": review.get("package_path"), "title": review.get("title"), }, } if "lastUpdated" in spec: payload["lastUpdated"] = spec["lastUpdated"] return payload def default_body(spec: dict[str, Any]) -> str: return ( f"{spec['hook']}\n\n" f"The safer move is simple: {spec['saferMove']}\n" ) def build_website_companion_package( *, review_package_path: Path, companion_spec_path: Path, website_root: Path, run_generator: bool = False, force: bool = False, prefer_media_cache: bool = True, ) -> WebsiteCompanionPackage: review = load_json(review_package_path) spec = load_json(companion_spec_path) candidate = build_website_candidate(review, spec, review_package_path=review_package_path.resolve()) video_path = resolve_review_video_path(review, prefer_media_cache=prefer_media_cache) gate = build_update_gate(review, candidate, video_path) plan = export_companion_candidate( review_package_path=review_package_path, companion_spec_path=companion_spec_path, website_root=website_root, run_generator=run_generator, force=force, prefer_media_cache=prefer_media_cache, ) redflag_markdown = website_root.resolve() / "src" / "content" / "redflags" / f"{plan.slug}.md" thumbnail = website_root.resolve() / "public" / "redflags" / "thumbnails" / f"{plan.slug}-start.webp" notes = [] if not gate.allowed: notes.append("update_allowed remains false until approval/hash/safety gate is satisfied") return WebsiteCompanionPackage( candidate_id=plan.candidate_id, version=review.get("version"), page_slug=plan.slug, website_path=website_root.resolve(), page_payload_json=plan.candidate_json_path, video_url=candidate.get("videoUrl"), thumbnail_path=thumbnail if thumbnail.exists() else None, redflag_markdown_path=redflag_markdown if redflag_markdown.exists() else None, safety_summary=candidate.get("saferMove", ""), seo_summary=candidate.get("seoTitle") or candidate.get("metaDescription") or "", tool_relevance=candidate.get("toolRelevance", ""), update_allowed=False, video_sha256=gate.video_sha256, posting_pack_sha256=gate.posting_pack_sha256, website_payload_sha256=gate.website_payload_sha256, review_package_path=review_package_path.resolve(), generator_command=plan.generator_command, notes=tuple(notes), ) def export_companion_candidate( *, review_package_path: Path, companion_spec_path: Path, website_root: Path, run_generator: bool = False, force: bool = False, prefer_media_cache: bool = True, ) -> CompanionExportPlan: review = load_json(review_package_path) spec = load_json(companion_spec_path) candidate = build_website_candidate(review, spec, review_package_path=review_package_path.resolve()) slug = candidate["slug"] candidate_id = str(review.get("candidate_id") or candidate["id"]) website_root = website_root.resolve() candidate_json_path = website_root / "data" / "candidates" / f"{candidate_id}.json" video_path = resolve_review_video_path(review, prefer_media_cache=prefer_media_cache) if run_generator and video_path is None: raise FileNotFoundError("No review video path exists in review package output/media_cache_path") write_json(candidate_json_path, candidate) generator_command: tuple[str, ...] | None = None if run_generator: rel_candidate = candidate_json_path.relative_to(website_root) cmd = ["npm", "run", "generate:companion", "--", "--candidate", str(rel_candidate), "--video", str(video_path)] if force: cmd.append("--force") result = subprocess.run(cmd, cwd=website_root, text=True, capture_output=True, check=False) if result.returncode != 0: raise RuntimeError( "Website companion generator failed:\n" f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" ) generator_command = tuple(cmd) return CompanionExportPlan( candidate_id=candidate_id, slug=slug, candidate_json_path=candidate_json_path, video_path=video_path, generator_command=generator_command, )