1211 lines
60 KiB
Python
1211 lines
60 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 prune # почистить journal.jsonl от подтверждённых записей старше 30 дней
|
||
"""
|
||
import argparse
|
||
import base64
|
||
import fcntl
|
||
import http.client
|
||
import json
|
||
import os
|
||
import signal
|
||
import ssl
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
AGENT_VERSION = 3
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# HTTP
|
||
# ---------------------------------------------------------------------------
|
||
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:
|
||
url = self.base + path
|
||
if params:
|
||
url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
|
||
data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else 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 data is not None:
|
||
req.add_header("Content-Type", "application/json; charset=utf-8")
|
||
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 call(self, method, path, body=None, params=None):
|
||
for attempt in range(1, 6):
|
||
try:
|
||
return super().call(method, path, body, params)
|
||
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 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("ответ не объект")
|
||
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
|
||
status, text = int(resp.get("status") or 0), resp.get("body") or ""
|
||
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) -> dict:
|
||
return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version,
|
||
"last_jira_call_at": last_jira_call_at}, timeout=max(self.timeout, 45))
|
||
|
||
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"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Журнал: только дозапись; 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
|
||
|
||
|
||
class ConfigGap(Exception):
|
||
"""В настройках хаба нет того, что нужно команде (R-11): команда падает с понятным текстом, а не молча теряет поле."""
|
||
|
||
|
||
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 = None
|
||
|
||
def run(self, cmd: dict) -> dict:
|
||
self.jira.reset_counters()
|
||
self._sprints = None
|
||
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 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 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:
|
||
it = (self.s.get("issue_types") or {}).get(kind) or {}
|
||
return {"id": str(it["id"])} if it.get("id") else {"name": it.get("name")}
|
||
|
||
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:
|
||
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": self.issue_type_ref(p["issue_type"]),
|
||
"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 (cur.get(self.fid("description")) or "") != (p["description"] or ""):
|
||
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 (existing.get("body") or "") == 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) -> Dict[str, dict]:
|
||
"""Имя → спринт (active и future) доски board_id; без доски — пусто."""
|
||
if self._sprints is None:
|
||
out, start = {}, 0
|
||
board = self.s.get("board_id")
|
||
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 = out
|
||
return self._sprints
|
||
|
||
def do_ensure_sprint(self, p: dict, cmd: dict) -> dict:
|
||
key, want = p["key"], p["sprint"]
|
||
cur = ((self.jira.get(AGILE + "/issue/%s" % key, fields="sprint")["fields"].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})
|
||
if not self.s.get("board_id"):
|
||
return self._out("failed", error="board_id не задан в настройках хаба — спринты недоступны")
|
||
if cur == want:
|
||
return self._out("done", {"sprint": want, "changed": False})
|
||
sp = self.sprints().get(want)
|
||
if not sp:
|
||
return self._out("failed", error="нет спринта «%s» среди активных и будущих доски %s: %s"
|
||
% (want, self.s["board_id"], ", ".join(sorted(self.sprints())) or "—"))
|
||
self.jira.post(AGILE + "/sprint/%s/issue" % sp["id"], {"issues": [key]})
|
||
return self._out("done", {"sprint": want, "from": cur})
|
||
|
||
# --- refresh ---------------------------------------------------------------
|
||
STATE_FIELDS = ["summary", "description", "status", "priority", "assignee", "reporter", "duedate", "labels",
|
||
"issuetype", "created", "updated", "issuelinks"]
|
||
|
||
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
|
||
|
||
def sprint_names(self, keys: List[str]) -> Dict[str, str]:
|
||
# Источник спринта — Agile по активным и будущим спринтам доски (R-6): id поля спринта в настройках хаба нет,
|
||
# поэтому у задач закрытых спринтов sprint = None (принятое ограничение).
|
||
names = {}
|
||
if not keys or not self.s.get("board_id"):
|
||
return names
|
||
for name, sp in self.sprints().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] = []
|
||
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:
|
||
return self._out("done", {"issues": 0, "neighbors": 0, "pages": 0})
|
||
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.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
|
||
names = self.sprint_names([r["key"] for r in everything])
|
||
issues, links, seen = [], [], set()
|
||
for r in everything:
|
||
st = self.to_state(r)
|
||
st["sprint"] = names.get(r["key"])
|
||
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.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]
|
||
return self._out("done", result)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте
|
||
# ---------------------------------------------------------------------------
|
||
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]
|
||
|
||
@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)."""
|
||
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())
|
||
return
|
||
self.jira.last_call_at = time.monotonic() - ago
|
||
self.jira.last_call_iso = at.astimezone(timezone.utc).replace(microsecond=0).isoformat()
|
||
log("темп: последний вызов Jira не позже %s (по журналу)" % self.jira.last_call_iso)
|
||
|
||
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"]
|
||
if not out.pop("retryable", False):
|
||
break
|
||
if attempt >= self.RETRY_ATTEMPTS:
|
||
out = {"status": "failed", "result": None,
|
||
"error": "результат неизвестен (обрыв связи с Jira), повтор безопасен: %s" % out.get("error")}
|
||
break
|
||
attempt += 1
|
||
log(" ! обрыв связи с Jira, повтор %d/%d через проверку: %s" % (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:
|
||
resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.jira.last_call_iso)
|
||
except (ConnectionError, HttpError) as e:
|
||
self.sleep(self.hub_wait(e, max(self.poll, 1)))
|
||
continue
|
||
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
|
||
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) -> int:
|
||
lock = acquire_lock(journal_path)
|
||
if lock is None:
|
||
log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path)
|
||
return 1
|
||
try:
|
||
return _run(cfg, journal_path)
|
||
finally:
|
||
lock.close()
|
||
|
||
|
||
def _run(cfg: dict, journal_path: str) -> int:
|
||
runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path))
|
||
|
||
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("остановлен")
|
||
return 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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 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", "prune"])
|
||
args = ap.parse_args(argv)
|
||
cfg = load_config(args.config)
|
||
if args.cmd == "check":
|
||
return cmd_check(cfg)
|
||
if args.cmd == "prune":
|
||
removed = Journal(args.journal).prune(30)
|
||
log("журнал: удалено записей %d" % removed)
|
||
return 0
|
||
return cmd_run(cfg, args.journal)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|