HubClient.wiki_versions, фейк маршрута версий, общая отправка пачек (#7)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
4bea83cdae
commit
37ebe8be5a
52
src/agent.py
52
src/agent.py
@ -579,6 +579,10 @@ class HubClient(Http):
|
||||
"""Пачка страниц wiki_sync в стадию хаба (до 50 страниц, тело до 16 МБ)."""
|
||||
return self.call("POST", "/agent/wiki/pages", {"command_id": command_id, "pages": pages})
|
||||
|
||||
def wiki_versions(self, command_id, versions: list) -> dict:
|
||||
"""Пачка версий страниц wiki_sync в стадию хаба (до 50 версий, тело до 16 МБ)."""
|
||||
return self.call("POST", "/agent/wiki/versions", {"command_id": command_id, "versions": versions})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Журнал: только дозапись; started → finished → acked по id команды
|
||||
@ -739,6 +743,7 @@ def _parse_at(value) -> float:
|
||||
# Дерево страниц Eva
|
||||
# ---------------------------------------------------------------------------
|
||||
MAX_PAGES = 50 # страниц в пачке (предел хаба, stage.MAX_PAGES)
|
||||
MAX_VERSIONS = 50 # версий в пачке (предел хаба)
|
||||
MAX_BATCH_BYTES = 4 * 1024 * 1024 # JSON пачки; предел тела хаба — 16 МБ
|
||||
TREE_FIELDS = ["id", "code", "name", "parent", "tree_parent", "project", "orderno"]
|
||||
VERSION_FIELDS = ["code", "cmf_created_at", "has_published", "cmf_author"] # список версий — без текста
|
||||
@ -831,6 +836,29 @@ class HubError(Exception):
|
||||
self.retryable = isinstance(cause, ConnectionError) or (isinstance(cause, HttpError) and cause.code >= 500)
|
||||
|
||||
|
||||
class _Batcher:
|
||||
"""Пачки для стадии хаба: не больше max_items элементов и max_bytes JSON; send получает накопленный список."""
|
||||
|
||||
def __init__(self, send: Callable[[list], None], max_items: int, max_bytes: int = MAX_BATCH_BYTES):
|
||||
self.send, self.max_items, self.max_bytes = send, max_items, max_bytes
|
||||
self.batch = [] # type: list
|
||||
self.size = 0
|
||||
self.sent = 0
|
||||
|
||||
def add(self, item: dict) -> None:
|
||||
isize = len(json.dumps(item, ensure_ascii=False).encode("utf-8")) + 2
|
||||
if self.batch and (len(self.batch) >= self.max_items or self.size + isize > self.max_bytes):
|
||||
self.flush()
|
||||
self.batch.append(item)
|
||||
self.size += isize
|
||||
|
||||
def flush(self) -> None:
|
||||
if self.batch:
|
||||
self.send(self.batch)
|
||||
self.sent += len(self.batch)
|
||||
self.batch, self.size = [], 0
|
||||
|
||||
|
||||
class Executor:
|
||||
def __init__(self, eva: EvaClient, hub: Optional[HubClient]):
|
||||
self.eva, self.hub = eva, hub
|
||||
@ -869,14 +897,14 @@ class Executor:
|
||||
except (ConnectionError, HttpError) as e:
|
||||
raise HubError(e) from None
|
||||
|
||||
def _send_pages(self, command_id, batch: list) -> None:
|
||||
"""Пачка страниц в стадию хаба. 413 (тело больше предела хаба) — ошибка с кодом страницы или границами пачки (M-3)."""
|
||||
def _send_batch(self, send: Callable[[Any, list], dict], command_id, batch: list, one: str, many: str) -> None:
|
||||
"""Пачка в стадию хаба. 413 (тело больше предела хаба) — ошибка с кодом элемента или границами пачки (M-3)."""
|
||||
try:
|
||||
self.hub.wiki_pages(command_id, batch)
|
||||
send(command_id, batch)
|
||||
except HttpError as e:
|
||||
if e.code == 413:
|
||||
what = ("страница %s" % batch[0]["code"] if len(batch) == 1 else
|
||||
"пачка страниц %s…%s (%d шт.)" % (batch[0]["code"], batch[-1]["code"], len(batch)))
|
||||
what = ("%s %s" % (one, batch[0]["code"]) if len(batch) == 1 else
|
||||
"%s %s…%s (%d шт.)" % (many, batch[0]["code"], batch[-1]["code"], len(batch)))
|
||||
e = HttpError(413, "%s больше 16 МБ — хаб не принял: %s" % (what, e.text[:300]), e.headers)
|
||||
raise HubError(e) from None
|
||||
except ConnectionError as e:
|
||||
@ -895,8 +923,8 @@ class Executor:
|
||||
root_code = root.get("code") or root_code
|
||||
base = self.eva.base
|
||||
codes = [] # type: List[str]
|
||||
batch = [] # type: list
|
||||
size = 0
|
||||
pages = _Batcher(lambda b: self._send_batch(self.hub.wiki_pages, cmd["id"], b, "страница", "пачка страниц"),
|
||||
MAX_PAGES)
|
||||
for doc, parent_code in walk(root, children):
|
||||
code = doc["code"]
|
||||
full = root if code == root_code else self.eva.get_doc(code, ["code", "name", "text"])
|
||||
@ -910,15 +938,9 @@ class Executor:
|
||||
"parent_code": parent_code, "orderno": _orderno(doc.get("orderno")),
|
||||
"url": urllib.parse.urljoin(base + "/", "project/Document/" + code),
|
||||
"html": html, "attachments": attachments, "links": links}
|
||||
psize = len(json.dumps(page, ensure_ascii=False).encode("utf-8")) + 2
|
||||
if batch and (len(batch) >= MAX_PAGES or size + psize > MAX_BATCH_BYTES):
|
||||
self._send_pages(cmd["id"], batch)
|
||||
batch, size = [], 0
|
||||
batch.append(page)
|
||||
size += psize
|
||||
pages.add(page)
|
||||
codes.append(code)
|
||||
if batch:
|
||||
self._send_pages(cmd["id"], batch)
|
||||
pages.flush()
|
||||
title = (root.get("name") or "").strip() or root_code
|
||||
return self._out("done", {"root_title": title, "codes": codes, "eva_calls": self.eva.calls})
|
||||
|
||||
|
||||
@ -6,6 +6,7 @@ import re
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
||||
|
||||
@ -59,6 +60,49 @@ def check_pages(command, command_id, pages):
|
||||
return None
|
||||
|
||||
|
||||
def check_versions(command, command_id, versions):
|
||||
"""Форма пачки версий как в приёме версий хаба (change wiki-history): (код, тело ошибки) или None."""
|
||||
def bad(detail, status=422, error="bad_versions"):
|
||||
return status, {"error": error, "detail": detail}
|
||||
|
||||
if isinstance(command_id, bool) or not isinstance(command_id, int):
|
||||
return bad("command_id — целое", error="bad_request")
|
||||
if not isinstance(versions, list) or not versions or len(versions) > MAX_PAGES:
|
||||
return bad("versions — список от 1 до %d версий" % MAX_PAGES)
|
||||
if command is None:
|
||||
return bad("command %s" % command_id, 404, "not_found")
|
||||
if command["type"] != "wiki_sync" or command["status"] != "claimed":
|
||||
return bad("команда %s не выдана агенту как wiki_sync" % command_id, 409, "not_claimed")
|
||||
for i, v in enumerate(versions, 1):
|
||||
code = v.get("code") if isinstance(v, dict) else None
|
||||
if not isinstance(code, str) or not code.strip():
|
||||
return bad("версия %d: code" % i)
|
||||
page = v.get("page_code")
|
||||
if not isinstance(page, str) or not page.strip():
|
||||
return bad("%s: page_code" % code)
|
||||
try:
|
||||
ok = isinstance(v.get("created_at"), str) and datetime.fromisoformat(v["created_at"]).tzinfo is not None
|
||||
except ValueError:
|
||||
ok = False
|
||||
if not ok:
|
||||
return bad("%s: created_at" % code)
|
||||
a = v.get("author")
|
||||
if not isinstance(a, dict) or not all(a.get(k) is None or isinstance(a.get(k), str) for k in ("login", "name")):
|
||||
return bad("%s: author" % code)
|
||||
if not isinstance(v.get("html"), str):
|
||||
return bad("%s: html" % code)
|
||||
att, links = v.get("attachments"), v.get("links")
|
||||
if not isinstance(att, list) or not all(
|
||||
isinstance(x, dict) and isinstance(x.get("ref"), str) and isinstance(x.get("url"), str)
|
||||
and isinstance(x.get("name") or "", str) for x in att):
|
||||
return bad("%s: attachments" % code)
|
||||
if not isinstance(links, list) or not all(
|
||||
isinstance(x, dict) and isinstance(x.get("ref"), str) and isinstance(x.get("code"), str)
|
||||
for x in links):
|
||||
return bad("%s: links" % code)
|
||||
return None
|
||||
|
||||
|
||||
class FakeHub:
|
||||
token = "agent-token"
|
||||
|
||||
@ -78,6 +122,10 @@ class FakeHub:
|
||||
self.file_requests = 0
|
||||
self.pages = [] # тела POST /agent/wiki/pages
|
||||
self.reject_pages = None # (код, тело) — ответ на /agent/wiki/pages ошибкой
|
||||
self.versions = [] # тела принятых POST /agent/wiki/versions
|
||||
self.reject_versions = None # (код, тело) — ответ на /agent/wiki/versions ошибкой
|
||||
self.fail_versions = 0 # сколько ближайших /agent/wiki/versions ответить 503
|
||||
self.wiki_posts = [] # пути POST /agent/wiki/pages|versions по порядку (и отклонённых)
|
||||
self.claim_wait = 0 # с: при пустой очереди держать claim, как хаб
|
||||
self.uploaded = [] # содержимое, присланное агентом POST /agent/files
|
||||
self.reject_files = None # (код, тело) — ответ на POST /agent/files ошибкой; "drop" — оборвать без ответа
|
||||
@ -173,7 +221,21 @@ class FakeHub:
|
||||
return (200, self.release) if self.release else (404, {"error": "no_release"})
|
||||
if path == "/agent/settings":
|
||||
return 200, self.settings
|
||||
if path == "/agent/wiki/versions" and method == "POST":
|
||||
self.wiki_posts.append(path)
|
||||
if self.fail_versions > 0:
|
||||
self.fail_versions -= 1
|
||||
return 503, {"error": "busy"}
|
||||
if self.reject_versions:
|
||||
return self.reject_versions
|
||||
cid = body.get("command_id")
|
||||
bad = check_versions(self.commands.get(cid) if isinstance(cid, int) else None, cid, body.get("versions"))
|
||||
if bad:
|
||||
return bad
|
||||
self.versions.append(body)
|
||||
return 200, {"accepted": len(body.get("versions") or [])}
|
||||
if path == "/agent/wiki/pages" and method == "POST":
|
||||
self.wiki_posts.append(path)
|
||||
if self.reject_pages:
|
||||
return self.reject_pages
|
||||
cid = body.get("command_id")
|
||||
|
||||
@ -195,6 +195,8 @@ class ClientsTest(unittest.TestCase):
|
||||
hub.claim(1, 2, None)
|
||||
hub.wiki_pages(7, [{"code": "DOC-1"}])
|
||||
hub.upload_file(b"abc")
|
||||
hub.wiki_versions(7, [{"code": "DOCH-1"}])
|
||||
self.assertEqual(calls[3], ("POST", "/agent/wiki/versions", {"command_id": 7, "versions": [{"code": "DOCH-1"}]}, None))
|
||||
self.assertEqual(calls[0][:2], ("POST", "/agent/claim"))
|
||||
self.assertEqual(calls[0][2]["last_call_at"], None)
|
||||
self.assertGreaterEqual(calls[0][3], 45)
|
||||
|
||||
@ -5,8 +5,24 @@ import unittest
|
||||
|
||||
from _common import SETTINGS, agent, eva_cfg, no_sleep
|
||||
from fake_eva import FakeEva
|
||||
import fake_hub
|
||||
from fake_hub import FakeHub
|
||||
|
||||
GOOD = {"page_code": "DOC-1", "code": "DOCH-1", "created_at": "2026-10-27T17:59:00+03:00",
|
||||
"author": {"login": None, "name": None}, "html": "", "attachments": [], "links": []}
|
||||
|
||||
|
||||
class FakeHubVersionsTest(unittest.TestCase):
|
||||
def test_check_versions(self):
|
||||
claimed = {"type": "wiki_sync", "status": "claimed"}
|
||||
self.assertIsNone(fake_hub.check_versions(claimed, 1, [GOOD]))
|
||||
self.assertEqual(fake_hub.check_versions(claimed, 1, [GOOD] * 51)[0], 422)
|
||||
self.assertEqual(fake_hub.check_versions({"type": "wiki_sync", "status": "done"}, 1, [GOOD])[0], 409)
|
||||
self.assertEqual(fake_hub.check_versions(claimed, 1, [dict(GOOD, created_at="2026-10-27T17:59:00")]),
|
||||
(422, {"error": "bad_versions", "detail": "DOCH-1: created_at"}))
|
||||
self.assertEqual(fake_hub.check_versions(claimed, 1, [dict(GOOD, author={"login": 1, "name": None})]),
|
||||
(422, {"error": "bad_versions", "detail": "DOCH-1: author"}))
|
||||
|
||||
|
||||
class SyncBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user