MVP stage 2: dedup (atom_relations) + compositions (telegram channel)

db/schema.sql

@@ -98,3 +98,20 @@ CREATE TABLE runs (
 -- Индексы
 CREATE INDEX ON jobs (status, scheduled_at);
 CREATE INDEX ON runs (stage, created_at DESC);
+
+-- Связи между атомами (дедup + lineage)
+CREATE TABLE atom_relations (
+  id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  atom_id_a       UUID REFERENCES atoms(id) ON DELETE CASCADE,
+  atom_id_b       UUID REFERENCES atoms(id) ON DELETE CASCADE,
+  relation_type   TEXT NOT NULL,   -- duplicate | related | superset | contradiction | parent_of
+  similarity      NUMERIC(4,3),    -- cosine similarity (0-1)
+  shared_topics   TEXT[] DEFAULT '{}',
+  confidence      NUMERIC(3,2),
+  source          TEXT,            -- semantic | lexical | manual
+  created_at      TIMESTAMPTZ DEFAULT now(),
+  UNIQUE (atom_id_a, atom_id_b, relation_type)
+);
+CREATE INDEX ON atom_relations (atom_id_a, relation_type);
+CREATE INDEX ON atom_relations (atom_id_b, relation_type);
+CREATE INDEX ON atom_relations (similarity DESC);

stage2/compositions.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""
+MVP stage 2: compositions — собрать atom_versions под второй канал (Telegram).
+
+Для каждого атома генерирует версию под Telegram (plain text, до 1024 символов),
+если atom_versions для этого канала ещё нет.
+
+bge-m3 (dim=1024) — единственная embed-модель (для semantic-версий).
+Postgres (atom-db) — atom_versions как источник.
+
+Запуск:
+    python3 stage2/compositions.py --channel telegram --top 20
+    python3 stage2/compositions.py --channel telegram --limit 50
+"""
+
+import argparse
+import json
+import os
+import sys
+import time
+
+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"),
+}
+MAX_LEN = 1024  # Telegram limit
+
+
+def get_conn():
+    return psycopg2.connect(**POSTGRES)
+
+
+def get_atoms(conn, limit: int) -> list[dict]:
+    """Атомы для генерации (draft, не занятые)."""
+    with conn.cursor() as cur:
+        cur.execute("""
+            SELECT a.id, a.type, a.plain_text, a.word_count, a.project_id
+            FROM atoms a
+            WHERE a.plain_text IS NOT NULL
+            ORDER BY a.created_at DESC
+            LIMIT %s
+        """, (limit,))
+        cols = [d[0] for d in cur.description]
+        return [dict(zip(cols, row)) for row in cur.fetchall()]
+
+
+def has_version(conn, atom_id: str, channel: str) -> bool:
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT 1 FROM atom_versions WHERE atom_id=%s AND channel=%s LIMIT 1",
+            (atom_id, channel),
+        )
+        return cur.fetchone() is not None
+
+
+def build_telegram_version(atom: dict) -> str:
+    """Генерация версии под Telegram (plain text)."""
+    text = atom["plain_text"]
+    if len(text) > MAX_LEN:
+        text = text[:MAX_LEN].rstrip() + "…"
+
+    # Типизированное оформление
+    t = atom["type"]
+    if t in ("stat",):
+        return f"📊 {text}"
+    if t in ("qa",):
+        return f"❓ {text}"
+    if t in ("cta",):
+        return f"👉 {text}"
+    if t in ("hook",):
+        return f"🔥 {text}"
+    if t in ("comparison",):
+        return f"⚖️ {text}"
+    if t in ("instruction",):
+        return f"✅ {text}"
+    if t in ("quote",):
+        return f'» {text}'
+    return text
+
+
+def insert_versions(conn, atom_id: str, content: str) -> bool:
+    """Вставить atom_versions под telegram (plain)."""
+    with conn.cursor() as cur:
+        cur.execute("""
+            INSERT INTO atom_versions
+            (atom_id, channel, format, content, quality_score)
+            VALUES (%s, %s, %s, %s, %s)
+            ON CONFLICT (atom_id, channel, format) DO NOTHING
+        """, (atom_id, "telegram", "plain", content, 7.0))
+        committed = cur.rowcount > 0
+    conn.commit()
+    return committed
+
+
+def main():
+    ap = argparse.ArgumentParser(description="MVP stage 2: compositions (telegram)")
+    ap.add_argument("--channel", default="telegram", help="канал")
+    ap.add_argument("--format", default="plain", help="формат")
+    ap.add_argument("--limit", type=int, default=20, help="сколько атомов")
+    args = ap.parse_args()
+
+    conn = get_conn()
+    atoms = get_atoms(conn, args.limit)
+    print(f"  • {len(atoms)} атомов для {args.channel}")
+
+    n = 0
+    start = time.time()
+    for atom in atoms:
+        if has_version(conn, str(atom["id"]), args.channel):
+            continue
+        content = build_telegram_version(atom)
+        if insert_versions(conn, str(atom["id"]), content):
+            n += 1
+    conn.close()
+
+    print(f"  ✓ atom_versions: {n} за {time.time()-start:.1f}с")
+    print(f"\nMVP stage 2 compositions OK: {n} версий под {args.channel}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

stage2/dedup.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""
+MVP stage 2: dedup atoms across projects.
+
+Для каждого атома ищет топ-K по cosine similarity в Qdrant.
+Если similarity > threshold (0.95) → relation_type='duplicate'.
+Сохраняет связи в atom_relations (Postgres).
+
+bge-m3 (dim=1024) — единственная embed-модель.
+Qdrant — эмбеддинги.
+Postgres (atom-db) — atom_relations как источник связей.
+
+Запуск:
+    python3 stage2/dedup.py --threshold 0.95 --top-k 5
+    python3 stage2/dedup.py --threshold 0.98 --top-k 3
+"""
+
+import argparse
+import json
+import os
+import sys
+import time
+import uuid
+
+import psycopg2
+from qdrant_client import QdrantClient
+from qdrant_client.models import QueryResponse
+
+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"),
+}
+COLLECTION = "content_atoms"
+DIM = 1024
+
+
+def get_conn():
+    return psycopg2.connect(**POSTGRES)
+
+
+def query_qdrant(client: QdrantClient, embedding: list[float], top_k: int = 5) -> list[dict]:
+    """Поиск топ-K по cosine similarity."""
+    try:
+        resp: QueryResponse = client.query_points(
+            collection_name=COLLECTION,
+            query=embedding,
+            limit=top_k,
+            with_payload=True,
+        )
+    except Exception as e:
+        print(f"  ✗ Qdrant query_points failed: {e}")
+        return []
+
+    results = []
+    for hit in resp.points:
+        payload = hit.payload
+        results.append({
+            "atom_id": payload.get("atom_id"),
+            "similarity": float(hit.score),
+            "project": payload.get("project"),
+            "type": payload.get("type"),
+        })
+    return results
+
+
+def get_all_atoms(conn) -> list[dict]:
+    """Все атомы из atoms."""
+    with conn.cursor() as cur:
+        cur.execute("""
+            SELECT id, project_id, plain_text, word_count
+            FROM atoms
+            WHERE plain_text IS NOT NULL
+            ORDER BY created_at
+        """)
+        cols = [d[0] for d in cur.description]
+        return [dict(zip(cols, row)) for row in cur.fetchall()]
+
+
+def dedup(client: QdrantClient, atoms: list[dict], threshold: float, top_k: int) -> list[dict]:
+    """Дедуп: для каждого атома ищет дубликаты."""
+    import urllib.request
+    relations = []
+    n = 0
+    for atom in atoms:
+        n += 1
+        # embed
+        body = atom["plain_text"]
+        if len(body) > 4000:
+            body = body[:4000]
+        try:
+            req = urllib.request.Request(
+                "http://127.0.0.1:11434/api/embeddings",
+                data=json.dumps({"model": "bge-m3", "prompt": body}).encode(),
+                headers={"Content-Type": "application/json"},
+            )
+            with urllib.request.urlopen(req, timeout=60) as resp:
+                emb = json.loads(resp.read().decode())["embedding"]
+        except Exception as e:
+            print(f"  ✗ embed failed for {atom['id']}: {e}")
+            continue
+        if len(emb) != DIM:
+            print(f"  ✗ wrong dim {len(emb)} for {atom['id']}")
+            continue
+
+        hits = query_qdrant(client, emb, top_k)
+        for hit in hits:
+            if hit["atom_id"] == str(atom["id"]):
+                continue
+            # только atom_id из atoms (UUID), не из старых эмбеддингов спейка
+            try:
+                uuid.UUID(hit["atom_id"])
+            except ValueError:
+                continue
+            if hit["similarity"] >= threshold:
+                relations.append({
+                    "atom_id_a": str(atom["id"]),
+                    "atom_id_b": hit["atom_id"],
+                    "relation_type": "duplicate",
+                    "similarity": round(hit["similarity"], 4),
+                    "shared_topics": [],
+                    "confidence": 0.9,
+                    "source": "semantic",
+                })
+    return relations
+
+
+def insert_relations(conn, relations: list[dict]) -> int:
+    """Вставить atom_relations. ON CONFLICT DO NOTHING."""
+    n = 0
+    with conn.cursor() as cur:
+        for rel in relations:
+            cur.execute("""
+                INSERT INTO atom_relations
+                (atom_id_a, atom_id_b, relation_type, similarity, shared_topics, confidence, source)
+                VALUES (%s, %s, %s, %s, %s, %s, %s)
+                ON CONFLICT (atom_id_a, atom_id_b, relation_type) DO NOTHING
+            """, (
+                rel["atom_id_a"], rel["atom_id_b"], rel["relation_type"],
+                rel["similarity"], rel["shared_topics"], rel["confidence"], rel["source"],
+            ))
+            n += cur.rowcount
+    conn.commit()
+    return n
+
+
+def main():
+    ap = argparse.ArgumentParser(description="MVP stage 2: dedup atoms across projects")
+    ap.add_argument("--threshold", type=float, default=0.95, help="cosine similarity threshold")
+    ap.add_argument("--top-k", type=int, default=5, help="top-K for Qdrant search")
+    args = ap.parse_args()
+
+    print(f"Дедup: threshold={args.threshold}, top_k={args.top_k}")
+
+    # 1. Атомы
+    conn = get_conn()
+    atoms = get_all_atoms(conn)
+    print(f"  • {len(atoms)} атомов")
+
+    # 2. Qdrant
+    client = QdrantClient(host="127.0.0.1", port=6333)
+
+    # 3. Дедup
+    start = time.time()
+    relations = dedup(client, atoms, args.threshold, args.top_k)
+    print(f"  ✓ {len(relations)} связей за {time.time()-start:.1f}с")
+
+    # 4. Postgres
+    n = insert_relations(conn, relations)
+    conn.close()
+    print(f"  ✓ Postgres: {n} atom_relations")
+
+    # 5. Итог
+    print(f"\nMVP stage 2 OK: {len(atoms)} атомов → {n} дупликатов")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())