MVP stage 3: media assets (video, atom_assets, publications, idempotent)

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:
+def clear_collection(client: QdrantClient) -> None:
-    if client.collection_exists(COLLECTION):
+    """Полностью очистить коллекцию 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)")
 

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())