diff --git a/stage1/extract_to_postgres.py b/stage1/extract_to_postgres.py index 793880c..6db80af 100644 --- a/stage1/extract_to_postgres.py +++ b/stage1/extract_to_postgres.py @@ -188,9 +188,10 @@ def ensure_project(conn, project: str) -> str: def insert_atoms(conn, project_id: str, atoms: list[Atom]) -> int: - """Вставить атомы в atoms. Возвращает количество.""" + """Вставить атомы в atoms. Дубли проекта — удалить перед вставкой. Возвращает количество.""" n = 0 with conn.cursor() as cur: + cur.execute("DELETE FROM atoms WHERE project_id = %s", (project_id,)) for atom in atoms: row = atom.to_row() cur.execute(""" @@ -242,9 +243,13 @@ def generate_web_version(atom: Atom) -> str: # --------------------------------------------------------------------------- # Qdrant # --------------------------------------------------------------------------- -def ensure_collection(client: QdrantClient) -> None: - if client.collection_exists(COLLECTION): +def clear_collection(client: QdrantClient) -> None: + """Полностью очистить коллекцию Qdrant (идемпотентность stage 1).""" + import os + if os.environ.get("QDRANT_KEEP"): return + if client.collection_exists(COLLECTION): + client.delete_collection(collection_name=COLLECTION) client.create_collection( collection_name=COLLECTION, vectors_config=VectorParams(size=DIM, distance=Distance.COSINE), @@ -305,7 +310,7 @@ def main(): # 3. Qdrant (эмбеддинги + дедуп) start = time.time() client = QdrantClient(host="127.0.0.1", port=6333) - ensure_collection(client) + clear_collection(client) n_emb = upsert_qdrant(client, all_atoms) print(f" ✓ Qdrant: {n_emb} эмбеддингов за {time.time()-start:.1f}с (~{ (time.time()-start)/n_emb*1000:.0f}ms/atom)") diff --git a/stage3/media_assets.py b/stage3/media_assets.py new file mode 100644 index 0000000..f1de003 --- /dev/null +++ b/stage3/media_assets.py @@ -0,0 +1,260 @@ +#!/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 1 FROM publications WHERE composition_id = %s LIMIT 1", + (comp_id,), + ) + if cur.fetchone(): + return { + "views": 1200, + "ctr": 0.045, + "clicks": 54, + "conversions": 4, + "revenue": 600.0, + } + views = 1200 + ctr = 0.045 + clicks = int(views * ctr) + conv = int(clicks * 0.08) + revenue = round(conv * 150.0, 2) # $150 / conversion + + 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": views, + "ctr": ctr, + "clicks": clicks, + "conversions": conv, + "revenue": revenue, + } + ), + ), + ) + conn.commit() + + return { + "views": views, + "clicks": clicks, + "conversions": conv, + "revenue": revenue, + } + + +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"]) + + print( + f" • {proj['slug']}: {asset['atoms']} атомов, " + f"{asset['duration_sec']:.0f}с видео, " + f"{pub['views']} views → {pub['conversions']} conv, " + f"${pub['revenue']}" + ) + total_revenue += 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())