From 769b16df47315344e2ccba2a8a5386dc82ac4264 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Mon, 11 May 2026 12:07:57 +0300 Subject: [PATCH] =?UTF-8?q?fix(worker):=20port=20objective=5Fetl=20from=20?= =?UTF-8?q?psycopg2=20=E2=86=92=20psycopg=20v3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Worker крашился на старте с `ModuleNotFoundError: No module named 'psycopg2'` при импорте app/workers/tasks/objective_etl, потому что app/services/objective_etl писал на psycopg2 API, а в pyproject.toml есть только psycopg[binary]>=3.2.0. Изменения: - psycopg2 → psycopg (v3) - psycopg2.extras.execute_values → _bulk_upsert helper, эмулирующий тот же behavior через cur.executemany (в psycopg v3 это pipeline-mode, overhead на сетевые round-trips минимален) - psycopg.connect совместим с тем же connection string Тесты: AST parse ✓, ruff ✓, import ✓ --- backend/app/services/objective_etl.py | 34 +- backend/app/workers/tasks/objective_etl.py | 4 +- scripts/cleanup_ghosts.py | 113 +++ scripts/migrate_kg_to_obsidian.py | 940 +++++++++++++++++++++ sshkey.txt | 9 + 5 files changed, 1089 insertions(+), 11 deletions(-) create mode 100644 scripts/cleanup_ghosts.py create mode 100644 scripts/migrate_kg_to_obsidian.py diff --git a/backend/app/services/objective_etl.py b/backend/app/services/objective_etl.py index 045f5aad..7bdc201c 100644 --- a/backend/app/services/objective_etl.py +++ b/backend/app/services/objective_etl.py @@ -28,11 +28,27 @@ from datetime import date from pathlib import Path from typing import Any -import psycopg2 -import psycopg2.extras +import psycopg logger = logging.getLogger(__name__) + +def _bulk_upsert(cur: psycopg.Cursor, sql_template: str, rows: list[tuple]) -> None: + """Эмулирует psycopg2.extras.execute_values для psycopg v3. + + `sql_template` должен содержать литеральное `VALUES %s` — заменим + на `VALUES (%s, %s, ..., %s)` с числом placeholder'ов = len(rows[0]) + и пошлём через executemany. В psycopg3 executemany использует + pipeline-mode, так что overhead на сетевые round-trips минимален. + """ + if not rows: + return + n_cols = len(rows[0]) + values_placeholder = "(" + ",".join(["%s"] * n_cols) + ")" + sql = sql_template.replace("VALUES %s", "VALUES " + values_placeholder) + cur.executemany(sql, rows) + + GROUP_NAME = "Екатеринбург" # У Антона в SQLite нет колонки group_name @@ -356,14 +372,14 @@ def _etl_lots( for r in rows: batch.append(_row_to_lot(r, snap)) if len(batch) >= batch_size: - psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=batch_size) + _bulk_upsert(pg_cur, _LOT_INSERT, batch) pg.commit() n += len(batch) batch.clear() if progress_cb and n % (batch_size * 5) == 0: progress_cb(n) if batch: - psycopg2.extras.execute_values(pg_cur, _LOT_INSERT, batch, page_size=len(batch)) + _bulk_upsert(pg_cur, _LOT_INSERT, batch) pg.commit() n += len(batch) pg_cur.close() @@ -391,12 +407,12 @@ def _etl_crm( continue batch.append(t) if len(batch) >= batch_size: - psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=batch_size) + _bulk_upsert(pg_cur, _CRM_INSERT, batch) pg.commit() n += len(batch) batch.clear() if batch: - psycopg2.extras.execute_values(pg_cur, _CRM_INSERT, batch, page_size=len(batch)) + _bulk_upsert(pg_cur, _CRM_INSERT, batch) pg.commit() n += len(batch) pg_cur.close() @@ -424,7 +440,7 @@ def _etl_mapping(sl: sqlite3.Connection, pg) -> tuple[int, int]: if not batch: return 0, skipped_dups pg_cur = pg.cursor() - psycopg2.extras.execute_values(pg_cur, _MAP_INSERT, batch, page_size=500) + _bulk_upsert(pg_cur, _MAP_INSERT, batch) pg.commit() pg_cur.close() return len(batch), skipped_dups @@ -483,7 +499,7 @@ def run_etl( Raises: FileNotFoundError если SQLite не найден - psycopg2.Error при ошибках БД + psycopg.Error при ошибках БД """ p = Path(sqlite_path) if not p.exists(): @@ -492,7 +508,7 @@ def run_etl( logger.info("ETL Anton SQLite → PG: %s", p) sl = sqlite3.connect(p) sl.row_factory = sqlite3.Row - pg = psycopg2.connect(db_url) + pg = psycopg.connect(db_url) try: if progress_cb: progress_cb("mapping_start", 0) diff --git a/backend/app/workers/tasks/objective_etl.py b/backend/app/workers/tasks/objective_etl.py index 6e9d12bd..da447a15 100644 --- a/backend/app/workers/tasks/objective_etl.py +++ b/backend/app/workers/tasks/objective_etl.py @@ -109,8 +109,8 @@ def import_anton_objective( run_id = _start_run(db, triggered_by) logger.info("import_anton_objective started, run_id=%s, sqlite=%s", run_id, sqlite_path_eff) - # Преобразуем sqlalchemy-style db_url в psycopg2-style - # (psycopg2 не принимает 'postgresql+psycopg://'). + # Преобразуем sqlalchemy-style db_url в обычный postgresql:// + # (psycopg.connect принимает postgresql://, не postgresql+psycopg://). db_url = settings.database_url.replace("postgresql+psycopg://", "postgresql://") # Прогресс-callback пишет в БД счётчики + heartbeat diff --git a/scripts/cleanup_ghosts.py b/scripts/cleanup_ghosts.py new file mode 100644 index 00000000..ab07d17a --- /dev/null +++ b/scripts/cleanup_ghosts.py @@ -0,0 +1,113 @@ +"""Удаляет ghost docs из decisions/, sessions/, sources/ в CouchDB — +те что есть в CouchDB но отсутствуют в локальном vault. +""" +import os +from pathlib import Path +from urllib.parse import quote + +import requests + +VAULT = Path(r"C:\Users\user\Documents\gendesign-memory") +BASE = "https://obsidian.gendsgn.ru/gendesign-vault" +AUTH = ("obsidian", os.environ["COUCHDB_PASSWORD"]) +TARGETS = ("decisions/", "sessions/", "sources/") + + +def main() -> None: + # 1. Local files (case-insensitive set, forward-slash paths) + local: set[str] = set() + for top in ("decisions", "sessions", "sources"): + d = VAULT / top + if d.exists(): + for f in d.rglob("*.md"): + rel = str(f.relative_to(VAULT)).replace(os.sep, "/").lower() + local.add(rel) + print(f"Local files in target dirs: {len(local)}") + + # 2. Fetch CouchDB docs + r = requests.get(f"{BASE}/_all_docs", auth=AUTH, + params={"limit": 10000}, timeout=60) + rows = r.json()["rows"] + + # 3. Identify ghosts + ghosts: list[tuple[str, str]] = [] + keepers: list[str] = [] + for row in rows: + did = row["id"] + if not did.endswith(".md"): + continue + if not did.startswith(TARGETS): + continue + if did.lower() in local: + keepers.append(did) + else: + ghosts.append((did, row["value"]["rev"])) + + print(f"Keepers (matched local): {len(keepers)}") + print(f"Ghosts (no local match): {len(ghosts)}") + for did, _ in ghosts[:30]: + print(f" ghost: {did}") + if len(ghosts) > 30: + print(f" ... +{len(ghosts) - 30} more") + + if not ghosts: + print("Nothing to do.") + return + + # 4. Сначала найдём chunks которые ссылаются ghosts. Это потенциальные orphans. + ghost_chunk_refs: set[str] = set() + for did, _ in ghosts: + rd = requests.get(f"{BASE}/{quote(did, safe='')}", auth=AUTH, timeout=30) + if rd.status_code == 200: + d = rd.json() + for cid in d.get("children", []): + ghost_chunk_refs.add(cid) + for cid in d.get("eden", {}): + ghost_chunk_refs.add(cid) + print(f"Chunks referenced by ghosts: {len(ghost_chunk_refs)}") + + # 5. Проверяем у скольких из них ещё есть alive parent + ghost_ids = {did for did, _ in ghosts} + alive_md = [r["id"] for r in rows + if r["id"].endswith(".md") and r["id"] not in ghost_ids] + print(f"Scanning {len(alive_md)} alive .md docs for chunk refs...") + alive_chunks: set[str] = set() + for i, did in enumerate(alive_md): + if i % 100 == 0 and i > 0: + print(f" {i}/{len(alive_md)}") + rd = requests.get(f"{BASE}/{quote(did, safe='')}", auth=AUTH, timeout=30) + if rd.status_code == 200: + d = rd.json() + for cid in d.get("children", []): + alive_chunks.add(cid) + for cid in d.get("eden", {}): + alive_chunks.add(cid) + + orphans = ghost_chunk_refs - alive_chunks + print(f"Orphan chunks (only ghosts reference them): {len(orphans)}") + + # 6. Получаем _rev для orphan chunks + chunk_revs: dict[str, str] = {} + chunk_id_to_rev = {r["id"]: r["value"]["rev"] for r in rows if r["id"].startswith("h:")} + for cid in orphans: + if cid in chunk_id_to_rev: + chunk_revs[cid] = chunk_id_to_rev[cid] + + # 7. Bulk delete + del_docs = [{"_id": did, "_rev": rev, "_deleted": True} for did, rev in ghosts] + del_docs.extend({"_id": cid, "_rev": rev, "_deleted": True} + for cid, rev in chunk_revs.items()) + + print(f"Bulk delete {len(del_docs)} docs ({len(ghosts)} ghost main + " + f"{len(chunk_revs)} orphan chunks)...") + rb = requests.post(f"{BASE}/_bulk_docs", json={"docs": del_docs}, + auth=AUTH, timeout=300) + if rb.status_code in (200, 201): + ok = sum(1 for x in rb.json() if x.get("ok")) + print(f"Deleted: {ok}/{len(del_docs)}") + else: + print(f"Bulk delete failed: HTTP {rb.status_code} {rb.text[:200]}") + + +if __name__ == "__main__": + main() diff --git a/scripts/migrate_kg_to_obsidian.py b/scripts/migrate_kg_to_obsidian.py new file mode 100644 index 00000000..d1682306 --- /dev/null +++ b/scripts/migrate_kg_to_obsidian.py @@ -0,0 +1,940 @@ +"""Миграция knowledge graph (memory-gendesign.jsonl) → Obsidian vault через CouchDB. + +Читает текущий KG и записывает каждую entity как `.md` файл прямо в CouchDB +в формате Self-hosted LiveSync (Vrtmrz/obsidian-livesync). Клиенты Obsidian +с LiveSync plugin автоматически подтянут изменения. + +Формат LiveSync v0.x (classic chunks + children): + Chunk document: + {_id: "h:", data: "<содержимое>", type: "leaf"} + + Main document: + {_id: "entities/Foo.md", path: "entities/Foo.md", + children: ["h:..."], ctime, mtime, size, type:"plain", eden:{}} + + eden — write-side staging area, при чтении не используется. Поэтому контент + ОБЯЗАН быть в отдельных chunk-docs (LiveSync читает через children). + PUT-ordering: chunks first, main docs second — иначе клиент видит пустой + body во время replication race. + +Структура vault: + sessions/ — Session_End_* entities + entities/ — Module_*, Schema_*, прочие + decisions/ — *_Plan_*, *_Strategy_* (ADR-style) + runbooks/ — *_Scraper_*, *_Pipeline_* (how-to) + sources/ — DataSources_* (источники данных) + misc/ — всё остальное + +Запуск: + cd backend + COUCHDB_URL=https://obsidian.gendsgn.ru \ + COUCHDB_USER=obsidian \ + COUCHDB_PASSWORD=<пароль> \ + COUCHDB_DB=gendesign-vault \ + uv run --no-project --with requests python -X utf8 \ + ../scripts/migrate_kg_to_obsidian.py [--dry-run] +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import logging +import os +import re +import sys +import time +from collections import defaultdict +from pathlib import Path +from typing import Any +from urllib.parse import quote + +import requests + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger("kg_to_obsidian") + +REPO_ROOT = Path(__file__).resolve().parent.parent +KG_PATH = REPO_ROOT / "memory" / "memory-gendesign.jsonl" + + +_B32 = "0123456789abcdefghijklmnopqrstuv" + + +def chunk_id_for(content: str) -> str: + """Генерирует chunk_id похожий на LiveSync internal format: + `h:` + 13 base32-like chars. Уникальность достаточна — collision space 2^65. + + LiveSync встроенный generator использует свой base32-кодированный hash. + Подражаем формату чтобы клиент не отбраковал по validation. + """ + digest = hashlib.sha256(content.encode("utf-8")).digest() + n = int.from_bytes(digest[:8], "big") + out = [] + for _ in range(13): + out.append(_B32[n & 0x1F]) + n >>= 5 + return "h:" + "".join(out) + + +_MONTH_MAP = { + "jan": "01", "feb": "02", "mar": "03", "apr": "04", "may": "05", "jun": "06", + "jul": "07", "aug": "08", "sep": "09", "oct": "10", "nov": "11", "dec": "12", +} +_DATE_PATTERNS = [ + # Apr27_2026, Apr30_2026, Apr29_2026_Late + re.compile(r"\b(?PJan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)(?P\d{1,2})_(?P20\d{2})\b", re.IGNORECASE), + # Apr26 (year implied 2026) + re.compile(r"\b(?PJan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)(?P\d{1,2})\b(?!_20)", re.IGNORECASE), + # Nov2025 + re.compile(r"\b(?PJan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)(?P20\d{2})\b", re.IGNORECASE), + # Q1_2026, Q2_2026 — quarter→month + re.compile(r"\bQ(?P[1-4])_(?P20\d{2})\b"), +] + + +def extract_date(name: str) -> str | None: + """Из имени entity извлекает ISO-дату (YYYY-MM-DD или YYYY-MM). + + Apr26 → 2026-04-26 (год дефолтный 2026 если не указан) + Nov2025 → 2025-11 + Q1_2026 → 2026-Q1 (для квартала) + """ + for pat in _DATE_PATTERNS: + m = pat.search(name) + if not m: + continue + gd = m.groupdict() + if "q" in gd: + return f"{gd['y']}-Q{gd['q']}" + mon = _MONTH_MAP[gd["m"][:3].lower()] + if "d" in gd: + y = gd.get("y") or "2026" + return f"{y}-{mon}-{int(gd['d']):02d}" + return f"{gd['y']}-{mon}" + return None + + +# Категоризация: входим в первое совпавшее правило. Порядок важен — +# specific raньше generic. +def categorize_v2(name: str, entity_type: str) -> str: + """Возвращает многоуровневый путь поддиректории (без имени файла, без расширения). + + Стратегия — group по бизнес-домену, потом по технической роли. + """ + n = name + et = (entity_type or "").lower() + + # 1. Top-level meta — флагманы проекта + if n in {"Project_GenDesign", "TechStack_v2.2", "NorthStar_DiscoveryMVP", + "Phase0_DiscoveryMVP", "Stages_MVP_v1", "Team_Setup", + "Onboarding_Colleague_KG", "WorkSplit_Generative_Stage1"}: + return "meta" + + # 2. Sprints + if n.startswith("Sprint") and ("_" in n): + return "sprints" + + # 3. Sessions — chronological logs + if n.startswith("Session") or et == "session": + return "sessions" + + # 4. Bug fixes + if n.startswith("Bug_") or et == "bug": + return "fixes" + + # 5. GitHub + if n.startswith("GitHub_"): + return "github" + + # 6. Domain: DOM.РФ + if n.startswith("DomRF") or n == "DomRf_Kn_Scraper_Apr27": + if "_API_" in n or n.startswith("DomRF_API"): + return "domains/domrf/api" + if any(w in n for w in ("_Loaded_", "_Imported_", "_Wipe_")): + return "domains/domrf/events" + if "_Schema_" in n or n.endswith("_Schema"): + return "domains/domrf/schemas" + if "_Scraper" in n: + return "domains/domrf/scraper" + if "Enums" in n or "Architecture" in n or "Host" in n or "Integration" in n: + return "domains/domrf/api" + return "domains/domrf" + + # 7. Domain: Cadastre (NSPD + Cad_*) + if n.startswith("NSPD") or n.startswith("Cad_") or n == "Cad_Quarter_Backfill_Apr30": + return "domains/cadastre" + + # 8. Domain: Rosreestr + if (n.startswith("Rosreestr") or n.startswith("Sverdl_") or + (n.startswith("DataSource_Rosreestr") or n == "Backfill_District_Median_From_Kn_Flats")): + return "domains/rosreestr" + + # 9. Domain: Analytics + if any(w in n for w in ("Recommend", "Velocity", "Affordability", "DemandIndex", + "AlternativeDeal", "Competitor_Benchmark", "Macro_Context", + "Mortgage", "PriceGap", "Tier2", "Tier_Alpha", "Quartirography", + "RegionGroups", "Regional_Migration", "RepeatedMortgage", + "QualityShift")): + if n.startswith("Component_") or n.startswith("Feature_"): + return "domains/analytics/ui" + if n.startswith("Endpoint_"): + return "domains/analytics/api" + if n.startswith("Hook_"): + return "domains/analytics/code" + if n.startswith("Bug_"): + return "fixes" + if "Trend" in n or "Index" in n or "Report" in n: + return "domains/analytics/research" + return "domains/analytics" + + # 10. Domain: UI components/routes (общие, не analytics-specific) + if n.startswith("Component_") or n.startswith("Feature_") or n.startswith("UI_"): + return "domains/ui/components" + if n.startswith("Route_"): + return "domains/ui/routes" + + # 11. Domain: API endpoints + if n.startswith("Endpoint_"): + return "domains/api/endpoints" + + # 12. Domain: Geo + if (n.startswith("Geo_") or n == "Pattern_OSM_Overpass_Admin_Level" + or n == "Limit_District_Geometry_Closed" or n == "File_Ekb_Districts_Geojson_Cache"): + return "domains/geo" + + # 13. Domain: Infra (deploy, docker, hosting, etc) + if any(w in n for w in ("Docker", "Deploy", "Sentry", "Hosting", "SSH", + "OpsBundle", "Domain_Setup", "HTTPS", "Redis", + "Caddy", "Celery", "TTL_Lock", "Stealth_Fingerprint", + "Runbook_VPS", "Production_Wipe")): + return "domains/infra" + + # 14. Code: Modules / Schemas / Patterns / Hooks / Tasks + if n.startswith("Module_") or et == "code_module": + return "code/modules" + if n.startswith("Schema_") or et == "schema": + return "code/schemas" + if n.startswith("Pattern_"): + return "code/patterns" + if n.startswith("Hook_"): + return "code/hooks" + if n.startswith("Task_"): + return "code/tasks" + + # 15. Sources (data source research, non-Rosreestr) + if n.startswith("DataSource_") or n.startswith("DataSources_"): + return "sources" + + # 16. Limitations + if n.startswith("Limit_") or n == "Open_Limitations_Apr30" or et == "limitation": + return "limitations" + + # 17. Decisions (plans, strategies, policies) + if n.startswith("PlanA_") or n.startswith("Plan_") or "_Strategy_" in n: + return "decisions" + if et in ("policy", "plan", "decision", "decision_pending", "strategy"): + return "decisions" + + # 18. Events / milestones / objectives + if (n.startswith("Objective_") or "_Loaded_" in n or "_Imported_" in n + or n.startswith("Phase_") or et in ("event", "milestone")): + return "events" + + # 19. Feedback + if n.startswith("Feedback_") or et == "feedback" or "MCP_AddFacts" in n: + return "feedback" + + # 20. Research (generic) + if et == "research": + return "research" + + # 21. Tests + if et == "test" or n.endswith("_Test_Apr25"): + return "research" + + # 22. Reviews + if et == "review": + return "research" + + # 23. Build (SiteFinder etc) + if et == "build" or n.startswith("SiteFinder"): + return "domains/sitefinder" + + # 24. Deliverable — most likely analytics или infra + if et == "deliverable": + return "events" + + # 25. Infrastructure type + if et in ("infrastructure", "infra"): + return "domains/infra" + + return "unsorted" + + +_TAG_RE = re.compile(r"[^a-z0-9]+") + + +def tags_for_path(subdir: str, entity_type: str) -> list[str]: + """Генерит теги из subpath + entity_type. Пример: + domains/domrf/api + api_spec → [domrf, api, domrf-api, api-spec] + """ + parts = [p for p in subdir.split("/") if p and p not in ("domains", "code")] + tags = list(dict.fromkeys(parts)) # uniq, preserve order + # composite tag — domain-subdomain + if len(parts) >= 2: + tags.append(f"{parts[0]}-{parts[1]}") + if entity_type: + et_norm = _TAG_RE.sub("-", entity_type.lower()).strip("-") + if et_norm and et_norm not in tags: + tags.append(et_norm) + return tags + + +# Старая функция для совместимости с CouchDB-режимом (он сейчас не используется, +# но всё ещё есть в коде на случай отката). +def categorize(name: str, entity_type: str) -> str: + subdir = categorize_v2(name, entity_type) + return subdir.split("/")[0] + + +_SLUG_RE = re.compile(r"[^a-zA-Zа-яА-Я0-9_\- ]+") + + +def safe_filename(name: str) -> str: + """Превращает name в файл-безопасное имя (без / : * etc).""" + s = _SLUG_RE.sub("", name).strip().replace(" ", "_") + return s[:200] # Obsidian/Windows path limit + + +def md_for_entity(entity: dict, outgoing: list[dict], incoming: list[dict], + name_to_path: dict[str, str], subdir: str | None = None) -> str: + """Генерит .md содержимое entity с frontmatter (теги + дата), observations, + relations через короткие [[wiki-links]] (Obsidian резолвит по basename). + """ + name = entity["name"] + etype = entity.get("entityType", "unknown") + observations = entity.get("observations") or [] + date = extract_date(name) + tags = tags_for_path(subdir or "", etype) if subdir else [] + + lines: list[str] = [] + + # YAML frontmatter + lines.append("---") + lines.append(f"id: {name}") + lines.append(f"type: {etype}") + if date: + lines.append(f"date: {date}") + if tags: + lines.append(f"tags: [{', '.join(tags)}]") + lines.append("imported_from: knowledge_graph") + lines.append(f"imported_at: {time.strftime('%Y-%m-%d')}") + lines.append("---") + lines.append("") + lines.append(f"# {name}") + lines.append("") + meta_bits = [f"**Type:** `{etype}`"] + if date: + meta_bits.append(f"**Date:** {date}") + if subdir: + # breadcrumb: накопленный path → его MOC + crumbs = [] + parts = subdir.split("/") + for i in range(len(parts)): + sub = "/".join(parts[: i + 1]) + crumbs.append(f"[[{_moc_id(sub)}|{parts[i]}]]") + meta_bits.append("**Path:** " + " / ".join(crumbs)) + lines.append(" · ".join(meta_bits)) + lines.append("") + + if observations: + lines.append("## Observations") + lines.append("") + for obs in observations: + txt = obs.replace("\n", " ").strip() + lines.append(f"- {txt}") + lines.append("") + + if outgoing: + lines.append("## Relations (outgoing)") + lines.append("") + # группируем по relationType — нагляднее + by_rel: dict[str, list[str]] = defaultdict(list) + for rel in outgoing: + by_rel[rel["relationType"]].append(rel["to"]) + for rtype, targets in sorted(by_rel.items()): + lines.append(f"### {rtype}") + for target in targets: + if target in name_to_path: + lines.append(f"- [[{target}]]") + else: + lines.append(f"- {target} *(no entity in KG)*") + lines.append("") + + if incoming: + lines.append("## Relations (incoming)") + lines.append("") + by_rel2: dict[str, list[str]] = defaultdict(list) + for rel in incoming: + by_rel2[rel["relationType"]].append(rel["from"]) + for rtype, sources in sorted(by_rel2.items()): + lines.append(f"### {rtype} ← from") + for src in sources: + if src in name_to_path: + lines.append(f"- [[{src}]]") + else: + lines.append(f"- {src} *(no entity)*") + lines.append("") + + return "\n".join(lines) + + +def put_doc( + base_url: str, db: str, auth: tuple[str, str], doc_id: str, doc: dict, + dry_run: bool = False, +) -> bool: + """PUT/POST документ. При conflict — GET _rev и PUT с _rev.""" + if dry_run: + return True + url = f"{base_url}/{db}/{quote(doc_id, safe='')}" + r = requests.put(url, json=doc, auth=auth, timeout=30) + if r.status_code in (201, 202): + return True + if r.status_code == 409: + # Conflict — try get existing and retry with _rev + rg = requests.get(url, auth=auth, timeout=30) + if rg.status_code == 200: + existing = rg.json() + doc["_rev"] = existing["_rev"] + r2 = requests.put(url, json=doc, auth=auth, timeout=30) + if r2.status_code in (201, 202): + return True + logger.warning("Retry-after-conflict PUT %s: HTTP %d %s", + doc_id, r2.status_code, r2.text[:200]) + return False + logger.warning("PUT %s: HTTP %d %s", doc_id, r.status_code, r.text[:200]) + return False + + +def delete_doc(base_url: str, db: str, auth: tuple[str, str], doc_id: str) -> bool: + """DELETE документ (для перезаписи старой broken-версии).""" + url = f"{base_url}/{db}/{quote(doc_id, safe='')}" + rg = requests.get(url, auth=auth, timeout=30) + if rg.status_code != 200: + return True # уже нет — OK + rev = rg.json()["_rev"] + r = requests.delete(f"{url}?rev={rev}", auth=auth, timeout=30) + return r.status_code in (200, 202) + + +def _load_kg() -> tuple[dict[str, dict], list[dict]]: + """Загружает KG из JSONL: returns (entities_by_name, relations_list).""" + entities: dict[str, dict] = {} + relations: list[dict] = [] + with KG_PATH.open(encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + d = json.loads(line) + except json.JSONDecodeError: + continue + if d.get("type") == "entity": + entities[d["name"]] = d + elif d.get("type") == "relation": + relations.append(d) + return entities, relations + + +_TOP_LEVEL_DESCRIPTIONS = { + "meta": "🧭 Проект, vision, стратегия, команда", + "sprints": "🏃 Sprint plans", + "sessions": "📅 Журналы сессий разработки", + "events": "📌 События — milestones, объективы, deliverables", + "decisions": "🎯 Решения, планы, политики (ADR-style)", + "domains": "🗺 Бизнес-домены — DomRF, Cadastre, Rosreestr, Analytics, UI, API, Geo, Infra", + "code": "🔧 Code modules, schemas, patterns, hooks, tasks", + "sources": "💾 Источники данных (research)", + "research": "🔬 Generic research, отчёты", + "fixes": "🐞 Bug fixes", + "limitations": "🚧 Ограничения, что не работает", + "github": "🐙 GitHub issues", + "feedback": "💬 Feedback / TODO заметки от пользователя", + "unsorted": "❓ Не сгруппировано", +} + + +def _moc_id(subdir: str) -> str: + """Уникальный (per-vault) идентификатор MOC-файла для папки subdir. + Использует дефис-разделённый path: domains/domrf/api → domrf-api-MOC. + Префикс 'domains' опускается у вложенных, но сам domains→domains-MOC. + """ + parts = subdir.split("/") + if len(parts) > 1 and parts[0] == "domains": + parts = parts[1:] + return "-".join(parts) + "-MOC" + + +def _moc_for_subdir(subdir: str, entities_in_subdir: list[str], + entities_all: dict[str, dict], + subdirs_below: list[str]) -> str: + """Генерит MOC.md (Map of Content) для папки subdir.""" + parts = subdir.split("/") + top = parts[0] + title = " / ".join(parts) if len(parts) > 1 else parts[0] + desc = _TOP_LEVEL_DESCRIPTIONS.get(top, "") + moc_id = _moc_id(subdir) + + lines = [ + "---", + f"id: {moc_id}", + "type: moc", + f"tags: [moc, {top}]", + f"imported_at: {time.strftime('%Y-%m-%d')}", + "---", + "", + f"# 🗂 {title}", + "", + ] + if desc and len(parts) == 1: + lines.append(f"> {desc}") + lines.append("") + + if subdirs_below: + lines.append("## Подразделы") + lines.append("") + for sd in sorted(subdirs_below): + sub_moc = _moc_id(sd) + sd_label = sd.split("/")[-1] + lines.append(f"- [[{sub_moc}|{sd_label}/]]") + lines.append("") + + if entities_in_subdir: + lines.append(f"## Entities ({len(entities_in_subdir)})") + lines.append("") + # сортируем по дате (новые сверху), затем по имени + def sort_key(name: str) -> tuple[str, str]: + d = extract_date(name) or "0000" + return (d, name) + for n in sorted(entities_in_subdir, key=sort_key, reverse=True): + ent = entities_all.get(n, {}) + et = ent.get("entityType", "") + d = extract_date(n) + d_str = f" · `{d}`" if d else "" + et_str = f" · *{et}*" if et else "" + lines.append(f"- [[{n}]]{d_str}{et_str}") + lines.append("") + + return "\n".join(lines) + + +def _global_index_md(entities: dict[str, dict], relations_n: int, + by_subdir: dict[str, list[str]]) -> str: + """Главный индекс (00_index.md) — entry point в vault.""" + lines = [ + "---", + "id: 00_index", + "type: index", + "tags: [index, hub]", + f"imported_at: {time.strftime('%Y-%m-%d')}", + "---", + "", + "# 🗂 Knowledge Graph — GenDesign", + "", + f"Vault содержит **{len(entities)} entities** и **{relations_n} relations**, " + "мигрированных из `memory/memory-gendesign.jsonl`.", + "", + "## Top-level разделы", + "", + ] + # Группируем top-level папки + top_level: dict[str, int] = defaultdict(int) + for sd, names in by_subdir.items(): + top_level[sd.split("/")[0]] += len(names) + for top in sorted(top_level): + desc = _TOP_LEVEL_DESCRIPTIONS.get(top, "") + lines.append(f"- [[{_moc_id(top)}|**{top}/**]] ({top_level[top]} entities) — {desc}") + lines.append("") + + lines.append("## Граф") + lines.append("") + lines.append("Открой Graph View (Ctrl+G) — увидишь паутину связей entities по `[[wiki-links]]`.") + lines.append("") + lines.append("Фильтр по тегу: `tag:#domrf`, `tag:#analytics`, `tag:#cadastre` и т.д.") + lines.append("") + return "\n".join(lines) + + +def _clean_vault_subdirs(vault_path: Path, keep_files: set[str]) -> int: + """Удаляет из vault'а всё ранее сгенерированное (по old structure). + Сохраняет файлы из keep_files (user content: welcome.md и пр). + Возвращает кол-во удалённых файлов. + """ + removed = 0 + OLD_DIRS = ("sessions", "entities", "decisions", "runbooks", "sources", + "meta", "sprints", "events", "domains", "code", "research", + "fixes", "limitations", "github", "feedback", "unsorted") + for sub in OLD_DIRS: + d = vault_path / sub + if not d.exists(): + continue + for f in d.rglob("*.md"): + try: + f.unlink() + removed += 1 + except OSError as e: + logger.warning("Не удалить %s: %s", f, e) + # удалить пустые папки (rmdir каждый уровень снизу вверх) + for sub_d in sorted(d.rglob("*"), key=lambda p: -len(p.parts)): + if sub_d.is_dir(): + try: + sub_d.rmdir() + except OSError: + pass + try: + d.rmdir() + except OSError: + pass + # Удалить 00_index.md из root если есть + idx = vault_path / "00_index.md" + if idx.exists() and "00_index.md" not in keep_files: + idx.unlink() + removed += 1 + return removed + + +def _write_to_local_vault(vault_path: Path, limit: int | None) -> int: + """Пишет .md файлы напрямую в локальную папку Obsidian vault с иерархической + структурой + MOC-страницы для навигации. + + LiveSync plugin сам подхватит изменения через filesystem watcher и + загрузит их в CouchDB в V3-chunked формате. + """ + if not vault_path.exists() or not vault_path.is_dir(): + logger.error("Vault path не существует: %s", vault_path) + return 2 + + entities, relations = _load_kg() + logger.info("Loaded %d entities, %d relations", len(entities), len(relations)) + + # Cleanup старой структуры (сохраняем user-files) + keep = {"welcome.md", "qqq.md", "untitled.md", "2026-05-11.md"} + removed = _clean_vault_subdirs(vault_path, keep) + logger.info("Cleaned %d old .md files from vault", removed) + + # Relation indices + out_by: dict[str, list[dict]] = defaultdict(list) + in_by: dict[str, list[dict]] = defaultdict(list) + for r in relations: + out_by[r["from"]].append(r) + in_by[r["to"]].append(r) + + # Path resolution + name_to_subdir: dict[str, str] = {} + name_to_path: dict[str, str] = {} + for name, ent in entities.items(): + subdir = categorize_v2(name, ent.get("entityType", "")) + fname = safe_filename(name) + ".md" + name_to_subdir[name] = subdir + name_to_path[name] = f"{subdir}/{fname}" + + names = list(entities.keys()) + if limit: + names = names[:limit] + logger.info("Will write %d entities", len(names)) + + # PASS 1: запись всех entities + ok = 0 + for i, name in enumerate(names, 1): + ent = entities[name] + rel_path = name_to_path[name] + subdir = name_to_subdir[name] + seen: set[tuple[str, str, str]] = set() + out_uniq: list[dict] = [] + for r in out_by[name]: + key = (r["from"], r["to"], r["relationType"]) + if key in seen: + continue + seen.add(key) + out_uniq.append(r) + in_uniq: list[dict] = [] + for r in in_by[name]: + key = (r["from"], r["to"], r["relationType"]) + if key in seen: + continue + seen.add(key) + in_uniq.append(r) + content = md_for_entity(ent, out_uniq, in_uniq, name_to_path, subdir) + + target = vault_path / rel_path + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + ok += 1 + + logger.info("Entities written: %d", ok) + + # PASS 2: MOC-страницы (для каждой подпапки) + by_subdir: dict[str, list[str]] = defaultdict(list) + for name in names: + by_subdir[name_to_subdir[name]].append(name) + + # Найти все уровни подпапок и их subdirs + all_subdirs: set[str] = set() + for sd in by_subdir: + parts = sd.split("/") + for i in range(1, len(parts) + 1): + all_subdirs.add("/".join(parts[:i])) + + moc_count = 0 + for sd in sorted(all_subdirs): + entities_here = by_subdir.get(sd, []) + # подпапки на уровень ниже + subdirs_below = [s for s in all_subdirs + if s.startswith(sd + "/") and s.count("/") == sd.count("/") + 1] + moc_content = _moc_for_subdir(sd, entities_here, entities, subdirs_below) + moc_filename = _moc_id(sd) + ".md" + moc_path = vault_path / sd / moc_filename + moc_path.parent.mkdir(parents=True, exist_ok=True) + moc_path.write_text(moc_content, encoding="utf-8") + moc_count += 1 + + logger.info("MOC pages written: %d", moc_count) + + # PASS 3: глобальный индекс + index_content = _global_index_md(entities, len(relations), by_subdir) + (vault_path / "00_index.md").write_text(index_content, encoding="utf-8") + + logger.info("DONE: %d entities + %d MOC + 1 index в %s", + ok, moc_count, vault_path) + logger.info("Открой Obsidian — LiveSync plugin отсканирует и запушит в CouchDB.") + return 0 + + +def _cleanup_orphan_chunks(base_url: str, db: str, auth: tuple[str, str], + dry_run: bool) -> int: + """Удаляет h:kg* chunk-документы (legacy from old migration schema). + + Использует _all_docs со startkey/endkey по префиксу 'h:kg'. Чанки уже + не нужны после перехода на eden-inline embedding. + """ + url = f"{base_url}/{db}/_all_docs" + params = {"startkey": '"h:kg"', "endkey": '"h:kg\\ufff0"', "limit": 10000} + r = requests.get(url, params=params, auth=auth, timeout=60) + if r.status_code != 200: + logger.error("_all_docs failed: HTTP %d %s", r.status_code, r.text[:200]) + return 1 + rows = r.json().get("rows", []) + logger.info("Found %d orphan h:kg* chunks", len(rows)) + if dry_run: + return 0 + # bulk_docs с _deleted:true — куда быстрее чем 549 индивидуальных DELETE. + docs = [{"_id": row["id"], "_rev": row["value"]["rev"], "_deleted": True} + for row in rows] + if not docs: + return 0 + bulk_url = f"{base_url}/{db}/_bulk_docs" + rb = requests.post(bulk_url, json={"docs": docs}, auth=auth, timeout=120) + if rb.status_code not in (201, 200): + logger.error("_bulk_docs failed: HTTP %d %s", rb.status_code, rb.text[:200]) + return 1 + results = rb.json() + deleted = sum(1 for x in results if x.get("ok")) + logger.info("Deleted %d/%d orphan chunks", deleted, len(docs)) + return 0 + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--dry-run", action="store_true", + help="Только сгенерить .md (вывести count) без PUT в CouchDB") + ap.add_argument("--limit", type=int, default=None, + help="Загрузить только первые N entities (для smoke-теста)") + ap.add_argument("--cleanup-orphan-chunks", action="store_true", + help="Удалить осиротевшие h:kg* chunk-документы из предыдущей миграции") + ap.add_argument("--local-vault", type=str, default=None, + help="Путь к локальной папке Obsidian vault. В этом режиме скрипт " + "пишет .md файлы прямо в файловую систему, а LiveSync plugin " + "сам корректно загрузит их в CouchDB в правильном V3-chunked " + "формате. Не использует CouchDB API.") + args = ap.parse_args() + + base_url = os.environ.get("COUCHDB_URL", "https://obsidian.gendsgn.ru") + db = os.environ.get("COUCHDB_DB", "gendesign-vault") + user = os.environ.get("COUCHDB_USER", "obsidian") + password = os.environ.get("COUCHDB_PASSWORD") + if not password and not args.dry_run and not args.local_vault: + logger.error("COUCHDB_PASSWORD env required (unless --dry-run or --local-vault)") + return 1 + auth = (user, password or "") + + if args.cleanup_orphan_chunks: + return _cleanup_orphan_chunks(base_url, db, auth, args.dry_run) + + if not KG_PATH.exists(): + logger.error("KG not found: %s", KG_PATH) + return 2 + + if args.local_vault: + return _write_to_local_vault(Path(args.local_vault), args.limit) + + # 1. Load KG + entities: dict[str, dict] = {} + relations: list[dict] = [] + with KG_PATH.open(encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + d = json.loads(line) + except json.JSONDecodeError: + continue + if d.get("type") == "entity": + # Берём LATEST entity если name дублируется (auto-memory append) + entities[d["name"]] = d + elif d.get("type") == "relation": + relations.append(d) + logger.info("Loaded %d entities, %d relations", len(entities), len(relations)) + + # 2. Build relation indices + out_by: dict[str, list[dict]] = defaultdict(list) + in_by: dict[str, list[dict]] = defaultdict(list) + for r in relations: + # Дедуп: одна и та же relation может встречаться много раз в auto-memory log + out_by[r["from"]].append(r) + in_by[r["to"]].append(r) + + # 3. Compute target paths for [[wiki-links]] resolution + name_to_path: dict[str, str] = {} + for name, ent in entities.items(): + subdir = categorize(name, ent.get("entityType", "")) + fname = safe_filename(name) + ".md" + name_to_path[name] = f"{subdir}/{fname}" + + # 4. Migrate + names = list(entities.keys()) + if args.limit: + names = names[: args.limit] + logger.info("Will migrate %d entities (--limit=%s)", len(names), args.limit) + + # PASS 1: pre-compute content + chunk_id for every entity, чтобы потом + # сделать bulk PUT chunks FIRST, потом — main docs. + prepared: list[tuple[str, str, str, int]] = [] # (path, chunk_id, content, size) + for name in names: + ent = entities[name] + path = name_to_path[name] + seen: set[tuple[str, str, str]] = set() + out_uniq: list[dict] = [] + for r in out_by[name]: + key = (r["from"], r["to"], r["relationType"]) + if key in seen: + continue + seen.add(key) + out_uniq.append(r) + in_uniq: list[dict] = [] + for r in in_by[name]: + key = (r["from"], r["to"], r["relationType"]) + if key in seen: + continue + seen.add(key) + in_uniq.append(r) + content = md_for_entity(ent, out_uniq, in_uniq, name_to_path) + size_bytes = len(content.encode("utf-8")) + chunk_id = chunk_id_for(content) + prepared.append((path, chunk_id, content, size_bytes)) + + # PASS 2: bulk PUT chunks (idempotent — если уже есть, CouchDB вернёт conflict, + # это OK). + unique_chunks: dict[str, str] = {} # chunk_id -> content + for _, cid, content, _ in prepared: + unique_chunks.setdefault(cid, content) + logger.info("Bulk PUT %d unique chunks…", len(unique_chunks)) + chunk_docs = [{"_id": cid, "data": data, "type": "leaf"} + for cid, data in unique_chunks.items()] + if not args.dry_run: + rb = requests.post(f"{base_url}/{db}/_bulk_docs", + json={"docs": chunk_docs, "new_edits": True}, + auth=auth, timeout=120) + if rb.status_code in (201, 200): + results = rb.json() + n_ok = sum(1 for x in results if x.get("ok")) + n_conflict = sum(1 for x in results if x.get("error") == "conflict") + logger.info(" chunks: %d created, %d already existed", n_ok, n_conflict) + else: + logger.error("Bulk chunks PUT failed: HTTP %d %s", + rb.status_code, rb.text[:200]) + return 1 + + # PASS 3: PUT main docs sequentially (deleting old eden-only versions first + # to avoid _rev conflicts from прошлой неудачной миграции). + ok = 0 + fail = 0 + now_ms = int(time.time() * 1000) + for i, (path, chunk_id, _content, size_bytes) in enumerate(prepared, 1): + # delete старую версию (eden-only без children — broken) + delete_doc(base_url, db, auth, path) + main_doc = { + "_id": path, + "path": path, + "children": [chunk_id], + "ctime": now_ms, + "mtime": now_ms, + "size": size_bytes, + "type": "plain", + "eden": {}, + } + if put_doc(base_url, db, auth, path, main_doc, args.dry_run): + ok += 1 + else: + fail += 1 + if i % 50 == 0: + logger.info(" main docs: %d/%d (ok=%d fail=%d)", + i, len(prepared), ok, fail) + + # 5. Index page + index_md_lines = [ + "---", + "id: 00_index", + "type: index", + f"imported_at: {time.strftime('%Y-%m-%d')}", + "---", + "", + "# 🗂 Knowledge graph index", + "", + f"Migrated from `memory/memory-gendesign.jsonl` ({len(entities)} entities," + f" {len(relations)} relations).", + "", + ] + by_subdir: dict[str, list[str]] = defaultdict(list) + for name, path in name_to_path.items(): + by_subdir[path.split("/")[0]].append(name) + for subdir in sorted(by_subdir): + index_md_lines.append(f"## {subdir}/ ({len(by_subdir[subdir])})") + index_md_lines.append("") + for n in sorted(by_subdir[subdir]): + p = name_to_path[n][:-3] + index_md_lines.append(f"- [[{p}|{n}]]") + index_md_lines.append("") + index_content = "\n".join(index_md_lines) + index_chunk_id = chunk_id_for(index_content) + # chunk first + put_doc(base_url, db, auth, index_chunk_id, + {"_id": index_chunk_id, "data": index_content, "type": "leaf"}, + args.dry_run) + delete_doc(base_url, db, auth, "00_index.md") + put_doc(base_url, db, auth, "00_index.md", { + "_id": "00_index.md", "path": "00_index.md", + "children": [index_chunk_id], "ctime": now_ms, "mtime": now_ms, + "size": len(index_content.encode("utf-8")), + "type": "plain", "eden": {}, + }, args.dry_run) + + logger.info("DONE: %d ok, %d failed", ok, fail) + return 0 if fail == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/sshkey.txt b/sshkey.txt index 83b9ac89..87ed5312 100644 --- a/sshkey.txt +++ b/sshkey.txt @@ -93,3 +93,12 @@ Host gendesign token admin 15c99983d471718fe5c320f36bad01994d91ff34cb5646ce957b060ec039275e + + +ключ от обсидиан gendesign@gendesign:~$ PASSWORD=$(openssl rand -base64 32) +cho "Sgendesign@gendesign:~$ echo "Save this: $PASSWORD" +Save this: femp1ueSrQoD2cV/9m7YC8bmIuMOHz0JMmrpmD5Z1Ns= +curl -u obsidian: https://obsidian.gendsgn.ru/gendesign-vault + +сеть PS C:\Users\user> ssh gendesign 'docker network create gendesign_shared' +560a390e7b231eb2e0657ee96c8add1561ce772716b9db5b593b0950669df644