#!/usr/bin/env python3 """Create prepared-package folders from recent rendered videos. This script intentionally copies media into local storage only. The repository .gitignore excludes storage media and video binaries. """ from __future__ import annotations import argparse import hashlib import json import re import shutil import sys from dataclasses import dataclass from pathlib import Path from urllib import request as urlrequest VIDEO_EXTS = {".mp4", ".mov", ".webm"} REVIEW_PACKAGE_NAMES = {"review_package_chatterbox.json", "review_package.json"} @dataclass class Candidate: video_path: Path review_package: Path | None metadata: dict score: int def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def load_json(path: Path | None) -> dict: if not path or not path.exists(): return {} return json.loads(path.read_text(encoding="utf-8")) def slug_title(value: str) -> str: cleaned = re.sub(r"[_-]+", " ", value) cleaned = re.sub(r"\b(chatterbox|gianna|clone|premium|director|pass|pure|ai|v\d+|1080x1920|review|preview)\b", " ", cleaned, flags=re.I) cleaned = re.sub(r"\s+", " ", cleaned).strip(" -_") return cleaned[:80].title() or "Prepared Video" def find_review_package(video_path: Path) -> Path | None: for name in REVIEW_PACKAGE_NAMES: p = video_path.parent / name if p.exists(): return p packages = sorted(video_path.parent.glob("review_package*.json")) return packages[0] if packages else None def select_video_from_metadata(meta: dict, fallback: Path) -> Path: outputs = meta.get("outputs") or {} for key in ("mp4", "media_cache_mp4"): value = outputs.get(key) if value and Path(value).exists(): return Path(value) for key in ("output", "media_cache_path"): value = meta.get(key) if value and Path(value).exists(): return Path(value) return fallback def extract_hashtags(*texts: str) -> list[str]: tags: list[str] = [] for text in texts: for tag in re.findall(r"#[\wÄÖÜäöüß]+", text or ""): if tag not in tags: tags.append(tag) return tags def manifest_from_candidate(candidate: Candidate, package_id: str) -> dict: meta = candidate.metadata youtube = meta.get("youtube_private_upload") or {} title = youtube.get("title") or meta.get("title") or meta.get("hook_line") or slug_title(candidate.video_path.stem) description = youtube.get("description") or meta.get("description") script = meta.get("script") metadata_note = None if not description and script: description = f"Short guide based on the approved script:\n\n{script[:650]}" if not description: description = "metadata_missing: No source description found. Please write the YouTube description during review." metadata_note = "metadata_missing" tiktok_caption = meta.get("tiktok_caption") or meta.get("suggested_tiktok_caption") if not tiktok_caption and script: tiktok_caption = script[:220].rstrip() + "…" if not tiktok_caption: tiktok_caption = "metadata_missing: Please write the TikTok caption during review." metadata_note = "metadata_missing" hashtags = extract_hashtags(tiktok_caption, description) if not hashtags: hashtags = ["#OnlineSafety", "#ScamAwareness"] guide = meta.get("website_guide_link") or (meta.get("format_metadata") or {}).get("website_target") series = meta.get("theme") or (meta.get("format_metadata") or {}).get("format_family") or "Prepared video" manifest = { "package_id": package_id, "source": "prepared_package", "working_title": title, "language": "en", "topic_id": None, "series": series, "suggested_tiktok_caption": tiktok_caption, "suggested_youtube_title": title, "suggested_youtube_description": description, "suggested_hashtags": hashtags, "youtube_tags": [tag.lstrip("#") for tag in hashtags], "disclosure": { "synthetic_media": True, "commercial_content": False, }, "status": "ready_for_review", "script": script, "transcript": meta.get("transcript"), "source_metadata": { "candidate_id": meta.get("candidate_id"), "version": meta.get("version"), "source_review_package": str(candidate.review_package) if candidate.review_package else None, "source_video": str(candidate.video_path), "source_sha256": meta.get("sha256") or sha256_file(candidate.video_path), "youtube_video_id": youtube.get("video_id"), "youtube_shorts_url": youtube.get("shorts_url"), "website_guide_link": guide, "note": metadata_note, }, } return manifest def find_candidates(source_dir: Path) -> list[Candidate]: candidates: list[Candidate] = [] for video in source_dir.rglob("*"): if not video.is_file() or video.suffix.lower() not in VIDEO_EXTS: continue lower = str(video).lower() if "video_no_audio" in lower or "/scene_" in lower or "contact_sheet" in lower: continue score = 0 if "review" in video.name.lower() or "preview" in video.name.lower(): score += 10 review_package = find_review_package(video) metadata = load_json(review_package) if metadata: score += 20 selected = select_video_from_metadata(metadata, video) if selected != video and selected.exists(): video = selected if "youtube_private_upload" in metadata: score += 5 candidates.append(Candidate(video_path=video, review_package=review_package, metadata=metadata, score=score)) dedup: dict[str, Candidate] = {} for item in candidates: key = str(item.video_path.resolve()) existing = dedup.get(key) if existing is None or (item.score, item.video_path.stat().st_mtime) > (existing.score, existing.video_path.stat().st_mtime): dedup[key] = item return sorted(dedup.values(), key=lambda c: (c.video_path.stat().st_mtime, c.score), reverse=True) def copy_candidate(candidate: Candidate, target_dir: Path, index: int, dry_run: bool) -> dict: package_id = f"package_real_{index:03d}" folder = target_dir / package_id manifest = manifest_from_candidate(candidate, package_id) out = { "package_id": package_id, "video": str(candidate.video_path), "review_package": str(candidate.review_package) if candidate.review_package else None, "title": manifest["working_title"], "metadata_note": manifest["source_metadata"].get("note"), } if dry_run: return out folder.mkdir(parents=True, exist_ok=True) shutil.copy2(candidate.video_path, folder / "video.mp4") (folder / "manifest.json").write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") if manifest.get("script"): (folder / "transcript.txt").write_text(manifest["script"], encoding="utf-8") return out def trigger_import(api_base: str) -> dict: req = urlrequest.Request(f"{api_base.rstrip('/')}/api/imports/scan", method="POST") with urlrequest.urlopen(req, timeout=30) as response: # noqa: S310 - local/dev URL by operator input return json.loads(response.read().decode("utf-8")) def main() -> int: parser = argparse.ArgumentParser(description="Import latest rendered videos as prepared packages.") parser.add_argument("--source-dir", required=True, type=Path) parser.add_argument("--target-dir", required=True, type=Path) parser.add_argument("--limit", type=int, default=3) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--import", dest="do_import", action="store_true") parser.add_argument("--api-base", default="http://127.0.0.1:8012") args = parser.parse_args() candidates = [c for c in find_candidates(args.source_dir) if c.score >= 10] selected = candidates[: args.limit] if not selected: print(json.dumps({"ok": False, "error": "No rendered video candidates found"}, indent=2)) return 1 results = [copy_candidate(candidate, args.target_dir, index + 1, args.dry_run) for index, candidate in enumerate(selected)] payload: dict = {"ok": True, "dry_run": args.dry_run, "selected": results} if args.do_import and not args.dry_run: payload["import_result"] = trigger_import(args.api_base) print(json.dumps(payload, indent=2, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())