Refs/heads/main

shorts_pipeline/config.yaml

@@ -0,0 +1,59 @@
+app:
+  timezone: "UTC"
+  log_file: "pipeline.log"
+  db_path: "shorts_pipeline.db"
+  gpu_device_id: 0
+
+ai:
+  provider: "openai" # openai or anthropic
+  openai_api_key: ""
+  anthropic_api_key: ""
+  model: "gpt-4o"
+  ideas_per_batch: 20
+
+niches:
+  active: ["curiosidades", "psicologia", "did_you_know", "ciencia_simples", "fatos_chocantes"]
+  prompt_templates:
+    curiosidades: "Generate viral curiosities facts with surprising twists for Brazilian audience."
+    psicologia: "Generate practical psychology micro-insights with emotional hooks."
+    did_you_know: "Generate did-you-know style surprising facts with high curiosity loops."
+    ciencia_simples: "Generate simple science explanations with clear metaphors and wow moments."
+    fatos_chocantes: "Generate shocking but safe facts that are true and verifiable."
+
+voice:
+  elevenlabs_api_key: ""
+  elevenlabs_voice_id: "EXAVITQu4vr4xnSDxMaL"
+  elevenlabs_model_id: "eleven_multilingual_v2"
+  concurrency: 10
+  fallback_engine: "coqui" # coqui or piper
+  piper_model_path: ""
+
+video:
+  pexels_api_key: ""
+  pixabay_api_key: ""
+  width: 1080
+  height: 1920
+  min_cut_seconds: 2
+  max_cut_seconds: 4
+  fps: 30
+  subtitle_font: "assets/fonts/Inter-Bold.ttf"
+  subtitle_size: 68
+  subtitle_color: "#FFFFFF"
+  branding_intro: "assets/branding/intro.mp4"
+  branding_outro: "assets/branding/outro.mp4"
+  logo_path: "assets/branding/logo.png"
+  primary_color: "#4A83DD"
+  nvenc_preset: "p4"
+
+upload:
+  youtube_client_secrets_file: "client_secret.json"
+  youtube_token_file: "youtube_token.json"
+  channel_default_privacy: "private"
+  publish_hours_utc: [10, 14, 18, 21]
+  max_videos_per_day: 4
+  default_language: "pt"
+
+analytics:
+  min_age_hours: 48
+  report_output_json: "output/weekly_report.json"
+  report_output_txt: "output/weekly_report.txt"

shorts_pipeline/database/db.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+from pathlib import Path
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+BASE_DIR = Path(__file__).resolve().parents[1]
+
+
+def build_engine(db_path: str):
+    uri = f"sqlite:///{BASE_DIR / db_path}"
+    return create_engine(uri, connect_args={"check_same_thread": False}, future=True)
+
+
+def build_session_factory(engine):
+    return sessionmaker(bind=engine, expire_on_commit=False)
+
+
+@contextmanager
+def session_scope(session_factory):
+    session = session_factory()
+    try:
+        yield session
+        session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()

shorts_pipeline/database/models.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, Text
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class Script(Base):
+    __tablename__ = "scripts"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    niche: Mapped[str] = mapped_column(String(128), index=True)
+    idea: Mapped[str] = mapped_column(Text)
+    hook: Mapped[str] = mapped_column(Text)
+    content: Mapped[str] = mapped_column(Text)
+    retention_hook: Mapped[str] = mapped_column(Text)
+    status: Mapped[str] = mapped_column(String(32), default="pending", index=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+
+    video: Mapped["Video"] = relationship(back_populates="script", uselist=False)
+
+
+class Video(Base):
+    __tablename__ = "videos"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    script_id: Mapped[int] = mapped_column(ForeignKey("scripts.id"), unique=True)
+    audio_path: Mapped[str | None] = mapped_column(Text, nullable=True)
+    video_path: Mapped[str | None] = mapped_column(Text, nullable=True)
+    thumbnail_path: Mapped[str | None] = mapped_column(Text, nullable=True)
+    youtube_video_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    title: Mapped[str | None] = mapped_column(Text, nullable=True)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    tags: Mapped[str | None] = mapped_column(Text, nullable=True)
+    scheduled_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    uploaded_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    status: Mapped[str] = mapped_column(String(32), default="pending", index=True)
+
+    script: Mapped[Script] = relationship(back_populates="video")
+    analytics: Mapped[list["Analytics"]] = relationship(back_populates="video")
+
+
+class Analytics(Base):
+    __tablename__ = "analytics"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    video_id: Mapped[int] = mapped_column(ForeignKey("videos.id"), index=True)
+    captured_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    views: Mapped[int] = mapped_column(Integer, default=0)
+    ctr: Mapped[float] = mapped_column(Float, default=0.0)
+    avg_view_duration: Mapped[float] = mapped_column(Float, default=0.0)
+    retention_rate: Mapped[float] = mapped_column(Float, default=0.0)
+
+    video: Mapped[Video] = relationship(back_populates="analytics")

shorts_pipeline/main.py

@@ -0,0 +1,80 @@
+"""
+README / Setup
+1) Install Python 3.11+, FFmpeg with NVENC support, CUDA drivers (NVIDIA RTX), and Tesseract optional.
+2) pip install -r requirements.txt
+3) Fill config.yaml keys: OpenAI/Anthropic, ElevenLabs, Pexels/Pixabay, YouTube OAuth files.
+4) Place branding assets under assets/branding and font under assets/fonts.
+5) Run once: python main.py (it bootstraps DB and schedules jobs).
+6) Keep this process alive (systemd/supervisor/docker). It runs autonomous loops.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from pathlib import Path
+
+import yaml
+from apscheduler.schedulers.blocking import BlockingScheduler
+
+from database.db import build_engine, build_session_factory, session_scope
+from database.models import Base
+from modules.idea_generator import IdeaGenerator
+from modules.uploader import Uploader
+from modules.video_assembler import VideoAssembler
+from modules.voice_synthesizer import VoiceSynthesizer
+
+
+def configure_logging(log_file: str):
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
+        handlers=[logging.FileHandler(log_file), logging.StreamHandler()],
+    )
+
+
+def load_config():
+    with open("config.yaml", "r", encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+
+def run_pipeline(cfg: dict, sf):
+    logger = logging.getLogger("pipeline")
+    with session_scope(sf) as session:
+        try:
+            IdeaGenerator(cfg).generate_and_store(session)
+        except Exception:
+            logger.exception("Idea generation step failed")
+    with session_scope(sf) as session:
+        try:
+            asyncio.run(VoiceSynthesizer(cfg).synthesize_pending(session))
+        except Exception:
+            logger.exception("Voice step failed")
+    with session_scope(sf) as session:
+        try:
+            VideoAssembler(cfg).assemble_ready(session)
+        except Exception:
+            logger.exception("Video assembly step failed")
+    with session_scope(sf) as session:
+        try:
+            Uploader(cfg).upload_due(session)
+        except Exception:
+            logger.exception("Upload step failed")
+
+
+def main():
+    Path("output").mkdir(exist_ok=True)
+    cfg = load_config()
+    configure_logging(cfg["app"]["log_file"])
+    engine = build_engine(cfg["app"]["db_path"])
+    Base.metadata.create_all(engine)
+    sf = build_session_factory(engine)
+
+    scheduler = BlockingScheduler(timezone=cfg["app"]["timezone"])
+    scheduler.add_job(lambda: run_pipeline(cfg, sf), "interval", hours=6, id="full_pipeline", max_instances=1, coalesce=True)
+    run_pipeline(cfg, sf)
+    scheduler.start()
+
+
+if __name__ == "__main__":
+    main()

shorts_pipeline/modules/analytics.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timedelta
+
+from googleapiclient.discovery import build
+
+from database.models import Analytics, Script, Video
+
+LOGGER = logging.getLogger(__name__)
+
+
+class AnalyticsModule:
+    def __init__(self, config: dict, creds):
+        self.cfg = config
+        self.api = build("youtubeAnalytics", "v2", credentials=creds)
+
+    def run(self, session):
+        cutoff = datetime.utcnow() - timedelta(hours=self.cfg["analytics"]["min_age_hours"])
+        rows = session.query(Video, Script).join(Script, Video.script_id == Script.id).filter(Video.status == "uploaded", Video.uploaded_at < cutoff).all()
+        scored = []
+        for video, script in rows:
+            try:
+                # Simplified demo query; dimensions and metrics can be expanded.
+                resp = self.api.reports().query(ids="channel==MINE", startDate=(video.uploaded_at.date()).isoformat(), endDate=datetime.utcnow().date().isoformat(), metrics="views,averageViewDuration,averageViewPercentage", filters=f"video=={video.youtube_video_id}").execute()
+                vals = resp.get("rows", [[0, 0.0, 0.0]])[0]
+                metric = Analytics(video_id=video.id, views=int(vals[0]), avg_view_duration=float(vals[1]), retention_rate=float(vals[2]), ctr=0.0)
+                session.add(metric)
+                video.status = "analyzed"
+                script.status = "analyzed"
+                scored.append({"script_id": script.id, "niche": script.niche, "hook": script.hook, "views": metric.views, "retention_rate": metric.retention_rate})
+            except Exception as exc:
+                LOGGER.exception("Analytics failed for video=%s err=%s", video.youtube_video_id, exc)
+        scored.sort(key=lambda x: (x["views"], x["retention_rate"]), reverse=True)
+        self._write_report(scored)
+
+    def _write_report(self, scored: list[dict]):
+        with open(self.cfg["analytics"]["report_output_json"], "w", encoding="utf-8") as f:
+            json.dump({"generated_at": datetime.utcnow().isoformat(), "top_videos": scored[:20]}, f, ensure_ascii=False, indent=2)
+        with open(self.cfg["analytics"]["report_output_txt"], "w", encoding="utf-8") as f:
+            f.write("Weekly Shorts Performance Report\n\n")
+            for i, row in enumerate(scored[:20], 1):
+                f.write(f"{i}. Script {row['script_id']} | {row['niche']} | Views={row['views']} | Retention={row['retention_rate']:.2f}\n")

shorts_pipeline/modules/idea_generator.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from anthropic import Anthropic
+from openai import OpenAI
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+from database.models import Script
+
+LOGGER = logging.getLogger(__name__)
+
+
+class IdeaGenerator:
+    def __init__(self, config: dict[str, Any]):
+        self.config = config
+        self.ai_cfg = config["ai"]
+
+    @retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=30))
+    def _llm_call(self, prompt: str) -> str:
+        if self.ai_cfg["provider"] == "anthropic":
+            client = Anthropic(api_key=self.ai_cfg["anthropic_api_key"])
+            msg = client.messages.create(model="claude-3-7-sonnet-latest", max_tokens=4000, messages=[{"role": "user", "content": prompt}])
+            return msg.content[0].text
+        client = OpenAI(api_key=self.ai_cfg["openai_api_key"])
+        resp = client.responses.create(model=self.ai_cfg["model"], input=prompt)
+        return resp.output_text
+
+    def generate_and_store(self, session):
+        for niche in self.config["niches"]["active"]:
+            template = self.config["niches"]["prompt_templates"][niche]
+            prompt = (
+                f"{template}\nGenerate {self.ai_cfg['ideas_per_batch']} short video scripts in JSON array. "
+                "Each object keys: idea, hook, content, retention_hook. Keep content <= 25 seconds speech."
+            )
+            try:
+                raw = self._llm_call(prompt)
+                scripts = json.loads(raw[raw.find("[") : raw.rfind("]") + 1])
+            except Exception as exc:
+                LOGGER.exception("Failed generating scripts for niche=%s: %s", niche, exc)
+                continue
+            for row in scripts:
+                session.add(
+                    Script(
+                        niche=niche,
+                        idea=row["idea"],
+                        hook=row["hook"],
+                        content=row["content"],
+                        retention_hook=row["retention_hook"],
+                        status="pending",
+                    )
+                )
+            LOGGER.info("Stored %s scripts for niche=%s", len(scripts), niche)

shorts_pipeline/modules/thumbnail_generator.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from PIL import Image, ImageDraw, ImageFont
+
+
+def generate_thumbnail(text: str, out_path: str, font_path: str, color: str):
+    img = Image.new("RGB", (1080, 1920), color="#111111")
+    draw = ImageDraw.Draw(img)
+    font = ImageFont.truetype(font_path, 96)
+    draw.rounded_rectangle((60, 640, 1020, 1280), radius=35, fill=color)
+    draw.text((100, 760), text[:70], fill="white", font=font)
+    Path(out_path).parent.mkdir(parents=True, exist_ok=True)
+    img.save(out_path, "PNG")