#!/usr/bin/env python3 """ MVP stage 3: media assets — медиа-активы + привязка атомов + публикации. Сборка: atom (stat/claim) → video asset → atom_assets (таймкоды) → compositions (video) → publications (YouTube) + metrics. Медиа-активы генерируются как видео-сценарий на основе atom_versions под youtube_script: каждый stat/claim → свой отрезок видео (таймкод). Запуск: python3 stage3/media_assets.py --project rabota-ufa python3 stage3/media_assets.py --all """ import argparse import json import os import sys import uuid import psycopg2 POSTGRES = { "host": os.environ.get("POSTGRES_HOST", "127.0.0.1"), "port": int(os.environ.get("POSTGRES_PORT", "5434")), "user": os.environ.get("POSTGRES_USER", "atom"), "dbname": os.environ.get("POSTGRES_DB", "content_atom"), "password": os.environ.get("POSTGRES_PASSWORD", "atompass123"), } def get_conn(): return psycopg2.connect(**POSTGRES) def get_projects(conn) -> list[dict]: with conn.cursor() as cur: cur.execute("SELECT id, slug FROM projects") cols = [d[0] for d in cur.description] return [dict(zip(cols, row)) for row in cur.fetchall()] def get_media_atoms(conn, project_id: str) -> list[dict]: """Атомы, подходящие для видео (stat/claim/qa).""" with conn.cursor() as cur: cur.execute( """ SELECT id, type, COALESCE(body->>'statement', body->>'question', plain_text) FROM atoms WHERE project_id=%s AND type IN ('stat', 'claim', 'qa') ORDER BY type """, (project_id,), ) cols = [d[0] for d in cur.description] return [dict(zip(cols, row)) for row in cur.fetchall()] def create_video_asset(conn, project_id: str, project_slug: str, atoms: list[dict]) -> dict | None: """Создаёт video asset + atom_assets (таймкоды) + video composition.""" if not atoms: return None asset_id = None total_sec = 0 # Чистим старые медиа для проекта (идемпотентность stage 3) with conn.cursor() as cur: cur.execute("DELETE FROM atom_assets WHERE asset_id IN (SELECT id FROM assets WHERE project_id = %s)", (project_id,)) cur.execute("DELETE FROM assets WHERE project_id = %s", (project_id,)) conn.commit() # Запись актива (video) — ДО atom_assets (FK) for atom in atoms: total_sec += 12 if atom["type"] in ("stat", "claim") else 8 # Запись актива (video) — ДО atom_assets (FK) with conn.cursor() as cur: cur.execute( """ INSERT INTO assets (project_id, asset_type, format, uri, title, duration_sec, meta) SELECT %s, 'video', 'mp4', %s, %s, %s, %s WHERE NOT EXISTS ( SELECT 1 FROM assets WHERE project_id = %s AND asset_type = 'video' ) RETURNING id """, ( project_id, "https://cdn.content-atom.local/" + project_slug + "-video.mp4", f"{project_slug} — видео (media stage 3)", total_sec, json.dumps({"channel": "youtube", "atoms": len(atoms)}), project_id, ), ) row = cur.fetchone() if row: asset_id = str(row[0]) conn.commit() # Привязываем атомы к активу по таймкодам cursor = 0 for atom in atoms: start = cursor dur = 12 if atom["type"] in ("stat", "claim") else 8 end = start + dur cursor = end with conn.cursor() as cur: cur.execute( """ INSERT INTO atom_assets (atom_id, asset_id, start_time_sec, end_time_sec, role) VALUES (%s, %s, %s, %s, 'narration') ON CONFLICT (atom_id, asset_id, start_time_sec) DO NOTHING """, (atom["id"], asset_id, start, end), ) conn.commit() # composition: video (сборка из atom_ids) atom_ids = [a["id"] for a in atoms] recipe = { "channel": "youtube", "format": "video", "per_atom_seconds": 12, "atoms": [{"id": a["id"], "type": a["type"]} for a in atoms], } with conn.cursor() as cur: cur.execute( """ INSERT INTO compositions (project_id, channel, slug, title, status, recipe, atom_ids) VALUES (%s, 'youtube', %s, %s, 'ready', %s, %s::uuid[]) ON CONFLICT (project_id, channel, slug) DO UPDATE SET recipe = EXCLUDED.recipe, atom_ids = EXCLUDED.atom_ids, status = 'ready' """, ( project_id, project_slug + "-video", f"{project_slug} — видео (media stage 3)", json.dumps(recipe), atom_ids, ), ) conn.commit() return {"asset": asset_id, "duration_sec": total_sec, "atoms": len(atoms)} def create_publication(conn, project_id: str, project_slug: str) -> dict: """Создаёт публикацию на YouTube с метриками (демо-данные).""" with conn.cursor() as cur: cur.execute( """ INSERT INTO compositions (project_id, channel, slug, title, status, recipe, atom_ids) VALUES (%s, 'youtube', %s, %s, 'published', '{}', '{}') ON CONFLICT (project_id, channel, slug) DO UPDATE SET status = 'published' RETURNING id """, ( project_id, project_slug + "-video", f"{project_slug} — видео", ), ) comp_id = cur.fetchone()[0] # Проверить, есть ли уже публикация (идемпотентность) cur.execute( "SELECT id FROM publications WHERE composition_id = %s AND target LIKE %s LIMIT 1", (comp_id, "%video%"), ) row = cur.fetchone() if row: pub_id = row[0] cur.execute( """ UPDATE publications SET metrics = %s WHERE id = %s """, ( json.dumps( { "views": 1200, "ctr": 0.045, "clicks": 54, "conversions": 4, "revenue": 600.0, } ), pub_id, ), ) else: cur.execute( """ INSERT INTO publications (composition_id, target, external_id, published_at, metrics) VALUES (%s, %s, %s, now(), %s) """, ( comp_id, "youtube.com/watch?v=" + project_slug + "-video", "vid-" + project_slug, json.dumps( { "views": 1200, "ctr": 0.045, "clicks": 54, "conversions": 4, "revenue": 600.0, } ), ), ) conn.commit() return { "views": 1200, "clicks": 54, "conversions": 4, "revenue": 600.0, } def create_podcast_asset(conn, project_id: str, project_slug: str, atoms: list[dict]) -> dict | None: """Создаёт audio asset (подкаст) + atom_assets (таймкоды) + podcast composition.""" if not atoms: return None asset_id = None total_sec = 0 # Чистим старые audio-активы для проекта (идемпотентность) with conn.cursor() as cur: cur.execute( "DELETE FROM atom_assets WHERE asset_id IN (SELECT id FROM assets WHERE project_id = %s AND asset_type = 'audio')", (project_id,), ) cur.execute("DELETE FROM assets WHERE project_id = %s AND asset_type = 'audio'", (project_id,)) conn.commit() # Длительность эпизода — ~10 сек на атом (речевой формат) for atom in atoms: total_sec += 10 # Запись audio актива (подкаст) with conn.cursor() as cur: cur.execute( """ INSERT INTO assets (project_id, asset_type, format, uri, title, duration_sec, meta) SELECT %s, 'audio', 'mp3', %s, %s, %s, %s WHERE NOT EXISTS ( SELECT 1 FROM assets WHERE project_id = %s AND asset_type = 'audio' ) RETURNING id """, ( project_id, "https://cdn.content-atom.local/" + project_slug + "-podcast.mp3", f"{project_slug} — подкаст (media stage 3)", total_sec, json.dumps({"channel": "podcast", "atoms": len(atoms)}), project_id, ), ) row = cur.fetchone() if row: asset_id = str(row[0]) conn.commit() # Привязываем атомы к эпизоду по таймкодам cursor = 0 for atom in atoms: start = cursor dur = 10 end = start + dur cursor = end with conn.cursor() as cur: cur.execute( """ INSERT INTO atom_assets (atom_id, asset_id, start_time_sec, end_time_sec, role) VALUES (%s, %s, %s, %s, 'narration') ON CONFLICT (atom_id, asset_id, start_time_sec) DO NOTHING """, (atom["id"], asset_id, start, end), ) conn.commit() # composition: podcast atom_ids = [a["id"] for a in atoms] recipe = { "channel": "podcast", "format": "audio", "per_atom_seconds": 10, "atoms": [{"id": a["id"], "type": a["type"]} for a in atoms], } with conn.cursor() as cur: cur.execute( """ INSERT INTO compositions (project_id, channel, slug, title, status, recipe, atom_ids) VALUES (%s, 'podcast', %s, %s, 'ready', %s, %s::uuid[]) ON CONFLICT (project_id, channel, slug) DO UPDATE SET recipe = EXCLUDED.recipe, atom_ids = EXCLUDED.atom_ids, status = 'ready' """, ( project_id, project_slug + "-podcast", f"{project_slug} — подкаст (media stage 3)", json.dumps(recipe), atom_ids, ), ) conn.commit() return {"asset": asset_id, "duration_sec": total_sec, "atoms": len(atoms)} def create_podcast_publication(conn, project_id: str, project_slug: str) -> dict: """Создаёт публикацию подкаста (RSS/feed) с метриками.""" with conn.cursor() as cur: cur.execute( """ INSERT INTO compositions (project_id, channel, slug, title, status, recipe, atom_ids) VALUES (%s, 'podcast', %s, %s, 'published', '{}', '{}') ON CONFLICT (project_id, channel, slug) DO UPDATE SET status = 'published' RETURNING id """, ( project_id, project_slug + "-podcast", f"{project_slug} — подкаст", ), ) comp_id = cur.fetchone()[0] # Проверить, есть ли уже публикация (идемпотентность) cur.execute( "SELECT id FROM publications WHERE composition_id = %s AND target LIKE %s LIMIT 1", (comp_id, "%podcast%"), ) row = cur.fetchone() if row: pub_id = row[0] cur.execute( """ UPDATE publications SET metrics = %s WHERE id = %s """, ( json.dumps( { "downloads": 320, "subscribers": 24, "listen_rate": 0.78, "conversions": 3, "revenue": 450.0, } ), pub_id, ), ) else: cur.execute( """ INSERT INTO publications (composition_id, target, external_id, published_at, metrics) VALUES (%s, %s, %s, now(), %s) """, ( comp_id, "https://feeds.content-atom.local/" + project_slug + "-podcast", "pod-" + project_slug, json.dumps( { "downloads": 320, "subscribers": 24, "listen_rate": 0.78, "conversions": 3, "revenue": 450.0, } ), ), ) conn.commit() return { "downloads": 320, "subscribers": 24, "listen_rate": 0.78, "conversions": 3, "revenue": 450.0, } def main(): ap = argparse.ArgumentParser(description="MVP stage 3: media assets") ap.add_argument("--project", default=None, help="project slug") ap.add_argument("--all", dest="all", action="store_true", help="все проекты") args = ap.parse_args() conn = get_conn() projects = get_projects(conn) total_revenue = 0.0 for proj in projects: if args.project and proj["slug"] != args.project: continue atoms = get_media_atoms(conn, proj["id"]) asset = create_video_asset(conn, proj["id"], proj["slug"], atoms) pub = create_publication(conn, proj["id"], proj["slug"]) pod_asset = create_podcast_asset(conn, proj["id"], proj["slug"], atoms) pod_pub = create_podcast_publication(conn, proj["id"], proj["slug"]) print( f" • {proj['slug']}: {asset['atoms']} атомов, " f"{asset['duration_sec']:.0f}с видео, {pod_asset['duration_sec']:.0f}с подкаст, " f"{pub['views']} views → {pub['conversions']} conv, " f"{pod_pub['downloads']} downloads подкаст, " f"${pub['revenue'] + pod_pub['revenue']}" ) total_revenue += pub["revenue"] + pod_pub["revenue"] conn.close() print( f"\nMVP stage 3 media assets OK: " f"{len(projects)} проектов, ${total_revenue:.2f} revenue" ) return 0 if __name__ == "__main__": import argparse sys.exit(main())