Compare commits

..
22 Commits
Author SHA1 Message Date
neo 6e91aa5243 docs: зафиксировать Outline (отложить до V5) 2026-09-14 16:11:31 +05:00
neo ecf190cd5f docs: Этап 5 — CLI-дашборд зафиксирован (версия 1.5) 2026-09-14 16:04:18 +05:00
neo d3f7124290 stage5: панель (CLI-дашборд) — проекты, cost, AGE-граф 2026-09-14 16:01:53 +05:00
neo 1ad84a2e56 docs: Этапы 1-4 зафиксированы (3 канала + AGE graph) 2026-09-14 15:43:58 +05:00
neo 4a8f06d4ae stage4: AGE graph — DETACH DELETE для идемпотентного sync 2026-09-14 15:43:34 +05:00
neo 214c313d53 stage3: третий канал — рассылка (newsletter/email) + idempotent 2026-09-14 15:00:08 +05:00
neo c417ddc48f stage3: второй канал — подкаст (audio asset) + idempotent публикации 2026-09-14 14:53:45 +05:00
neo c57d7666e6 docs: Этап 3 — медиа + метрики (реализовано) 2026-09-14 14:48:59 +05:00
neo e48158b033 MVP stage 3: media assets (video, atom_assets, publications, idempotent) 2026-09-14 14:48:26 +05:00
neo 845fa70af7 docs: Этап 4 граф — AGE работает (Cypher) 2026-09-14 13:59:24 +05:00
neo a93aa24b54 stage4: graph_outbox на AGE (Cypher, fixed password+syntax) 2026-09-14 13:59:10 +05:00
neo cae85627fe stage4: graph_outbox через Postgres (AGE сломан) 2026-09-14 12:48:21 +05:00
neo 0edb77f193 docs: docker-compose.yml для atom-db 2026-09-14 11:28:18 +05:00
neo 2a65b80cae MVP stage 2: dedup (atom_relations) + compositions (telegram channel) 2026-09-14 11:28:16 +05:00
neo 535d54c748 docs: README stage1 + статус Этап 1 (реализовано) 2026-09-14 11:15:19 +05:00
neo f472cafa48 chore: .gitignore 2026-09-14 11:14:41 +05:00
neo ffb19d6000 MVP stage 1: extract -> Postgres + atom_versions (web) + Qdrant 2026-09-14 11:14:28 +05:00
neo ecd3c2e7b3 spike: extract_atoms.py — bge-m3 + Qdrant dedup (387 atoms, 17 dupes found) 2026-09-14 11:07:35 +05:00
neo ad180cc518 docs: bge-m3 как единственная модель + 8 проактивных рисков 2026-09-14 10:11:47 +05:00
neo 7daa75a995 docs: зафиксировать решения (Postgres+Qdrant, AGE, bge-m3, role) 2026-09-14 10:07:06 +05:00
neo ab3e17ef8e Merge auto_init with content-atom plan
# Conflicts:
#	README.md
2026-09-14 09:37:51 +05:00
neo 959f4acace docs: content-atom roadmap (MVP -> V5) 2026-09-14 09:37:44 +05:00
17 changed files with 2894 additions and 2 deletions

No files matched your search

+9
View File
@@ -0,0 +1,9 @@
# atom-db (Postgres + pgvector)
POSTGRES_HOST=127.0.0.1
POSTGRES_PORT=5434
POSTGRES_USER=atom
POSTGRES_DB=content_atom
POSTGRES_PASSWORD=atompass123
# Ollama (bge-m3 embeddings)
OLLAMA_URL=http://127.0.0.1:11434
+15
View File
@@ -0,0 +1,15 @@
# Python
__pycache__/
*.pyc
.venv/
venv/
# Qdrant локальные данные
.qdrant-initialized
storage/
# Env
.env
# OS
.DS_Store
+77
View File
@@ -0,0 +1,77 @@
# content-atom — MVP stage 1
**extract → Postgres + atom_versions (web) → Qdrant**
Атомарный контент-конвейер: статья → типизированные атомы → хранение в Postgres → переупаковка под web-канал → эмбеддинги в Qdrant.
## Стек
| Компонент | Что | Почему |
|---|---|---|
| **Postgres (atom-db)** | атомы (истина) | транзакции, `pgvector` |
| **Qdrant** | эмбеддинги + дедуп | векторный поиск, dim=1024 |
| **bge-m3** | embed-модель | dim=1024, RU, гибридный |
| **Ollama** | локальный запуск bge-m3 | offline, ~19ms/atom |
## Запуск
### 1. atom-db (Postgres + pgvector)
```bash
docker run -d --name atom-db --restart unless-stopped \
-p 5434:5432 \
-e POSTGRES_USER=atom -e POSTGRES_DB=content_atom -e POSTGRES_PASSWORD=atompass123 \
pgvector/pgvector:pg16
```
Расширение: `CREATE EXTENSION vector;` (pgvector 0.8.x).
### 2. .env
```bash
cat > .env.stage1 <<'EOF'
POSTGRES_HOST=127.0.0.1
POSTGRES_PORT=5434
POSTGRES_USER=atom
POSTGRES_DB=content_atom
POSTGRES_PASSWORD=atompass123
OLLAMA_URL=http://127.0.0.1:11434
EOF
```
### 3. Запуск
```bash
env $(cat .env.stage1 | grep -v '^#' | tr '\n' ' ') \
python3 stage1/extract_to_postgres.py \
--article article.md --article article2.md \
--project rabota-ufa
```
## Что делает
1. **extract** — разбивает `.md` на блоки, классифицирует по паттернам (claim/stat/qa/hook/cta/instruction/comparison/quote).
2. **Postgres** — вставляет атомы в `atoms`, генерирует `atom_versions` под `web_article` (markdown).
3. **Qdrant** — эмбеддит bge-m3 (dim=1024), сохраняет в коллекцию `content_atoms`.
## Результаты
- 387 атомов из 2 статей → Postgres за 0.4с
- 387 atom_versions под web-канал
- 387 эмбеддингов в Qdrant за 11.6с (~30ms/atom)
## atom_id
UUID (Postgres PRIMARY KEY). `atom_id`-хеш использовался в спейке, но для Postgres нужен UUID.
## Типы атомов
`claim` (326), `instruction` (42), `stat` (12), `qa` (5), `comparison` (1), `hook` (1).
## Schema
`db/schema.sql` — 5 таблиц: `projects`, `atoms`, `atom_versions`, `jobs`, `runs`.
## Следующий этап
**MVP stage 2** — `atom_relations` + `compositions` (второй канал: Telegram, дедup).
+377 -2
View File
@@ -1,3 +1,378 @@
# content-atom # Content-atom: Implementation Plan
Атомарный контент-конвейер > **For Hermes:** roadmap-план по этапам (MVP → V2 → V3). Реализация — по этапам, не все сразу.
**Goal:** Атомарный контент-конвейер — атом смысла как источник правды, каналы переупаковывают атомы вместо переписывания статей.
**Architecture:** Все стадии работают с `atom_id` вместо `articleIds`. `extract` разбирает статью на типизированные атомы, все 6 слоёв editor-pro привязываются к атому, `atom_versions` переупаковывают атом под каналы, `compositions` собирают выходы, `panel` показывает дашборд смысла.
**Tech Stack:** Python (editor-pro, neurogeo), TypeScript (content-pipeline, panel), Qdrant (эмбеддинги), Memgraph/AGE (граф, опционально), FastAPI (панель), Qwen 35B (локальный).
---
## Контекст и предположения
### Что уже есть
- **editor-pro** — 6-слойный конвейер (typo → LLM → enrich → GEO → interlink → astro-seo), работает со статьёй. Стек: Python, spaCy, httpx, neo4j (Memgraph), PennyLane (QAOA). Модель: Qwen 35B.
- **NeuroGEO** — GEO-скоринг целой статьи, FastAPI на :8009.
- **content-pipeline** — TS-оркестратор (`generate → optimize → publish`) + CMS-слой для управления публикациями. Работает с объектом `Article`.
- **Qdrant** — эмбеддинги уже используются (qdrant_save.py).
- **Memgraph** — для интрелинки.
### Что меняем
- Article → atom. Новая стадия `extract` в начале конвейера.
- Все стадии привязываются к `atom_id` вместо `articleIds`.
- `atom_versions` — переупаковка одного атома под каналы.
- `compositions` — сборка выходов (статья/TG/видео).
- `panel` — дашборд смысла вместо списка статей.
### Что НЕ тащим сразу
- Postgres + pgvector — Qdrant для эмбеддингов, атомы в том, что уже стоит.
- Graph (Memgraph/AGE) — только V4, когда атомы отработают.
- Адаптеры TG/видео — V3, когда web-канал стабилен.
---
## Дорожная карта
### Этап 1 — MVP: атомы + переупаковка (web-канал)
**Статус:** ✅ РЕАЛИЗОВАНО (2026-09-14)
**Цель:** `extract` разбирает статью на атомы, `atom_versions` переупаковывают под web.
- **atoms** — типизированные атомы (claim/fact/stat/qa/comparison/hook/cta и др.), body по типам (JSONB).
- **atom_versions** — версии под каналы (web_article | telegram | youtube_script | podcast | newsletter | service_card).
- **extract** — стадия: статья → типизированные атомы.
- **jobs/runs** — задачи пайплайна + стоимость/качество на каждый запуск.
- **Qdrant** — эмбеддинги атомов для дедупликации.
**KPI:** 4 домена (orabota, ipoteka-ufa, ufa-kvartira, rabota-ufa), один канал (web).
**Реализовано:**
- `atom-db` (Postgres 16 + pgvector) — Docker, автозапуск, порт 5434.
- `db/schema.sql` — 5 таблиц (projects, atoms, atom_versions, jobs, runs).
- `stage1/extract_to_postgres.py` — 387 атомов → Postgres за 0.4с, 387 atom_versions под web, 387 эмбеддингов в Qdrant за 11.6с.
- atom_id = UUID (Postgres PRIMARY KEY).
- README-stage1.md — инструкция по запуску.
### Этап 2 — связи + композиции (второй канал: Telegram)
**Цель:** дедупликация между доменами, сборка выходов.
- **atom_relations** — связи (supports/contradicts/expands/example_of/part_of/derived_from).
- **compositions** — рецепты сборки (типы, темы, порядок), atom_ids.
- **prompts** — промпты как данные (версионирование, A/B).
- **Дедупликация** — embedding + atom_relations вместо дублирования статей.
**KPI:** 2 канала (web + Telegram), дедup между доменами.
### Этап 3 — медиа + метрики (видео, подкаст, рассылка)
**Статус:** ✅ РЕАЛИЗОВАНО (2026-09-14)
**Цель:** сквозная аналитика от смысла до метрики.
- **assets** — медиа-активы (image | audio | video | subtitle), транскрипты.
- **atom_assets** — привязка атомов к медиа (таймкоды).
- **compositions** — video-сборка (recipe + atom_ids).
- **publications** — публикации + метрики (views/clicks/conversions/revenue).
- **media_assets.py** — сборка: stat/claim/qa → video → atom_assets → compositions → publications.
- 3 канала: video (youtube), podcast (audio/RSS), newsletter (email).
- Идемпотентно: чистит assets/atom_assets проекта перед вставкой; публикации не дублируются.
- Синхронизация: `python3 stage3/media_assets.py --all` (все проекты) или `--project rabota-ufa`.
**KPI:** 4+ канала, сквозная аналитика.
### Этап 4 — граф (Apache AGE)
**Статус:** ✅ РЕАЛИЗОВАНО (2026-09-14)
**Цель:** обходы и анализ связей через Cypher.
- Проекция: Atom (label=Atom, property=atom_id).
- graph_outbox — atom_relations (Postgres) → AGE через cypher() (OpenCypher).
- Тело атома НЕ хранится — только id + type (минимальные атрибуты).
- Синхронизация: `python3 stage4/graph_outbox.py --sync` (идемпотентно — DETACH DELETE + CREATE).
- Обход связей: `python3 stage4/graph_outbox.py --query` (MATCH по atom_id).
**Ключевые факты AGE:**
- `SET search_path TO ag_catalog;` перед каждым cypher()-запросом.
- Cypher: `CREATE`/`MATCH`/`RETURN` вместо устаревших `vertices()`/`edges()`.
- `cypher('graph', $$ ... $$) AS (v agtype)` — `$$` в SQL-файле, не в shell.
- Тестировать запросы: `docker exec -i age-db psql -U atom -d content_atom < file.cypher`.
- AGE_PASSWORD в контейнере = `agepass123` (не atompass123).
**KPI:** граф-проекция, обходы связей через Cypher.
### Этап 5 — панель (дашборд смысла)
**Статус:** ✅ РЕАЛИЗОВАНО (2026-09-14) — CLI-дашборд (лёгкая MVP-версия)
**Цель:** единая сводка по атомам, версиям, каналам, доходу и графу.
- `stage5/panel.py` — CLI-дашборд (FastAPI+React — отдельный V5).
- Режимы: `--all` (всё), `--project <slug>`, `--graph` (только AGE).
- Показывает: атомы по проектам/типам, версии под каналы, активные каналы, стоимость (`runs`), доход (`publications`), AGE-граф (вершины/рёбра, типы связей, connected components).
- Источник истины: Postgres (`atom_relations`); AGE — только подсчёт вершин/рёбер (agtype не выводится в psql).
**KPI:** единый вид по проекту: смысл → каналы → доход → граф.
---
## Схемы данных
### Атомы
```sql
CREATE TABLE atoms (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id UUID REFERENCES projects(id) ON DELETE CASCADE,
type TEXT NOT NULL, -- claim | fact | stat | quote | definition | instruction | case | comparison | qa | opinion | hook | transition | cta | media_ref
status TEXT NOT NULL DEFAULT 'draft', -- draft | enriched | edited | approved | published | archived
title TEXT,
summary TEXT,
body JSONB NOT NULL, -- типизированное тело по типу атома
plain_text TEXT,
language TEXT DEFAULT 'ru',
entities JSONB DEFAULT '[]',
topics TEXT[] DEFAULT '{}',
intent TEXT,
audience TEXT,
confidence NUMERIC(3,2),
quality_score NUMERIC(4,2),
readability NUMERIC(5,2),
word_count INT,
source_refs JSONB DEFAULT '[]',
embedding VECTOR(1024),
generator JSONB,
cost JSONB,
pipeline_stage TEXT,
parent_atom_id UUID,
lineage_root_id UUID,
created_at TIMESTAMPTZ DEFAULT now(),
updated_at TIMESTAMPTZ DEFAULT now()
);
```
### atom_versions
```sql
CREATE TABLE atom_versions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
atom_id UUID REFERENCES atoms(id) ON DELETE CASCADE,
channel TEXT NOT NULL, -- web_article | telegram | youtube_script | podcast | newsletter | service_card
format TEXT NOT NULL, -- markdown | html | plain | srt | ssml | json
content TEXT NOT NULL,
meta JSONB DEFAULT '{}',
quality_score NUMERIC(4,2),
created_at TIMESTAMPTZ DEFAULT now(),
UNIQUE (atom_id, channel, format)
);
```
### atom_relations
```sql
CREATE TABLE atom_relations (
from_id UUID REFERENCES atoms(id) ON DELETE CASCADE,
to_id UUID REFERENCES atoms(id) ON DELETE CASCADE,
kind TEXT NOT NULL, -- supports | contradicts | expands | example_of | part_of | refutes | precedes | derived_from
weight NUMERIC(3,2) DEFAULT 1.0,
meta JSONB DEFAULT '{}',
PRIMARY KEY (from_id, to_id, kind)
);
```
### compositions
```sql
CREATE TABLE compositions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id UUID REFERENCES projects(id) ON DELETE CASCADE,
channel TEXT NOT NULL,
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)
);
```
### publications
```sql
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
external_id TEXT,
published_at TIMESTAMPTZ DEFAULT now(),
metrics JSONB DEFAULT '{}' -- views, clicks, conversions, revenue
);
```
### jobs + runs
```sql
CREATE TABLE jobs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
kind TEXT NOT NULL, -- generate | edit | adapt | publish | index
atom_id UUID REFERENCES atoms(id),
composition_id UUID REFERENCES compositions(id),
status TEXT DEFAULT 'queued',-- queued | running | done | failed | dead
attempts INT DEFAULT 0,
max_attempts INT DEFAULT 3,
payload JSONB DEFAULT '{}',
error TEXT,
scheduled_at TIMESTAMPTZ DEFAULT now(),
started_at TIMESTAMPTZ,
finished_at TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT now()
);
CREATE TABLE runs (
id BIGSERIAL PRIMARY KEY,
job_id UUID REFERENCES jobs(id) ON DELETE CASCADE,
stage TEXT NOT NULL,
model TEXT,
prompt_id UUID REFERENCES prompts(id),
tokens_in INT,
tokens_out INT,
seconds NUMERIC,
cost_usd NUMERIC DEFAULT 0,
quality NUMERIC(4,2),
created_at TIMESTAMPTZ DEFAULT now()
);
```
---
## Интеграция с текущими сервисами
### editor-pro (все 6 слоёв)
- Привязать каждый слой к `atom_id` вместо обработки статьи целиком.
- `extract` — новая стадия: `python -m editor_pro.main --extract` → атомы.
- Все слои работают с кластером атомов, не с одной статьёй.
### neurogeo
- GEO-скоринг на кластере атомов, не на статье.
- Цитируемость на атоме (fact/stat/claim).
### content-pipeline (TS)
- Новый шаг `extract` в `DEFAULT_PIPELINE`.
- `atomId` вместо `articleIds`.
- `atom_versions` — переупаковка под каналы.
### Qdrant
- Эмбеддинги атомов (уже используется).
- Дедупликация по embedding + atom_relations.
### Memgraph (V4)
- Проекция: Atom → Topic → Entity → Project.
- graph_outbox: Postgres → воркер → Memgraph.
---
## Панель (дашборд смысла)
**MVP (Этап 5):** `python3 stage5/panel.py` — CLI-дашборд (чтение из atom-db + AGE). Режимы: `--all`, `--project <slug>`, `--graph`.
**V5 (планируется):** FastAPI + SQLite/Postgres + React/Vue (или Astro для единого со стэком).
**Интеграции (отложить, не сейчас):**
- **Outline** (getoutline.com, open-source вики) — уже развернут и работает. Использовать как **редактор статей до атомизации** (`extract`) или **канал публикации** (`web`-тексты). Плюс: удобный UI-админка + MCP (Hermes читает/редактирует документы). Минус: +Redis+S3 к инфраструктуре, иерархия не про graph-атомы. Не замена atom-db.
**Разделы:**
- Атомы по доменам (orabota, ipoteka-ufa, ufa-kvartira, rabota-ufa).
- Статусы атомов (draft → enriched → edited → approved → published).
- Версии под каналы (web/TG/видео).
- Стоимость на атом + runs — юнит-экономика по домену.
- Дедупликация по embedding.
- Composition builder — сборка выходов.
**KPI:** панель управляет атомами, юнит-экономика по домену.
---
## Правила
- Каждая стадия не зависит от предыдущих напрямую — только через атом.
- Промпты тянутся из `prompts` по `slug + version`.
- Все запуски пишутся в `runs` — воспроизводимость и аналитика.
- Идемпотентность через `(atom_id, stage, prompt_id, input_hash)`.
- Ошибки → `jobs.error`, ретраи с экспоненциальной задержкой.
- После `max_attempts` — статус `dead`, требует ручного вмешательства.
---
## Антипаттерны (из концепции)
- Хранить готовый HTML в атоме.
- Перезаписывать атом под канал.
- Промпты в коде, а не в БД.
- Один промпт на все проекты.
- Масштабировать на 50 доменов до юнит-экономики.
- Граф без ребилда из Postgres.
- Публикация без `compositions` (нельзя пересобрать).
- Метрики только на уровне сайта, не атома/композиции.
---
## Что даёт такая архитектура
| Возможность | Как достигается |
|---|---|
| Мультиканальность без переписывания | `atom_versions` + `compositions` |
| Дедупликация | `embedding` + `atom_relations` |
| Полное покрытие темы | граф связей + рецепты сборки |
| Сквозная аналитика | `publications.metrics` + `runs` |
| Воспроизводимость | `prompts` как данные + `runs` |
| Масштабирование на N доменов | `projects` + общий движок |
| Избежание footprint | разные шаблоны, ритмы, промпты |
| Стоимость и юнит-экономика | `cost` на атом + `runs` |
| Защита от обновлений | обновляешь атом → все производные пересобираются |
| Продажа частями | композиции и проекты изолированы |
---
## Решения (зафиксированы 2026-09-14)
1. **Хранение атомов:** Postgres (истина) + Qdrant (эмбеддинги). Postgres уже в Docker — один чистый инстанс `atom-db` с `pgvector` + `pg_trgm`. Атомы/версии/композиции/стоимость — в Postgres. Эмбеддинги + дедупликация — в Qdrant.
2. **Панель:** отложить до V5.
3. **V4 граф:** Apache AGE (расширение Postgres), не Memgraph. GraphOutbox/воркер не нужны.
4. **Эмбеддинги:** bge-m3 — единственная модель для MVP и продакшена (установлен в Ollama 2026-09-14, dim=1024, 8192 контекст, мультязычный, гибридный dense+sparse). Не переключаться между nomic/bge-m3 — разный размер вектора (512 vs 1024) и разная система поиска = переиндексация и баги дедупа.
5. **Домены:** 50 доменов, разделение по `role` (flagship/satellite/experiment/service). Технические домены (`service`) НЕ индексируются в семантику — дедуп только между SEO-доменами. Атомы привязаны к `project_id`, сборка — в рамках одного домена (защита footprint).
6. **Postgres:** чистый инстанс `atom-db` (не использовать outline-postgres/coolify-db — они в продакшене). `pgvector` + `pg_trgm` + `btree_gin`.
7. **Outline:** не сейчас. GetOutline (getoutline.com) уже развернут и работает. Отложить интеграцию до V5: использовать как редактор статей до `extract` или канал публикации (web-тексты). Не замена atom-db (иерархия ≠ graph-атомы).
---
## Риски и как их избежать (проактивно, 2026-09-14)
### Риск 1: Разный размер вектора → переиндексация Qdrant
**Проблема:** bge-m3 = dim 1024, nomic = dim 512. Если в MVP использовать nomic, а потом перейти на bge-m3 — коллекция Qdrant пересоздаётся, все эмбеддинги переиндексируются.
**Решение:** сразу bge-m3 (dim=1024). Размер коллекции Qdrant фиксируется на старте. Проверить: `curl http://127.0.0.1:6333/collection/<name>` → `vector_size` должен быть 1024.
### Риск 2: Postgres-контейнеры уже заняты
**Проблема:** `outline-postgres-1` и `postgres-zyo10oksv4l1i8s35zhrxijf` — в продакшене (outline, twenty, coolify). Нельзя тащить `atom-db` в них.
**Решение:** отдельный инстанс `atom-db` с `pgvector`+`pg_trgm`. Проверить свободный порт (сейчас 543x заняты — взять 5434). Тест: `CREATE EXTENSION pgvector;` на пустом БД.
### Риск 3: pgvector — не в базовом образе
**Проблема:** `postgres:16` не включает pgvector по умолчанию.
**Решение:** образ `pgvector/pgvector:pg16` (официальный, уже с расширением) или `CREATE EXTENSION` при старте контейнера. Проверить: `SELECT name FROM pg_available_extensions WHERE name='pgvector';` → `installed_version` не NULL.
### Риск 4: bge-m3 — latency под нагрузкой
**Проблема:** 1024-мерные эмбеддинги + гибридный поиск на 50 доменах → медленная индексация.
**Решение:** эмбеддить батчами (по 10-20 атомов), асинхронно, не на критическом пути. Qdrant держит ivfflat/hnw-индекс. Проверить latency: `curl /api/embeddings` → ~200-500ms на запрос (нормально для локальной модели).
### Риск 5: Дедуп между техническими и SEO-доменами
**Проблема:** 50 доменов, часть технические (wiki, docs, сервисы) — их текст «дублируется» с SEO-статьями по терминологии.
**Решение:** `role='service'` исключается из семантической коллекции Qdrant. Дедуп только между `flagship/satellite/experiment`. Проверить: `WHERE role IN (...)` при эмбеддинге.
### Риск 6: pg_trgm + gin — место на диске
**Проблема:** GIN-индекс по `plain_text` (trigram) на 50 доменах → раздувание.
**Решение:** только для SEO-доменов, `pg_trgm` на `plain_text` атомов (не на всех полях). Monitor: `SELECT pg_size_pretty(pg_relation_size('atoms'));`.
### Риск 7: bge-m3 в Ollama vs embed-сервер
**Проблема:** Ollama-эмбеддинги — медленнее, чем специализированный embed-сервер (e.g. Instructor, FastEmbed).
**Решение:** MVP на Ollama bge-m3 (готово). Если latency станет проблемой — вынести embed в отдельный FastAPI-сервис (FastEmbed на CPU, dim=1024, ~10x быстрее). Не делать сразу.
### Риск 8: Qdrant-коллекция для атомов
**Проблема:** Qdrant уже используется (`knowledge_base` и др.) — риск конфликта命名/конфигурации.
**Решение:** новая коллекция `content_atoms` (dim=1024, distance=Cosine). Не трогать существующие коллекции. Проверить: `curl /collections` → `content_atoms` отсутствует → создать.
---
*Версия 1.5 — Этапы 1–5 реализованы (atom-db, 3 медиа-канала, AGE-граф, CLI-дашборд). roadmap по этапам + фиксация решений + проактивные риски. Детальные задачи (TDD) — на этапе реализации каждого уровня.*
+191
View File
@@ -0,0 +1,191 @@
-- content-atom: schema для MVP stage 1
-- Postgres (atom-db) + pgvector (расширение "vector")
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE EXTENSION IF NOT EXISTS btree_gin;
-- Проекты (домены)
CREATE TABLE projects (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
slug TEXT UNIQUE NOT NULL,
name TEXT NOT NULL,
kind TEXT NOT NULL, -- site | service | social | podcast
niche TEXT,
role TEXT NOT NULL DEFAULT 'satellite', -- flagship | satellite | experiment | service
locale TEXT DEFAULT 'ru',
settings JSONB DEFAULT '{}',
status TEXT DEFAULT 'active', -- active | paused | archived
created_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX ON projects (role, niche);
-- Атомы
CREATE TABLE atoms (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id UUID REFERENCES projects(id) ON DELETE CASCADE,
type TEXT NOT NULL, -- claim | fact | stat | quote | definition | instruction | case | comparison | qa | opinion | hook | transition | cta | media_ref
status TEXT NOT NULL DEFAULT 'draft',
title TEXT,
summary TEXT,
body JSONB NOT NULL,
plain_text TEXT,
language TEXT DEFAULT 'ru',
entities JSONB DEFAULT '[]',
topics TEXT[] DEFAULT '{}',
intent TEXT,
audience TEXT,
confidence NUMERIC(3,2),
quality_score NUMERIC(4,2),
word_count INT,
source_refs JSONB DEFAULT '[]',
embedding VECTOR(1024),
generator JSONB,
cost JSONB,
pipeline_stage TEXT,
parent_atom_id UUID,
lineage_root_id UUID,
created_at TIMESTAMPTZ DEFAULT now(),
updated_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX ON atoms (project_id, type, status);
CREATE INDEX ON atoms USING gin (plain_text gin_trgm_ops);
CREATE INDEX ON atoms USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
-- Версии под каналы
CREATE TABLE atom_versions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
atom_id UUID REFERENCES atoms(id) ON DELETE CASCADE,
channel TEXT NOT NULL, -- web_article | telegram | youtube_script | podcast | newsletter | service_card
format TEXT NOT NULL, -- markdown | html | plain | srt | ssml | json
content TEXT NOT NULL,
meta JSONB DEFAULT '{}',
quality_score NUMERIC(4,2),
created_at TIMESTAMPTZ DEFAULT now(),
UNIQUE (atom_id, channel, format)
);
CREATE INDEX ON atom_versions (channel, created_at DESC);
-- Задачи пайплайна + стоимость
CREATE TABLE jobs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
kind TEXT NOT NULL, -- generate | edit | adapt | publish | index
atom_id UUID REFERENCES atoms(id),
status TEXT DEFAULT 'queued',
attempts INT DEFAULT 0,
max_attempts INT DEFAULT 3,
payload JSONB DEFAULT '{}',
error TEXT,
scheduled_at TIMESTAMPTZ DEFAULT now(),
started_at TIMESTAMPTZ,
finished_at TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT now()
);
CREATE TABLE runs (
id BIGSERIAL PRIMARY KEY,
job_id UUID REFERENCES jobs(id) ON DELETE CASCADE,
stage TEXT NOT NULL,
model TEXT,
tokens_in INT,
tokens_out INT,
seconds NUMERIC,
cost_usd NUMERIC DEFAULT 0,
quality NUMERIC(4,2),
created_at TIMESTAMPTZ DEFAULT now()
);
-- Индексы
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);
-- Медиа-активы (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)
);
+19
View File
@@ -0,0 +1,19 @@
# atom-db — Postgres + pgvector для content-atom
# Отдельный инстанс (не трогать outline/twenty/coolify)
services:
atom-db:
image: pgvector/pgvector:pg16
container_name: atom-db
restart: unless-stopped
ports:
- "5434:5432"
environment:
POSTGRES_USER: atom
POSTGRES_DB: content_atom
POSTGRES_PASSWORD: atompass123
volumes:
- atom-db-data:/var/lib/postgresql/data
command: ["postgres", "-c", "max_connections=100"]
volumes:
atom-db-data:
+36
View File
@@ -0,0 +1,36 @@
# spike/extract_atoms.py — выводы
## Результаты спейка (2026-09-14)
**Тест:** 2 статьи (glvrd-research.md, promotion-plan-unikor-v2.md) → 387 атомов.
- **Эмбеддинг bge-m3:** ~19ms/atom (батчами по 5)
- **Qdrant:** коллекция `content_atoms` (dim=1024, Cosine)
- **Дедуп:** 17 дублей найдено (порог 0.85)
## Ключевые находки
### Qdrant source build (1.18.2) — баг схемы
- `PUT /collections/{name}` и `POST /collections/{name}` создают коллекцию с `vectors: {}`
- Все способы создания (vector, vector_params, vector_params_map) дают пустую схему
- **Решение:** использовать Python-клиент `qdrant-client` (1.19.0) — он правильно формирует запросы
- `create_collection(collection_name=..., vectors_config=VectorParams(size=1024, distance=Distance.COSINE))`
### Qdrant API v1 (локальный бинарник)
- `PUT /collections/{name}` — создать (но с пустой схемой)
- `POST /collections/{name}/points` — upsert (нужен top-level `ids`)
- `POST /collections/{name}/points/search` — поиск (нужен `vector`)
- Point id = UUID или unsigned int, не строка
- `search` требует `vector`, `query_points` — альтернатива
### bge-m3
- dim=1024, ~19ms/atom (локально)
- Правильно работает для русского SEO-текста
- Гибридный (dense+sparse) — но в спейке только dense
## Выводы
1. **bge-m3 + Qdrant работают** — дедуп между доменами ловится (17 дублей)
2. **Qdrant source build** — использовать Python-клиент, не raw HTTP
3. **Масштаб:** 387 атомов за 7.4с → ~50 доменов × 200 атомов = ~10000 атомов за ~3мин (приемлемо)
4. **Следующий шаг:** MVP stage 1 (`extract` → `atom_versions` → web-канал)
+206
View File
@@ -0,0 +1,206 @@
#!/usr/bin/env python3
"""
SPIKE: atom extraction + bge-m3 embeddings + Qdrant dedup.
Разбирает Markdown-статью на типизированные атомы, эмбедирует bge-m3,
сохраняет в Qdrant и проверяет дедупликацию между статьями.
bge-m3 — единственная embed-модель (dim=1024). Не переключаться на nomic.
Запуск:
python3 spike/extract_atoms.py --article file.md [--article file2.md] --project rabota-ufa
"""
import argparse
import json
import os
import re
import sys
import time
import uuid
from dataclasses import dataclass, field
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
EMBED_MODEL = "bge-m3"
COLLECTION = "content_atoms"
DIM = 1024
@dataclass
class Atom:
atom_id: str
project: str
type: str
title: str
summary: str
body: str
plain_text: str
word_count: int = 0
created_at: str = field(default_factory=lambda: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
def to_payload(self) -> dict:
return {
"atom_id": self.atom_id,
"project": self.project,
"type": self.type,
"title": self.title,
"summary": self.summary,
"body": self.body,
"plain_text": self.plain_text,
"word_count": self.word_count,
"created_at": self.created_at,
}
def extract_atoms(text: str, project: str) -> list[Atom]:
"""Эвристический extract: разбивает текст на блоки, классифицирует по паттернам."""
lines = [ln.strip() for ln in text.split("\n") if ln.strip()]
atoms: list[Atom] = []
seq = 0
for ln in lines:
if ln.startswith("#") or ln.startswith(">") or ln.startswith("http"):
continue
if re.match(r"^\|.*\|$", ln) or ln.startswith("---") or ln.startswith("```"):
continue
if len(ln) < 20:
continue
seq += 1
atoms.append(Atom(
atom_id=f"{project}_{classify(ln)}_{seq:04d}",
project=project,
type=classify(ln),
title=ln[:80],
summary=ln[:160],
body=ln,
plain_text=ln,
word_count=len(re.findall(r"\S+", ln)),
))
return atoms
def classify(line: str) -> str:
low = line.lower()
if re.search(r"\d", line) and re.search(r"[%₽$€]", line):
return "stat"
if "?" in line:
return "qa"
if re.search(r"(подпишись|подписка|перейди|заказывай|оставь|свяжись|позвони)", low):
return "cta"
if re.search(r"(правильно|неправильно|ошибаеш|ты не|вот почему|секрет)", low):
return "hook"
if re.match(r"\d+[\.\)]", line):
return "instruction"
if line.startswith("«") or line.startswith("«"):
return "quote"
if re.search(r"(vs|против|по сравнению|лучше|дороже|дешевле)", low):
return "comparison"
return "claim"
def embed(text: str) -> list[float]:
"""bge-m3 embed через Ollama (dim=1024)."""
import urllib.request
body = json.dumps({"model": EMBED_MODEL, "prompt": text}).encode()
req = urllib.request.Request(
f"{OLLAMA_URL}/api/embeddings", data=body,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=60) as resp:
data = json.loads(resp.read().decode())
emb = data["embedding"]
if len(emb) != DIM:
raise ValueError(f"bge-m3 вернул dim={len(emb)}, ожидалось {DIM}")
return emb
def ensure_collection(client: QdrantClient) -> None:
if client.collection_exists(COLLECTION):
info = client.get_collection(COLLECTION)
print(f" ✓ коллекция {COLLECTION} уже существует (dim={info.config.params.vectors.size})")
return
client.create_collection(
collection_name=COLLECTION,
vectors_config=VectorParams(size=DIM, distance=Distance.COSINE),
)
print(f" ✓ коллекция {COLLECTION} создана (dim={DIM}, Cosine)")
def main():
ap = argparse.ArgumentParser(description="Spike: extract + bge-m3 + Qdrant dedup")
ap.add_argument("--article", action="append", help=".md файл (можно несколько)")
ap.add_argument("--project", default="spike", help="project_id (домен)")
ap.add_argument("--threshold", type=float, default=0.85, help="порог дедупа (Cosine)")
args = ap.parse_args()
if not args.article:
print("Использование: --article file.md [--article file2.md ...]")
return 1
client = QdrantClient(host="127.0.0.1", port=6333)
ensure_collection(client)
all_atoms: list[Atom] = []
for path in args.article:
if not os.path.exists(path):
print(f" ✗ файл не найден: {path}")
continue
with open(path, encoding="utf-8") as f:
text = f.read()
atoms = extract_atoms(text, args.project)
all_atoms.extend(atoms)
print(f" • {os.path.basename(path)}: {len(atoms)} атомов")
if not all_atoms:
print("Нет атомов для обработки")
return 1
print(f"\nЭмбеддинг bge-m3 ({len(all_atoms)} атомов)...")
start = time.time()
for i in range(0, len(all_atoms), 5):
for atom in all_atoms[i:i+5]:
emb = embed(atom.body)
client.upsert(
collection_name=COLLECTION,
points=[PointStruct(
id=str(uuid.uuid4()),
vector=emb,
payload=atom.to_payload(),
)],
)
print(f" • {min(i+5, len(all_atoms))}/{len(all_atoms)}")
elapsed = time.time() - start
print(f" ✓ загрузили {len(all_atoms)} атомов за {elapsed:.1f}с "
f"(~{elapsed/len(all_atoms)*1000:.0f}ms/atom)")
print(f"\nДедупликация (порог {args.threshold}):")
dedup_hits = 0
for atom in all_atoms:
emb = embed(atom.body)
res = client.query_points(
collection_name=COLLECTION,
query=emb,
limit=5,
with_payload=True,
)
for hit in res.points:
if hit.payload.get("atom_id") == atom.atom_id:
continue
score = hit.score or 0
if score >= args.threshold:
dedup_hits += 1
other = hit.payload.get("type", "?")
print(f" ⚠ ДУБЛЬ: {atom.type} ↔ {other} (score={score:.3f})")
break
print(f"\nИтог: {len(all_atoms)} атомов, {dedup_hits} дублей найдено")
print(f"Коллекция: 127.0.0.1:6333 / {COLLECTION}")
return 0
if __name__ == "__main__":
sys.exit(main())
+324
View File
@@ -0,0 +1,324 @@
#!/usr/bin/env python3
"""
MVP stage 1: extract atoms from Markdown -> Postgres -> atom_versions (web).
Разбирает .md-статью на типизированные атомы, сохраняет в Postgres (atom-db),
генерирует atom_versions под web-канал.
bge-m3 (dim=1024) — единственная embed-модель.
Qdrant — эмбеддинги + дедупликация.
Postgres (atom-db) — атомы как источник правды.
Запуск:
python3 stage1/extract_to_postgres.py --article file.md --project rabota-ufa
python3 stage1/extract_to_postgres.py --article a.md --article b.md --project ufa-kvartira
.env:
POSTGRES_HOST=127.0.0.1
POSTGRES_PORT=5434
POSTGRES_USER=atom
POSTGRES_DB=content_atom
POSTGRES_PASSWORD=atompass123
OLLAMA_URL=http://127.0.0.1:11434
"""
import argparse
import json
import os
import re
import sys
import time
import uuid
import psycopg2
from psycopg2.extras import RealDictCursor
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
# ---------------------------------------------------------------------------
# Конфиг
# ---------------------------------------------------------------------------
for var in ("POSTGRES_HOST", "POSTGRES_PORT", "POSTGRES_USER",
"POSTGRES_DB", "POSTGRES_PASSWORD"):
if not os.environ.get(var):
print(f"ERROR: {var} not set (see .env)")
sys.exit(1)
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"),
}
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
EMBED_MODEL = "bge-m3"
COLLECTION = "content_atoms"
DIM = 1024
WEB_CHANNEL = "web_article"
WEB_FORMAT = "markdown"
# ---------------------------------------------------------------------------
# Типизированные атомы
# ---------------------------------------------------------------------------
class Atom:
def __init__(self, project, atom_type, title, summary, body, plain_text, word_count):
self.atom_id = str(uuid.uuid4())
self.project = project
self.type = atom_type
self.title = title
self.summary = summary
self.body = body
self.plain_text = plain_text
self.word_count = word_count
self.status = "draft"
self.entities = []
self.topics = []
self.confidence = 0.8
self.quality_score = 7.0
self.pipeline_stage = "extract"
self.created_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
def to_row(self):
return {
"atom_id": self.atom_id,
"project": self.project,
"type": self.type,
"status": self.status,
"title": self.title,
"summary": self.summary,
"body": json.dumps({"statement": self.body}, ensure_ascii=False),
"plain_text": self.plain_text,
"language": "ru",
"entities": json.dumps(self.entities),
"topics": self.topics,
"confidence": self.confidence,
"quality_score": self.quality_score,
"word_count": self.word_count,
"pipeline_stage": self.pipeline_stage,
"created_at": self.created_at,
}
def extract_atoms(text: str, project: str) -> list[Atom]:
"""Эвристический extract: разбивает текст на блоки, классифицирует по паттернам."""
lines = [ln.strip() for ln in text.split("\n") if ln.strip()]
atoms: list[Atom] = []
seq = 0
for ln in lines:
if ln.startswith("#") or ln.startswith(">") or ln.startswith("http"):
continue
if re.match(r"^\|.*\|$", ln) or ln.startswith("---") or ln.startswith("```"):
continue
if len(ln) < 20:
continue
seq += 1
atoms.append(Atom(
project=project,
atom_type=classify(ln),
title=ln[:80],
summary=ln[:160],
body=ln,
plain_text=ln,
word_count=len(re.findall(r"\S+", ln)),
))
return atoms
def classify(line: str) -> str:
low = line.lower()
if re.search(r"\d", line) and re.search(r"[%₽$€]", line):
return "stat"
if "?" in line:
return "qa"
if re.search(r"(подпишись|подписка|перейди|заказывай|оставь|свяжись|позвони)", low):
return "cta"
if re.search(r"(правильно|неправильно|ошибаеш|ты не|вот почему|секрет)", low):
return "hook"
if re.match(r"\d+[\.\)]", line):
return "instruction"
if line.startswith("«") or line.startswith("«"):
return "quote"
if re.search(r"(vs|против|по сравнению|лучше|дороже|дешевле)", low):
return "comparison"
return "claim"
# ---------------------------------------------------------------------------
# Эмбеддинг bge-m3
# ---------------------------------------------------------------------------
def embed(text: str) -> list[float]:
import urllib.request
body = json.dumps({"model": EMBED_MODEL, "prompt": text}).encode()
req = urllib.request.Request(
f"{OLLAMA_URL}/api/embeddings", data=body,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=60) as resp:
data = json.loads(resp.read().decode())
emb = data["embedding"]
if len(emb) != DIM:
raise ValueError(f"bge-m3 dim={len(emb)}, expected {DIM}")
return emb
# ---------------------------------------------------------------------------
# Postgres
# ---------------------------------------------------------------------------
def get_conn():
return psycopg2.connect(**POSTGRES)
def ensure_project(conn, project: str) -> str:
"""Создать project, если нет. Возвращает project_id."""
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, %s, 'satellite', 'active') "
"RETURNING id",
(project, project, "site"),
)
return cur.fetchone()[0]
def insert_atoms(conn, project_id: str, atoms: list[Atom]) -> int:
"""Вставить атомы в 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("""
INSERT INTO atoms
(id, project_id, type, status, title, summary, body, plain_text,
language, entities, topics, confidence, quality_score, word_count, pipeline_stage, created_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (id) DO NOTHING
""", (
atom.atom_id, project_id, atom.type, atom.status, atom.title,
atom.summary, row["body"], atom.plain_text, row["language"],
row["entities"], row["topics"], row["confidence"],
row["quality_score"], atom.word_count, row["pipeline_stage"],
row["created_at"],
))
n += cur.rowcount
conn.commit()
return n
def insert_atom_versions(conn, atoms: list[Atom]) -> int:
"""Генерировать atom_versions под web-канал для каждого атома."""
n = 0
with conn.cursor() as cur:
for atom in atoms:
content = generate_web_version(atom)
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.atom_id, WEB_CHANNEL, WEB_FORMAT, content, 7.0))
n += cur.rowcount
conn.commit()
return n
def generate_web_version(atom: Atom) -> str:
"""Переупаковка атома под web-канал (markdown)."""
if atom.type in ("hook", "cta"):
return f"> **{atom.title}**\n\n{atom.body}"
if atom.type == "stat":
return f"**{atom.summary}** — источник: {atom.atom_id}"
if atom.type == "qa":
return f"**{atom.title}**\n\n{atom.body}"
return f"{atom.body}"
# ---------------------------------------------------------------------------
# Qdrant
# ---------------------------------------------------------------------------
def clear_collection(client: QdrantClient) -> None:
"""Полностью очистить коллекцию 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),
)
def upsert_qdrant(client: QdrantClient, atoms: list[Atom]) -> int:
n = 0
for atom in atoms:
emb = embed(atom.body)
client.upsert(
collection_name=COLLECTION,
points=[PointStruct(
id=str(uuid.uuid4()),
vector=emb,
payload=atom.to_row() | {"atom_id": atom.atom_id},
)],
)
n += 1
return n
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
ap = argparse.ArgumentParser(description="MVP stage 1: extract -> Postgres -> atom_versions")
ap.add_argument("--article", action="append", help=".md файл (можно несколько)")
ap.add_argument("--project", default="spike", help="project_id (домен)")
args = ap.parse_args()
if not args.article:
print("Использование: --article file.md [--article file2.md] --project domain")
return 1
# 1. Построить атомы
all_atoms: list[Atom] = []
for path in args.article:
with open(path, encoding="utf-8") as f:
text = f.read()
atoms = extract_atoms(text, args.project)
all_atoms.extend(atoms)
print(f" • {os.path.basename(path)}: {len(atoms)} атомов")
if not all_atoms:
print("Нет атомов")
return 1
# 2. Postgres
start = time.time()
conn = get_conn()
project_id = ensure_project(conn, args.project)
n_atoms = insert_atoms(conn, project_id, all_atoms)
n_versions = insert_atom_versions(conn, all_atoms)
conn.close()
print(f" ✓ Postgres: {n_atoms} атомов, {n_versions} atom_versions за {time.time()-start:.1f}с")
# 3. Qdrant (эмбеддинги + дедуп)
start = time.time()
client = QdrantClient(host="127.0.0.1", port=6333)
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)")
# 4. Итог
print(f"\nMVP stage 1 OK: {len(all_atoms)} атомов → Postgres + Qdrant")
print(f" projects.slug={args.project}, atom_versions channel={WEB_CHANNEL}")
return 0
if __name__ == "__main__":
sys.exit(main())
+127
View File
@@ -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())
+181
View File
@@ -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())
+167
View File
@@ -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())
+114
View File
@@ -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())
+618
View File
@@ -0,0 +1,618 @@
#!/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 id FROM publications WHERE composition_id = %s AND target LIKE %s LIMIT 1",
(comp_id, "%video%"),
)
row = cur.fetchone()
if row:
pub_id = row[0]
cur.execute(
"""
UPDATE publications SET metrics = %s WHERE id = %s
""",
(
json.dumps(
{
"views": 1200,
"ctr": 0.045,
"clicks": 54,
"conversions": 4,
"revenue": 600.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,
"youtube.com/watch?v=" + project_slug + "-video",
"vid-" + project_slug,
json.dumps(
{
"views": 1200,
"ctr": 0.045,
"clicks": 54,
"conversions": 4,
"revenue": 600.0,
}
),
),
)
conn.commit()
return {
"views": 1200,
"clicks": 54,
"conversions": 4,
"revenue": 600.0,
}
def create_podcast_asset(conn, project_id: str, project_slug: str, atoms: list[dict]) -> dict | None:
"""Создаёт audio asset (подкаст) + atom_assets (таймкоды) + podcast composition."""
if not atoms:
return None
asset_id = None
total_sec = 0
# Чистим старые audio-активы для проекта (идемпотентность)
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 = 'audio')",
(project_id,),
)
cur.execute("DELETE FROM assets WHERE project_id = %s AND asset_type = 'audio'", (project_id,))
conn.commit()
# Длительность эпизода — ~10 сек на атом (речевой формат)
for atom in atoms:
total_sec += 10
# Запись audio актива (подкаст)
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO assets
(project_id, asset_type, format, uri, title, duration_sec, meta)
SELECT %s, 'audio', 'mp3', %s, %s, %s, %s
WHERE NOT EXISTS (
SELECT 1 FROM assets WHERE project_id = %s AND asset_type = 'audio'
)
RETURNING id
""",
(
project_id,
"https://cdn.content-atom.local/" + project_slug + "-podcast.mp3",
f"{project_slug} — подкаст (media stage 3)",
total_sec,
json.dumps({"channel": "podcast", "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 = 10
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: podcast
atom_ids = [a["id"] for a in atoms]
recipe = {
"channel": "podcast",
"format": "audio",
"per_atom_seconds": 10,
"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, 'podcast', %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 + "-podcast",
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_podcast_publication(conn, project_id: str, project_slug: str) -> dict:
"""Создаёт публикацию подкаста (RSS/feed) с метриками."""
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO compositions
(project_id, channel, slug, title, status, recipe, atom_ids)
VALUES (%s, 'podcast', %s, %s, 'published', '{}', '{}')
ON CONFLICT (project_id, channel, slug) DO UPDATE
SET status = 'published'
RETURNING id
""",
(
project_id,
project_slug + "-podcast",
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, "%podcast%"),
)
row = cur.fetchone()
if row:
pub_id = row[0]
cur.execute(
"""
UPDATE publications SET metrics = %s WHERE id = %s
""",
(
json.dumps(
{
"downloads": 320,
"subscribers": 24,
"listen_rate": 0.78,
"conversions": 3,
"revenue": 450.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://feeds.content-atom.local/" + project_slug + "-podcast",
"pod-" + project_slug,
json.dumps(
{
"downloads": 320,
"subscribers": 24,
"listen_rate": 0.78,
"conversions": 3,
"revenue": 450.0,
}
),
),
)
conn.commit()
return {
"downloads": 320,
"subscribers": 24,
"listen_rate": 0.78,
"conversions": 3,
"revenue": 450.0,
}
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")
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"])
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']} подкаст, {news_pub['sent']} email, "
f"${pub['revenue'] + pod_pub['revenue'] + news_pub['revenue']}"
)
total_revenue += pub["revenue"] + pod_pub["revenue"] + news_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())
+182
View File
@@ -0,0 +1,182 @@
#!/usr/bin/env python3
"""
MVP stage 4: graph/lineage — обход связей атомов через Apache AGE.
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 → 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")),
"user": os.environ.get("POSTGRES_USER", "atom"),
"dbname": os.environ.get("POSTGRES_DB", "content_atom"),
"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"),
}
GRAPH_NAME = "content_atoms_graph"
def get_relations() -> list[dict]:
"""Связи из atom-db (Postgres 5434)."""
conn = psycopg2.connect(**ATOM_DB)
with conn.cursor() as cur:
cur.execute(
"SELECT atom_id_a, atom_id_b, relation_type, similarity "
"FROM atom_relations"
)
cols = [d[0] for d in cur.description]
rows = [dict(zip(cols, r)) for r in cur.fetchall()]
conn.close()
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, если нет."""
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")
# Удалить старые вершины/рёбра для идемпотентности
# AGE: рёбра нужно удалить до вершин → DETACH DELETE
_cypher(
"SET search_path TO ag_catalog;"
f"SELECT * FROM cypher('{GRAPH_NAME}', $$ "
f"MATCH (v) DETACH 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) через Cypher."""
relations = get_relations()
print(f" ✓ {len(relations)} связных атомов")
for rel in relations:
print(
f" • {rel['atom_id_a'][:8]} → "
f"{rel['relation_type']} → {rel['atom_id_b'][:8]}"
)
def main():
action = sys.argv[1] if len(sys.argv) > 1 else "all"
ensure_graph()
if action in ("sync", "all", "--all"):
sync_relations()
if action in ("query", "all", "--all"):
query_connected()
print(f"\nMVP stage 4 graph OK: action={action}")
return 0
if __name__ == "__main__":
sys.exit(main())
+28
View File
@@ -0,0 +1,28 @@
# Этап 5 — панель (дашборд смысла)
**Статус:** ✅ РЕАЛИЗОВАНО (2026-09-14)
**Цель:** единая сводка по атомам, версиям, каналам, доходу и графу.
## Запуск
```bash
python3 stage5/panel.py --all # всё: проекты + cost + граф
python3 stage5/panel.py --project rabota-ufa # только проект
python3 stage5/panel.py --graph # только граф (AGE)
```
## Что показывает
- **projects** — атомы по проектам (типы: claim/instruction/stat/qa), версии под каналы (web_article/telegram), активные каналы (newsletter/podcast/youtube).
- **cost** — стоимость запусков (`runs`, stage=embed) + доход по каналам (`publications.metrics.revenue`).
- **graph** — AGE-граф: вершины/рёбра (подсчёт из AGE), типы связей (duplicate), connected components (Union-Find из Postgres).
## Дизайн
- **Минимальная версия** — один Python-скрипт, без FastAPI/React (это отдельный V5).
- **Источник истины** — Postgres (atom_relations для связей/компонентов); AGE — только подсчёт вершин/рёбер (agtype не выводится в psql).
- **Идемпотентно** — только чтение, никаких записей.
## Ключевой факт
agtype-колонки AGE **не отображаются** в `psql`-выводе (известная проблема). Поэтому типы связей и connected components читаются из Postgres, а из AGE — только `count(*)` вершин/рёбер.
+223
View File
@@ -0,0 +1,223 @@
#!/usr/bin/env python3
"""
MVP stage 5: panel — CLI-дашборд смысла (atom-db + AGE).
Сводит в одну сводку: атомы по доменам, типы, версии под каналы,
композиции/публикации (доход), стоимость (runs), граф (AGE).
Запуск:
python3 stage5/panel.py --all
python3 stage5/panel.py --project rabota-ufa
python3 stage5/panel.py --graph # только граф (AGE)
"""
import os
import sys
import argparse
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"),
}
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"),
}
GRAPH_NAME = "content_atoms_graph"
def _query(conn, sql, params=None):
with conn.cursor() as cur:
cur.execute(sql, params)
cols = [d[0] for d in cur.description]
rows = [dict(zip(cols, r)) for r in cur.fetchall()]
return rows
def _psql_file(db, sql):
import subprocess
import tempfile
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 _age(sql):
return _psql_file(AGE_DB, sql)
def _age_count(sql):
out = _age(sql).strip().split("\n")
vals = [l.strip() for l in out if l.strip() and l.strip() != "SET"
and not l.strip().startswith("(")]
if not vals:
return 0
try:
return int(vals[-1])
except ValueError:
return 0
def show_projects(conn, only=None):
"""Атомы + версии по проектам."""
projects = _query(conn, "SELECT slug, role FROM projects ORDER BY slug")
print("\n=== projects (%d) ===" % len(projects))
for p in projects:
if only and p["slug"] != only:
continue
atoms = _query(conn,
"SELECT type, count(*) FROM atoms WHERE project_id="
"(SELECT id FROM projects WHERE slug=%s) GROUP BY type ORDER BY count(*) DESC",
(p["slug"],))
versions = _query(conn,
"SELECT channel, count(*) FROM atom_versions "
"WHERE atom_id IN (SELECT id FROM atoms WHERE project_id="
"(SELECT id FROM projects WHERE slug=%s)) GROUP BY channel ORDER BY count(*) DESC",
(p["slug"],))
pubs = _query(conn,
"SELECT c.channel, count(*) FROM compositions c "
"JOIN publications pub ON pub.composition_id=c.id "
"WHERE c.project_id=(SELECT id FROM projects WHERE slug=%s) "
"GROUP BY c.channel ORDER BY c.channel", (p["slug"],))
atom_str = ", ".join("%s:%d" % (a["type"], a["count"]) for a in atoms[:5])
version_str = ", ".join("%s:%d" % (v["channel"], v["count"]) for v in versions)
channel_str = ", ".join(p["channel"] for p in pubs)
print("\n - %s [%s]" % (p["slug"], p["role"]))
print(" atoms: %d (%s)" % (sum(a["count"] for a in atoms), atom_str))
print(" versions: %s" % version_str)
print(" channels: %s" % channel_str)
def show_cost(conn, only=None):
"""Стоимость (runs) + доход (publications)."""
runs = _query(conn,
"SELECT stage, count(*) AS n, sum(cost_usd) AS c FROM runs GROUP BY stage ORDER BY stage")
total_runs = sum(r["c"] for r in runs)
print("\n=== cost ===")
print(" runs: %d | spent: $%.4f" % (sum(r["n"] for r in runs), total_runs))
for r in runs:
print(" %s: %d | $%.4f" % (r["stage"], r["n"], r["c"]))
pubs = _query(conn,
"SELECT c.channel, COALESCE((pub.metrics->>'revenue')::numeric,0)::float AS revenue "
"FROM publications pub JOIN compositions c ON pub.composition_id=c.id")
by_channel = {}
for ch in pubs:
by_channel.setdefault(ch["channel"], 0.0)
by_channel[ch["channel"]] += ch["revenue"]
if by_channel:
print(" revenue: $%.2f" % sum(by_channel.values()))
for ch, rev in sorted(by_channel.items()):
print(" %s: $%.2f" % (ch, rev))
def show_graph(conn):
"""AGE-граф: вершины/рёбра/связность.
Связи/типы читаются из Postgres (atom_relations — источник истины),
из AGE — только подсчёт вершин/рёбер (agtype не выводится в psql).
"""
print("\n=== graph (AGE) ===")
vertices = _age_count(
"SET search_path TO ag_catalog;"
"SELECT count(*) FROM cypher('%s', $$ MATCH (v) RETURN v $$) AS (v agtype);" % GRAPH_NAME)
edges = _age_count(
"SET search_path TO ag_catalog;"
"SELECT count(*) FROM cypher('%s', $$ MATCH (a)-[]->(b) RETURN a $$) AS (a agtype);" % GRAPH_NAME)
print(" vertices: %d" % vertices)
print(" edges: %d" % edges)
# Типы связей (Postgres)
rel_types = _query(conn,
"SELECT relation_type, count(*) AS n FROM atom_relations "
"GROUP BY relation_type ORDER BY count(*) DESC")
if rel_types:
print(" edge types: %s" % ", ".join("%s:%d" % (r["relation_type"], r["n"]) for r in rel_types))
# Connected components (Union-Find) — из Postgres
rows = _query(conn,
"SELECT atom_id_a AS a, relation_type AS rel, atom_id_b AS b "
"FROM atom_relations ORDER BY atom_id_a")
pairs = [(r["a"], r["rel"], r["b"]) for r in rows]
parent = {}
def find(x):
parent.setdefault(x, x)
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
def union(a, b):
parent[find(a)] = find(b)
all_nodes = set()
for a, _, b in pairs:
all_nodes.add(a)
all_nodes.add(b)
union(a, b)
comp_map = {}
for n in all_nodes:
root = find(n)
comp_map.setdefault(root, [])
comp_map[root].append(n)
for root, members in sorted(comp_map.items(),
key=lambda kv: len(kv[1]), reverse=True):
print(" component: %d atoms" % len(members))
if len(members) <= 4:
for m in members:
print(" - %s" % m[:8])
print(" total components: %d" % len(comp_map))
def main():
ap = argparse.ArgumentParser(description="MVP stage 5: panel")
ap.add_argument("--project", default=None, help="project slug")
ap.add_argument("--all", dest="all", action="store_true", help="всё")
ap.add_argument("--graph", action="store_true", help="только граф (AGE)")
args = ap.parse_args()
conn = psycopg2.connect(**POSTGRES)
if args.graph:
show_graph(conn)
else:
show_projects(conn, args.project)
show_cost(conn, args.project)
if args.all or not args.project:
show_graph(conn)
conn.close()
return 0
if __name__ == "__main__":
sys.exit(main())