from dataclasses import dataclass, field from pathlib import Path import json import re from sqlmodel import Session, select from app.core.config import Settings from app.models.core import ActivityEvent, ContentScript, ImportLog, PostDraft, ProductionQueueItem, Provider, VideoAsset, VideoStatus from app.services.video_metadata import copy_video_to_processed, generate_thumbnail def _extract_youtube_id(payload: dict) -> str | None: text = json.dumps(payload, ensure_ascii=False) for pattern in [r"youtube\.com/shorts/([A-Za-z0-9_-]{6,})", r"youtu\.be/([A-Za-z0-9_-]{6,})", r"[?&]v=([A-Za-z0-9_-]{6,})"]: match = re.search(pattern, text) if match: return match.group(1) for key in ["youtube_video_id", "video_id", "external_youtube_id"]: value = payload.get(key) if isinstance(value, str) and value.strip(): return value.strip() return None @dataclass class ImportResult: imported: int = 0 failed: int = 0 skipped: int = 0 video_ids: list[str] = field(default_factory=list) logs: list[dict] = field(default_factory=list) errors: list[str] = field(default_factory=list) class PreparedPackageImporter: def __init__(self, *, settings: Settings, engine): self.settings = settings self.engine = engine self.incoming = settings.storage_root / "incoming" self.processed = settings.storage_root / "processed" self.thumbnails = settings.storage_root / "thumbnails" def scan_once(self) -> ImportResult: self.incoming.mkdir(parents=True, exist_ok=True) result = ImportResult() for folder in sorted(p for p in self.incoming.iterdir() if p.is_dir()): try: video_id = self._import_folder(folder) if video_id: result.imported += 1 result.video_ids.append(video_id) result.logs.append({"folder": folder.name, "status": "imported", "video_id": video_id}) else: result.skipped += 1 result.logs.append({"folder": folder.name, "status": "skipped", "message": "manifest missing"}) except Exception as exc: result.failed += 1 message = f"{folder.name}: {exc}" result.errors.append(message) result.logs.append({"folder": folder.name, "status": "failed", "error": str(exc)}) (folder / "import_error.log").write_text(str(exc), encoding="utf-8") self._log(package_id=folder.name, folder_name=folder.name, status="failed", error=str(exc)) return result def import_manifest_payload(self, manifest: dict, video_path: Path | None = None) -> str: package_id = manifest.get("package_id") or manifest.get("candidate_id") if not package_id: raise ValueError("package_id is required") if video_path is None: raise ValueError("video_path is required for MVP API import") target_dir = self.processed / package_id target_video = target_dir / video_path.name meta = copy_video_to_processed(video_path, target_video) return self._create_records(manifest=manifest, target_video=target_video, meta=meta, folder_name=package_id) def _import_folder(self, folder: Path) -> str | None: manifest_path = folder / "manifest.json" if not manifest_path.exists(): return None manifest = json.loads(manifest_path.read_text(encoding="utf-8")) package_id = manifest.get("package_id") or manifest.get("candidate_id") or folder.name manifest["package_id"] = package_id source_video = next((folder / name for name in ("video.mp4", "video.mov", "video.webm") if (folder / name).exists()), None) if source_video is None: raise ValueError("video file missing") target_video = self.processed / package_id / source_video.name meta = copy_video_to_processed(source_video, target_video) return self._create_records(manifest=manifest, target_video=target_video, meta=meta, folder_name=folder.name) def _log(self, *, package_id: str | None, folder_name: str | None, status: str, video_asset_id: str | None = None, message: str | None = None, error: str | None = None): with Session(self.engine) as session: session.add(ImportLog(package_id=package_id, folder_name=folder_name, status=status, video_asset_id=video_asset_id, message=message, error=error)) session.commit() def _quality_payload(self, manifest: dict) -> dict: quality = manifest.get("quality") if isinstance(manifest.get("quality"), dict) else {} manifest_quality_report = manifest.get("quality_report") if isinstance(manifest.get("quality_report"), dict) else {} reports = { "quality_report": manifest_quality_report or quality, "voice_qa": manifest.get("voice_qa") if isinstance(manifest.get("voice_qa"), dict) else {}, "visual_qa": manifest.get("visual_qa") if isinstance(manifest.get("visual_qa"), dict) else {}, "production_report": manifest.get("production_report") if isinstance(manifest.get("production_report"), dict) else {}, } checks = quality.get("checks") if isinstance(quality.get("checks"), dict) else {} visual_qa = reports["visual_qa"] required_visual_checks = [ "premium_ai_styleframes_used", "styleframes_text_free", "no_large_top_title", "no_unnecessary_lower_third_dim", "no_powerpoint_layout", "renderer_owned_text_listed", ] # Backward-compatible hardening: only enforce the new checklist when a package declares checks, # but always fail explicit blocker fields from visual_qa. blockers: list[str] = [] if checks: for key in required_visual_checks: if checks.get(key) is False: blockers.append(key) if visual_qa.get("large_top_title_detected") is True: checks["no_large_top_title"] = False blockers.append("no_large_top_title") if visual_qa.get("unnecessary_lower_third_dim_detected") is True or visual_qa.get("lower_third_dim_detected") == "unnecessary": checks["no_unnecessary_lower_third_dim"] = False blockers.append("no_unnecessary_lower_third_dim") if visual_qa.get("powerpoint_layout_detected") is True or visual_qa.get("powerpoint_look") is True: checks["no_powerpoint_layout"] = False blockers.append("no_powerpoint_layout") if "renderer_owned_text" in visual_qa and not visual_qa.get("renderer_owned_text"): checks["renderer_owned_text_listed"] = False blockers.append("renderer_owned_text_listed") if blockers: quality = {**quality, "checks": {**checks}, "visual_gate": "failed", "overall": "failed", "blocked_by": sorted(set(blockers))} reports["quality_report"] = {**reports["quality_report"], "checks": {**checks}, "visual_gate": "failed", "overall": "failed", "blocked_by": sorted(set(blockers))} overall = str(quality.get("overall") or reports["quality_report"].get("overall") or "").lower() reports["overall"] = overall reports["gates"] = quality return reports def _status_from_manifest(self, manifest: dict) -> VideoStatus: queue_item_id = manifest.get("queue_item_id") source = manifest.get("source", "prepared_package") quality = self._quality_payload(manifest) status_value = manifest.get("status") or "ready_for_review" if source == "prepared_package" and queue_item_id: if quality.get("overall") != "passed": return VideoStatus.needs_internal_fix try: return VideoStatus(status_value) except Exception: return VideoStatus.ready_for_review def _create_records(self, *, manifest: dict, target_video: Path, meta: dict, folder_name: str | None = None) -> str: package_id = manifest.get("package_id") or manifest.get("candidate_id") if not package_id: raise ValueError("package_id is required") status = self._status_from_manifest(manifest) quality_payload = self._quality_payload(manifest) disclosure = manifest.get("disclosure") or manifest.get("commercial_disclosure") or {} synthetic_media = bool(disclosure.get("synthetic_media", manifest.get("is_aigc", True))) commercial_content = bool(disclosure.get("commercial_content", disclosure.get("promotes_own_brand", False) or disclosure.get("promotes_third_party", False))) with Session(self.engine) as session: existing = session.exec(select(VideoAsset).where(VideoAsset.candidate_id == package_id)).first() if existing: self._write_log_in_session(session, package_id, folder_name, "skipped", existing.id, "duplicate package_id", None) return existing.id duplicate = session.exec(select(VideoAsset).where(VideoAsset.checksum == meta["checksum"])).first() if duplicate: self._write_log_in_session(session, package_id, folder_name, "skipped", duplicate.id, "duplicate checksum", None) return duplicate.id thumb = self.thumbnails / f"{package_id}.jpg" generate_thumbnail(target_video, thumb) video = VideoAsset( candidate_id=package_id, package_id=package_id, source=manifest.get("source", "prepared_package"), file_path=str(target_video), original_file_name=target_video.name, mime_type=meta.get("mime_type"), file_size=meta.get("file_size") or 0, checksum=meta["checksum"], duration_sec=meta.get("duration_sec"), width=meta.get("width"), height=meta.get("height"), fps=meta.get("fps"), aspect_ratio=meta.get("aspect_ratio"), thumbnail_path=str(thumb), transcript=manifest.get("transcript"), script=manifest.get("script"), language=manifest.get("language"), working_title=manifest.get("working_title"), topic_id=manifest.get("topic_id"), series=manifest.get("series"), is_aigc=synthetic_media, commercial_promotes_own_brand=commercial_content, commercial_promotes_third_party=False, quality_score=manifest.get("quality_score"), hook_score=manifest.get("hook_score"), status=status, ) session.add(video) session.commit() session.refresh(video) session.add(ActivityEvent(entity_type="video", entity_id=video.id, event_type="imported", label="Package imported", payload_json={"package_id": package_id, "source": video.source, "quality_overall": quality_payload.get("overall") or None})) if quality_payload.get("quality_report") or quality_payload.get("voice_qa") or quality_payload.get("visual_qa") or quality_payload.get("overall"): session.add(ActivityEvent(entity_type="video", entity_id=video.id, event_type="quality_report_imported", label="Quality report imported", payload_json=quality_payload)) queue_item_id = manifest.get("queue_item_id") raw_source_metadata = manifest.get("source_metadata") source_metadata: dict = dict(raw_source_metadata) if isinstance(raw_source_metadata, dict) else {} source_metadata.update({ "package_id": package_id, "queue_item_id": queue_item_id, "family_id": manifest.get("family_id"), "idea_id": manifest.get("idea_id"), "content_script_id": manifest.get("content_script_id"), "source": manifest.get("source", "prepared_package"), "test_package": bool(manifest.get("test_package", False)), }) youtube_id = _extract_youtube_id(manifest) if youtube_id: video.external_youtube_id = youtube_id session.add(video) session.commit() session.add(PostDraft( video_asset_id=video.id, provider=Provider.tiktok, caption=manifest.get("suggested_tiktok_caption"), hashtags_json=manifest.get("suggested_hashtags") or [], metadata_json={"synthetic_media": synthetic_media, "commercial_content": commercial_content, "requires_consent": True, "source_metadata": source_metadata if isinstance(source_metadata, dict) else {"source": str(source_metadata)}}, source_metadata_json=source_metadata if isinstance(source_metadata, dict) else {"source": str(source_metadata)}, status="draft", )) session.add(PostDraft( video_asset_id=video.id, provider=Provider.youtube, title=manifest.get("suggested_youtube_title") or manifest.get("working_title"), description=manifest.get("suggested_youtube_description"), tags_json=manifest.get("youtube_tags") or manifest.get("suggested_hashtags") or [], privacy_status="private", metadata_json={"containsSyntheticMedia": synthetic_media, "commercialContent": commercial_content, "selfDeclaredMadeForKids": False, "source_metadata": source_metadata if isinstance(source_metadata, dict) else {"source": str(source_metadata)}}, external_post_id=youtube_id, source_metadata_json=source_metadata if isinstance(source_metadata, dict) else {"source": str(source_metadata)}, status="draft", )) if queue_item_id: queue_item = session.get(ProductionQueueItem, queue_item_id) if queue_item: queue_item.status = "package_ready" if status == VideoStatus.ready_for_review else "needs_internal_fix" queue_item.actual_package_id = package_id queue_item.video_asset_id = video.id queue_item.attached_package_path = str(target_video.parent) queue_item.expected_package_manifest = str(target_video.parent / "manifest.json") if status != VideoStatus.ready_for_review: queue_item.error = "Quality gate missing or failed; package not eligible for active review" else: queue_item.error = None session.add(queue_item) event_type = "production_package_imported" if status == VideoStatus.ready_for_review else "production_quality_failed" label = "Production package imported" if status == VideoStatus.ready_for_review else "Production package failed quality gate" session.add(ActivityEvent(entity_type="production_queue", entity_id=queue_item.id, event_type=event_type, label=label, payload_json={"package_id": package_id, "video_asset_id": video.id, "manifest_queue_item_id": queue_item_id, "quality": quality_payload})) session.add(ActivityEvent(entity_type="video", entity_id=video.id, event_type=event_type, label=label, payload_json={"queue_item_id": queue_item.id, "package_id": package_id, "quality": quality_payload})) self._write_log_in_session(session, package_id, folder_name, "imported", video.id, "package imported", None) session.commit() return video.id def _write_log_in_session(self, session: Session, package_id: str | None, folder_name: str | None, status: str, video_asset_id: str | None, message: str | None, error: str | None): session.add(ImportLog(package_id=package_id, folder_name=folder_name, status=status, video_asset_id=video_asset_id, message=message, error=error))