from pathlib import Path import hashlib import os import subprocess from sqlmodel import Session, select from app.core.config import get_settings from app.db.session import get_engine, init_db from app.models.core import Idea, ImportLog, JobStatus, PostDraft, Provider, PublishJob, Theme, VideoAsset, VideoStatus def _ensure_demo_video(path: Path): path.parent.mkdir(parents=True, exist_ok=True) if path.exists(): return try: subprocess.run([ "ffmpeg", "-y", "-f", "lavfi", "-i", "color=c=0x0b1020:s=720x1280:d=1", "-vf", "drawtext=text=Prepared Package:fontcolor=white:fontsize=48:x=(w-text_w)/2:y=(h-text_h)/2", "-pix_fmt", "yuv420p", str(path) ], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except Exception: path.write_bytes(b"demo video placeholder") def seed_demo() -> dict: settings=get_settings() init_db() settings.storage_root.mkdir(parents=True, exist_ok=True) video_path=settings.storage_root / "processed" / "package_demo_ready" / "video.mp4" _ensure_demo_video(video_path) checksum=hashlib.sha256(video_path.read_bytes()).hexdigest() with Session(get_engine()) as session: if session.exec(select(Theme)).first(): return {"seeded": False, "message": "demo data already present"} topics=[ Theme(title="Delivery fee traps", description="Shorts about small payment requests and fake delivery notices.", status="planned", priority=90, default_language="de"), Theme(title="Fake support calls", description="How to spot pressure tactics and external payment requests.", status="in_production", priority=75, default_language="de"), Theme(title="Invoice detail changes", description="Business-payment red flags and safer verification steps.", status="reviewed", priority=60, default_language="de"), ] session.add_all(topics); session.commit() for t in topics: session.refresh(t) ideas=[ Idea(topic_id=topics[0].id, title="Tiny customs fee SMS", short_explanation="Use a small-fee hook and show safe app-first verification.", keywords=["delivery","fee","sms"], source_links=["https://example.com/reference-delivery"], priority=95, status="approved", pillar="Everyday Red Flags"), Idea(topic_id=topics[1].id, title="Caller asks for remote access", short_explanation="Explain why support never needs screen-control tools.", keywords=["support","remote access"], priority=80, status="needs_research", pillar="Everyday Red Flags"), Idea(topic_id=topics[2].id, title="Changed bank details", short_explanation="Show callback verification before paying invoices.", keywords=["invoice","bank details"], priority=70, status="planned", pillar="Business Safety"), Idea(title="Gift card pressure from boss", short_explanation="Classic impersonation scenario for office workers.", keywords=["gift cards","boss"], priority=65, status="idea", pillar="Workplace Red Flags"), Idea(title="Subscription renewal popup", short_explanation="Browser popup pushing urgent payment update.", keywords=["popup","subscription"], priority=55, status="idea", pillar="Everyday Red Flags"), ] session.add_all(ideas); session.commit() v1=VideoAsset(candidate_id="package_demo_ready", package_id="package_demo_ready", source="prepared_package", source_label="demo", is_demo=True, import_batch_label="demo", file_path=str(video_path), original_file_name="video.mp4", mime_type="video/mp4", file_size=video_path.stat().st_size, checksum=checksum, working_title="Small delivery fee warning", language="de", topic_id=topics[0].id, series="Everyday Red Flags", is_aigc=True, status=VideoStatus.ready_for_review) v2=VideoAsset(candidate_id="package_demo_needs_metadata", package_id="package_demo_needs_metadata", source="prepared_package", source_label="demo", is_demo=True, import_batch_label="demo", file_path=str(video_path), original_file_name="video.mp4", mime_type="video/mp4", file_size=video_path.stat().st_size, checksum=checksum+"-demo2", working_title="Support call pressure", language="de", topic_id=topics[1].id, series="Everyday Red Flags", is_aigc=True, status=VideoStatus.needs_metadata) v3=VideoAsset(candidate_id="package_demo_scheduled", package_id="package_demo_scheduled", source="prepared_package", source_label="demo", is_demo=True, import_batch_label="demo", file_path=str(video_path), original_file_name="video.mp4", mime_type="video/mp4", file_size=video_path.stat().st_size, checksum=checksum+"-demo3", working_title="Changed invoice details", language="de", topic_id=topics[2].id, series="Business Safety", is_aigc=True, status=VideoStatus.scheduled) session.add_all([v1,v2,v3]); session.commit(); session.refresh(v1); session.refresh(v2); session.refresh(v3) for v in [v1,v2,v3]: session.add(PostDraft(video_asset_id=v.id, provider=Provider.tiktok, caption="Editable TikTok caption #safety", hashtags_json=["#safety","#scamalert"], metadata_json={"synthetic_media": True, "commercial_content": False}, status="scheduled" if v.status==VideoStatus.scheduled else "draft")) session.add(PostDraft(video_asset_id=v.id, provider=Provider.youtube, title=v.working_title, description="Editable YouTube description for private upload review.", tags_json=["safety","shorts"], privacy_status="private", metadata_json={"containsSyntheticMedia": True}, status="scheduled" if v.status==VideoStatus.scheduled else "draft")) session.commit() failed_draft = session.exec(select(PostDraft).where(PostDraft.video_asset_id == v2.id, PostDraft.provider == Provider.youtube)).first() session.add(ImportLog(package_id=v1.package_id, folder_name=v1.package_id, status="imported", video_asset_id=v1.id, message="demo package imported")) session.add(ImportLog(package_id="package_demo_failed", folder_name="package_demo_failed", status="failed", error="manifest missing video.mp4")) if failed_draft: session.add(PublishJob(post_draft_id=failed_draft.id, provider=Provider.youtube, job_type="youtube_private_upload", status=JobStatus.failed, error_code="demo_failure", error_message="Demo failed upload job for UI review")) session.commit() return {"seeded": True, "topics": len(topics), "ideas": len(ideas), "packages": 3} if __name__ == "__main__": print(seed_demo())