390 lines
19 KiB
Python
390 lines
19 KiB
Python
"""Синхронизация вики: сборка Markdown-копии из стадии, подмена каталога, строки wiki_page/вложений/FTS.
|
||
|
||
Порядок отчёта `done` (design 10): `prepare` вызывается до транзакции отчёта (CommandHooks.before_result) — готовит
|
||
каталог `wiki/.<метка>.new/` и выделяет id вложений; `commit` (в on_result) пишет строки и последним шагом подменяет
|
||
каталог. Перевод HTML → Markdown уже сделан при приёме пачек (stage.accept)."""
|
||
import json
|
||
import logging
|
||
import re
|
||
import shutil
|
||
import threading
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
from ...core import db
|
||
from ...core.paths import secure_dir
|
||
from . import convert, paths, search
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
_BAD_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
||
NAME_CHARS = 80 # длина названия, символов (как safe_name в eva_wiki_export.py)
|
||
NAME_BYTES = 200 # и байт UTF-8: у файловой системы предел имени 255 байт, «NN_» и «_k» помещаются
|
||
|
||
|
||
class SyncFailed(Exception):
|
||
"""Синхронизация не удалась: текст идёт в error команды и sync_error проекта."""
|
||
|
||
|
||
@dataclass
|
||
class Prepared:
|
||
cmd_id: int
|
||
label: str
|
||
root: str
|
||
discarded: str | None = None
|
||
error: str | None = None
|
||
rows: list = field(default_factory=list)
|
||
attachments: list = field(default_factory=list) # (page_code, url, name)
|
||
kept_ids: set = field(default_factory=set)
|
||
new_ids: list = field(default_factory=list) # id вложений, добавленных подготовкой
|
||
new_dir: Path | None = None
|
||
root_title: str | None = None
|
||
|
||
|
||
_PREPARED: dict[int, Prepared] = {}
|
||
_lock = threading.Lock()
|
||
|
||
|
||
# --- раскладка копии ----------------------------------------------------------
|
||
|
||
def safe_name(title: str) -> str:
|
||
"""Название каталога: запрещённые символы → «_», до 80 символов и 200 байт, «.» и пробелы по краям убраны."""
|
||
name = _BAD_CHARS.sub("_", title or "")[:NAME_CHARS]
|
||
name = name.encode("utf-8")[:NAME_BYTES].decode("utf-8", "ignore").strip(" .")
|
||
return name or "untitled"
|
||
|
||
|
||
def _sorted(codes, by_code):
|
||
return sorted(codes, key=lambda c: (by_code[c]["orderno"] is None, by_code[c]["orderno"] or 0, c))
|
||
|
||
|
||
def _tree(order, by_code) -> dict[str, list[str]]:
|
||
"""Потомки по parent_code; страница без родителя в наборе (или в цикле) становится дочерней корню."""
|
||
root = order[0]
|
||
children = {c: [] for c in order}
|
||
parent = {}
|
||
for c in order[1:]:
|
||
p = by_code[c].get("parent_code")
|
||
parent[c] = p if p in children and p != c else root
|
||
children[parent[c]].append(c)
|
||
|
||
def reach():
|
||
seen, stack = set(), [root]
|
||
while stack:
|
||
c = stack.pop()
|
||
if c not in seen:
|
||
seen.add(c)
|
||
stack.extend(children[c])
|
||
return seen
|
||
|
||
seen = reach()
|
||
for c in order[1:]:
|
||
if c not in seen: # цикл родителей: оторванное звено цепляем к корню
|
||
children[parent[c]].remove(c)
|
||
parent[c] = root
|
||
children[root].append(c)
|
||
seen = reach()
|
||
return children
|
||
|
||
|
||
def _inside(target: Path, path: Path) -> None:
|
||
if not path.resolve().is_relative_to(target.resolve()):
|
||
raise ValueError(f"путь {path} вне каталога копии")
|
||
|
||
|
||
def build_copy(stage_pages: list[dict], codes: list[str], target: Path, ids: dict[tuple[str, str], int]) -> list[dict]:
|
||
"""Собрать копию дерева в `target`: index.md страниц и _tree.json. Корень — codes[0]. Возвращает строки wiki_page
|
||
(с полем markdown — текст с подставленными id вложений). ValueError — страница или путь негодны."""
|
||
by_code = {p["code"]: p for p in stage_pages}
|
||
order = list(dict.fromkeys(codes))
|
||
if not order:
|
||
raise ValueError("пустой список страниц")
|
||
for c in order:
|
||
if c not in by_code:
|
||
raise ValueError(f"нет страницы {c}")
|
||
children = _tree(order, by_code)
|
||
secure_dir(target)
|
||
synced_at = db.now()
|
||
rows, tree = [], []
|
||
stack = [(order[0], Path(), 0, 0)]
|
||
used: dict[str, set[str]] = {} # родитель → занятые имена (без регистра)
|
||
while stack:
|
||
code, parent_rel, depth, pos = stack.pop()
|
||
page = by_code[code]
|
||
taken = used.setdefault(str(parent_rel), set())
|
||
name, k = safe_name(page["title"]), 1
|
||
base = name
|
||
while name.lower() in taken:
|
||
k += 1
|
||
name = f"{base}_{k}"
|
||
taken.add(name.lower())
|
||
rel = parent_rel / (name if depth == 0 else f"{pos:02d}_{name}")
|
||
full = target / rel
|
||
_inside(target, full)
|
||
try:
|
||
text = convert.fill_ids(page["markdown"], [ids[(code, a["url"])] for a in page["attachments"]])
|
||
except (ValueError, KeyError) as e:
|
||
raise ValueError(f"{code}: {e}") from None
|
||
secure_dir(full)
|
||
title_line = " ".join(str(page["title"]).split())
|
||
head = ["---", f"code: {json.dumps(code, ensure_ascii=False)}", f"title: {json.dumps(page['title'], ensure_ascii=False)}",
|
||
f"url: {json.dumps(page.get('url'), ensure_ascii=False)}", f"synced_at: {json.dumps(synced_at)}", "---", "",
|
||
f"# {title_line}", ""]
|
||
(full / "index.md").write_text("\n".join(head) + "\n" + text, encoding="utf-8")
|
||
rows.append({"code": code, "eva_id": page.get("eva_id"), "title": page["title"], "parent_code": page.get("parent_code"),
|
||
"orderno": page.get("orderno"), "depth": depth, "path": rel.as_posix(), "url": page.get("url"),
|
||
"size": len(text), "synced_at": synced_at, "markdown": text})
|
||
tree.append({"code": code, "title": page["title"], "depth": depth, "path": rel.as_posix()})
|
||
kids = _sorted(children[code], by_code)
|
||
for i, kid in reversed(list(enumerate(kids, 1))): # стек: первый ребёнок — следующим
|
||
stack.append((kid, rel, depth + 1, i))
|
||
(target / "_tree.json").write_text(json.dumps(tree, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
return rows
|
||
|
||
|
||
# --- подготовка (до транзакции отчёта) ---------------------------------------
|
||
|
||
def _siblings(conn, label: str) -> tuple[Path, Path, Path]:
|
||
cur = paths.project_dir(conn, label)
|
||
return cur, cur.with_name(f".{cur.name}.new"), cur.with_name(f".{cur.name}.old")
|
||
|
||
|
||
def _allocate_ids(conn, label, pages) -> tuple[dict, list, list]:
|
||
"""id вложений страниц: существующие по (страница, url) сохраняются, новые вставляются короткой транзакцией."""
|
||
ids, new_ids, attach = {}, [], []
|
||
existing = {(r["page_code"], r["url"]): r["id"] for r in conn.execute(
|
||
"SELECT id, page_code, url FROM wiki_attachment WHERE label = ?", (label,))}
|
||
with db.transaction(conn):
|
||
for page in pages:
|
||
for a in page["attachments"]:
|
||
key = (page["code"], a["url"])
|
||
if key not in existing:
|
||
cur = conn.execute("INSERT INTO wiki_attachment(label, page_code, url, name) VALUES (?, ?, ?, ?)",
|
||
(label, key[0], key[1], a["name"]))
|
||
existing[key] = cur.lastrowid
|
||
new_ids.append(cur.lastrowid)
|
||
ids[key] = existing[key]
|
||
attach.append((page["code"], a["url"], a["name"]))
|
||
return ids, new_ids, attach
|
||
|
||
|
||
def _abandon(conn, p: Prepared) -> None:
|
||
"""Подготовка не пригодилась: каталог .new и добавленные ею вложения убираются."""
|
||
if p.new_dir is not None:
|
||
shutil.rmtree(p.new_dir, ignore_errors=True)
|
||
if p.new_ids:
|
||
marks = ",".join("?" * len(p.new_ids))
|
||
conn.execute(f"DELETE FROM wiki_attachment WHERE id IN ({marks}) AND state = 'none'", p.new_ids)
|
||
p.new_ids = []
|
||
|
||
|
||
def prepare(conn, cmd: dict, result: dict | None) -> Prepared:
|
||
"""До транзакции: проверить корень и список кодов, выделить id вложений, собрать `.new`. Ошибка — в p.error."""
|
||
label, root = cmd["params"]["label"], cmd["params"]["root"]
|
||
p = Prepared(cmd["id"], label, root)
|
||
row = conn.execute("SELECT root_code FROM wiki_project WHERE label = ?", (label,)).fetchone()
|
||
if row is None or row["root_code"] != root:
|
||
p.discarded = "корень сменился"
|
||
return p
|
||
try:
|
||
codes = (result or {}).get("codes")
|
||
if not isinstance(codes, list) or not codes or not all(isinstance(c, str) for c in codes):
|
||
raise SyncFailed("в отчёте нет списка codes")
|
||
codes = list(dict.fromkeys(codes))
|
||
if codes[0] != root:
|
||
raise SyncFailed(f"первой в codes должна быть страница корня {root}")
|
||
stage = {r["code"]: json.loads(r["payload"]) for r in conn.execute(
|
||
"SELECT code, payload FROM wiki_sync_stage WHERE command_id = ?", (cmd["id"],))}
|
||
for c in codes:
|
||
if c not in stage:
|
||
raise SyncFailed(f"нет страницы {c}")
|
||
pages = [stage[c] for c in codes]
|
||
_, p.new_dir, _ = _siblings(conn, label)
|
||
secure_dir(p.new_dir.parent)
|
||
shutil.rmtree(p.new_dir, ignore_errors=True)
|
||
ids, p.new_ids, p.attachments = _allocate_ids(conn, label, pages)
|
||
p.kept_ids = set(ids.values())
|
||
p.rows = build_copy(pages, codes, p.new_dir, ids)
|
||
title = (result or {}).get("root_title")
|
||
p.root_title = title if isinstance(title, str) and title.strip() else p.rows[0]["title"]
|
||
except SyncFailed as e:
|
||
p.error = str(e)
|
||
except ValueError as e: # fill_ids, путь вне копии
|
||
p.error = str(e)
|
||
except OSError as e:
|
||
p.error = f"не удалось собрать копию: {e}"
|
||
except Exception as e: # noqa: BLE001 сбой не должен оставлять команду без отчёта
|
||
log.exception("wiki_sync %s: сборка копии", cmd["id"])
|
||
p.error = f"внутренняя ошибка сборки копии: {type(e).__name__}: {e}"
|
||
if p.error:
|
||
_abandon(conn, p)
|
||
return p
|
||
|
||
|
||
def before_result(conn, cmd, status, result, files_dir) -> None:
|
||
if cmd["type"] != "wiki_sync" or status != "done":
|
||
return
|
||
p = prepare(conn, cmd, result)
|
||
with _lock:
|
||
_PREPARED[cmd["id"]] = p
|
||
|
||
|
||
def _prepared(conn, cmd, result) -> Prepared:
|
||
with _lock:
|
||
p = _PREPARED.get(cmd["id"])
|
||
return p if p is not None else prepare(conn, cmd, result) # в обход before_result: в транзакции, но верно
|
||
|
||
|
||
def check_result(conn, cmd, status, result, files_dir) -> tuple[str, str | None]:
|
||
if cmd["type"] != "wiki_sync" or status != "done":
|
||
return status, None
|
||
p = _prepared(conn, cmd, result)
|
||
with _lock:
|
||
_PREPARED[cmd["id"]] = p
|
||
if p.discarded:
|
||
if isinstance(result, dict):
|
||
result["discarded"] = p.discarded
|
||
return "done", None
|
||
if p.error:
|
||
return "failed", p.error
|
||
return "done", None
|
||
|
||
|
||
# --- применение (в транзакции отчёта) ----------------------------------------
|
||
|
||
def _swap(cur: Path, new: Path, old: Path) -> None:
|
||
"""wiki/<метка> → .old, .new → <метка>, .old удаляется. При сбое прежний каталог возвращается на место."""
|
||
shutil.rmtree(old, ignore_errors=True)
|
||
had = cur.exists()
|
||
try:
|
||
if had:
|
||
cur.rename(old)
|
||
new.rename(cur)
|
||
except OSError:
|
||
if had and not cur.exists() and old.exists():
|
||
old.rename(cur)
|
||
raise
|
||
shutil.rmtree(old, ignore_errors=True)
|
||
|
||
|
||
def _fail(conn, cmd, error) -> None:
|
||
"""Команда уже записана как done, но копия не применилась: failed + sync_error."""
|
||
conn.execute("UPDATE command SET status = 'failed', error = ? WHERE id = ?", (error[:1500], cmd["id"]))
|
||
on_failed_or_reported_failed(conn, cmd, error)
|
||
|
||
|
||
def on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_dir) -> None:
|
||
if cmd["type"] != "wiki_sync":
|
||
return
|
||
with _lock:
|
||
p = _PREPARED.pop(cmd["id"], None)
|
||
if status != "done":
|
||
error = conn.execute("SELECT error FROM command WHERE id = ?", (cmd["id"],)).fetchone()["error"]
|
||
on_failed_or_reported_failed(conn, cmd, error or "синхронизация не удалась")
|
||
return
|
||
if p is None or p.error:
|
||
error = p.error if p else "копия не подготовлена"
|
||
on_failed_or_reported_failed(conn, cmd, error)
|
||
return
|
||
row = conn.execute("SELECT root_code FROM wiki_project WHERE label = ?", (p.label,)).fetchone()
|
||
if p.discarded or row is None or row["root_code"] != p.root: # корень сменился, пока шла подготовка
|
||
_abandon(conn, p)
|
||
if not p.discarded:
|
||
out = dict(result or {}, discarded="корень сменился")
|
||
conn.execute("UPDATE command SET result = ? WHERE id = ?", (json.dumps(out, ensure_ascii=False), cmd["id"]))
|
||
conn.execute("DELETE FROM wiki_sync_stage WHERE command_id = ?", (cmd["id"],))
|
||
conn.execute("UPDATE wiki_project SET failed_at = ? WHERE label = ?", (now, p.label))
|
||
return
|
||
try:
|
||
with db.transaction(conn):
|
||
_commit(conn, p, now)
|
||
except OSError as e:
|
||
_abandon(conn, p)
|
||
_fail(conn, cmd, f"не удалось подменить копию: {e}")
|
||
|
||
|
||
def _commit(conn, p: Prepared, now: str) -> None:
|
||
search.replace(conn, p.label, [(r["code"], r["title"], r["markdown"]) for r in p.rows])
|
||
conn.execute("DELETE FROM wiki_page WHERE label = ?", (p.label,))
|
||
conn.executemany(
|
||
"INSERT INTO wiki_page(label, code, eva_id, title, parent_code, orderno, depth, path, url, size, synced_at) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
[(p.label, r["code"], r["eva_id"], r["title"], r["parent_code"], r["orderno"], r["depth"], r["path"], r["url"],
|
||
r["size"], r["synced_at"]) for r in p.rows])
|
||
conn.executemany("UPDATE wiki_attachment SET name = ? WHERE label = ? AND page_code = ? AND url = ?",
|
||
[(name, p.label, code, url) for code, url, name in p.attachments])
|
||
for r in conn.execute("SELECT id FROM wiki_attachment WHERE label = ?", (p.label,)).fetchall():
|
||
if r["id"] not in p.kept_ids:
|
||
conn.execute("DELETE FROM wiki_attachment WHERE id = ?", (r["id"],))
|
||
conn.execute("UPDATE wiki_project SET synced_at = ?, pages_count = ?, root_title = ?, sync_error = NULL, failed_at = NULL "
|
||
"WHERE label = ?", (now, len(p.rows), p.root_title, p.label))
|
||
conn.execute("DELETE FROM wiki_sync_stage WHERE command_id = ?", (p.cmd_id,))
|
||
cur, new, old = _siblings(conn, p.label)
|
||
_swap(cur, new, old) # последним: сбой откатывает строки (savepoint)
|
||
|
||
|
||
def on_failed_or_reported_failed(conn, cmd: dict, error: str | None) -> None:
|
||
"""Отчёт failed или предел выдач: sync_error и failed_at проекта (если корень команды ещё текущий), стадия очищена."""
|
||
if cmd["type"] != "wiki_sync":
|
||
return
|
||
conn.execute("DELETE FROM wiki_sync_stage WHERE command_id = ?", (cmd["id"],))
|
||
conn.execute("UPDATE wiki_project SET sync_error = ?, failed_at = ? WHERE label = ? AND root_code = ?",
|
||
((error or "")[:1500], db.now(), cmd["params"]["label"], cmd["params"]["root"]))
|
||
|
||
|
||
def on_failed(conn, cmd, now) -> None:
|
||
"""Предел выдач: ошибка уже записана в команду."""
|
||
row = conn.execute("SELECT error FROM command WHERE id = ?", (cmd["id"],)).fetchone()
|
||
on_failed_or_reported_failed(conn, cmd, row["error"] if row else None)
|
||
|
||
|
||
def on_claim(conn, cmd, now, redelivery) -> None:
|
||
"""Повторная выдача: пачки прошлой выдачи не нужны, агент пришлёт заново."""
|
||
if cmd["type"] == "wiki_sync" and redelivery:
|
||
conn.execute("DELETE FROM wiki_sync_stage WHERE command_id = ?", (cmd["id"],))
|
||
|
||
|
||
def after_result(conn, cmd, status) -> None:
|
||
with _lock:
|
||
_PREPARED.pop(cmd["id"], None)
|
||
|
||
|
||
# --- уборка ----------------------------------------------------------------
|
||
|
||
def finish_pending_swaps(root: Path) -> None:
|
||
"""При старте: `.<имя>.new` без каталога `<имя>` — переименовать (подмена прервана); `.new` рядом с `<имя>` —
|
||
остаток неудачной сборки, удаляется; `.<имя>.old` — удаляется. Каталоги, которые собирает идущая подготовка, не трогаются."""
|
||
root = Path(root)
|
||
if not root.is_dir():
|
||
return
|
||
with _lock:
|
||
busy = {p.new_dir for p in _PREPARED.values() if p.new_dir}
|
||
for entry in sorted(root.iterdir()):
|
||
n = entry.name
|
||
if not n.startswith(".") or entry in busy:
|
||
continue
|
||
try:
|
||
if n.endswith(".new"):
|
||
target = root / n[1:-4]
|
||
if target.exists():
|
||
shutil.rmtree(entry, ignore_errors=True)
|
||
else:
|
||
entry.rename(target)
|
||
elif n.endswith(".old"):
|
||
shutil.rmtree(entry)
|
||
except OSError as e:
|
||
log.warning("вики: не удалось довести подмену %s: %s", entry, e)
|
||
|
||
|
||
def housekeeping(conn) -> None:
|
||
"""Стадия команд, которые уже не в работе (отмена, терминальный статус), очищается; отменённая синхронизация
|
||
сдвигает failed_at проекта, чтобы автосинхронизация не ставила пакеты каждый тик."""
|
||
conn.execute("DELETE FROM wiki_sync_stage WHERE command_id NOT IN "
|
||
"(SELECT id FROM command WHERE status IN ('queued', 'claimed'))")
|
||
for r in conn.execute("SELECT label, synced_at, failed_at FROM wiki_project").fetchall():
|
||
last = max((t for t in (r["synced_at"], r["failed_at"]) if t), default="")
|
||
c = conn.execute("SELECT MAX(finished_at) AS t FROM command WHERE type = 'wiki_sync' AND status = 'cancelled' "
|
||
"AND json_extract(params, '$.label') = ?", (r["label"],)).fetchone()["t"]
|
||
if c and c > last:
|
||
conn.execute("UPDATE wiki_project SET failed_at = ? WHERE label = ?", (c, r["label"]))
|