from __future__ import annotations import json from datetime import datetime from pathlib import Path from src.config import Settings from src.generators.ai_image import ImageGenerator from src.generators.assets import AssetProducer from src.generators.visuals import VisualRenderer from src.notify.telegram import TelegramNotifier from src.services.llm import create_llm from src.services.creative import CreativeDirector from src.services.ideation import generate_ideas from src.services.knowledge import KnowledgeService from src.services.news import NewsService from src.storage.drive import DriveUploader from src.util.logging_setup import setup_logging from src.util.text import slugify class ContentPipeline: def __init__(self, settings: Settings): self.s = settings self.log = setup_logging(settings.storage_logs) self.llm = create_llm( provider=settings.llm_provider, anthropic_key=settings.anthropic_api_key, claude_model=settings.claude_model, openai_key=settings.openai_api_key, openai_model=settings.openai_model, ) self.log.info("LLM provider: %s", settings.llm_provider) self.news = NewsService() self.knowledge = KnowledgeService(settings.data_profiles, settings.data_erp) self.creative = CreativeDirector( Path(__file__).resolve().parents[1] / "storage" / "creative_history.json" ) self.renderer = VisualRenderer( brand_name=settings.brand_name, primary=settings.brand_primary, accent=settings.brand_accent, bg=settings.brand_bg, fonts_dir=Path(__file__).resolve().parents[1] / "data" / "fonts", ) img_gen = None if settings.enable_ai_images: if settings.openai_api_key: img_gen = ImageGenerator(settings.openai_api_key, model=settings.openai_image_model) else: self.log.warning("ENABLE_AI_IMAGES=true but OPENAI_API_KEY missing — using template/news visuals") sora_gen = None if settings.enable_video and settings.video_engine.startswith("sora") and settings.openai_api_key: from src.generators.sora_video import SoraVideoGenerator sora_gen = SoraVideoGenerator( settings.openai_api_key, model=settings.openai_video_model, default_seconds="12", ) self.producer = AssetProducer( self.renderer, image_generator=img_gen, sora_generator=sora_gen, enable_video=settings.enable_video, video_engine=settings.video_engine, video_max_seconds=min(12, max(4, settings.video_max_seconds)), enable_ai_images=settings.enable_ai_images, enable_audio=settings.enable_audio, ) self.log.info( "Assets: image_model=%s video_engine=%s video_model=%s audio=%s", settings.openai_image_model, settings.video_engine, settings.openai_video_model, settings.enable_audio, ) def run(self) -> dict: ts = datetime.now().strftime("%Y%m%d_%H%M%S") batch_dir = self.s.output_dir / f"batch_{ts}" batch_dir.mkdir(parents=True, exist_ok=True) self.log.info("Starting batch %s (dry_run=%s)", ts, self.s.dry_run) news_items = self.news.fetch(max_items=28) self.log.info( "Fetched %d AI/IT news items (top: %s)", len(news_items), news_items[0].title[:80] if news_items else "none", ) profiles = self.knowledge.profiles_text() briefs = self.creative.assign_batch(self.s.batch_size) self.log.info( "Creative briefs: %s", ", ".join(f"{b.layout}/{b.palette_name}" for b in briefs), ) ideas = generate_ideas( self.llm, brand_name=self.s.brand_name, profiles=profiles, news_block=self.news.as_prompt_block(news_items), batch_size=self.s.batch_size, enable_video=self.s.enable_video, creative_briefs=briefs, recent_styles=self.creative.recent_summary(), ) self.log.info("Generated %d ideas", len(ideas)) self.creative.record(briefs) results = [] for idea in ideas: folder = batch_dir / f"{idea.id}_{slugify(idea.topic or idea.hook)}" self.log.info("Producing %s (%s)", idea.id, idea.format) meta = self.producer.produce(idea, folder) results.append(meta) manifest = { "batch_id": ts, "created_at": datetime.now().isoformat(timespec="seconds"), "brand": self.s.brand_name, "ideas": results, "news_count": len(news_items), } (batch_dir / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8" ) drive_info = None if not self.s.dry_run and self.s.google_drive_folder_id: self.log.info("Uploading to Google Drive…") uploader = DriveUploader( self.s.google_drive_folder_id, oauth_client_secrets=self.s.google_oauth_client_secrets, oauth_token_file=self.s.google_oauth_token_file, service_account_file=self.s.google_service_account_file if self.s.google_use_shared_drive else None, use_shared_drive=self.s.google_use_shared_drive, ) # ts = YYYYMMDD_HHMMSS → date folder YYYY-MM-DD, batch folder batch_HHMMSS date_folder = f"{ts[:4]}-{ts[4:6]}-{ts[6:8]}" batch_folder = f"batch_{ts[9:]}" drive_info = uploader.upload_tree( batch_dir, date_folder=date_folder, batch_folder=batch_folder, ) manifest["drive"] = { "folder_id": drive_info["folder_id"], "folder_url": drive_info["folder_url"], "date_folder": drive_info["date_folder"], "date_folder_url": drive_info["date_folder_url"], "batch_folder": drive_info["batch_folder"], } (batch_dir / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8" ) elif not self.s.dry_run: self.log.warning("Skipping Drive upload (GOOGLE_DRIVE_FOLDER_ID not set)") if not self.s.dry_run and self.s.telegram_bot_token: self.log.info("Sending Telegram notification…") notifier = TelegramNotifier(self.s.telegram_bot_token, self.s.telegram_chat_id) notifier.send(self._telegram_message(manifest, drive_info)) elif not self.s.dry_run: self.log.warning("Skipping Telegram (credentials missing)") self.log.info("Batch complete: %s", batch_dir) return manifest def _telegram_message(self, manifest: dict, drive_info: dict | None) -> str: lines = [ f"✅ Content batch ready — {self.s.brand_name}", f"Batch: {manifest['batch_id']}", f"Items: {len(manifest.get('ideas', []))}", "", ] for idea in manifest.get("ideas", []): hook = idea.get("hook") or idea.get("topic") lines.append(f"• [{idea.get('format')}] {hook}") plats = idea.get("platforms") or {} if plats: lines.append(f" 📱 LinkedIn · IG · X · Threads · Facebook") if drive_info: lines += [ "", f"📁 {drive_info.get('date_folder')}/{drive_info.get('batch_folder')}", f"Drive: {drive_info['folder_url']}", ] else: lines += ["", f"Local: storage/runs/batch_{manifest['batch_id']}"] return "\n".join(lines)