from __future__ import annotations import json from pathlib import Path from src.generators.ai_image import ImageGenerator from src.generators.audio import AudioModule from src.generators.image_source import FetchedImage, fetch_news_image from src.generators.sora_video import SoraVideoGenerator from src.generators.video import VideoDirector from src.generators.visuals import VisualRenderer from src.services.ideation import PLATFORMS, ContentIdea class AssetProducer: def __init__( self, renderer: VisualRenderer, *, image_generator: ImageGenerator | None = None, sora_generator: SoraVideoGenerator | None = None, enable_video: bool = True, video_engine: str = "sora", video_max_seconds: int = 12, enable_ai_images: bool = True, enable_audio: bool = False, ): self.renderer = renderer self.image_generator = image_generator self.sora = sora_generator self.enable_video = enable_video self.video_engine = (video_engine or "sora").strip().lower() self.enable_ai_images = enable_ai_images and image_generator is not None self.ffmpeg_video = VideoDirector(max_seconds=video_max_seconds, enable_audio=enable_audio) self.audio = AudioModule(enabled=enable_audio) def produce(self, idea: ContentIdea, out_dir: Path) -> dict: out_dir.mkdir(parents=True, exist_ok=True) meta = { "id": idea.id, "format": idea.format, "topic": idea.topic, "hook": idea.hook, "news_url": idea.news_url, "news_title": idea.news_title, "news_category": idea.news_category, "creative_direction": idea.creative_direction, "layout": idea.layout, "palette": idea.palette, "palette_name": idea.palette_name, "typography": idea.typography, "background_style": idea.background_style, "illustration_style": idea.illustration_style, "why_it_matters": idea.why_it_matters, "key_fact": idea.key_fact, "visual_brief": idea.visual_brief, "source_refs": idea.source_refs, "platforms": {}, "image_source": None, "image_model": None, "video_engine": None, "video_model": None, "audio": self.audio.plan().__dict__, "files": [], } style_kwargs = { "layout": idea.layout, "palette": idea.palette, "typography": idea.typography, "creative_key": f"{idea.id}-{idea.palette_name}-{idea.layout}", } size = (1080, 1350) if idea.aspect == "portrait" else (1080, 1080) if idea.format == "carousel": meta["files"].extend(self._produce_carousel(idea, out_dir, style_kwargs, meta, size)) elif idea.format == "video" and self.enable_video: meta["files"].extend(self._produce_video(idea, out_dir, style_kwargs, meta)) meta["video_script"] = idea.video_script meta["video_style"] = idea.video_style else: hero = self._produce_single(idea, out_dir, style_kwargs, size) meta["image_source"] = hero.source meta["image_model"] = getattr(self.image_generator, "model", None) if self.image_generator else None meta["files"].append(hero.path) for platform in PLATFORMS: post = idea.platform(platform) fname = f"{platform}.txt" (out_dir / fname).write_text(post.as_publish_text(), encoding="utf-8") meta["platforms"][platform] = { "post": post.post, "hashtags": post.hashtags, "file": fname, } meta["files"].append(fname) (out_dir / "copy.json").write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") meta["files"].append("copy.json") return meta def _generate_bg(self, idea: ContentIdea, out_dir: Path, prompt: str | None = None) -> tuple[Path | None, str]: prompt_text = (prompt or idea.image_prompt or idea.visual_brief or idea.hook).strip() if self.enable_ai_images and prompt_text: ai_path = out_dir / f"bg_ai_{abs(hash(prompt_text)) % 10_000_000}.png" try: full = self.image_generator.build_prompt( base_prompt=prompt_text, category=idea.news_category, illustration_style=idea.illustration_style, background_style=idea.background_style, palette=idea.palette, hook=idea.hook, ) self.image_generator.generate(full, ai_path, aspect=idea.aspect or "square") return ai_path, self.image_generator.model except Exception as exc: (out_dir / "ai_image_error.txt").write_text(str(exc), encoding="utf-8") raw = out_dir / "bg_news.jpg" fetched = fetch_news_image(idea.news_url, raw) if fetched: return raw, "news_og" return None, "generated" def _produce_single(self, idea: ContentIdea, out_dir: Path, style_kwargs: dict, size: tuple[int, int]) -> FetchedImage: bg, source = self._generate_bg(idea, out_dir) out = out_dir / "hero.png" body = idea.why_it_matters or idea.key_fact or idea.visual_brief or idea.topic self.renderer.compose_hero( out, hook=idea.hook, body=body[:160], footer=idea.news_source_label(), bg_image=bg, size=size, **style_kwargs, ) return FetchedImage(path=out.name, source=source) def _produce_carousel(self, idea: ContentIdea, out_dir: Path, style_kwargs: dict, meta: dict, size) -> list[str]: slides = idea.slides or [ {"title": idea.hook, "body": idea.why_it_matters, "role": "hook"}, {"title": "What happened", "body": idea.topic, "role": "what"}, {"title": "Why it matters", "body": idea.why_it_matters, "role": "why"}, {"title": "The numbers", "body": idea.key_fact, "role": "stats"}, {"title": "Your move", "body": "Save this for later.", "role": "cta"}, ] shared_bg, source = self._generate_bg(idea, out_dir) meta["image_source"] = source meta["image_model"] = source if source in {"news_og", "generated"} else source alt_bg = None if self.enable_ai_images and len(slides) >= 4: mid = slides[len(slides) // 2] alt_prompt = mid.get("image_prompt") or f"{idea.topic} โ€” visual metaphor, {idea.illustration_style}" alt_bg, _ = self._generate_bg(idea, out_dir, prompt=alt_prompt) from src.services.creative import LAYOUTS files = [] start = LAYOUTS.index(idea.layout) if idea.layout in LAYOUTS else 0 for i, slide in enumerate(slides): path = out_dir / f"slide_{i + 1:02d}.png" layout = LAYOUTS[(start + i) % len(LAYOUTS)] bg = alt_bg if (alt_bg and i >= len(slides) // 2) else shared_bg self.renderer.compose_hero( path, hook=slide.get("title", ""), body=slide.get("body", "")[:180], footer=f"{i + 1}/{len(slides)} ยท {idea.news_source_label()[:36]}", layout=layout, palette=idea.palette, typography=idea.typography, bg_image=bg, creative_key=f"{idea.id}-c{i}-{layout}", size=size, ) files.append(path.name) return files def _produce_video(self, idea: ContentIdea, out_dir: Path, style_kwargs: dict, meta: dict) -> list[str]: """Prefer Sora original footage; optional ffmpeg fallback (not preferred).""" script = idea.video_script or [] video_path = out_dir / "short.mp4" if self.video_engine == "sora" and self.sora is not None: try: prompt = self.sora.build_prompt( hook=idea.hook, topic=idea.topic, why_it_matters=idea.why_it_matters, key_fact=idea.key_fact, creative_direction=idea.creative_direction, video_style=idea.video_style, news_category=idea.news_category, beats=script, ) (out_dir / "sora_prompt.txt").write_text(prompt, encoding="utf-8") # Sora max clip length is typically 12s โ€” pack the full narrative into one cinematic clip result = self.sora.generate( prompt, video_path, seconds="12", aspect="portrait" if idea.aspect != "landscape" else "landscape", ) # No audio mux by design final = self.audio.attach(result["path"], out_dir) meta["video_engine"] = "sora" meta["video_model"] = result.get("model") meta["sora_video_id"] = result.get("video_id") meta["image_source"] = "sora_native" return [final.name, "sora_prompt.txt"] except Exception as exc: (out_dir / "sora_error.txt").write_text(str(exc), encoding="utf-8") meta["sora_error"] = str(exc)[:500] # fall through to ffmpeg only if explicitly allowed if self.video_engine == "sora": # Keep failing clearly for Sora-primary mode unless fallback set via engine=sora_with_fallback raise return self._produce_video_ffmpeg_fallback(idea, out_dir, style_kwargs, meta) def _produce_video_ffmpeg_fallback(self, idea, out_dir, style_kwargs, meta) -> list[str]: script = idea.video_script or [ {"beat": "hook", "on_screen": idea.hook, "voiceover": idea.why_it_matters, "visual": idea.image_prompt, "motion": "punch_in", "seconds": 2.5}, {"beat": "details", "on_screen": idea.key_fact or "Details", "voiceover": idea.topic, "visual": "", "motion": "zoom_in", "seconds": 5}, {"beat": "cta", "on_screen": "Follow for daily AI drops", "voiceover": "", "visual": "", "motion": "zoom_out", "seconds": 3}, ] frames_dir = out_dir / "frames" frames_dir.mkdir(exist_ok=True) shared_bg, source = self._generate_bg(idea, out_dir) meta["image_source"] = source from src.services.creative import LAYOUTS frame_paths, captions, motions, durations = [], [], [], [] start = LAYOUTS.index(idea.layout) if idea.layout in LAYOUTS else 0 for i, beat in enumerate(script): frame = frames_dir / f"frame_{i + 1:02d}.png" self.renderer.compose_hero( frame, hook=beat.get("on_screen") or idea.hook, body=(beat.get("voiceover") or "")[:120], footer=idea.news_source_label(), layout=LAYOUTS[(start + i * 2) % len(LAYOUTS)], palette=idea.palette, typography=idea.typography, bg_image=shared_bg, creative_key=f"{idea.id}-v{i}", ) frame_paths.append(frame) captions.append((beat.get("on_screen") or "")[:70]) motions.append(beat.get("motion") or "zoom_in") durations.append(float(beat.get("seconds") or 4)) video_path = out_dir / "short.mp4" # Build without relying on audio emphasis โ€” AudioModule stays off self.ffmpeg_video.build( frame_paths, captions, video_path, motion_style=idea.video_style or "cinematic", motions=motions, durations=durations, ) meta["video_engine"] = "ffmpeg_fallback" meta["video_model"] = None return [video_path.name]