diff --git a/db/schema.sql b/db/schema.sql index 80df4f2..a7355e4 100644 --- a/db/schema.sql +++ b/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); diff --git a/stage2/compositions.py b/stage2/compositions.py new file mode 100644 index 0000000..cfb6aa1 --- /dev/null +++ b/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()) diff --git a/stage2/dedup.py b/stage2/dedup.py new file mode 100644 index 0000000..108122c --- /dev/null +++ b/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())