stage3: третий канал — рассылка (newsletter/email) + idempotent

stage3/media_assets.py

@@ -401,6 +401,177 @@ def create_podcast_publication(conn, project_id: str, project_slug: str) -> dict
     }
 
 
+def create_newsletter_asset(conn, project_id: str, project_slug: str, atoms: list[dict]) -> dict | None:
+    """Создаёт newsletter asset (html) + composition + email publication.
+
+    Рассылка — сборка из top-атомов (stat/claim/qa), отсортированных по весу.
+    Каждый атом → свой блок email-рассылки (таймкод = индекс в письме).
+    """
+    if not atoms:
+        return None
+
+    # Берём top-атомы (макс. 25 для рассылки)
+    top = atoms[:25]
+    asset_id = None
+
+    # Чистим старые newsletter-активы для проекта
+    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 = 'email')",
+            (project_id,),
+        )
+        cur.execute("DELETE FROM assets WHERE project_id = %s AND asset_type = 'email'", (project_id,))
+        conn.commit()
+
+    # Длина письма — ~400 символов на атом (email-формат)
+    total_chars = sum(min(len(a["type"]) + 200, 400) for a in top)
+
+    with conn.cursor() as cur:
+        cur.execute(
+            """
+            INSERT INTO assets
+                (project_id, asset_type, format, uri, title, duration_sec, meta)
+            SELECT %s, 'email', 'html', %s, %s, %s, %s
+            WHERE NOT EXISTS (
+                SELECT 1 FROM assets WHERE project_id = %s AND asset_type = 'email'
+            )
+            RETURNING id
+            """,
+            (
+                project_id,
+                "https://cdn.content-atom.local/" + project_slug + "-newsletter.html",
+                f"{project_slug} — рассылка (media stage 3)",
+                total_chars,
+                json.dumps({"channel": "newsletter", "atoms": len(top)}),
+                project_id,
+            ),
+        )
+        row = cur.fetchone()
+        if row:
+            asset_id = str(row[0])
+        conn.commit()
+
+    # Привязываем атомы к письму (индекс = таймкод)
+    for idx, atom in enumerate(top):
+        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, %s)
+                ON CONFLICT (atom_id, asset_id, start_time_sec) DO NOTHING
+                """,
+                (atom["id"], asset_id, float(idx), float(idx + 1), "digest"),
+            )
+    conn.commit()
+
+    # composition: newsletter
+    atom_ids = [a["id"] for a in top]
+    recipe = {
+        "channel": "newsletter",
+        "format": "html",
+        "max_atoms": 25,
+        "atoms": [{"id": a["id"], "type": a["type"]} for a in top],
+    }
+    with conn.cursor() as cur:
+        cur.execute(
+            """
+            INSERT INTO compositions
+                (project_id, channel, slug, title, status, recipe, atom_ids)
+            VALUES (%s, 'newsletter', %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 + "-newsletter",
+                f"{project_slug} — рассылка (media stage 3)",
+                json.dumps(recipe),
+                atom_ids,
+            ),
+        )
+    conn.commit()
+
+    return {"asset": asset_id, "chars": total_chars, "atoms": len(top)}
+
+
+def create_newsletter_publication(conn, project_id: str, project_slug: str) -> dict:
+    """Создаёт публикацию рассылки (email) с метриками."""
+    with conn.cursor() as cur:
+        cur.execute(
+            """
+            INSERT INTO compositions
+                (project_id, channel, slug, title, status, recipe, atom_ids)
+            VALUES (%s, 'newsletter', %s, %s, 'published', '{}', '{}')
+            ON CONFLICT (project_id, channel, slug) DO UPDATE
+            SET status = 'published'
+            RETURNING id
+            """,
+            (
+                project_id,
+                project_slug + "-newsletter",
+                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, "%newsletter%"),
+        )
+        row = cur.fetchone()
+        if row:
+            pub_id = row[0]
+            cur.execute(
+                """
+                UPDATE publications SET metrics = %s WHERE id = %s
+                """,
+                (
+                    json.dumps(
+                        {
+                            "sent": 500,
+                            "open_rate": 0.42,
+                            "clicks": 60,
+                            "conversions": 5,
+                            "revenue": 750.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://newsletter.content-atom.local/" + project_slug,
+                    "news-" + project_slug,
+                    json.dumps(
+                        {
+                            "sent": 500,
+                            "open_rate": 0.42,
+                            "clicks": 60,
+                            "conversions": 5,
+                            "revenue": 750.0,
+                        }
+                    ),
+                ),
+            )
+    conn.commit()
+
+    return {
+        "sent": 500,
+        "open_rate": 0.42,
+        "clicks": 60,
+        "conversions": 5,
+        "revenue": 750.0,
+    }
+
+
 def main():
     ap = argparse.ArgumentParser(description="MVP stage 3: media assets")
     ap.add_argument("--project", default=None, help="project slug")
@@ -420,15 +591,18 @@ def main():
         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"])
+        news_asset = create_newsletter_asset(conn, proj["id"], proj["slug"], atoms)
+        news_pub = create_newsletter_publication(conn, proj["id"], proj["slug"])
 
         print(
             f"  • {proj['slug']}: {asset['atoms']} атомов, "
             f"{asset['duration_sec']:.0f}с видео, {pod_asset['duration_sec']:.0f}с подкаст, "
+            f"{news_asset['atoms']} в рассылке, "
             f"{pub['views']} views → {pub['conversions']} conv, "
-            f"{pod_pub['downloads']} downloads подкаст, "
+            f"{pod_pub['downloads']} подкаст, {news_pub['sent']} email, "
-            f"${pub['revenue'] + pod_pub['revenue']}"
+            f"${pub['revenue'] + pod_pub['revenue'] + news_pub['revenue']}"
         )
-        total_revenue += pub["revenue"] + pod_pub["revenue"]
+        total_revenue += pub["revenue"] + pod_pub["revenue"] + news_pub["revenue"]
 
     conn.close()
     print(