1661 lines
87 KiB
Python
1661 lines
87 KiB
Python
#!/usr/bin/env python3
|
||
"""jira-agent — клиент моста Jira для виртуалки с доступом к Jira банка.
|
||
|
||
Опрашивает хаб (jira-hub) на Mac mini, выполняет команды «желаемый результат» в Jira по правилу
|
||
«проверь, потом сделай», ведёт журнал на диске и возвращает хабу состояние задач. Только стандартная библиотека,
|
||
Python 3.9. Спека: jira-hub/openspec/specs/agent/spec.md.
|
||
Связь с хабом — HTTP (`hub_url = http://…`) или общая папка (`hub_url = folder:<путь>`), если сеть закрыта VPN.
|
||
|
||
python3 agent.py check # доступ к хабу и Jira, без записи
|
||
python3 agent.py run # основной цикл (по умолчанию)
|
||
python3 agent.py selftest # самопроверка новой версии перед обновлением: версия + check; код 0 или 3
|
||
python3 agent.py prune # почистить journal.jsonl от подтверждённых записей старше 30 дней
|
||
|
||
Самообновление — через запускатель run.py (спека agent, «Самообновление агента»).
|
||
"""
|
||
import argparse
|
||
import base64
|
||
import fcntl
|
||
import hashlib
|
||
import http.client
|
||
import json
|
||
import os
|
||
import re
|
||
import signal
|
||
import ssl
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Callable, Dict, List, Optional, Tuple
|
||
|
||
AGENT_VERSION = 12
|
||
EXIT_LOCKED = 73 # «журнал занят другим процессом»: для run.py это не сбой версии (design 8)
|
||
EXIT_UPDATE = 75 # «обновление готово»: run.py подменит agent.py на agent.py.new
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
API = "/rest/api/2"
|
||
AGILE = "/rest/agile/1.0"
|
||
|
||
|
||
def now_iso() -> str:
|
||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||
|
||
|
||
def log(msg: str) -> None:
|
||
print("[%s] %s" % (datetime.now().strftime("%H:%M:%S"), msg), flush=True)
|
||
|
||
|
||
def load_config(path: str) -> dict:
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
cfg = json.load(f)
|
||
except FileNotFoundError:
|
||
sys.exit("нет файла настроек %s — скопируйте agent.example.json в agent.json и заполните" % path)
|
||
for key in ("hub_url", "hub_token", "jira"):
|
||
if not cfg.get(key):
|
||
sys.exit("в %s нет обязательного поля %s" % (path, key))
|
||
if not cfg["jira"].get("base_url"):
|
||
sys.exit("в %s нет jira.base_url" % path)
|
||
if cfg["hub_url"].startswith("folder:") and not os.path.isabs(cfg["hub_url"][len("folder:"):]):
|
||
sys.exit("hub_url: после folder: нужен абсолютный путь общей папки, например "
|
||
"folder:/Volumes/My Shared Files/RSHB/jira-bridge")
|
||
cfg.setdefault("timeout", 60)
|
||
cfg.setdefault("poll_seconds", 5)
|
||
cfg.setdefault("verify_ssl", True)
|
||
return cfg
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Самообновление: update.json рядом с agent.py (спека agent «Самообновление агента», design 7)
|
||
# ---------------------------------------------------------------------------
|
||
def read_update_state(path: str) -> Optional[dict]:
|
||
"""Состояние обновления; нет файла или мусор — None."""
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
except FileNotFoundError:
|
||
return None
|
||
except (OSError, ValueError):
|
||
log("update.json не читается — считаю, что состояния нет")
|
||
return None
|
||
return data if isinstance(data, dict) else None
|
||
|
||
|
||
def write_update_state(path: str, state: str, from_version, to_version, sha256, error=None) -> dict:
|
||
data = {"state": state, "from_version": from_version, "to_version": to_version, "sha256": sha256,
|
||
"error": error, "at": now_iso()}
|
||
tmp = path + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False)
|
||
os.replace(tmp, path)
|
||
return data
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# HTTP
|
||
# ---------------------------------------------------------------------------
|
||
def wire_filename(name: str) -> str:
|
||
"""Имя для filename="…" в multipart: кавычка и переводы строк — %22, %0D, %0A, как у браузеров (HTML5);
|
||
остальное — UTF-8 как есть (Jira Server читает filename в кодировке запроса)."""
|
||
return name.replace('"', "%22").replace("\r", "%0D").replace("\n", "%0A")
|
||
|
||
|
||
def multipart_file(boundary: str, filename: str, data: bytes) -> bytes:
|
||
"""Тело multipart/form-data с одной частью file (спека agent «Тип задачи и вложения»)."""
|
||
disposition = "form-data; name=\"file\"; filename=\"%s\"; filename*=UTF-8''%s" % (
|
||
wire_filename(filename), urllib.parse.quote(filename, safe=""))
|
||
head = "--%s\r\nContent-Disposition: %s\r\nContent-Type: application/octet-stream\r\n\r\n" % (boundary, disposition)
|
||
return head.encode("utf-8") + data + ("\r\n--%s--\r\n" % boundary).encode("ascii")
|
||
|
||
|
||
class HttpError(Exception):
|
||
def __init__(self, code: int, text: str, headers=None):
|
||
super().__init__("HTTP %d: %s" % (code, text[:300]))
|
||
self.code, self.text, self.headers = code, text, headers or {}
|
||
|
||
|
||
class Http:
|
||
def __init__(self, base: str, headers: Dict[str, str], timeout: float = 60, ctx=None):
|
||
self.base, self.headers, self.timeout, self.ctx = base.rstrip("/"), headers, timeout, ctx
|
||
|
||
def __repr__(self):
|
||
return "<%s %s>" % (self.__class__.__name__, self.base)
|
||
|
||
def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any:
|
||
data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
|
||
return self.send(method, path, data, "application/json; charset=utf-8" if data is not None else None,
|
||
params, timeout)
|
||
|
||
def send(self, method: str, path: str, data: Optional[bytes], content_type: Optional[str], params=None,
|
||
timeout=None) -> Any:
|
||
"""Запрос с готовым телом (JSON или multipart); ответ — JSON. Ошибки — HttpError и ConnectionError."""
|
||
url = self.base + path
|
||
if params:
|
||
url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
|
||
req = urllib.request.Request(url, data=data, method=method)
|
||
for k, v in self.headers.items():
|
||
req.add_header(k, v)
|
||
req.add_header("Accept", "application/json")
|
||
if content_type is not None:
|
||
req.add_header("Content-Type", content_type)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout or self.timeout, context=self.ctx) as resp:
|
||
raw = resp.read()
|
||
status = resp.status
|
||
except urllib.error.HTTPError as e:
|
||
self._touch()
|
||
try:
|
||
headers = {k.lower(): v for k, v in e.headers.items()}
|
||
text = e.read().decode("utf-8", "replace")
|
||
except (OSError, ConnectionError, TimeoutError, http.client.HTTPException) as err:
|
||
raise ConnectionError("%s: обрыв при чтении ответа: %s" % (self.base, err)) from None
|
||
finally:
|
||
e.close()
|
||
raise HttpError(e.code, text, headers) from None
|
||
except (urllib.error.URLError, ConnectionError, TimeoutError, OSError, http.client.HTTPException) as e:
|
||
raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None
|
||
self._touch()
|
||
try:
|
||
return json.loads(raw.decode("utf-8")) if raw else {}
|
||
except ValueError as e: # включая UnicodeDecodeError и JSONDecodeError
|
||
raise HttpError(status, "ответ не JSON: %s; начало: %s" % (e, raw[:200].decode("utf-8", "replace"))) from None
|
||
|
||
def _touch(self) -> None:
|
||
"""Вызывается на каждый полученный HTTP-ответ (любой код)."""
|
||
|
||
|
||
class JiraClient(Http):
|
||
"""Все вызовы Jira идут здесь: счётчики вызовов и 429/503, время последнего вызова, повтор по Retry-After."""
|
||
|
||
def __init__(self, cfg: dict):
|
||
j = cfg["jira"]
|
||
auth = j.get("auth") or {}
|
||
kind = auth.get("type") or ("bearer" if auth.get("token") else "basic")
|
||
if kind == "bearer":
|
||
if not auth.get("token"):
|
||
sys.exit("jira.auth.token пуст")
|
||
header = "Bearer " + auth["token"]
|
||
else:
|
||
if not (auth.get("user") and auth.get("password")):
|
||
sys.exit("jira.auth.user / jira.auth.password пусты")
|
||
header = "Basic " + base64.b64encode(("%s:%s" % (auth["user"], auth["password"])).encode()).decode()
|
||
ctx = None
|
||
if not cfg.get("verify_ssl", True):
|
||
ctx = ssl.create_default_context()
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = ssl.CERT_NONE
|
||
elif cfg.get("ca_bundle"):
|
||
ctx = ssl.create_default_context(cafile=cfg["ca_bundle"])
|
||
super().__init__(j["base_url"], {"Authorization": header, "X-Atlassian-Token": "no-check"}, cfg.get("timeout", 60), ctx)
|
||
self.calls = 0
|
||
self.rate_limited = 0
|
||
self.last_call_at = None # type: Optional[float] (time.monotonic)
|
||
self.last_call_iso = None # type: Optional[str]
|
||
self.sleep = time.sleep
|
||
|
||
def _touch(self) -> None:
|
||
self.calls += 1
|
||
self.last_call_at, self.last_call_iso = time.monotonic(), now_iso()
|
||
|
||
def reset_counters(self) -> None:
|
||
self.calls, self.rate_limited = 0, 0
|
||
|
||
def _retry(self, request: Callable[[], Any]) -> Any:
|
||
"""429/503 — повтор по Retry-After (до 5 запросов, не более 4 ожиданий); прочие ошибки — наружу."""
|
||
for attempt in range(1, 6):
|
||
try:
|
||
return request()
|
||
except HttpError as e:
|
||
if e.code in (429, 503) and attempt < 5:
|
||
wait = 60 * attempt
|
||
try:
|
||
wait = max(0, int(e.headers.get("retry-after")))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
self.rate_limited += 1
|
||
log(" ! HTTP %d, жду %d с (попытка %d/5)" % (e.code, wait, attempt))
|
||
self.sleep(wait)
|
||
continue
|
||
if e.code in (429, 503):
|
||
self.rate_limited += 1
|
||
raise
|
||
|
||
def call(self, method, path, body=None, params=None):
|
||
return self._retry(lambda: Http.call(self, method, path, body, params))
|
||
|
||
def upload(self, key: str, filename: str, data: bytes) -> Any:
|
||
"""Вложение в задачу: multipart, X-Atlassian-Token: no-check (заголовок клиента), таймаут не меньше 120 с.
|
||
Содержимое не логируется."""
|
||
boundary = uuid.uuid4().hex
|
||
body = multipart_file(boundary, filename, data)
|
||
return self._retry(lambda: self.send("POST", API + "/issue/%s/attachments" % key, body,
|
||
"multipart/form-data; boundary=" + boundary,
|
||
timeout=max(self.timeout, 120)))
|
||
|
||
def get(self, path, **params):
|
||
return self.call("GET", path, params=params or None)
|
||
|
||
def post(self, path, body):
|
||
return self.call("POST", path, body)
|
||
|
||
def put(self, path, body):
|
||
return self.call("PUT", path, body)
|
||
|
||
def delete(self, path):
|
||
return self.call("DELETE", path)
|
||
|
||
|
||
class FolderTransport:
|
||
"""Обмен с хабом через общую папку (hub_url = folder:<путь>), когда сеть до хаба закрыта (VPN на виртуалке).
|
||
Запрос — файл req/<id>.json, ответ кладёт реле хаба в resp/<id>.json. Ошибки те же, что у Http.call:
|
||
нет ответа к сроку или нет папки — ConnectionError, код не 2xx — HttpError. Спека: specs/agent, folder-transport."""
|
||
POLL = 0.1
|
||
BAD_JSON_WAIT = 2.0
|
||
|
||
def __init__(self, root: str, authorization: str):
|
||
self.root, self.authorization = root, authorization
|
||
|
||
def __repr__(self):
|
||
return "<FolderTransport %s>" % self.root
|
||
|
||
def call(self, method: str, path: str, body=None, timeout: float = 60) -> Any:
|
||
if not os.path.isdir(self.root):
|
||
raise ConnectionError("нет папки %s" % self.root)
|
||
req_dir, resp_dir = os.path.join(self.root, "req"), os.path.join(self.root, "resp")
|
||
rid = uuid.uuid4().hex
|
||
try:
|
||
os.makedirs(req_dir, exist_ok=True)
|
||
os.makedirs(resp_dir, exist_ok=True)
|
||
deadline = time.time() + timeout
|
||
self._write(req_dir, rid + ".json", {
|
||
"id": rid, "method": method, "path": path, "body": body, "authorization": self.authorization,
|
||
"deadline": datetime.fromtimestamp(deadline, timezone.utc).isoformat()})
|
||
except OSError as e:
|
||
raise ConnectionError("папка %s недоступна: %s" % (self.root, e)) from None
|
||
resp_path = os.path.join(resp_dir, rid + ".json")
|
||
end = time.monotonic() + timeout
|
||
while time.monotonic() < end:
|
||
if os.path.exists(resp_path):
|
||
return self._take(resp_path)
|
||
time.sleep(self.POLL)
|
||
try:
|
||
os.remove(os.path.join(req_dir, rid + ".json"))
|
||
except OSError:
|
||
pass
|
||
raise ConnectionError("не ответил через папку %s за %.0f с" % (self.root, timeout))
|
||
|
||
@staticmethod
|
||
def _write(directory: str, name: str, payload: dict) -> None:
|
||
tmp = os.path.join(directory, "." + name + ".tmp")
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(payload, f, ensure_ascii=False)
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
os.replace(tmp, os.path.join(directory, name))
|
||
|
||
def _take(self, path: str) -> Any:
|
||
end = time.monotonic() + self.BAD_JSON_WAIT
|
||
while True:
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
resp = json.load(f)
|
||
if not isinstance(resp, dict):
|
||
raise ValueError("ответ не объект")
|
||
status, body = resp.get("status"), resp.get("body")
|
||
if type(status) is not int or not (body is None or isinstance(body, str)):
|
||
raise ValueError("неверные типы status/body") # bool — подкласс int, поэтому type(), не isinstance (M-2 agent#8)
|
||
break
|
||
except (ValueError, OSError):
|
||
if time.monotonic() >= end:
|
||
raise ConnectionError("нечитаемый ответ хаба в %s" % path) from None
|
||
time.sleep(self.POLL)
|
||
try:
|
||
os.remove(path)
|
||
except OSError:
|
||
pass
|
||
text = body or "" # status и body проверены выше (M-5 #18)
|
||
if not 200 <= status < 300:
|
||
raise HttpError(status, text)
|
||
try:
|
||
return json.loads(text) if text else {}
|
||
except ValueError as e:
|
||
raise HttpError(status, "ответ не JSON: %s; начало: %s" % (e, text[:200])) from None
|
||
|
||
|
||
class HubClient(Http):
|
||
def __init__(self, cfg: dict):
|
||
auth = "Bearer " + cfg["hub_token"]
|
||
super().__init__(cfg["hub_url"], {"Authorization": auth}, cfg.get("timeout", 60))
|
||
url = cfg["hub_url"]
|
||
self.folder = FolderTransport(url[len("folder:"):], auth) if url.startswith("folder:") else None
|
||
|
||
def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any:
|
||
if self.folder is None:
|
||
return super().call(method, path, body, params, timeout)
|
||
if params:
|
||
path += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
|
||
return self.folder.call(method, path, body, timeout or self.timeout)
|
||
|
||
def claim(self, agent_version, settings_version, last_jira_call_at, update_state=None, launcher_version=None) -> dict:
|
||
return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version,
|
||
"last_jira_call_at": last_jira_call_at, "update_state": update_state,
|
||
"launcher_version": launcher_version}, timeout=max(self.timeout, 45))
|
||
|
||
def release(self) -> dict:
|
||
return self.call("GET", "/agent/release")
|
||
|
||
def settings(self) -> dict:
|
||
return self.call("GET", "/agent/settings")
|
||
|
||
def command(self, command_id) -> dict:
|
||
return self.call("GET", "/agent/commands/%d" % int(command_id))
|
||
|
||
def result(self, command_id, payload: dict) -> dict:
|
||
return self.call("POST", "/agent/commands/%d/result" % int(command_id), payload)
|
||
|
||
def state(self, payload: dict) -> dict:
|
||
return self.call("POST", "/agent/state", payload)
|
||
|
||
def known_keys(self) -> List[str]:
|
||
return self.call("GET", "/agent/known-keys")["keys"]
|
||
|
||
def file(self, sha256: str) -> dict:
|
||
"""Файл вложения из хранилища хаба: {file, size, content (base64)}; нет файла — HttpError 404 no_file."""
|
||
return self.call("GET", "/agent/files/%s" % sha256)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Журнал: только дозапись; started → finished → acked по id команды
|
||
# ---------------------------------------------------------------------------
|
||
class Journal:
|
||
def __init__(self, path: str):
|
||
self.path = path
|
||
|
||
def _torn_tail(self) -> bool:
|
||
"""Файл не пуст и не заканчивается переводом строки (оборванная запись)."""
|
||
try:
|
||
if os.path.getsize(self.path) == 0:
|
||
return False
|
||
except FileNotFoundError:
|
||
return False
|
||
with open(self.path, "rb") as f:
|
||
f.seek(-1, os.SEEK_END)
|
||
return f.read(1) != b"\n"
|
||
|
||
def append(self, record: dict) -> None:
|
||
record = dict(record)
|
||
record.setdefault("at", now_iso())
|
||
data = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8")
|
||
if self._torn_tail():
|
||
data = b"\n" + data
|
||
fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
|
||
with os.fdopen(fd, "ab") as f:
|
||
f.write(data)
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
|
||
def _scan(self):
|
||
"""Возвращает (годные записи, номера битых строк). Годная — dict с command_id."""
|
||
good, bad = [], []
|
||
if not os.path.exists(self.path):
|
||
return good, bad
|
||
with open(self.path, "rb") as f:
|
||
for n, raw in enumerate(f, 1):
|
||
line = raw.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
r = json.loads(line.decode("utf-8"))
|
||
except ValueError:
|
||
r = None
|
||
if isinstance(r, dict) and "command_id" in r:
|
||
good.append(r)
|
||
else:
|
||
bad.append(n)
|
||
return good, bad
|
||
|
||
def records(self) -> List[dict]:
|
||
good, bad = self._scan()
|
||
for n in bad:
|
||
log("журнал: строка %d не читается, пропускаю" % n)
|
||
return good
|
||
|
||
def last_at(self) -> Optional[datetime]:
|
||
"""Самое позднее время записи журнала (UTC). Записи без времени или с непонятным временем пропускаются."""
|
||
latest = None
|
||
for r in self.records():
|
||
try:
|
||
t = datetime.fromisoformat(r["at"])
|
||
except (KeyError, TypeError, ValueError):
|
||
continue
|
||
if t.tzinfo is None:
|
||
continue
|
||
if latest is None or t > latest:
|
||
latest = t
|
||
return latest
|
||
|
||
def claimed(self, command_id) -> None:
|
||
"""Команда получена из claim (до паузы): с этого момента хаб считает её выданной агенту."""
|
||
self.append({"kind": "claimed", "command_id": int(command_id)})
|
||
|
||
def started(self, command_id) -> None:
|
||
self.append({"kind": "started", "command_id": int(command_id)})
|
||
|
||
def finished(self, command_id, status, result, error, jira_calls, rate_limited) -> None:
|
||
self.append({"kind": "finished", "command_id": int(command_id), "status": status, "result": result,
|
||
"error": error, "jira_calls": int(jira_calls or 0), "rate_limited": int(rate_limited or 0)})
|
||
|
||
def acked(self, command_id) -> None:
|
||
self.append({"kind": "acked", "command_id": int(command_id)})
|
||
|
||
def unacked_finished(self) -> List[dict]:
|
||
last, acked = {}, set()
|
||
for r in self.records():
|
||
if r.get("kind") == "finished":
|
||
last[r["command_id"]] = r
|
||
acked.discard(r["command_id"])
|
||
elif r.get("kind") == "acked":
|
||
acked.add(r["command_id"])
|
||
return [r for cid, r in last.items() if cid not in acked]
|
||
|
||
def unfinished_started(self) -> List[int]:
|
||
"""Порядковый учёт: claimed/started открывают id, finished/acked закрывает; порядок — первое открытие."""
|
||
opened = {} # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия
|
||
for r in self.records():
|
||
kind, cid = r.get("kind"), r["command_id"]
|
||
if kind in ("claimed", "started"):
|
||
opened[cid] = True
|
||
elif kind in ("finished", "acked") and cid in opened:
|
||
opened[cid] = False
|
||
return [cid for cid, is_open in opened.items() if is_open]
|
||
|
||
def prune(self, days: int = 30) -> int:
|
||
"""Удалить записи id до его последнего acked, если тот старше days дней (возраст — по acked)
|
||
и после него у id нет started/finished. Возвращает число удалённых годных записей; битые строки
|
||
отбрасываются при перезаписи и учитываются в логе."""
|
||
good, bad = self._scan()
|
||
if bad:
|
||
log("журнал: удаляю битых строк %d" % len(bad))
|
||
cutoff = datetime.now(timezone.utc).timestamp() - days * 86400
|
||
last_ack, last_open = {}, {}
|
||
for i, r in enumerate(good):
|
||
cid = r["command_id"]
|
||
if r.get("kind") == "acked":
|
||
last_ack[cid] = i
|
||
elif r.get("kind") in ("claimed", "started", "finished"):
|
||
last_open[cid] = i
|
||
drop_upto = {} # command_id -> индекс последнего acked: всё до него включительно удаляется
|
||
for cid, i in last_ack.items():
|
||
if last_open.get(cid, -1) > i:
|
||
continue
|
||
try:
|
||
ts = _parse_at(good[i].get("at"))
|
||
except (ValueError, TypeError):
|
||
continue
|
||
if ts < cutoff:
|
||
drop_upto[cid] = i
|
||
keep = [r for i, r in enumerate(good)
|
||
if not (r["command_id"] in drop_upto and i <= drop_upto[r["command_id"]])]
|
||
tmp = self.path + ".tmp"
|
||
try:
|
||
os.remove(tmp)
|
||
except FileNotFoundError:
|
||
pass
|
||
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||
with os.fdopen(fd, "wb") as f:
|
||
for r in keep:
|
||
f.write((json.dumps(r, ensure_ascii=False) + "\n").encode("utf-8"))
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
os.replace(tmp, self.path)
|
||
return len(good) - len(keep)
|
||
|
||
|
||
def _parse_at(value) -> float:
|
||
"""Метка времени журнала в эпоху; наивное время трактуется как UTC."""
|
||
dt = datetime.fromisoformat(value)
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.timestamp()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Исполнитель: команда — желаемый результат; прочитать → сравнить → записать разницу
|
||
# ---------------------------------------------------------------------------
|
||
def _date(value) -> Optional[str]:
|
||
"""Даты Jira (duedate — YYYY-MM-DD, кастомные — с временем) сравниваем по дню."""
|
||
return (value or "")[:10] or None
|
||
|
||
|
||
def _nl(value) -> str:
|
||
r"""Текст для сравнения: Jira хранит правленное в UI с \r\n, команды приходят с \n (T3.9, проверено 13.10.2026)."""
|
||
return (value or "").replace("\r\n", "\n")
|
||
|
||
|
||
class ConfigGap(Exception):
|
||
"""В настройках хаба нет того, что нужно команде (R-11): команда падает с понятным текстом, а не молча теряет поле."""
|
||
|
||
|
||
class HubError(Exception):
|
||
"""Ошибка хаба внутри команды (refresh: known_keys, state): источник — хаб, не Jira (M-6 #18, design 12).
|
||
Повтор — только при недоступности хаба и 5xx; refresh в Jira не пишет, повтор безопасен."""
|
||
|
||
def __init__(self, cause: Exception):
|
||
super().__init__("хаб: %s" % cause)
|
||
self.retryable = isinstance(cause, ConnectionError) or (isinstance(cause, HttpError) and cause.code >= 500)
|
||
|
||
|
||
FILE_GONE = "файл удалён с хаба, загрузите заново"
|
||
|
||
|
||
class Executor:
|
||
def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None):
|
||
self.jira, self.s, self.hub = jira, settings, hub
|
||
self.sleep = time.sleep
|
||
self._symmetric = None
|
||
self._sprints = {} # str(доска) → имя → спринт; на одну команду
|
||
|
||
def run(self, cmd: dict) -> dict:
|
||
self.jira.reset_counters()
|
||
self._sprints = {}
|
||
handler = getattr(self, "do_" + str(cmd.get("type")), None)
|
||
if handler is None:
|
||
return self._out("failed", error="неизвестный тип команды %s" % cmd.get("type"))
|
||
try:
|
||
return handler(cmd.get("params") or {}, cmd)
|
||
except ConfigGap as e:
|
||
return self._out("failed", error=str(e))
|
||
except HubError as e:
|
||
out = self._out("failed", error=str(e))
|
||
out["retryable"] = e.retryable
|
||
out["peer"] = "хабом"
|
||
return out
|
||
except HttpError as e:
|
||
out = self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500]))
|
||
out["retryable"] = e.code >= 500 # исход неизвестен: Runner повторит через проверку (R-10)
|
||
return out
|
||
except ConnectionError as e:
|
||
out = self._out("failed", error=str(e))
|
||
out["retryable"] = True
|
||
return out
|
||
|
||
def _out(self, status, result=None, error=None) -> dict:
|
||
return {"status": status, "result": result, "error": error,
|
||
"jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited}
|
||
|
||
def _hub(self, fn, *args):
|
||
"""Вызов хаба из команды: его сбой — HubError, чтобы не выдать его за обрыв связи с Jira."""
|
||
try:
|
||
return fn(*args)
|
||
except (ConnectionError, HttpError) as e:
|
||
raise HubError(e) from None
|
||
|
||
def gap(self, cmd_type: str) -> None:
|
||
"""Зазор между записями внутри одной команды — по политике хаба для этого типа."""
|
||
seconds = int((self.s.get("pacing") or {}).get(cmd_type, 0) or 0)
|
||
if seconds > 0:
|
||
log(" пауза %d с" % seconds)
|
||
self.sleep(seconds)
|
||
|
||
def fid(self, logical: str) -> str:
|
||
return (self.s.get("field_ids") or {}).get(logical, logical)
|
||
|
||
def has_field(self, logical: str) -> bool:
|
||
return logical in ("summary", "description", "priority", "duedate", "labels") or bool((self.s.get("field_ids") or {}).get(logical))
|
||
|
||
def priority_value(self, name: str) -> dict:
|
||
mapped = (self.s.get("priority_map") or {}).get(name, name)
|
||
return {self.s.get("priority_value_key") or "name": mapped}
|
||
|
||
def issue_type_ref(self, kind: str) -> dict:
|
||
"""Тип задачи Jira из issue_types.<kind>: {id}, если id задан, иначе {name}; нет ни того, ни другого —
|
||
ошибка конфигурации, а не {"name": None}, на которое Jira ответит 400 без понятной причины (спека agent v11)."""
|
||
it = (self.s.get("issue_types") or {}).get(kind)
|
||
if isinstance(it, dict) and it.get("id"):
|
||
return {"id": str(it["id"])}
|
||
if isinstance(it, dict) and it.get("name"):
|
||
return {"name": it["name"]}
|
||
raise ConfigGap("нет настройки issue_types.%s" % kind)
|
||
|
||
def epic_field_id(self) -> str:
|
||
ep = self.s.get("epic_link") or {}
|
||
if ep.get("mode", "field") != "field":
|
||
raise ConfigGap("epic_key передан, но epic_link.mode = %r (поддержан только «field»)" % ep.get("mode"))
|
||
if not ep.get("field_id"):
|
||
raise ConfigGap("epic_key передан, но не настроено epic_link.field_id")
|
||
return ep["field_id"]
|
||
|
||
def epic_fields(self, epic_key) -> dict:
|
||
return {self.epic_field_id(): epic_key} if epic_key else {}
|
||
|
||
def need_field(self, logical: str) -> str:
|
||
if not self.has_field(logical):
|
||
raise ConfigGap("поле передано, но не настроено field_ids.%s" % logical)
|
||
return self.fid(logical)
|
||
|
||
# --- ensure_issue ----------------------------------------------------------
|
||
def do_ensure_issue(self, p: dict, cmd: dict) -> dict:
|
||
issuetype = self.issue_type_ref(p["issue_type"]) # до поиска дубля: без настройки — ни одного вызова Jira
|
||
summary = (p["summary"] or "").strip()
|
||
jql = "project = %s AND reporter = currentUser() AND created >= -7d ORDER BY created DESC" % self.s["project_key"]
|
||
start = 0
|
||
while start < 1000:
|
||
found = self.jira.get(API + "/search", jql=jql, fields="summary", maxResults=200, startAt=start)
|
||
issues = found.get("issues", [])
|
||
for it in issues:
|
||
if (it.get("fields", {}).get("summary") or "").strip() == summary:
|
||
return self._out("done", {"key": it["key"], "existed": True})
|
||
start += len(issues)
|
||
if not issues or start >= int(found.get("total") or 0):
|
||
break
|
||
fields = {"project": {"key": self.s["project_key"]}, "issuetype": issuetype, "summary": p["summary"]}
|
||
if p.get("description") is not None:
|
||
fields[self.fid("description")] = p["description"]
|
||
if p.get("priority"):
|
||
fields[self.fid("priority")] = self.priority_value(p["priority"])
|
||
if p.get("duedate"):
|
||
fields[self.fid("duedate")] = p["duedate"]
|
||
for logical in ("planned_start", "planned_end"):
|
||
if p.get(logical):
|
||
fields[self.need_field(logical)] = p[logical]
|
||
if p.get("labels") is not None:
|
||
fields[self.fid("labels")] = list(p["labels"])
|
||
if p.get("assignee"):
|
||
fields["assignee"] = {"name": p["assignee"]}
|
||
fields.update(self.epic_fields(p.get("epic_key")))
|
||
if p["issue_type"] == "epic" and self.s.get("epic_name_field_id"):
|
||
fields[self.s["epic_name_field_id"]] = p["summary"]
|
||
created = self.jira.post(API + "/issue", {"fields": fields})
|
||
return self._out("done", {"key": created["key"], "created": True})
|
||
|
||
# --- ensure_fields ---------------------------------------------------------
|
||
def do_ensure_fields(self, p: dict, cmd: dict) -> dict:
|
||
key = p["key"]
|
||
cur = self.jira.get(API + "/issue/%s" % key)["fields"]
|
||
fields, changed = {}, []
|
||
if "summary" in p and (cur.get("summary") or "") != (p["summary"] or ""):
|
||
fields["summary"] = p["summary"]
|
||
changed.append("summary")
|
||
if "description" in p and _nl(cur.get(self.fid("description"))) != _nl(p["description"]):
|
||
fields[self.fid("description")] = p["description"]
|
||
changed.append("description")
|
||
if "priority" in p:
|
||
want = self.priority_value(p["priority"])
|
||
vkey = list(want)[0]
|
||
have = (cur.get(self.fid("priority")) or {}).get(vkey)
|
||
if (None if have is None else str(have)) != str(want[vkey]):
|
||
fields[self.fid("priority")] = want
|
||
changed.append("priority")
|
||
for logical in ("duedate", "planned_start", "planned_end"):
|
||
if logical not in p:
|
||
continue
|
||
if logical != "duedate" and p[logical] and not self.has_field(logical):
|
||
self.need_field(logical)
|
||
if self.has_field(logical) and _date(cur.get(self.fid(logical))) != _date(p[logical]):
|
||
fields[self.fid(logical)] = p[logical]
|
||
changed.append(logical)
|
||
if "epic_key" in p and (p["epic_key"] or (self.s.get("epic_link") or {}).get("field_id")):
|
||
fid = self.epic_field_id()
|
||
if (cur.get(fid) or None) != (p["epic_key"] or None):
|
||
fields[fid] = p["epic_key"]
|
||
changed.append("epic_key")
|
||
if "assignee" in p and ((cur.get("assignee") or {}).get("name") or None) != (p["assignee"] or None):
|
||
fields["assignee"] = {"name": p["assignee"]} if p["assignee"] else None
|
||
changed.append("assignee")
|
||
if "labels_add" in p or "labels_remove" in p:
|
||
labels = list(cur.get(self.fid("labels")) or [])
|
||
remove = set(p.get("labels_remove") or [])
|
||
new = [l for l in labels if l not in remove] + [l for l in (p.get("labels_add") or []) if l not in labels]
|
||
if new != labels:
|
||
fields[self.fid("labels")] = new
|
||
changed.append("labels")
|
||
if not fields:
|
||
return self._out("done", {"changed": []})
|
||
self.jira.put(API + "/issue/%s" % key, {"fields": fields})
|
||
return self._out("done", {"changed": changed})
|
||
|
||
# --- ensure_comment --------------------------------------------------------
|
||
def do_ensure_comment(self, p: dict, cmd: dict) -> dict:
|
||
key, marker, body = p["key"], p["marker"], p["body"]
|
||
heads = (marker, "*%s.*" % marker) # маркер — только в начале тела (формат агента: «*маркер.* текст»)
|
||
if not body.startswith(heads):
|
||
body = "*%s.* %s" % (marker, body)
|
||
existing, start = None, 0
|
||
while existing is None:
|
||
page = self.jira.get(API + "/issue/%s/comment" % key, startAt=start, maxResults=100)
|
||
comments = page.get("comments", [])
|
||
for c in comments:
|
||
if (c.get("body") or "").startswith(heads):
|
||
existing = c
|
||
break
|
||
start += len(comments)
|
||
if not comments or page.get("total") is None or start >= int(page["total"]):
|
||
break
|
||
if existing and _nl(existing.get("body")) == _nl(body):
|
||
return self._out("done", {"comment_id": existing["id"], "changed": False})
|
||
if existing:
|
||
self.jira.put(API + "/issue/%s/comment/%s" % (key, existing["id"]), {"body": body})
|
||
return self._out("done", {"comment_id": existing["id"], "changed": True})
|
||
c = self.jira.post(API + "/issue/%s/comment" % key, {"body": body})
|
||
return self._out("done", {"comment_id": c.get("id"), "created": True})
|
||
|
||
# --- ensure_status ---------------------------------------------------------
|
||
@staticmethod
|
||
def _pick_transition(trs: list, step: str) -> Optional[dict]:
|
||
"""Переход, ведущий в статус step. При неоднозначности предпочитается он; имя перехода — запасной вариант."""
|
||
for t in trs:
|
||
if (t.get("to") or {}).get("name") == step:
|
||
return t
|
||
for t in trs:
|
||
if t.get("name") == step:
|
||
return t
|
||
return None
|
||
|
||
def do_ensure_status(self, p: dict, cmd: dict) -> dict:
|
||
key, target = p["key"], p["target"]
|
||
if isinstance(target, str):
|
||
path = [target]
|
||
elif isinstance(target, list) and all(isinstance(s, str) for s in target):
|
||
path = list(target)
|
||
else:
|
||
return self._out("failed", error="target должен быть строкой или списком строк")
|
||
if not path or not all(path):
|
||
return self._out("failed", error="пустой путь статусов")
|
||
final = path[-1]
|
||
cur = self.jira.get(API + "/issue/%s" % key, fields="status")["fields"]["status"]["name"]
|
||
start = cur
|
||
if cur == final:
|
||
return self._out("done", {"from": start, "to": final, "steps": 0})
|
||
if cur in path:
|
||
path = path[path.index(cur) + 1:]
|
||
steps = 0
|
||
for step in path:
|
||
if cur == step:
|
||
continue # повтор того же статуса подряд в пути: перехода не нужно
|
||
trs = self.jira.get(API + "/issue/%s/transitions" % key).get("transitions", [])
|
||
hit = self._pick_transition(trs, step)
|
||
if hit is None:
|
||
avail = ", ".join("%s→%s" % (t.get("name"), (t.get("to") or {}).get("name")) for t in trs)
|
||
return self._out("failed", {"from": start, "reached": cur},
|
||
"нет перехода из «%s» в «%s»; доступны: %s" % (cur, step, avail or "—"))
|
||
if steps:
|
||
self.gap("ensure_status")
|
||
self.jira.post(API + "/issue/%s/transitions" % key, {"transition": {"id": hit["id"]}})
|
||
cur = (hit.get("to") or {}).get("name") or step
|
||
steps += 1
|
||
return self._out("done", {"from": start, "to": cur, "steps": steps})
|
||
|
||
# --- ensure_link -----------------------------------------------------------
|
||
def symmetric_types(self) -> set:
|
||
"""Типы, у которых inward == outward («Связанные»): направление не важно.
|
||
Ошибка HTTP пробрасывается наружу и не кэшируется: следующая команда попробует снова."""
|
||
if self._symmetric is None:
|
||
types = self.jira.get(API + "/issueLinkType").get("issueLinkTypes", [])
|
||
self._symmetric = {t["name"] for t in types if t.get("inward") == t.get("outward")}
|
||
return self._symmetric
|
||
|
||
def do_ensure_link(self, p: dict, cmd: dict) -> dict:
|
||
src, dst = p["source"], p["target"]
|
||
types = self.s.get("link_types") or {}
|
||
# намеренный fallback (рулинг R-5): имя, которого нет в link_types, уходит в Jira как есть
|
||
lt = types.get(p["link_type"], p["link_type"])
|
||
managed = set(types.values()) | {lt}
|
||
symmetric = self.symmetric_types()
|
||
raw = self.jira.get(API + "/issue/%s" % src, fields="issuelinks")["fields"].get("issuelinks") or []
|
||
links = [] # (тип, другая задача, направление, id)
|
||
for l in raw:
|
||
t = (l.get("type") or {}).get("name")
|
||
if l.get("outwardIssue"):
|
||
links.append((t, l["outwardIssue"]["key"], "out", l.get("id")))
|
||
if l.get("inwardIssue"):
|
||
links.append((t, l["inwardIssue"]["key"], "in", l.get("id")))
|
||
wanted = [x for x in links if x[1] == dst and x[0] == lt and (x[2] == "out" or lt in symmetric)]
|
||
conflicts = [x for x in links if x[1] == dst and x[0] in managed and x not in wanted]
|
||
if conflicts and not p.get("replace"):
|
||
desc = ", ".join("«%s» (%s)" % (t, "обратное направление" if t == lt else "другой тип") for t, _, _, _ in conflicts)
|
||
return self._out("failed", error="между %s и %s уже есть: %s; нужен replace" % (src, dst, desc))
|
||
removed = []
|
||
for t, other, direction, lid in conflicts:
|
||
if not lid:
|
||
return self._out("failed", {"removed": removed}, "у связи «%s» с %s нет id — снять нельзя" % (t, other))
|
||
if removed:
|
||
self.gap("ensure_link")
|
||
self.jira.delete(API + "/issueLink/%s" % lid)
|
||
removed.append(t)
|
||
if wanted:
|
||
return self._out("done", {"existed": True, "removed": removed})
|
||
if removed:
|
||
self.gap("ensure_link")
|
||
# inwardIssue — источник (получает outward-описание «Блокирует начало…»), outwardIssue — цель
|
||
self.jira.post(API + "/issueLink", {"type": {"name": lt}, "inwardIssue": {"key": src}, "outwardIssue": {"key": dst}})
|
||
return self._out("done", {"created": True, "removed": removed})
|
||
|
||
# --- ensure_sprint ---------------------------------------------------------
|
||
def sprints(self, board: Any) -> Dict[str, dict]:
|
||
"""Имя → спринт (active и future) доски board; без доски — пусто."""
|
||
if str(board) not in self._sprints:
|
||
out, start = {}, 0
|
||
while board:
|
||
page = self.jira.get(AGILE + "/board/%s/sprint" % board, state="active,future", startAt=start, maxResults=50)
|
||
for sp in page.get("values", []):
|
||
out[sp["name"]] = sp
|
||
if page.get("isLast", True) or not page.get("values"):
|
||
break
|
||
start += len(page["values"])
|
||
self._sprints[str(board)] = out
|
||
return self._sprints[str(board)]
|
||
|
||
def do_ensure_sprint(self, p: dict, cmd: dict) -> dict:
|
||
key, want = p["key"], p["sprint"]
|
||
# у задачи без спринта Jira банка отвечает без ключа fields (#13)
|
||
issue = self.jira.get(AGILE + "/issue/%s" % key, fields="sprint")
|
||
cur = (((issue.get("fields") or {}).get("sprint")) or {}).get("name")
|
||
if want == "backlog": # в бэклог доска не нужна
|
||
if cur is None:
|
||
return self._out("done", {"sprint": None, "changed": False})
|
||
self.jira.post(AGILE + "/backlog/issue", {"issues": [key]})
|
||
return self._out("done", {"sprint": None, "from": cur})
|
||
board = p.get("board_id") or self.s.get("board_id") # без доски в команде — хаб до проектов (v11)
|
||
if not board:
|
||
return self._out("failed", error="board_id нет ни в команде, ни в настройках хаба — спринты недоступны")
|
||
if cur == want:
|
||
return self._out("done", {"sprint": want, "changed": False})
|
||
sp = self.sprints(board).get(want)
|
||
if not sp:
|
||
return self._out("failed", error="нет спринта «%s» среди активных и будущих доски %s: %s"
|
||
% (want, board, ", ".join(sorted(self.sprints(board))) or "—"))
|
||
self.jira.post(AGILE + "/sprint/%s/issue" % sp["id"], {"issues": [key]})
|
||
return self._out("done", {"sprint": want, "from": cur})
|
||
|
||
# --- ensure_attachment -----------------------------------------------------
|
||
def hub_file(self, sha: str, size: int) -> Optional[bytes]:
|
||
"""Файл вложения у хаба. None — хаб его не знает (404). Повреждённый или нечитаемый ответ — HubError с повтором,
|
||
как обрыв связи с хабом. Содержимое файла в текст ошибки не попадает."""
|
||
try:
|
||
got = self.hub.file(sha)
|
||
except HttpError as e:
|
||
if e.code == 404:
|
||
return None
|
||
if 200 <= e.code < 300: # ответ не JSON: в тексте Http — начало тела, то есть содержимого
|
||
raise HubError(ConnectionError("ответ на файл %s не JSON" % sha)) from None
|
||
raise HubError(e) from None
|
||
except ConnectionError as e:
|
||
raise HubError(e) from None
|
||
content = got.get("content") if isinstance(got, dict) else None
|
||
data = None
|
||
if isinstance(content, str):
|
||
try:
|
||
data = base64.b64decode(content, validate=True)
|
||
except ValueError: # binascii.Error — подкласс ValueError
|
||
data = None
|
||
if data is None or len(data) != size or hashlib.sha256(data).hexdigest() != sha:
|
||
raise HubError(ConnectionError("файл %s пришёл повреждённым: sha256 или размер не сходятся" % sha))
|
||
return data
|
||
|
||
def do_ensure_attachment(self, p: dict, cmd: dict) -> dict:
|
||
key, sha, filename, size = p["key"], p["file"], p["filename"], p.get("size")
|
||
if size is None: # хаб не нашёл файл при выдаче
|
||
return self._out("failed", error=FILE_GONE)
|
||
names = {filename, wire_filename(filename)} # Jira могла сохранить имя в «браузерной» форме (R-23)
|
||
issue = self.jira.get(API + "/issue/%s" % key, fields="attachment")
|
||
for a in (issue.get("fields") or {}).get("attachment") or []:
|
||
if a.get("filename") in names and a.get("size") == size:
|
||
return self._out("done", {"attachment_id": str(a.get("id")), "existed": True})
|
||
data = self.hub_file(sha, size)
|
||
if data is None:
|
||
return self._out("failed", error=FILE_GONE)
|
||
log(" вложение «%s» (%d байт) → %s" % (filename, size, key))
|
||
created = self.jira.upload(key, filename, data)
|
||
first = created[0] if isinstance(created, list) and created and isinstance(created[0], dict) else {}
|
||
if not first.get("id"):
|
||
return self._out("failed", error="Jira не вернула id вложения в ответе на загрузку")
|
||
return self._out("done", {"attachment_id": str(first["id"]), "created": True})
|
||
|
||
# --- refresh ---------------------------------------------------------------
|
||
STATE_FIELDS = ["summary", "description", "status", "priority", "assignee", "reporter", "duedate", "labels",
|
||
"issuetype", "created", "updated", "issuelinks", "comment"]
|
||
|
||
def state_fields(self) -> str:
|
||
extra = [self.fid(x) for x in ("planned_start", "planned_end") if self.has_field(x)]
|
||
ep = (self.s.get("epic_link") or {}) # режим связи с эпиком — только field (mode=link не поддержан)
|
||
if ep.get("mode", "field") == "field" and ep.get("field_id"):
|
||
extra.append(ep["field_id"])
|
||
return ",".join(self.STATE_FIELDS + extra)
|
||
|
||
def search_all(self, jql: str) -> List[dict]:
|
||
issues, start = [], 0
|
||
while True:
|
||
page = self.jira.get(API + "/search", jql=jql, fields=self.state_fields(), startAt=start, maxResults=100)
|
||
got = page.get("issues") or []
|
||
issues += got
|
||
start += len(got)
|
||
if not got or start >= int(page.get("total") or 0):
|
||
return issues
|
||
|
||
def search_tolerant(self, keys: List[str], missing: List[str], errors: List[HttpError]) -> List[dict]:
|
||
"""key in (...): Jira Server отвечает 400 на несуществующий/недоступный ключ. Чанк делим пополам до
|
||
одиночного ключа; одиночный с 400 — «не найден» (в missing), не ошибка. Прочие HttpError пробрасываются.
|
||
Первый 400 копится в errors: вызывающий пробрасывает его, если не найден НИ ОДИН ключ (неверная конфигурация,
|
||
а не удалённые задачи).
|
||
Стоимость: до ~2·m·log2(n) вызовов на m отсутствующих ключей из n — принятое ограничение, рулинг R-7."""
|
||
if not keys:
|
||
return []
|
||
try:
|
||
return self.search_all("key in (%s)" % ",".join(keys))
|
||
except HttpError as e:
|
||
if e.code != 400:
|
||
raise
|
||
if not errors:
|
||
errors.append(e)
|
||
if len(keys) == 1:
|
||
missing.append(keys[0])
|
||
return []
|
||
mid = len(keys) // 2
|
||
return self.search_tolerant(keys[:mid], missing, errors) + self.search_tolerant(keys[mid:], missing, errors)
|
||
|
||
def to_state(self, raw: dict) -> dict:
|
||
f = raw.get("fields") or {}
|
||
|
||
def name(obj):
|
||
return (obj or {}).get("name")
|
||
|
||
ep = self.s.get("epic_link") or {}
|
||
epic = f.get(ep["field_id"]) if ep.get("mode", "field") == "field" and ep.get("field_id") else None
|
||
return {"key": raw["key"], "id": raw.get("id"), "issue_type": name(f.get("issuetype")), "summary": f.get("summary"),
|
||
"description": f.get("description"), "status": name(f.get("status")), "priority": name(f.get("priority")),
|
||
"assignee": name(f.get("assignee")), "reporter": name(f.get("reporter")), "duedate": _date(f.get("duedate")),
|
||
"planned_start": _date(f.get(self.fid("planned_start"))) if self.has_field("planned_start") else None,
|
||
"planned_end": _date(f.get(self.fid("planned_end"))) if self.has_field("planned_end") else None,
|
||
"epic_key": epic, "labels": list(f.get("labels") or []), "sprint": None,
|
||
"created": f.get("created"), "jira_updated": f.get("updated")}
|
||
|
||
def links_from(self, raw: dict) -> List[dict]:
|
||
key, out = raw["key"], []
|
||
for l in (raw.get("fields") or {}).get("issuelinks") or []:
|
||
t = (l.get("type") or {}).get("name")
|
||
if l.get("outwardIssue"):
|
||
out.append({"source_key": key, "target_key": l["outwardIssue"]["key"], "link_type": t})
|
||
if l.get("inwardIssue"):
|
||
out.append({"source_key": l["inwardIssue"]["key"], "target_key": key, "link_type": t})
|
||
return out
|
||
|
||
@staticmethod
|
||
def comment_state(c: dict) -> dict:
|
||
"""Комментарий для хаба: автор — отображаемое имя, иначе логин; текст и время — как отдала Jira."""
|
||
a = c.get("author") or {}
|
||
return {"id": str(c.get("id")), "author": a.get("displayName") or a.get("name") or None,
|
||
"created": c.get("created"), "updated": c.get("updated"), "body": c.get("body") or ""}
|
||
|
||
def comments_of(self, raw: dict) -> Optional[List[dict]]:
|
||
"""Комментарии задачи из ответа search. Если Jira отдала не все (len < total) — дочитать постранично
|
||
через /issue/{key}/comment и взять полный список вместо частичного. Комментарий без id отбрасывается (M-1 #30);
|
||
403/404 при дочитывании — None: задача уходит без comments, хаб оставит прежние (M-2 #30)."""
|
||
key = raw["key"]
|
||
block = (raw.get("fields") or {}).get("comment") or {}
|
||
got = list(block.get("comments") or [])
|
||
total = block.get("total")
|
||
if total is not None and len(got) < int(total):
|
||
got, start = [], 0
|
||
try:
|
||
while True:
|
||
page = self.jira.get(API + "/issue/%s/comment" % key, startAt=start, maxResults=100)
|
||
part = page.get("comments") or []
|
||
got += part
|
||
start += len(part)
|
||
if not part or page.get("total") is None or start >= int(page["total"]):
|
||
break
|
||
except HttpError as e:
|
||
if e.code not in (403, 404):
|
||
raise
|
||
log("refresh: %s — комментарии не дочитаны (HTTP %d), задача уходит без comments" % (key, e.code))
|
||
return None
|
||
out = []
|
||
for c in got:
|
||
if c.get("id") is None or str(c.get("id")) == "":
|
||
log("refresh: %s — комментарий без id пропущен" % key)
|
||
continue
|
||
out.append(self.comment_state(c))
|
||
return out
|
||
|
||
def board_type(self, board: Any) -> Optional[str]:
|
||
"""Тип доски Jira: scrum или kanban; иное и ошибка Jira — None с записью в лог (refresh продолжается)."""
|
||
try:
|
||
t = self.jira.get(AGILE + "/board/%s" % board).get("type")
|
||
except HttpError as e:
|
||
log("refresh: доска %s — тип не определён (HTTP %d)" % (board, e.code))
|
||
return None
|
||
if t in ("scrum", "kanban"):
|
||
return t
|
||
log("refresh: доска %s — тип «%s» не scrum/kanban" % (board, t))
|
||
return None
|
||
|
||
def refresh_boards(self, p: dict) -> Tuple[List[Any], Optional[Dict[str, Optional[str]]]]:
|
||
"""(scrum-доски для имён спринтов, типы запрошенных досок для результата или None).
|
||
Без boards в команде — хаб до проектов: доска из настроек, как v11."""
|
||
if "boards" not in p:
|
||
board = self.s.get("board_id")
|
||
return ([board] if board else []), None
|
||
scrum: List[Any] = []
|
||
types: Dict[str, Optional[str]] = {}
|
||
seen = set()
|
||
for b in p.get("boards") or []:
|
||
if not isinstance(b, dict): # хаб шлёт [{id, type}]; иное не роняет refresh (M-2)
|
||
continue
|
||
bid = b.get("id")
|
||
if bid is None or str(bid) in seen:
|
||
continue
|
||
seen.add(str(bid))
|
||
t = b.get("type")
|
||
if t is None:
|
||
t = types[str(bid)] = self.board_type(bid)
|
||
if t == "scrum":
|
||
scrum.append(bid)
|
||
return scrum, (types or None)
|
||
|
||
def board_sprints(self, board: Any, types: Dict[str, Optional[str]]) -> Dict[str, dict]:
|
||
"""Спринты scrum-доски для refresh. Доска не отдала спринты: стала канбаном — тип в types, доска пропускается
|
||
(R-36); иначе ошибка Jira с номером доски, refresh failed (R-32, I-1)."""
|
||
try:
|
||
return self.sprints(board)
|
||
except HttpError as e:
|
||
if self.board_type(board) == "kanban":
|
||
types[str(board)] = "kanban"
|
||
return {}
|
||
raise HttpError(e.code, "доска %s: %s" % (board, e.text), e.headers) from None
|
||
|
||
def sprint_names(self, keys: List[str], boards: List[Any],
|
||
types: Optional[Dict[str, Optional[str]]] = None) -> Dict[str, str]:
|
||
# Источник спринта — Agile по активным и будущим спринтам досок (R-6): id поля спринта в настройках хаба нет,
|
||
# поэтому у задач закрытых спринтов sprint = None (принятое ограничение). Задача в спринтах двух досок —
|
||
# побеждает последнее найденное имя.
|
||
names = {}
|
||
if not keys:
|
||
return names
|
||
types = {} if types is None else types
|
||
for board in boards:
|
||
for name, sp in self.board_sprints(board, types).items():
|
||
for i in range(0, len(keys), 100):
|
||
chunk = keys[i:i + 100]
|
||
start = 0
|
||
while True:
|
||
page = self.jira.get(AGILE + "/sprint/%s/issue" % sp["id"], jql="key in (%s)" % ",".join(chunk),
|
||
fields="key", startAt=start, maxResults=50)
|
||
got = page.get("issues") or []
|
||
for it in got:
|
||
names[it["key"]] = name
|
||
start += len(got)
|
||
if not got or start >= int(page.get("total") or 0):
|
||
break
|
||
return names
|
||
|
||
def do_refresh(self, p: dict, cmd: dict) -> dict:
|
||
missing: List[str] = []
|
||
errors: List[HttpError] = []
|
||
scrum, types = self.refresh_boards(p) # первым шагом: тип досок нужен хабу и при пустых keys
|
||
if p.get("jql"):
|
||
raws = self.search_all(p["jql"])
|
||
else:
|
||
keys = list(dict.fromkeys(k for k in (p.get("keys") or []) if k))
|
||
if not keys:
|
||
empty: Dict[str, Any] = {"issues": 0, "neighbors": 0, "pages": 0}
|
||
if types is not None:
|
||
empty["boards"] = types
|
||
return self._out("done", empty)
|
||
raws = []
|
||
for i in range(0, len(keys), 100):
|
||
raws += self.search_tolerant(keys[i:i + 100], missing, errors)
|
||
if not raws and errors: # ни один ключ не найден: это конфигурация/запрос, а не удалённые задачи
|
||
raise errors[0]
|
||
fetched = {r["key"] for r in raws}
|
||
known = set(self._hub(self.hub.known_keys)) if self.hub else set()
|
||
linked = set()
|
||
for r in raws:
|
||
for l in self.links_from(r):
|
||
linked.add(l["source_key"])
|
||
linked.add(l["target_key"])
|
||
absent = sorted(k for k in linked if k not in fetched and k not in known)
|
||
neighbors = []
|
||
for i in range(0, len(absent), 100):
|
||
neighbors += self.search_tolerant(absent[i:i + 100], missing, errors)
|
||
everything = raws + neighbors
|
||
types = dict(types or {})
|
||
names = self.sprint_names([r["key"] for r in everything], scrum, types)
|
||
issues, links, seen = [], [], set()
|
||
for r in everything:
|
||
st = self.to_state(r)
|
||
st["sprint"] = names.get(r["key"])
|
||
comments = self.comments_of(r)
|
||
if comments is not None: # None — нет доступа к комментариям: поле не передаём (M-2 #30)
|
||
st["comments"] = comments
|
||
issues.append(st)
|
||
for l in self.links_from(r):
|
||
sig = (l["source_key"], l["target_key"], l["link_type"])
|
||
if sig not in seen:
|
||
seen.add(sig)
|
||
links.append(l)
|
||
neighbor_keys = [r["key"] for r in neighbors]
|
||
pages = 0
|
||
for i in range(0, len(issues), 100):
|
||
chunk = issues[i:i + 100]
|
||
chunk_keys = {x["key"] for x in chunk}
|
||
pages += 1
|
||
self._hub(self.hub.state, {"command_id": cmd.get("id"), "agent_version": AGENT_VERSION, "issues": chunk,
|
||
"links": [l for l in links if l["source_key"] in chunk_keys or l["target_key"] in chunk_keys],
|
||
"neighbor_keys": [k for k in neighbor_keys if k in chunk_keys],
|
||
"page": pages, "last_page": i + 100 >= len(issues)})
|
||
result = {"issues": len(raws), "neighbors": len(neighbors), "pages": pages}
|
||
if missing:
|
||
result["missing"] = list(dict.fromkeys(missing))[:50]
|
||
if types:
|
||
result["boards"] = types
|
||
return self._out("done", result)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте
|
||
# ---------------------------------------------------------------------------
|
||
class Updater:
|
||
"""Самообновление агента (спека agent «Самообновление агента», design 7–12). Работает только под run.py."""
|
||
RETRY_SECONDS = 300 # временный сбой подготовки — повтор не раньше (R-8)
|
||
HINT_SECONDS = 3600 # подсказка «запускайте через run.py» — не чаще (R-10)
|
||
SELFTEST_TIMEOUT = 120
|
||
VERSION_RE = re.compile(r"^AGENT_VERSION = (\d+)[ \t]*\r?$", re.M)
|
||
SHA_RE = re.compile(r"[0-9a-f]{64}")
|
||
|
||
def __init__(self, hub, base_dir: str, launcher_version: Optional[int], enabled: bool = True,
|
||
config_path: Optional[str] = None):
|
||
self.hub, self.dir, self.launcher_version, self.enabled = hub, base_dir, launcher_version, enabled
|
||
self.config_path = config_path or os.path.join(base_dir, "agent.json")
|
||
self.state_path = os.path.join(base_dir, "update.json")
|
||
self.new_path = os.path.join(base_dir, "agent.py.new")
|
||
self.agent_path = os.path.join(base_dir, "agent.py")
|
||
self.retry_after = {} # (version, sha256) -> clock(), раньше которого не повторять
|
||
self.hints = {} # вид подсказки -> clock() последней записи в лог (не чаще HINT_SECONDS)
|
||
self.clock = time.monotonic
|
||
self.should_stop = lambda: False
|
||
self.run_selftest = self._run_selftest
|
||
|
||
def state(self) -> Optional[dict]:
|
||
return read_update_state(self.state_path)
|
||
|
||
def confirm(self) -> None:
|
||
"""Первый успешный claim новой версии: trial этой версии → ok (запускатель перестаёт ждать отката)."""
|
||
st = self.state()
|
||
if st and st.get("state") == "trial" and st.get("to_version") == AGENT_VERSION:
|
||
try:
|
||
write_update_state(self.state_path, "ok", st.get("from_version"), AGENT_VERSION, st.get("sha256"))
|
||
except OSError as e: # диск полон — не падать; trial остался, повтор на следующем claim (M-3)
|
||
log("не удалось записать update.json: %s" % e)
|
||
return
|
||
log("обновление до v%d подтверждено" % AGENT_VERSION)
|
||
|
||
def _hint(self, kind: str, msg: str) -> None:
|
||
now = self.clock()
|
||
at = self.hints.get(kind)
|
||
if at is None or now - at >= self.HINT_SECONDS:
|
||
self.hints[kind] = now
|
||
log(msg)
|
||
|
||
def check(self, release) -> bool:
|
||
"""Вызывается после claim без команды. True — agent.py.new проверен, update.json = staged: run выходит с 75."""
|
||
if not self.enabled or not isinstance(release, dict):
|
||
return False
|
||
version, sha = release.get("version"), release.get("sha256")
|
||
if type(version) is not int or not isinstance(sha, str) or not self.SHA_RE.fullmatch(sha):
|
||
self._hint("format", "хаб прислал неполную публикацию: version=%.80r, sha256=%.80r" % (version, sha)) # design 10
|
||
return False
|
||
if version <= AGENT_VERSION:
|
||
return False
|
||
st = self.state()
|
||
if st and st.get("state") in ("rejected", "rolled_back") and st.get("to_version") == version \
|
||
and st.get("sha256") == sha:
|
||
return False
|
||
if self.launcher_version is None:
|
||
self._hint("launcher", "доступна v%d; для самообновления запускайте через run.py" % version)
|
||
return False
|
||
until = self.retry_after.get((version, sha))
|
||
if until is not None and self.clock() < until:
|
||
return False
|
||
return self.prepare(version, sha)
|
||
|
||
def prepare(self, version: int, sha: str) -> bool:
|
||
log("доступна v%d — готовлю обновление" % version)
|
||
try:
|
||
rel = self.hub.release()
|
||
except (ConnectionError, HttpError) as e:
|
||
return self._transient(version, sha, "скачивание: %s" % e)
|
||
content = rel.get("content") if isinstance(rel, dict) else None
|
||
if not isinstance(content, str) or rel.get("version") != version or rel.get("sha256") != sha:
|
||
return self._transient(version, sha, "публикация сменилась между запросами")
|
||
data = content.encode("utf-8")
|
||
if hashlib.sha256(data).hexdigest() != sha:
|
||
return self._reject(version, sha, "sha256 не сходится")
|
||
try:
|
||
compile(content, "agent.py.new", "exec")
|
||
except (SyntaxError, ValueError) as e:
|
||
return self._reject(version, sha, "не компилируется: %s" % e)
|
||
if self.VERSION_RE.findall(content) != [str(version)]:
|
||
return self._reject(version, sha, "AGENT_VERSION в файле не %d" % version)
|
||
tmp = self.new_path + ".tmp"
|
||
try: # диск полон, нет прав — временный сбой, не трейсбек (design 9)
|
||
with open(tmp, "wb") as f:
|
||
f.write(data)
|
||
try:
|
||
os.chmod(tmp, os.stat(self.agent_path).st_mode & 0o777)
|
||
except OSError:
|
||
pass
|
||
os.replace(tmp, self.new_path)
|
||
except OSError as e:
|
||
return self._transient(version, sha, "файлы: %s" % e)
|
||
if self.should_stop():
|
||
self._drop_new()
|
||
return False
|
||
try:
|
||
code, out = self.run_selftest(self.new_path)
|
||
except OSError as e:
|
||
return self._transient(version, sha, "selftest: %s" % e)
|
||
if self.should_stop(): # Ctrl-C уронил и selftest — это не отказ версии
|
||
self._drop_new()
|
||
return False
|
||
if code == 0:
|
||
try:
|
||
write_update_state(self.state_path, "staged", AGENT_VERSION, version, sha)
|
||
except OSError as e:
|
||
return self._transient(version, sha, "update.json: %s" % e)
|
||
log("v%d готова, перезапуск через run.py" % version)
|
||
return True
|
||
if code is None:
|
||
return self._transient(version, sha, "selftest: таймаут %d с" % self.SELFTEST_TIMEOUT)
|
||
if code == 3:
|
||
return self._transient(version, sha, "selftest: хаб или Jira недоступны")
|
||
return self._reject(version, sha, "selftest: код %s: %s" % (code, out[-500:]))
|
||
|
||
def _run_selftest(self, path: str):
|
||
try:
|
||
p = subprocess.run([sys.executable, path, "--config", self.config_path, "selftest"], cwd=self.dir,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=self.SELFTEST_TIMEOUT)
|
||
except subprocess.TimeoutExpired:
|
||
return None, ""
|
||
return p.returncode, p.stdout.decode("utf-8", "replace")
|
||
|
||
def _drop_new(self) -> None:
|
||
for path in (self.new_path, self.new_path + ".tmp"):
|
||
try:
|
||
os.remove(path)
|
||
except OSError:
|
||
pass
|
||
|
||
def _transient(self, version, sha, reason) -> bool:
|
||
self._drop_new()
|
||
self.retry_after[(version, sha)] = self.clock() + self.RETRY_SECONDS
|
||
log("обновление до v%d отложено на %d мин: %s" % (version, self.RETRY_SECONDS // 60, reason))
|
||
return False
|
||
|
||
def _reject(self, version, sha, reason) -> bool:
|
||
self._drop_new()
|
||
try:
|
||
write_update_state(self.state_path, "rejected", AGENT_VERSION, version, sha, reason)
|
||
except OSError as e:
|
||
log("не удалось записать update.json: %s" % e)
|
||
log("v%d отвергнута: %s" % (version, reason))
|
||
return False
|
||
|
||
|
||
class Runner:
|
||
RETRY_ATTEMPTS = 3 # повторов команды после обрыва связи с Jira / 5xx (R-10)
|
||
RETRY_MIN_WAIT = 30 # секунд до повтора
|
||
|
||
def __init__(self, cfg: dict, hub: HubClient, jira: JiraClient, journal: Journal):
|
||
self.cfg, self.hub, self.jira, self.journal = cfg, hub, jira, journal
|
||
self.settings = None # type: Optional[dict]
|
||
self.executor = None # type: Optional[Executor]
|
||
self.stop = False
|
||
self.sleep = time.sleep
|
||
self.idle_limit = None # type: Optional[int]
|
||
self.updater = None # type: Optional[Updater]
|
||
self.exit_code = 0 # EXIT_UPDATE — выход для подмены версии
|
||
|
||
@property
|
||
def poll(self) -> float:
|
||
return max(float(self.cfg.get("poll_seconds", 5) or 0), 0)
|
||
|
||
def hub_wait(self, e: Exception, base: float) -> float:
|
||
"""Залогировать сбой хаба и вернуть, сколько ждать. 401/403 — токен неверен: повторять часто бессмысленно."""
|
||
if isinstance(e, HttpError) and e.code in (401, 403):
|
||
log("хаб: HTTP %d — неверный hub_token в agent.json, повтор через 60 с" % e.code)
|
||
return 60
|
||
log("хаб: %s, повтор через %d с" % (e, base))
|
||
return base
|
||
|
||
def load_settings(self) -> None:
|
||
self.settings = self.hub.settings()
|
||
self.executor = Executor(self.jira, self.settings, self.hub)
|
||
self.executor.sleep = self.sleep
|
||
log("настройки хаба v%s (проект %s)" % (self.settings.get("settings_version"), self.settings.get("project_key")))
|
||
|
||
def restore_pace(self) -> None:
|
||
"""После перезапуска время последнего вызова Jira неизвестно (живёт в памяти). Берём самое позднее время записи
|
||
журнала: finished пишется сразу после последнего вызова команды, оценка не раньше настоящего вызова (M-9).
|
||
Время в будущем (часы ушли назад) — считаем «сейчас»: пауза полная (M-6 #12). Оценка служит только паузе:
|
||
last_call_iso (уходит хабу в claim) ставит лишь настоящий вызов Jira (M-7 #12)."""
|
||
if self.jira.last_call_at is not None:
|
||
return
|
||
at = self.journal.last_at()
|
||
if at is None:
|
||
return
|
||
ago = (datetime.now(timezone.utc) - at).total_seconds()
|
||
if ago < 0:
|
||
log("журнал: время последней записи в будущем (%s) — пауза перед первой командой полная" % at.isoformat())
|
||
self.jira.last_call_at = time.monotonic()
|
||
return # вторая строка «не позже <будущее>» противоречила бы первой (M-2 #18)
|
||
self.jira.last_call_at = time.monotonic() - ago
|
||
log("темп: последний вызов Jira не позже %s (по журналу)"
|
||
% at.astimezone(timezone.utc).replace(microsecond=0).isoformat())
|
||
|
||
def send_result(self, rec: dict) -> bool:
|
||
"""Отчёт хабу; повторяется каждые 10 с, пока хаб не подтвердит. 404/409/422 — хаб отверг (команда отменена
|
||
или удалена): считаем закрытой, чтобы не зациклиться. При остановке по сигналу делается одна попытка."""
|
||
cid = rec["command_id"]
|
||
payload = {"status": rec["status"], "result": rec.get("result"), "error": rec.get("error"),
|
||
"jira_calls": rec.get("jira_calls", 0), "rate_limited": rec.get("rate_limited", 0),
|
||
"agent_version": AGENT_VERSION}
|
||
while True:
|
||
try:
|
||
self.hub.result(cid, payload)
|
||
self.journal.acked(cid)
|
||
return True
|
||
except HttpError as e:
|
||
if e.code in (404, 409, 422):
|
||
log("хаб отверг отчёт по #%s: %s — считаю закрытым" % (cid, e))
|
||
self.journal.acked(cid)
|
||
return True
|
||
wait = self.hub_wait(e, 10)
|
||
except ConnectionError as e:
|
||
wait = self.hub_wait(e, 10)
|
||
if self.stop:
|
||
log("остановка: отчёт по #%s не подтверждён, будет дослан при следующем старте" % cid)
|
||
return False
|
||
self.sleep(wait)
|
||
|
||
def wait_seconds(self, seconds) -> bool:
|
||
"""Спать заданное время шагами по секунде; False — остановлены во время ожидания."""
|
||
remaining = float(seconds or 0)
|
||
if remaining > 0:
|
||
log(" пауза %.0f с" % remaining)
|
||
while remaining > 0 and not self.stop:
|
||
step = min(remaining, 1.0)
|
||
self.sleep(step)
|
||
remaining -= step
|
||
return not self.stop
|
||
|
||
def wait_pause(self, pause_seconds) -> bool:
|
||
"""Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы.
|
||
Если вызовов Jira ещё не было — пауза не нужна."""
|
||
pause = float(pause_seconds or 0)
|
||
if pause > 0 and self.jira.last_call_at is not None:
|
||
return self.wait_seconds(pause - (time.monotonic() - self.jira.last_call_at))
|
||
return not self.stop
|
||
|
||
def run_once(self, cmd: dict) -> dict:
|
||
try:
|
||
return self.executor.run(cmd)
|
||
except Exception as e: # баг в обработчике не должен ронять цикл и зацикливать восстановление
|
||
log(" ! необработанная ошибка: %r" % e)
|
||
return {"status": "failed", "result": None,
|
||
"error": "внутренняя ошибка агента: %s: %s" % (type(e).__name__, str(e)[:300]),
|
||
"jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited}
|
||
|
||
def execute(self, cmd: dict, journal_claimed: bool = True) -> None:
|
||
cid = cmd["id"]
|
||
log("#%s %s %s" % (cid, cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200]))
|
||
if journal_claimed:
|
||
self.journal.claimed(cid) # сразу после claim, до паузы (R-9): остановка не теряет команду
|
||
if not self.wait_pause(cmd.get("pause_seconds")):
|
||
return
|
||
self.journal.started(cid) # started — только после паузы
|
||
calls = limited = attempt = 0
|
||
while True:
|
||
out = self.run_once(cmd)
|
||
calls += out["jira_calls"]
|
||
limited += out["rate_limited"]
|
||
peer = out.pop("peer", "Jira") # с кем оборвалась связь: Jira или хаб (refresh)
|
||
if not out.pop("retryable", False):
|
||
break
|
||
if attempt >= self.RETRY_ATTEMPTS:
|
||
out = {"status": "failed", "result": None,
|
||
"error": "результат неизвестен (обрыв связи с %s), повтор безопасен: %s" % (peer, out.get("error"))}
|
||
break
|
||
attempt += 1
|
||
log(" ! обрыв связи с %s, повтор %d/%d через проверку: %s" % (peer, attempt, self.RETRY_ATTEMPTS, out.get("error")))
|
||
# F-1: обрыв не обновляет last_call_at, поэтому ждём фиксированно от момента отказа, а не от последнего вызова
|
||
if not self.wait_seconds(max(float(cmd.get("pause_seconds") or 0), self.RETRY_MIN_WAIT)):
|
||
return # остановка: finished не пишем, команду подхватит recover
|
||
out["jira_calls"], out["rate_limited"] = calls, limited
|
||
self.journal.finished(cid, out["status"], out.get("result"), out.get("error"), calls, limited)
|
||
log(" → %s %s" % (out["status"], out.get("error") or json.dumps(out.get("result"), ensure_ascii=False)))
|
||
rec = dict(out)
|
||
rec["command_id"] = cid
|
||
self.send_result(rec)
|
||
|
||
def fetch_command(self, cid):
|
||
"""Команда из хаба по id; None — хаб такой не знает; при недоступности хаба ждём и повторяем."""
|
||
while True:
|
||
try:
|
||
return self.hub.command(cid)
|
||
except HttpError as e:
|
||
if e.code in (404, 409, 422):
|
||
log("#%s: хаб ответил %s" % (cid, e))
|
||
return None
|
||
wait = self.hub_wait(e, 10)
|
||
except ConnectionError as e:
|
||
wait = self.hub_wait(e, 10)
|
||
if self.stop:
|
||
raise InterruptedError(cid)
|
||
self.sleep(wait)
|
||
|
||
def recover(self) -> None:
|
||
for rec in self.journal.unacked_finished():
|
||
log("досылаю отчёт по #%s" % rec["command_id"])
|
||
self.send_result(rec)
|
||
for cid in self.journal.unfinished_started():
|
||
if self.stop:
|
||
return
|
||
try:
|
||
cmd = self.fetch_command(cid)
|
||
except InterruptedError:
|
||
return
|
||
if cmd is None:
|
||
self.journal.acked(cid)
|
||
continue
|
||
if cmd.get("status") != "claimed":
|
||
log("#%s уже %s в хабе — пропускаю" % (cid, cmd.get("status")))
|
||
self.journal.acked(cid)
|
||
continue
|
||
log("#%s выдана до перезапуска — выполняю через проверку" % cid)
|
||
self.execute(cmd, journal_claimed=False) # claimed/started для неё уже в журнале (FM-4); пауза — из хаба (M-9)
|
||
|
||
def reload_settings_before(self, cmd) -> bool:
|
||
"""Настройки сменились: перечитать ДО выполнения. False — остановка, команда не должна выполняться
|
||
(если она уже выдана хабом, фиксируем claimed — подхватит recover)."""
|
||
while True:
|
||
try:
|
||
self.load_settings()
|
||
return True
|
||
except (ConnectionError, HttpError) as e:
|
||
wait = self.hub_wait(e, max(self.poll, 1))
|
||
if self.stop:
|
||
if cmd:
|
||
self.journal.claimed(cmd["id"])
|
||
return False
|
||
if not cmd:
|
||
self.sleep(wait)
|
||
return False
|
||
self.sleep(wait)
|
||
|
||
def loop(self) -> None:
|
||
self.restore_pace()
|
||
idle, recovered = 0, False
|
||
while not self.stop:
|
||
if self.settings is None:
|
||
try:
|
||
self.load_settings()
|
||
except (ConnectionError, HttpError) as e:
|
||
self.sleep(self.hub_wait(e, max(self.poll, 1)))
|
||
continue
|
||
if not recovered:
|
||
self.recover()
|
||
recovered = not self.stop
|
||
continue
|
||
try:
|
||
upd = self.updater
|
||
resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.jira.last_call_iso,
|
||
upd.state() if upd else None, upd.launcher_version if upd else None)
|
||
except (ConnectionError, HttpError) as e:
|
||
self.sleep(self.hub_wait(e, max(self.poll, 1)))
|
||
continue
|
||
if self.updater:
|
||
self.updater.confirm()
|
||
cmd = resp.get("command")
|
||
if resp.get("settings_version") != self.settings.get("settings_version"):
|
||
if not self.reload_settings_before(cmd):
|
||
continue
|
||
if cmd:
|
||
idle = 0
|
||
self.execute(cmd)
|
||
continue
|
||
if self.updater and not self.stop and self.updater.check(resp.get("agent_release")):
|
||
if not self.stop: # Ctrl-C во время подготовки — обычная остановка; staged доделает run.py
|
||
self.exit_code = EXIT_UPDATE
|
||
return
|
||
idle += 1
|
||
if self.idle_limit is not None and idle >= self.idle_limit:
|
||
return
|
||
self.sleep(self.poll)
|
||
|
||
|
||
def acquire_lock(journal_path: str):
|
||
"""Исключительная блокировка <журнал>.lock: один процесс агента на журнал. Хаб довыдаёт команду без отчёта при
|
||
следующем claim, поэтому второй процесс выполнял бы ту же команду одновременно с первым. None — уже занято."""
|
||
f = open(journal_path + ".lock", "a")
|
||
try:
|
||
fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except OSError:
|
||
f.close()
|
||
return None
|
||
return f
|
||
|
||
|
||
def cmd_run(cfg: dict, journal_path: str, config_path: Optional[str] = None) -> int:
|
||
lock = acquire_lock(journal_path)
|
||
if lock is None:
|
||
log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path)
|
||
return EXIT_LOCKED
|
||
try:
|
||
return _run(cfg, journal_path, config_path)
|
||
finally:
|
||
lock.close()
|
||
|
||
|
||
def cmd_prune(journal_path: str) -> int:
|
||
"""Удалить подтверждённые записи старше 30 дней. Под той же блокировкой, что run: prune перезаписывает файл, и запись,
|
||
добавленная работающим агентом (finished — неотправленный отчёт), пропала бы (T2.3)."""
|
||
lock = acquire_lock(journal_path)
|
||
if lock is None:
|
||
log("агент запущен с журналом %s — prune невозможен, сначала остановите агента" % journal_path)
|
||
return 1
|
||
try:
|
||
removed = Journal(journal_path).prune(30)
|
||
log("журнал: удалено записей %d" % removed)
|
||
return 0
|
||
finally:
|
||
lock.close()
|
||
|
||
|
||
def launcher_version() -> Optional[int]:
|
||
"""Версия запускателя из окружения (run.py ставит JIRA_AGENT_LAUNCHER); без запускателя — None."""
|
||
try:
|
||
return int(os.environ["JIRA_AGENT_LAUNCHER"])
|
||
except (KeyError, ValueError):
|
||
return None
|
||
|
||
|
||
def _run(cfg: dict, journal_path: str, config_path: Optional[str] = None) -> int:
|
||
runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path))
|
||
runner.updater = Updater(runner.hub, HERE, launcher_version(), cfg.get("auto_update", True) is not False, config_path)
|
||
runner.updater.should_stop = lambda: runner.stop
|
||
|
||
def on_signal(signum, frame):
|
||
try:
|
||
if runner.stop:
|
||
log("повторный сигнал — выхожу немедленно")
|
||
else:
|
||
log("сигнал %s: доделываю текущую команду и выхожу" % signum)
|
||
except RuntimeError: # print внутри обработчика, прервавшего другой print (reentrant call)
|
||
pass
|
||
if runner.stop:
|
||
sys.exit(130)
|
||
runner.stop = True
|
||
|
||
signal.signal(signal.SIGINT, on_signal)
|
||
signal.signal(signal.SIGTERM, on_signal)
|
||
log("jira-agent v%d: хаб %s, Jira %s, журнал %s" % (AGENT_VERSION, cfg["hub_url"], cfg["jira"]["base_url"], journal_path))
|
||
runner.loop()
|
||
log("выход для обновления (код %d)" % EXIT_UPDATE if runner.exit_code == EXIT_UPDATE else "остановлен")
|
||
return runner.exit_code
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CLI
|
||
# ---------------------------------------------------------------------------
|
||
def cmd_check(cfg: dict) -> int:
|
||
ok = True
|
||
hub = HubClient(cfg)
|
||
via = ""
|
||
if hub.folder is not None:
|
||
hub.timeout = 10 # реле молчит дольше 10 с — хаб недоступен
|
||
via = " через папку %s" % hub.folder.root
|
||
try:
|
||
s = hub.settings()
|
||
log("хаб: доступен%s, настройки v%s, проект %s" % (via, s.get("settings_version"), s.get("project_key")))
|
||
except (ConnectionError, HttpError) as e:
|
||
log("хаб: НЕДОСТУПЕН — %s" % e)
|
||
ok = False
|
||
try:
|
||
me = JiraClient(cfg).get(API + "/myself")
|
||
log("Jira: доступен, пользователь %s" % me.get("name"))
|
||
except (ConnectionError, HttpError) as e:
|
||
log("Jira: НЕДОСТУПЕН — %s" % e)
|
||
ok = False
|
||
return 0 if ok else 1
|
||
|
||
|
||
def cmd_selftest(cfg: dict) -> int:
|
||
"""Самопроверка версии перед обновлением: её запускает предыдущая версия. Контракт кодов MUST сохраняться:
|
||
0 — хаб и Jira доступны, 3 — что-то недоступно (временно); любой другой код — файл негоден (design 10)."""
|
||
log("jira-agent v%d" % AGENT_VERSION)
|
||
return 0 if cmd_check(cfg) == 0 else 3
|
||
|
||
|
||
def main(argv=None) -> int:
|
||
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
ap.add_argument("--config", default=os.path.join(HERE, "agent.json"))
|
||
ap.add_argument("--journal", default=os.path.join(HERE, "journal.jsonl"))
|
||
ap.add_argument("cmd", nargs="?", default="run", choices=["run", "check", "selftest", "prune"])
|
||
args = ap.parse_args(argv)
|
||
cfg = load_config(args.config)
|
||
if args.cmd == "check":
|
||
return cmd_check(cfg)
|
||
if args.cmd == "selftest":
|
||
return cmd_selftest(cfg)
|
||
if args.cmd == "prune":
|
||
return cmd_prune(args.journal)
|
||
return cmd_run(cfg, args.journal, args.config)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|