515 lines
27 KiB
Python
515 lines
27 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 sqlite3
|
||
import threading
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
from ...core import db
|
||
from ...core.paths import secure_dir
|
||
from . import convert, history, 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
|
||
old_dir: Path | None = None # прежняя копия после подмены: удаляется в after_result
|
||
committed: bool = False
|
||
root_title: str | None = None
|
||
with_history: bool = False # агент прислал versions: историю обновить
|
||
versions: list = field(default_factory=list) # новые строки wiki_version (с added/removed)
|
||
recount: list = field(default_factory=list) # (added, removed, code) известных версий
|
||
|
||
|
||
_PREPARED: dict[int, Prepared] = {}
|
||
_BUSY: set[Path] = set() # каталоги .new идущих подготовок (с начала сборки)
|
||
_lock = threading.Lock()
|
||
_label_locks: dict[str, threading.Lock] = {} # подготовка…применение одного проекта — по одной
|
||
_tls = threading.local() # блокировка, взятая этим потоком в before_result
|
||
|
||
|
||
def _label_lock(label: str) -> threading.Lock:
|
||
with _lock:
|
||
return _label_locks.setdefault(label, 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, cmd_id: int) -> tuple[Path, Path, Path]:
|
||
"""(копия, каталог сборки этой команды, каталог прежней копии на время подмены)."""
|
||
cur = paths.project_dir(conn, label)
|
||
return cur, cur.with_name(f".{cur.name}.new.{cmd_id}"), cur.with_name(f".{cur.name}.old")
|
||
|
||
|
||
def _allocate_ids(conn, label, pages) -> tuple[dict, list, list]:
|
||
"""id вложений страниц: существующие по (страница, url) сохраняются, новые вставляются короткой транзакцией."""
|
||
ids, new_ids, attach = {}, [], []
|
||
with db.transaction(conn):
|
||
existing = {(r["page_code"], r["url"]): r["id"] for r in conn.execute(
|
||
"SELECT id, page_code, url FROM wiki_attachment WHERE label = ?", (label,))}
|
||
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 _plan_versions(conn, label: str, codes, staged) -> tuple[list[dict], list[tuple[int, int, str]]]:
|
||
"""Новые версии из стадии и пересчёт «+/−» (R-6: вне транзакции). Берутся версии страниц из codes с кодом, которого
|
||
у проекта ещё нет; по затронутой странице версии идут по (created_at, code), предыдущий текст первой — пустой.
|
||
Известная версия, чей предшественник новый, пересчитывается."""
|
||
marks = json.dumps(list(codes))
|
||
known = {r["code"] for r in conn.execute(
|
||
"SELECT code FROM wiki_version WHERE label = ? AND page_code IN (SELECT value FROM json_each(?))", (label, marks))}
|
||
fresh = [s for s in staged if s["page_code"] in set(codes) and s["code"] not in known]
|
||
by_page: dict[str, list[dict]] = {}
|
||
for s in fresh:
|
||
by_page.setdefault(s["page_code"], []).append(dict(s, new=True))
|
||
new_rows, recount = [], []
|
||
for page_code, items in by_page.items():
|
||
for r in conn.execute("SELECT code, created_at, markdown, added, removed FROM wiki_version "
|
||
"WHERE label = ? AND page_code = ?", (label, page_code)):
|
||
items.append(dict(r, new=False))
|
||
items.sort(key=lambda x: (x["created_at"], x["code"]))
|
||
prev, prev_new = "", False
|
||
for it in items:
|
||
if it["new"] or prev_new:
|
||
added, removed = history.line_delta(prev, it["markdown"])
|
||
if it["new"]:
|
||
new_rows.append({**{k: it[k] for k in ("code", "page_code", "created_at", "author_login", "author_name",
|
||
"markdown")}, "added": added, "removed": removed})
|
||
elif (added, removed) != (it["added"], it["removed"]):
|
||
recount.append((added, removed, it["code"]))
|
||
prev, prev_new = it["markdown"], it["new"]
|
||
return new_rows, recount
|
||
|
||
|
||
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)
|
||
try:
|
||
_, p.new_dir, _ = _siblings(conn, label, cmd["id"])
|
||
except RuntimeError as e: # база без файла
|
||
p.error = str(e)
|
||
return p
|
||
with _lock: # с этого момента ни finish_pending_swaps, ни уборка не тронут .new
|
||
_PREPARED[cmd["id"]] = p
|
||
_BUSY.add(p.new_dir)
|
||
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]
|
||
if "versions" in (result or {}):
|
||
m = result["versions"]
|
||
if isinstance(m, bool) or not isinstance(m, int) or m < 0:
|
||
raise SyncFailed("versions — неотрицательное целое")
|
||
staged = [json.loads(r["payload"]) for r in conn.execute(
|
||
"SELECT payload FROM wiki_version_stage WHERE command_id = ?", (cmd["id"],))]
|
||
if len(staged) < m:
|
||
raise SyncFailed(f"нет версий: пришло {len(staged)} из {m}")
|
||
p.versions, p.recount = _plan_versions(conn, label, codes, staged)
|
||
p.with_history = True
|
||
cur, _, old = _siblings(conn, label, cmd["id"])
|
||
secure_dir(p.new_dir.parent)
|
||
shutil.rmtree(p.new_dir, ignore_errors=True)
|
||
shutil.rmtree(old, 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 sqlite3.OperationalError:
|
||
raise # база занята: 500, агент повторит отчёт; abort_result всё уберёт
|
||
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:
|
||
"""До транзакции отчёта done: одна подготовка на проект (блокировка до after_result/abort_result)."""
|
||
if cmd["type"] != "wiki_sync" or status != "done":
|
||
return
|
||
lock = _label_lock(cmd["params"]["label"])
|
||
lock.acquire()
|
||
_tls.held = lock
|
||
row = conn.execute("SELECT status FROM command WHERE id = ?", (cmd["id"],)).fetchone()
|
||
if row is None or row["status"] != "claimed": # параллельный отчёт уже закрыл команду
|
||
return
|
||
prepare(conn, cmd, result)
|
||
|
||
|
||
def _release(cmd_id: int) -> Prepared | None:
|
||
with _lock:
|
||
p = _PREPARED.pop(cmd_id, None)
|
||
if p is not None and p.new_dir is not None:
|
||
_BUSY.discard(p.new_dir)
|
||
lock = getattr(_tls, "held", None)
|
||
if lock is not None:
|
||
_tls.held = None
|
||
lock.release()
|
||
return p
|
||
|
||
|
||
def abort_result(conn, cmd) -> None:
|
||
"""Отчёт не дошёл до конца: подготовка снимается — каталог сборки, добавленные ею вложения, блокировка."""
|
||
p = _release(cmd["id"])
|
||
if p is None:
|
||
return
|
||
if p.new_dir is not None and not p.committed:
|
||
shutil.rmtree(p.new_dir, ignore_errors=True)
|
||
if p.new_ids and not p.committed:
|
||
try:
|
||
with db.transaction(conn):
|
||
_abandon(conn, p)
|
||
except sqlite3.Error:
|
||
log.warning("wiki_sync %s: вложения подготовки не удалены", cmd["id"], exc_info=True)
|
||
|
||
|
||
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)
|
||
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) -> bool:
|
||
"""wiki/<метка> → .old, .new → <метка>; True, если прежняя копия была (её удалит after_result вне транзакции).
|
||
При сбое прежний каталог возвращается на место."""
|
||
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
|
||
return had
|
||
|
||
|
||
def _clear_stage(conn, cmd_id: int) -> None:
|
||
"""Стадия страниц и версий команды."""
|
||
conn.execute("DELETE FROM wiki_sync_stage WHERE command_id = ?", (cmd_id,))
|
||
conn.execute("DELETE FROM wiki_version_stage WHERE command_id = ?", (cmd_id,))
|
||
|
||
|
||
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.get(cmd["id"])
|
||
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"]))
|
||
_clear_stage(conn, 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)
|
||
p.committed = True
|
||
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))
|
||
if p.with_history:
|
||
conn.execute("DELETE FROM wiki_version WHERE label = ? AND page_code NOT IN (SELECT value FROM json_each(?))",
|
||
(p.label, json.dumps([r["code"] for r in p.rows])))
|
||
conn.executemany(
|
||
"INSERT OR IGNORE INTO wiki_version(label, code, page_code, created_at, author_login, author_name, markdown, "
|
||
"added, removed) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
[(p.label, v["code"], v["page_code"], v["created_at"], v["author_login"], v["author_name"], v["markdown"],
|
||
v["added"], v["removed"]) for v in p.versions])
|
||
conn.executemany("UPDATE wiki_version SET added = ?, removed = ? WHERE label = ? AND code = ?",
|
||
[(a, r, p.label, code) for a, r, code in p.recount])
|
||
conn.execute("INSERT OR REPLACE INTO wiki_history(label, history_at) VALUES (?, ?)", (p.label, now))
|
||
_clear_stage(conn, p.cmd_id)
|
||
cur, new, old = _siblings(conn, p.label, p.cmd_id)
|
||
if _swap(cur, new, old): # последним: сбой откатывает строки (savepoint)
|
||
p.old_dir = old
|
||
|
||
|
||
def on_failed_or_reported_failed(conn, cmd: dict, error: str | None) -> None:
|
||
"""Отчёт failed или предел выдач: sync_error и failed_at проекта (если корень команды ещё текущий), стадия очищена."""
|
||
if cmd["type"] != "wiki_sync":
|
||
return
|
||
_clear_stage(conn, 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:
|
||
_clear_stage(conn, cmd["id"])
|
||
|
||
|
||
def after_result(conn, cmd, status) -> None:
|
||
"""Вне транзакции: прежняя копия после подмены удаляется, блокировка проекта снимается."""
|
||
p = _release(cmd["id"])
|
||
if p is not None and p.old_dir is not None:
|
||
try:
|
||
shutil.rmtree(p.old_dir)
|
||
except OSError as e:
|
||
log.warning("вики: прежняя копия %s не удалена (уберёт следующий старт): %s", p.old_dir, e)
|
||
|
||
|
||
# --- уборка ----------------------------------------------------------------
|
||
|
||
_PENDING = re.compile(r"^\.(.+)\.(new(?:\.\d+)?|old)$", re.DOTALL)
|
||
|
||
|
||
def finish_pending_swaps(root: Path) -> None:
|
||
"""При старте: `.<имя>.new[.<id>]` с `_tree.json` (сборка закончена) без каталога `<имя>` — переименовать (подмена
|
||
прервана); без `_tree.json` или рядом с `<имя>` — остаток сборки, удаляется; `.<имя>.old` — удаляется. Каталоги
|
||
идущих подготовок не трогаются."""
|
||
root = Path(root)
|
||
if not root.is_dir():
|
||
return
|
||
with _lock:
|
||
busy = set(_BUSY)
|
||
for entry in sorted(root.iterdir()):
|
||
m = _PENDING.match(entry.name)
|
||
if m is None or entry in busy:
|
||
continue
|
||
try:
|
||
if m.group(2) == "old":
|
||
shutil.rmtree(entry)
|
||
continue
|
||
target = root / m.group(1)
|
||
if target.exists() or not (entry / "_tree.json").is_file():
|
||
shutil.rmtree(entry, ignore_errors=True)
|
||
else:
|
||
entry.rename(target)
|
||
except OSError as e:
|
||
log.warning("вики: не удалось довести подмену %s: %s", entry, e)
|
||
|
||
|
||
def housekeeping(conn) -> None:
|
||
"""Стадия команд, которые уже не в работе (отмена, терминальный статус), очищается; отменённая синхронизация
|
||
сдвигает failed_at проекта, чтобы автосинхронизация не ставила пакеты каждый тик."""
|
||
live = "(SELECT id FROM command WHERE status IN ('queued', 'claimed'))"
|
||
conn.execute(f"DELETE FROM wiki_sync_stage WHERE command_id NOT IN {live}")
|
||
conn.execute(f"DELETE FROM wiki_version_stage WHERE command_id NOT IN {live}")
|
||
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"]))
|