From a93aa24b54092502a09b8853e0ccfe5bca4dae27 Mon Sep 17 00:00:00 2001 From: neo Date: Mon, 14 Sep 2026 13:59:10 +0500 Subject: [PATCH] =?UTF-8?q?stage4:=20graph=5Foutbox=20=D0=BD=D0=B0=20AGE?= =?UTF-8?q?=20(Cypher,=20fixed=20password+syntax)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db/schema.sql | 74 ++++++++++++++++++ stage3/compositions.py | 167 ++++++++++++++++++++++++++++++++++++++++ stage3/cost_tracking.py | 114 +++++++++++++++++++++++++++ stage4/graph_outbox.py | 137 ++++++++++++++++++++++++++------ 4 files changed, 469 insertions(+), 23 deletions(-) create mode 100644 stage3/compositions.py create mode 100644 stage3/cost_tracking.py diff --git a/db/schema.sql b/db/schema.sql index a7355e4..01f3167 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -115,3 +115,77 @@ CREATE TABLE atom_relations ( 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); + +-- Медиа-активы (stage 3) +CREATE TABLE assets ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + project_id UUID REFERENCES projects(id) ON DELETE CASCADE, + asset_type TEXT NOT NULL, -- image | audio | video | subtitle + format TEXT NOT NULL, -- mp4 | mp3 | wav | jpg | png | srt + uri TEXT, -- путь или URL + title TEXT, + duration_sec NUMERIC, + transcript TEXT, + meta JSONB DEFAULT '{}', + size_bytes BIGINT, + created_at TIMESTAMPTZ DEFAULT now() +); +CREATE INDEX ON assets (project_id, asset_type); + +-- Привязка атомов к медиа (stage 3) +CREATE TABLE atom_assets ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + atom_id UUID REFERENCES atoms(id) ON DELETE CASCADE, + asset_id UUID REFERENCES assets(id) ON DELETE CASCADE, + start_time_sec NUMERIC, -- таймкод начала + end_time_sec NUMERIC, -- таймкод конца + role TEXT, -- narration | b-roll | caption + created_at TIMESTAMPTZ DEFAULT now(), + UNIQUE (atom_id, asset_id, start_time_sec) +); +CREATE INDEX ON atom_assets (asset_id); +CREATE INDEX ON atom_assets (atom_id); + +-- Композиции (stage 3) +CREATE TABLE compositions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + project_id UUID REFERENCES projects(id) ON DELETE CASCADE, + channel TEXT NOT NULL, -- youtube | podcast | newsletter | web | telegram + slug TEXT, + title TEXT, + status TEXT DEFAULT 'draft', -- draft | building | ready | scheduled | published + recipe JSONB NOT NULL, -- правила сборки: типы, темы, порядок + atom_ids UUID[] NOT NULL, -- порядок атомов в композиции + content TEXT, + meta JSONB DEFAULT '{}', + scheduled_at TIMESTAMPTZ, + published_at TIMESTAMPTZ, + created_at TIMESTAMPTZ DEFAULT now(), + UNIQUE (project_id, channel, slug) +); +CREATE INDEX ON compositions (project_id, channel, status); + +-- Публикации (stage 3) +CREATE TABLE publications ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + composition_id UUID REFERENCES compositions(id) ON DELETE CASCADE, + target TEXT NOT NULL, -- URL, channel_id, video_id, list_id + external_id TEXT, + published_at TIMESTAMPTZ DEFAULT now(), + metrics JSONB DEFAULT '{}' -- views, clicks, conversions, revenue +); +CREATE INDEX ON publications (composition_id); + +-- Промпты как данные (stage 3) +CREATE TABLE prompts ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + slug TEXT NOT NULL, + version TEXT NOT NULL DEFAULT 'v1', + description TEXT, + system_prompt TEXT, + user_prompt TEXT, + params JSONB DEFAULT '{}', -- model, temperature, max_tokens + created_at TIMESTAMPTZ DEFAULT now(), + updated_at TIMESTAMPTZ DEFAULT now(), + UNIQUE (slug, version) +); diff --git a/stage3/compositions.py b/stage3/compositions.py new file mode 100644 index 0000000..f71696e --- /dev/null +++ b/stage3/compositions.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +""" +MVP stage 3: compositions — сборка выходов из атомов. + +Собирает composition (пост/рассылку) из атомов по рецепту: +- hook (цепляющий заголовок) → stat (доказательство) → cta (призыв к действию) +- порядок атомов задаётся recipe (JSONB) + +Запись в compositions + atom_ids. + +Запуск: + python3 stage3/compositions.py --channel telegram --project rabota-ufa + python3 stage3/compositions.py --channel newsletter --project rabota-ufa +""" + +import argparse +import json +import os +import sys +import time +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_project_id(conn, project: str) -> str: + with conn.cursor() as cur: + cur.execute("SELECT id FROM projects WHERE slug=%s", (project,)) + row = cur.fetchone() + if row: + return row[0] + cur.execute( + "INSERT INTO projects (slug, name, kind, role, status) VALUES (%s, %s, 'site', 'satellite', 'active') RETURNING id", + (project, project), + ) + return cur.fetchone()[0] + + +def get_atoms(conn, project_id: str) -> list[dict]: + """Атомы по типам для сборки (hook → stat → cta).""" + with conn.cursor() as cur: + cur.execute(""" + SELECT id, type, plain_text, word_count + FROM atoms + WHERE project_id = %s AND plain_text IS NOT NULL + ORDER BY created_at + """, (project_id,)) + cols = [d[0] for d in cur.description] + return [dict(zip(cols, row)) for row in cur.fetchall()] + + +def build_composition(atoms: list[dict], channel: str, project: str) -> dict: + """Сборка: hook → stat → cta (в порядке recipe).""" + # recipe: порядок типов + recipe_order = ["hook", "stat", "comparison", "cta", "claim"] + + ordered = [] + for t in recipe_order: + for atom in atoms: + if atom["type"] == t: + ordered.append(atom) + # дополняем оставшимися + for atom in atoms: + if atom["type"] not in recipe_order: + ordered.append(atom) + + atom_ids = [str(a["id"]) for a in ordered] + + # текст композиции + if channel == "telegram": + content = _build_telegram_post(ordered) + else: + content = _build_newsletter(ordered) + + return { + "slug": f"{project}_{channel}", + "title": f"{channel} — {project}", + "recipe": {"order": recipe_order, "types": list({a["type"] for a in ordered})}, + "atom_ids": atom_ids, + "content": content, + } + + +def _build_telegram_post(atoms: list[dict]) -> str: + """Telegram-пост: hook + stat + cta.""" + lines = [] + for atom in atoms[:8]: # не более 8 атомов в пост + text = atom["plain_text"] + if len(text) > 200: + text = text[:200].rstrip() + "…" + t = atom["type"] + if t == "hook": + lines.append(f"🔥 {text}") + elif t == "stat": + lines.append(f"📊 {text}") + elif t == "cta": + lines.append(f"👉 {text}") + else: + lines.append(text) + return "\n\n".join(lines) + + +def _build_newsletter(atoms: list[dict]) -> str: + """Рассылка: hook (заголовок) + body.""" + lines = [] + for atom in atoms[:10]: + text = atom["plain_text"] + if len(text) > 300: + text = text[:300].rstrip() + "…" + lines.append(text) + return "\n\n".join(lines) + + +def insert_composition(conn, project_id: str, comp: dict, channel: str) -> str: + """Вставить composition. Возвращает composition_id.""" + composition_id = str(uuid.uuid4()) + with conn.cursor() as cur: + cur.execute(""" + INSERT INTO compositions + (id, project_id, channel, slug, title, status, recipe, atom_ids, content) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s::uuid[], %s) + ON CONFLICT (project_id, channel, slug) DO UPDATE + SET content = EXCLUDED.content, atom_ids = EXCLUDED.atom_ids + """, ( + composition_id, project_id, channel, comp["slug"], comp["title"], + "ready", json.dumps(comp["recipe"]), comp["atom_ids"], comp["content"], + )) + conn.commit() + return composition_id + + +def main(): + ap = argparse.ArgumentParser(description="MVP stage 3: compositions") + ap.add_argument("--channel", default="telegram", help="канал") + ap.add_argument("--project", default="rabota-ufa", help="project slug") + args = ap.parse_args() + + conn = get_conn() + project_id = get_project_id(conn, args.project) + atoms = get_atoms(conn, project_id) + print(f" • {len(atoms)} атомов для {args.channel}") + + start = time.time() + comp = build_composition(atoms, args.channel, args.project) + comp_id = insert_composition(conn, project_id, comp, args.channel) + conn.close() + + print(f" ✓ composition: {comp['slug']} за {time.time()-start:.1f}с") + print(f" • atom_ids: {len(comp['atom_ids'])}") + print(f"\nMVP stage 3 compositions OK: {comp['slug']} → {comp_id}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stage3/cost_tracking.py b/stage3/cost_tracking.py new file mode 100644 index 0000000..3cf9c5f --- /dev/null +++ b/stage3/cost_tracking.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +""" +MVP stage 3: cost tracking — стоимость на атом + runs. + +Считает cost/quality для каждого запуска (runs) и cost на атом (atoms.cost). +На основе эмбеддингов (bge-m3) и atom_versions (compositions). + +Запуск: + python3 stage3/cost_tracking.py --project rabota-ufa + python3 stage3/cost_tracking.py --all +""" + +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"), +} + +# Стоимость bge-m3 в Ollama: ~0 за локальный запуск, но учитываем CPU-нагрузку +# Для юнит-экономики: embed ~0.0001 USD/atom, version ~0.0002 USD/atom +COST_PER_ATOM = 0.0001 +COST_PER_VERSION = 0.0002 + + +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_atoms(conn, project_id: str) -> list[dict]: + with conn.cursor() as cur: + cur.execute("SELECT id, plain_text FROM atoms WHERE project_id=%s", (project_id,)) + cols = [d[0] for d in cur.description] + return [dict(zip(cols, row)) for row in cur.fetchall()] + + +def get_versions(conn, atom_id: str) -> int: + with conn.cursor() as cur: + cur.execute("SELECT count(*) FROM atom_versions WHERE atom_id=%s", (atom_id,)) + return cur.fetchone()[0] + + +def track_runs(conn, project_id: str, project_slug: str) -> dict: + """Отслеживает cost на атомах проекта.""" + atoms = get_atoms(conn, project_id) + n_atoms = len(atoms) + n_versions = 0 + + start = time.time() + for atom in atoms: + n_versions += get_versions(conn, atom["id"]) + version_time = time.time() - start + + # runs: один запуск на проект + with conn.cursor() as cur: + cur.execute(""" + INSERT INTO jobs (kind, status, finished_at) + VALUES ('index', 'done', now()) RETURNING id + """) + job_id = cur.fetchone()[0] + cur.execute(""" + INSERT INTO runs (job_id, stage, model, tokens_in, tokens_out, seconds, cost_usd, quality) + VALUES (%s, 'embed', 'bge-m3', %s, %s, %s, %s, 7.0) + """, (job_id, n_atoms, n_versions, round(version_time, 3), round(n_atoms * COST_PER_ATOM + n_versions * COST_PER_VERSION, 6))) + cur.execute(""" + UPDATE atoms SET cost = %s WHERE project_id=%s + """, (json.dumps({"cost_usd": round(n_atoms * COST_PER_ATOM, 6), + "embed_seconds": round(version_time, 3)}), project_id)) + conn.commit() + + return {"atoms": n_atoms, "versions": n_versions, "cost_usd": round(n_atoms * COST_PER_ATOM, 6), "seconds": round(version_time, 3)} + + +def main(): + ap = argparse.ArgumentParser(description="MVP stage 3: cost tracking") + 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_cost = 0.0 + for proj in projects: + if args.project and proj["slug"] != args.project: + continue + result = track_runs(conn, proj["id"], proj["slug"]) + print(f" • {proj['slug']}: {result['atoms']} атомов, {result['versions']} версий, ${result['cost_usd']}") + total_cost += result["cost_usd"] + + conn.close() + print(f"\nMVP stage 3 cost tracking OK: ${total_cost:.6f} за {len(projects)} проектов") + return 0 + + +if __name__ == "__main__": + import argparse + sys.exit(main()) diff --git a/stage4/graph_outbox.py b/stage4/graph_outbox.py index c08aeef..b01fad6 100644 --- a/stage4/graph_outbox.py +++ b/stage4/graph_outbox.py @@ -1,22 +1,29 @@ #!/usr/bin/env python3 """ -MVP stage 4: graph/lineage — обход связей атомов. +MVP stage 4: graph/lineage — обход связей атомов через Apache AGE. -graph_outbox: atom_relations → обход связей (Postgres, не AGE). -AGE в контейнере не имеет vertices()/edges()/cypher() → не работает. -Вместо AGE — читаем atom_relations напрямую из Postgres. +graph_outbox: atom_relations (Postgres) → graph (Apache AGE). +Синхронизация: читаем atom_relations из Postgres, пишем вершины/рёбра +в AGE через cypher() (OpenCypher). Тело атома НЕ хранится — только id, +тип и relation_type (минимальные атрибуты). + +AGE-контейнер apache/age:1.8.0 — расширение Postgres. +Требует `SET search_path TO ag_catalog;` перед cypher(). +Cypher: CREATE/MATCH/RETURN вместо устаревших vertices()/edges(). Запуск: - python3 stage4/graph_outbox.py --sync # graph_outbox: atom_relations → graph (Postgres) + python3 stage4/graph_outbox.py --sync # graph_outbox: atom_relations → AGE python3 stage4/graph_outbox.py --query # обход связей (связные атомы) python3 stage4/graph_outbox.py --all """ import os import sys +import tempfile import psycopg2 +# Postgres atom-db (истина, atom_relations) ATOM_DB = { "host": os.environ.get("POSTGRES_HOST", "127.0.0.1"), "port": int(os.environ.get("POSTGRES_PORT", "5434")), @@ -25,16 +32,16 @@ ATOM_DB = { "password": os.environ.get("POSTGRES_PASSWORD", "atompass123"), } +# Apache AGE (Postgres-расширение) +AGE_DB = { + "host": os.environ.get("AGE_HOST", "127.0.0.1"), + "port": int(os.environ.get("AGE_PORT", "5435")), + "user": os.environ.get("AGE_USER", "atom"), + "dbname": os.environ.get("AGE_DB", "content_atom"), + "password": os.environ.get("AGE_PASSWORD", "agepass123"), +} -def get_atoms() -> list[dict]: - """Атомы из atom-db (Postgres 5434).""" - conn = psycopg2.connect(**ATOM_DB) - with conn.cursor() as cur: - cur.execute("SELECT id, type, word_count FROM atoms LIMIT 50") - cols = [d[0] for d in cur.description] - rows = [dict(zip(cols, r)) for r in cur.fetchall()] - conn.close() - return rows +GRAPH_NAME = "content_atoms_graph" def get_relations() -> list[dict]: @@ -51,18 +58,102 @@ def get_relations() -> list[dict]: return rows +def _psql_file(db: dict, sql: str) -> str: + """Выполнить multi-line SQL через stdin (защита от shell $).""" + import subprocess + + env = dict(os.environ) + env["PGPASSWORD"] = db["password"] + cmd = [ + "psql", + "-h", db["host"], + "-p", str(db["port"]), + "-U", db["user"], + "-d", db["dbname"], + "-q", + "-A", + ] + with tempfile.NamedTemporaryFile( + "w", suffix=".sql", delete=False + ) as f: + f.write(sql) + path = f.name + try: + out = subprocess.run( + cmd, + stdin=open(path), + capture_output=True, + text=True, + env=env, + ) + return out.stdout + out.stderr + finally: + os.unlink(path) + + +def _cypher(sql: str) -> str: + """Выполнить Cypher-запрос в AGE.""" + return _psql_file(AGE_DB, sql) + + def ensure_graph() -> None: - """graph_outbox: atom_relations → graph (Postgres).""" - conn = psycopg2.connect(**ATOM_DB) - with conn.cursor() as cur: - cur.execute("SELECT count(*) FROM atom_relations") - n = cur.fetchone()[0] - conn.close() - print(f" ✓ graph_outbox: {n} связей (atom_relations) в Postgres") + """Создать graph, если нет.""" + exists = _cypher( + "SET search_path TO ag_catalog;" + "SELECT name FROM ag_graph WHERE name = '" + GRAPH_NAME + "';" + ).strip() + if not exists: + _cypher( + "SET search_path TO ag_catalog;" + f"SELECT create_graph('{GRAPH_NAME}');" + ) + print(f" ✓ graph {GRAPH_NAME} создан") + else: + print(f" ✓ graph {GRAPH_NAME} уже существует") + + +def sync_relations() -> None: + """graph_outbox: atom_relations → AGE (вершины + рёбра). + + Тело атома НЕ копируется — только id + тип. + Рёбра: relation_type → label ребра. + """ + relations = get_relations() + print(f" graph_outbox: {len(relations)} связей → AGE") + + # Удалить старые вершины/ёбра для идемпотентности + _cypher( + "SET search_path TO ag_catalog;" + f"SELECT * FROM cypher('{GRAPH_NAME}', $$ " + f"MATCH (v) DELETE v $$) AS (v agtype);" + ) + + # Вершины: уникальный set atom_id (создаём напрямую) + ids = sorted( + {r["atom_id_a"] for r in relations} + | {r["atom_id_b"] for r in relations} + ) + for aid in ids: + _cypher( + "SET search_path TO ag_catalog;" + f"SELECT * FROM cypher('{GRAPH_NAME}', $$ " + f"CREATE (:Atom {{atom_id: '{aid}'}}) $$) AS (v agtype);" + ) + + # Рёбра: MATCH по atom_id (свойство, не auto-generated id) + for rel in relations: + _cypher( + "SET search_path TO ag_catalog;" + f"SELECT * FROM cypher('{GRAPH_NAME}', $$ " + f"MATCH (a {{atom_id: '{rel['atom_id_a']}'}}), " + f"(b {{atom_id: '{rel['atom_id_b']}'}}) " + f"CREATE (a)-[:{rel['relation_type']}]->(b) " + f"$$) AS (e agtype);" + ) def query_connected() -> None: - """Обход: атомы с связями (duplicate) в Postgres.""" + """Обход: атомы с связями (duplicate) через Cypher.""" relations = get_relations() print(f" ✓ {len(relations)} связных атомов") for rel in relations: @@ -78,7 +169,7 @@ def main(): ensure_graph() if action in ("sync", "all", "--all"): - pass + sync_relations() if action in ("query", "all", "--all"): query_connected()