298 lines
18 KiB
Python
298 lines
18 KiB
Python
"""Смысл типов команд jira для очереди ядра: проверки постановки, паузы, отчёты, журнал синхронизаций (решения 4, 12)."""
|
||
import json
|
||
import logging
|
||
|
||
from ...core import files, projects, queue, settings
|
||
from ...core.errors import CommandError
|
||
from ...core.modules import CommandHooks
|
||
from . import attachments, commands, history, mirror, refresh
|
||
from . import projects as jira_projects
|
||
from . import settings as jira_settings
|
||
from .fetch import TooLarge
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
READ_TYPES = frozenset({"refresh", "fetch_attachment"}) # не пишут в Jira: без хвостового refresh и без резервной копии
|
||
|
||
|
||
# --- постановка ---------------------------------------------------------------
|
||
|
||
def check_setup(conn, cmds, files_dir) -> None:
|
||
"""Проверки постановки, которым нужны настройки хаба и хранилище файлов (bug-and-attachments, решения 6–7): 422
|
||
до создания пакета. Номер команды — как в присланном пакете (хвостовой refresh ещё не добавлен).
|
||
files_dir=None — хранилища нет: любой файл «не загружен»."""
|
||
bug = (settings.get(conn, "issue_types") or {}).get("bug")
|
||
bug_ok = isinstance(bug, dict) and bool(bug.get("name") or bug.get("id"))
|
||
for i, c in enumerate(cmds, 1):
|
||
p = c["params"]
|
||
if c["type"] == "ensure_issue" and p["issue_type"] == "bug" and not bug_ok:
|
||
raise CommandError(f"команда {i}: нет настройки issue_types.bug")
|
||
if c["type"] == "ensure_attachment":
|
||
if not files.is_sha(p["file"]):
|
||
raise CommandError(f"команда {i}: file — sha256 из POST /files (64 символа 0-9a-f)")
|
||
if files_dir is None or not files.exists(files_dir, p["file"]):
|
||
raise CommandError(f"команда {i}: файл {p['file']} не загружен: POST /files")
|
||
if not p["filename"].strip():
|
||
raise CommandError(f"команда {i}: filename пустой")
|
||
if "/" in p["filename"] or "\\" in p["filename"]:
|
||
raise CommandError(f"команда {i}: filename без / и \\ — имя вложения, а не путь")
|
||
if '"' in p["filename"] or any(ord(ch) < 32 or ord(ch) == 127 for ch in p["filename"]):
|
||
# кавычка, CR, LF ломают заголовок части multipart/form-data у агента (R-2)
|
||
raise CommandError(f"команда {i}: filename без кавычек и управляющих символов (CR, LF и др.)")
|
||
if c["type"] == "fetch_attachment":
|
||
if not p["attachment_id"].isdigit():
|
||
raise CommandError(f"команда {i}: attachment_id — цифры id вложения из issue")
|
||
row = conn.execute("SELECT size FROM issue_attachment WHERE attachment_id = ? AND key = ?",
|
||
(p["attachment_id"], p["key"])).fetchone()
|
||
if row is None:
|
||
raise CommandError(f"у {p['key']} нет вложения #{p['attachment_id']} в зеркале — обновите задачу")
|
||
if row["size"] is not None and row["size"] > files.MAX_SIZE: # size NULL — размер неизвестен, ставим
|
||
raise TooLarge(f"файл больше 10 МБ ({attachments.fmt_size(row['size'])}) — скачайте в Jira")
|
||
_assign_boards(conn, cmds)
|
||
|
||
|
||
def _assign_boards(conn, cmds) -> None:
|
||
"""Доска проекта задачи в params ensure_sprint (design #7): метки временного ключа — из его ensure_issue в пакете,
|
||
иначе из зеркала; 422 — задача вне проектов, разные доски, нет доски, канбан, тип ещё не известен."""
|
||
temp_labels = {c["params"]["temp_key"]: c["params"].get("labels") or [] for c in cmds if c["type"] == "ensure_issue"}
|
||
found = jira_projects.light_list(conn)
|
||
for i, c in enumerate(cmds, 1):
|
||
if c["type"] != "ensure_sprint":
|
||
continue
|
||
key = c["params"]["key"]
|
||
labels = temp_labels[key] if key in temp_labels else (mirror.labels_of(conn, key) or [])
|
||
own = [p for p in found if p["label"] in labels]
|
||
if not own:
|
||
raise CommandError(f"команда {i}: {key} не входит ни в один проект — доску для спринта не определить")
|
||
if len({p["board_id"] for p in own}) > 1:
|
||
raise CommandError(f"команда {i}: {key} в проектах {', '.join(p['label'] for p in own)} с разными досками")
|
||
p = own[0]
|
||
err = _board_error(p["label"], p["board_id"], p["board_type"])
|
||
if err:
|
||
raise CommandError(f"команда {i}: {err}")
|
||
c["params"] = {**c["params"], "board_id": p["board_id"]}
|
||
|
||
|
||
def _board_error(label, board_id, board_type) -> str | None:
|
||
"""Общие тексты 422 для ensure_sprint и команд спринта: доски нет, канбан, тип ещё не известен; None — доска годится."""
|
||
if board_id is None:
|
||
return f"у проекта {label} нет доски Jira"
|
||
if board_type == "kanban":
|
||
return f"у проекта {label} канбан-доска, спринтов нет"
|
||
if board_type is None:
|
||
return f"тип доски проекта {label} ещё не известен — дождитесь обновления из Jira"
|
||
return None
|
||
|
||
|
||
def _assign_project_board(conn, cmds, project) -> None:
|
||
"""Доска проекта пакета в params команд спринта (ensure_sprint_meta/state): спринт не про задачу, проект — из пакета.
|
||
Новые словари params — входные не мутируются. Пакет без проекта, нет доски, канбан, тип не известен — 422."""
|
||
board = None
|
||
for i, c in enumerate(cmds, 1):
|
||
if c["type"] not in commands.SPRINT_TYPES:
|
||
continue
|
||
if project is None:
|
||
raise CommandError(f"команда {i} ({c['type']}): команды спринта — только в пакете проекта")
|
||
if board is None:
|
||
found = next((p for p in jira_projects.light_list(conn) if p["label"] == project), None)
|
||
board = found or {"label": project, "board_id": None, "board_type": None}
|
||
err = _board_error(board["label"], board["board_id"], board["board_type"])
|
||
if err:
|
||
raise CommandError(f"команда {i}: {err}")
|
||
c["params"] = {**c["params"], "board_id": board["board_id"]}
|
||
|
||
|
||
def _check_project(conn, cmds, project) -> list[dict]:
|
||
"""Проверки пакета проекта (spec commands, «Пакет проекта»): новые словари команд — входные params не мутируются.
|
||
Номер команды — как в присланном пакете."""
|
||
if projects.get(conn, project) is None:
|
||
raise CommandError(f"проекта {project} нет")
|
||
others = {l.casefold() for l in projects.labels(conn) if l != project}
|
||
temps = {c["params"]["temp_key"] for c in cmds if c["type"] == "ensure_issue"}
|
||
|
||
def own(key):
|
||
return key in temps or project in (mirror.labels_of(conn, key) or [])
|
||
|
||
out = []
|
||
for i, c in enumerate(cmds, 1):
|
||
t, p = c["type"], dict(c["params"])
|
||
if t == "ensure_issue":
|
||
labels = []
|
||
for l in p.get("labels") or []:
|
||
if l.casefold() in others:
|
||
raise CommandError(f"команда {i}: метка другого проекта: {l}")
|
||
l = project if l.casefold() == project.casefold() else l
|
||
if l not in labels or l != project:
|
||
labels.append(l)
|
||
if project not in labels:
|
||
labels.append(project)
|
||
p["labels"] = labels
|
||
elif t == "ensure_link":
|
||
if not (own(p["source"]) or own(p["target"])):
|
||
raise CommandError(f"команда {i}: связь {p['source']} → {p['target']} не касается проекта {project}")
|
||
elif t in commands.SPRINT_TYPES:
|
||
pass # спринт — не задача проекта: доску ставит _assign_project_board
|
||
elif t not in READ_TYPES: # fetch_attachment — не запись в Jira, проекта не касается
|
||
if t == "ensure_fields":
|
||
if project.casefold() in {l.casefold() for l in p.get("labels_remove") or []}:
|
||
raise CommandError(f"команда {i}: снять метку своего проекта {project} нельзя")
|
||
added = []
|
||
for l in p.get("labels_add") or []:
|
||
if l.casefold() in others:
|
||
raise CommandError(f"команда {i}: метка другого проекта: {l}")
|
||
l = project if l.casefold() == project.casefold() else l
|
||
if l not in added:
|
||
added.append(l)
|
||
if "labels_add" in p:
|
||
p["labels_add"] = added
|
||
if not own(p["key"]):
|
||
raise CommandError(f"команда {i}: {p['key']} не в проекте {project}")
|
||
out.append({**c, "params": p})
|
||
return out
|
||
|
||
|
||
def check_batch(conn, cmds, project, files_dir) -> list[dict]:
|
||
if project is not None:
|
||
cmds = _check_project(conn, cmds, project)
|
||
check_setup(conn, cmds, files_dir)
|
||
_assign_project_board(conn, cmds, project)
|
||
return cmds
|
||
|
||
|
||
def tail(cmds) -> dict | None:
|
||
"""Хвостовой refresh пишущего пакета: ключи из всех команд, чтобы зеркало увидело результат."""
|
||
keys = []
|
||
for c in cmds:
|
||
keys += [k for k in commands.referenced_keys(c) if k not in keys]
|
||
if any(c["type"] not in READ_TYPES for c in cmds) and keys:
|
||
return {"type": "refresh", "params": {"keys": keys}}
|
||
return None
|
||
|
||
|
||
def pause(conn, ctype) -> int:
|
||
return int(jira_settings.pacing(conn).get(ctype, 0))
|
||
|
||
|
||
# --- выдача -------------------------------------------------------------------
|
||
|
||
def deliver_params(conn, cmd, files_dir) -> dict:
|
||
"""Параметры команды для агента: ensure_attachment получает size — текущий размер файла в хранилище или None,
|
||
если файла уже нет (решение 5); refresh — boards, доски проектов (design #7), и history — отметка журнала
|
||
(change-history). В базе params остаются без них: повтор команды проходит проверку как новая."""
|
||
if cmd["type"] == "refresh":
|
||
out = {**cmd["params"], "boards": jira_projects.boards(conn)}
|
||
hist = history.delivery_param(conn, cmd["params"]) # отметка журнала (change-history, design #5)
|
||
if hist is not None:
|
||
out["history"] = hist
|
||
return out
|
||
if cmd["type"] != "ensure_attachment":
|
||
return cmd["params"]
|
||
size = files.size(files_dir, cmd["params"].get("file")) if files_dir is not None else None
|
||
return {**cmd["params"], "size": size}
|
||
|
||
|
||
def drop_unresolved(cmd, unresolved) -> list[str]:
|
||
"""refresh с неразрешёнными временными ключами выполняется по остальным (временные выбрасываются из keys); остался
|
||
хоть один ключ — неразрешённых нет. cmd["params"] — уже после подстановки."""
|
||
if cmd["type"] == "refresh":
|
||
cmd["params"]["keys"] = [k for k in cmd["params"].get("keys", []) if not commands.is_temp(k)]
|
||
return [] if cmd["params"]["keys"] else unresolved
|
||
return unresolved
|
||
|
||
|
||
def on_claim(conn, cmd, now, redelivery) -> None:
|
||
if cmd["type"] != "refresh":
|
||
return
|
||
if redelivery:
|
||
# повторное выполнение загрузит то же заново: счётчики с нуля; started_at — граница пометки соседей,
|
||
# не сдвигается (M-5 #12)
|
||
conn.execute("UPDATE sync_run SET issues_count = 0, links_count = 0 WHERE command_id = ?", (cmd["id"],))
|
||
else:
|
||
p = cmd["params"]
|
||
conn.execute("INSERT INTO sync_run(started_at, kind, query, command_id) VALUES (?, ?, ?, ?)",
|
||
(now, "jql" if p.get("jql") else "keys", p.get("jql") or ",".join(p.get("keys", [])), cmd["id"]))
|
||
|
||
|
||
def on_failed(conn, cmd, now) -> None:
|
||
if cmd["type"] == "refresh": # загрузка закончилась ничем — журнал синхронизаций закрыт (M-4 #12)
|
||
conn.execute("UPDATE sync_run SET finished_at = ? WHERE command_id = ?", (now, cmd["id"]))
|
||
|
||
|
||
# --- отчёты -------------------------------------------------------------------
|
||
|
||
def check_result(conn, cmd, status, result, files_dir) -> tuple[str, str | None]:
|
||
"""fetch_attachment: файл должен лежать в хранилище с заявленным размером; иначе отчёт не принимаем (уборка успела
|
||
раньше, чужой размер). Ошибка None — отчёт принят как есть."""
|
||
if cmd["type"] == "fetch_attachment" and status == "done":
|
||
sha = result.get("file") if isinstance(result, dict) else None
|
||
real = files.size(files_dir, sha) if files_dir is not None else None
|
||
if real is None or real != result.get("size"):
|
||
return "failed", f"хаб: файл {sha} не найден в хранилище"
|
||
return status, None
|
||
|
||
|
||
def on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_dir) -> None:
|
||
ctype = cmd["type"]
|
||
if ctype == "ensure_issue" and status == "done" and result and result.get("key"):
|
||
mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"])
|
||
if ctype == "fetch_attachment" and status == "done": # check_result уже сверил файл и размер
|
||
conn.execute("INSERT OR REPLACE INTO attachment_file(attachment_id, file, size, fetched_at) VALUES (?, ?, ?, ?)",
|
||
(cmd["params"]["attachment_id"], result["file"], result["size"], now))
|
||
if ctype == "refresh":
|
||
conn.execute("UPDATE sync_run SET finished_at = ?, jira_calls = ?, rate_limited = ? WHERE command_id = ?",
|
||
(now, int(jira_calls or 0), int(rate_limited or 0), cmd["id"]))
|
||
if isinstance(result, dict) and "boards" in result:
|
||
jira_projects.record_board_types(conn, result["boards"]) # в любом статусе; без ключа (агент v11) типы не трогаем
|
||
full = status == "done" and refresh.is_full_refresh(conn, cmd["id"])
|
||
if full and history.accepted(result):
|
||
conn.execute("UPDATE sync_run SET history_at = ? WHERE command_id = ?", (now, cmd["id"]))
|
||
if full:
|
||
# полный прогон не обновил задачу — в базовую JQL она больше не входит: сосед. Пустой прогон (0 задач) —
|
||
# почти всегда сбой выборки, а не «все задачи ушли»: соседей не трогаем (M-12)
|
||
run = conn.execute("SELECT started_at, issues_count FROM sync_run WHERE command_id = ?", (cmd["id"],)).fetchone()
|
||
if run and run["issues_count"] > 0:
|
||
conn.execute("UPDATE issue SET neighbor = 1 WHERE fetched_at < ?", (run["started_at"],))
|
||
|
||
|
||
def after_result(conn, cmd, status) -> None:
|
||
if cmd["type"] == "refresh" and status == "done":
|
||
queue.reconcile(conn)
|
||
|
||
|
||
def on_late_result(conn, cmd, status, result, jira_calls, rate_limited) -> bool:
|
||
"""Поздний отчёт о созданной задаче по команде, которую хаб уже закрыл (force-отмена — M-5, предел выдач — M-4 #12):
|
||
ключ не теряем, статус не трогаем."""
|
||
if (cmd["status"] in ("cancelled", "failed") and cmd["type"] == "ensure_issue" and status == "done"
|
||
and result and result.get("key") and not cmd["result"]):
|
||
conn.execute("UPDATE command SET result = ?, jira_calls = ?, rate_limited = ? WHERE id = ?",
|
||
(json.dumps(result, ensure_ascii=False), int(jira_calls or 0), int(rate_limited or 0), cmd["id"]))
|
||
mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"])
|
||
log.warning("команда %s уже %s, но агент создал %s — ключ записан в результат",
|
||
cmd["id"], cmd["status"], result["key"])
|
||
return True
|
||
return False
|
||
|
||
|
||
def check_confirmed(conn, cmd) -> tuple[bool, dict | None]:
|
||
return commands.check_confirmed(conn, cmd, settings.get(conn, "link_types") or {})
|
||
|
||
|
||
def retry_params(cmd) -> dict:
|
||
"""Повтор ensure_sprint и команд спринта берёт доску заново (она в params ставится при постановке)."""
|
||
drop = cmd["type"] in ("ensure_sprint", *commands.SPRINT_TYPES)
|
||
return {k: v for k, v in cmd["params"].items() if not (drop and k == "board_id")}
|
||
|
||
|
||
def file_refs(conn) -> set[str]:
|
||
"""Файлы, на которые ссылаются ensure_attachment в queued или claimed: уборка их не трогает."""
|
||
return {r[0] for r in conn.execute("SELECT json_extract(params, '$.file') FROM command "
|
||
"WHERE type = 'ensure_attachment' AND status IN ('queued', 'claimed')")}
|
||
|
||
|
||
HOOKS = CommandHooks(
|
||
types=frozenset(commands.SPEC), read_types=READ_TYPES, validate=commands.validate, check_batch=check_batch,
|
||
tail=tail, pause=pause, deliver_params=deliver_params, temp_definer=("ensure_issue", "temp_key", "key"),
|
||
temp_keys_used=commands.temp_keys_used, substitute=commands.substitute, drop_unresolved=drop_unresolved,
|
||
retry_params=retry_params, check_confirmed=check_confirmed, on_claim=on_claim, on_failed=on_failed,
|
||
check_result=check_result, on_result=on_result, after_result=after_result, on_late_result=on_late_result)
|