299 lines
17 KiB
Python
299 lines
17 KiB
Python
"""Контракт команд: каждая — желаемый результат на одной задаче Jira (specs/commands/spec.md).
|
||
Здесь валидация, работа с ключами (в том числе временными new:<n>), оценка длительности и
|
||
проверка «результат уже виден в зеркале» для подтверждения по состоянию."""
|
||
import math
|
||
import re
|
||
from datetime import date
|
||
|
||
from ...core.errors import CommandError # noqa: F401 (реэкспорт: прежнее место)
|
||
from . import mirror
|
||
|
||
TEMP_KEY = re.compile(r"^new:\d+$")
|
||
FIELD_PARAMS = ["summary", "description", "priority", "duedate", "planned_start", "planned_end", "epic_key", "assignee"]
|
||
|
||
DATE = "YYYY-MM-DD"
|
||
FIELD_HINTS = {"duedate": DATE, "planned_start": DATE, "planned_end": DATE, "epic_key": "ключ или new:<n>",
|
||
"assignee": "логин Jira"}
|
||
|
||
# hints — формат значения для описания MCP batch_create, note — условие на набор параметров (#44)
|
||
SPEC = {
|
||
"ensure_issue": {"required": ["temp_key", "issue_type", "summary"],
|
||
"optional": ["description", "priority", "duedate", "planned_start", "planned_end", "labels",
|
||
"epic_key", "assignee"],
|
||
"keys": ["temp_key", "epic_key"],
|
||
"hints": {"temp_key": "new:<n>", "issue_type": "task|epic|bug", "labels": "список", **FIELD_HINTS}},
|
||
"ensure_fields": {"required": ["key"], "optional": FIELD_PARAMS + ["labels_add", "labels_remove", "story_points"],
|
||
"keys": ["key", "epic_key"],
|
||
"hints": {**FIELD_HINTS, "labels_add": "список", "labels_remove": "список",
|
||
"story_points": "число ≥ 0 или null"},
|
||
"note": "хотя бы одно поле"},
|
||
"ensure_comment": {"required": ["key", "marker", "body"], "optional": [], "keys": ["key"],
|
||
"hints": {"marker": "метка в начале комментария"}},
|
||
"ensure_status": {"required": ["key", "target"], "optional": [], "keys": ["key"],
|
||
"hints": {"target": "статус или список-путь"}},
|
||
"ensure_link": {"required": ["source", "target", "link_type"], "optional": ["replace"], "keys": ["source", "target"],
|
||
"hints": {"link_type": "логическое имя из link_types", "replace": "true|false"}},
|
||
"ensure_sprint": {"required": ["key", "sprint"], "optional": [], "keys": ["key"],
|
||
"hints": {"sprint": "имя спринта или backlog"}},
|
||
"ensure_attachment": {"required": ["key", "file", "filename"], "optional": [], "keys": ["key"],
|
||
"hints": {"file": "sha256 из POST /files; в MCP — path к локальному файлу",
|
||
"filename": "имя вложения"}},
|
||
"fetch_attachment": {"required": ["key", "attachment_id"], "optional": [], "keys": ["key"],
|
||
"hints": {"attachment_id": "id вложения из issue (цифры)"}},
|
||
"refresh": {"required": [], "optional": ["keys", "jql"], "keys": ["keys"], "hints": {"keys": "список"},
|
||
"note": "keys или jql"},
|
||
"ensure_sprint_meta": {"required": ["sprint"], "optional": ["rename_to", "start_date", "end_date", "goal"], "keys": [],
|
||
"hints": {"sprint": "имя спринта", "rename_to": "новое имя", "start_date": "YYYY-MM-DD",
|
||
"end_date": "YYYY-MM-DD", "goal": "цель спринта"}},
|
||
"ensure_sprint_state": {"required": ["sprint", "state"], "optional": ["move_open_to"], "keys": [],
|
||
"hints": {"state": "active|closed",
|
||
"move_open_to": "имя спринта или backlog; обязателен при closed"}},
|
||
"ensure_rank": {"required": ["key"], "optional": ["before", "after"], "keys": ["key", "before", "after"],
|
||
"hints": {"before": "ключ задачи, выше которой", "after": "ключ задачи, ниже которой"},
|
||
"note": "ровно одно из before, after"},
|
||
}
|
||
WRITE_TYPES = [t for t in SPEC if t not in ("refresh", "fetch_attachment")]
|
||
LIST_PARAMS = {"labels", "labels_add", "labels_remove", "keys"}
|
||
BOOL_PARAMS = {"replace"}
|
||
NUMBER_PARAMS = {"story_points"}
|
||
SPRINT_TYPES = ("ensure_sprint_meta", "ensure_sprint_state") # только в пакете проекта (check_batch)
|
||
DATE_RE = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}$")
|
||
|
||
|
||
def is_temp(key) -> bool:
|
||
return isinstance(key, str) and bool(TEMP_KEY.match(key))
|
||
|
||
|
||
def referenced_keys(cmd) -> list[str]:
|
||
out = []
|
||
for field in SPEC[cmd["type"]]["keys"]:
|
||
value = cmd["params"].get(field)
|
||
if isinstance(value, list):
|
||
out += [v for v in value if isinstance(v, str) and v]
|
||
elif isinstance(value, str) and value:
|
||
out.append(value)
|
||
return out
|
||
|
||
|
||
def temp_keys_used(cmd) -> list[str]:
|
||
"""Чужие временные ключи (свой temp_key у ensure_issue — не зависимость)."""
|
||
own = cmd["params"].get("temp_key") if cmd["type"] == "ensure_issue" else None
|
||
return [k for k in referenced_keys(cmd) if is_temp(k) and k != own]
|
||
|
||
|
||
def _finite(value) -> bool:
|
||
try:
|
||
return math.isfinite(value)
|
||
except OverflowError: # целое 10**400 не приводится к float
|
||
return False
|
||
|
||
|
||
def _type_error(ctype, name, value) -> str | None:
|
||
"""Проверка типа значения параметра (M-3 #12): иначе .strip() и подстановка ключей падают с 500.
|
||
null допустим везде — обязательность проверяется отдельно."""
|
||
if value is None:
|
||
return None
|
||
if name in LIST_PARAMS:
|
||
return None if _is_str_list(value) else f"{name} — список строк"
|
||
if name in BOOL_PARAMS:
|
||
return None if isinstance(value, bool) else f"{name} — true или false"
|
||
if name in NUMBER_PARAMS: # конечное число ≥ 0; bool, NaN и Infinity — не число (R-15)
|
||
ok = isinstance(value, (int, float)) and not isinstance(value, bool) and _finite(value) and value >= 0
|
||
return None if ok else f"{name} — число ≥ 0 или null"
|
||
if ctype == "ensure_status" and name == "target":
|
||
return None if isinstance(value, str) or _is_str_list(value) else f"{name} — строка или список строк"
|
||
return None if isinstance(value, str) else f"{name} — строка"
|
||
|
||
|
||
def _date(value) -> date | None:
|
||
"""Строка ГГГГ-ММ-ДД, которая задаёт настоящую дату; иначе None."""
|
||
if not isinstance(value, str) or not DATE_RE.match(value):
|
||
return None
|
||
try:
|
||
return date.fromisoformat(value)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _rule_error(ctype, p) -> str | None:
|
||
"""Смысловые проверки постановки (R-15): типы уже проверены в _type_error."""
|
||
if ctype == "ensure_sprint_meta":
|
||
if not p["sprint"].strip():
|
||
return "sprint — непустое имя"
|
||
if p.get("rename_to") is not None and not p["rename_to"].strip():
|
||
return "rename_to — непустое имя"
|
||
for name in ("start_date", "end_date"):
|
||
if p.get(name) is not None and _date(p[name]) is None:
|
||
return f"{name} — дата ГГГГ-ММ-ДД"
|
||
start, end = _date(p.get("start_date")), _date(p.get("end_date"))
|
||
if start and end and start > end:
|
||
return "start_date позже end_date"
|
||
if ctype == "ensure_sprint_state":
|
||
if not p["sprint"].strip():
|
||
return "sprint — непустое имя"
|
||
if p["state"] not in ("active", "closed"):
|
||
return "state — active или closed"
|
||
move = (p.get("move_open_to") or "").strip()
|
||
if p["state"] == "closed":
|
||
if not move:
|
||
return "move_open_to обязателен при state = closed"
|
||
if move == p["sprint"].strip():
|
||
return "move_open_to — другой спринт"
|
||
elif move:
|
||
return "move_open_to — только при state = closed"
|
||
if ctype == "ensure_rank":
|
||
before = (p.get("before") or "").strip() or None # пустая и пробельная строка — «не задано» (R-31)
|
||
after = (p.get("after") or "").strip() or None
|
||
if bool(before) == bool(after):
|
||
return "ровно одно из before, after"
|
||
if before == p["key"].strip():
|
||
return "before — другая задача"
|
||
if after == p["key"].strip():
|
||
return "after — другая задача"
|
||
return None
|
||
|
||
|
||
def _is_str_list(value) -> bool:
|
||
return isinstance(value, list) and all(isinstance(v, str) for v in value)
|
||
|
||
|
||
def _names(spec, names) -> str:
|
||
return ", ".join(f"{n} ({spec['hints'][n]})" if n in spec.get("hints", {}) else n for n in names)
|
||
|
||
|
||
def describe() -> str:
|
||
"""Параметры всех типов команд, строка на тип (необязательные — в [ ]): описание MCP batch_create из SPEC (#44)."""
|
||
lines = []
|
||
for ctype, spec in SPEC.items():
|
||
line = f"- {ctype}: {_names(spec, spec['required']) or '—'}"
|
||
if spec["optional"]:
|
||
line += f"; [{_names(spec, spec['optional'])}]"
|
||
if spec.get("note"):
|
||
line += f" — {spec['note']}"
|
||
lines.append(line)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _expected(spec) -> str:
|
||
"""Хвост ошибки «нет …» / «лишние параметры …»: какие параметры тип ожидает (#44)."""
|
||
parts = [f"ожидаются {', '.join(spec['required'])}"] if spec["required"] else []
|
||
if spec["optional"]:
|
||
parts.append(f"необязательные {', '.join(spec['optional'])}")
|
||
return "; ".join(parts)
|
||
|
||
|
||
def validate(commands) -> list[dict]:
|
||
if not isinstance(commands, list) or not commands:
|
||
raise CommandError("пакет без команд")
|
||
defined, out = set(), []
|
||
for i, cmd in enumerate(commands, 1):
|
||
if not isinstance(cmd, dict) or "type" not in cmd:
|
||
raise CommandError(f"команда {i}: нет type")
|
||
ctype, params = cmd["type"], cmd.get("params") or {}
|
||
if ctype not in SPEC:
|
||
raise CommandError(f"команда {i}: неизвестный тип «{ctype}»")
|
||
if not isinstance(params, dict):
|
||
raise CommandError(f"команда {i}: params не объект")
|
||
spec = SPEC[ctype]
|
||
missing = [f for f in spec["required"] if params.get(f) in (None, "", [])]
|
||
if missing:
|
||
raise CommandError(f"команда {i} ({ctype}): нет {', '.join(missing)}; {_expected(spec)}")
|
||
unknown = set(params) - set(spec["required"]) - set(spec["optional"])
|
||
if unknown:
|
||
raise CommandError(f"команда {i} ({ctype}): лишние параметры {', '.join(sorted(unknown))}; "
|
||
f"{_expected(spec)}")
|
||
bad = [e for e in (_type_error(ctype, n, v) for n, v in params.items()) if e]
|
||
if bad:
|
||
raise CommandError(f"команда {i} ({ctype}): " + "; ".join(bad))
|
||
rule = _rule_error(ctype, params)
|
||
if rule:
|
||
raise CommandError(f"команда {i} ({ctype}): {rule}")
|
||
if ctype == "ensure_fields" and not any(f in params for f in spec["optional"]):
|
||
raise CommandError(f"команда {i}: ensure_fields без полей")
|
||
if ctype == "refresh" and not (params.get("keys") or params.get("jql")):
|
||
raise CommandError(f"команда {i}: refresh без keys и jql")
|
||
if ctype == "ensure_issue":
|
||
if not is_temp(params["temp_key"]):
|
||
raise CommandError(f"команда {i}: temp_key должен быть new:<n>")
|
||
if params["temp_key"] in defined:
|
||
raise CommandError(f"команда {i}: temp_key {params['temp_key']} уже объявлен")
|
||
if params["issue_type"] not in ("task", "epic", "bug"):
|
||
raise CommandError(f"команда {i}: issue_type — task, epic или bug")
|
||
for key in temp_keys_used({"type": ctype, "params": params}):
|
||
if key not in defined:
|
||
raise CommandError(f"команда {i}: временный ключ {key} не объявлен ранее в пакете командой ensure_issue")
|
||
if ctype == "ensure_issue":
|
||
defined.add(params["temp_key"])
|
||
out.append({"type": ctype, "params": params})
|
||
summaries = [c["params"]["summary"].strip() for c in out if c["type"] == "ensure_issue"]
|
||
repeated = sorted({s for s in summaries if summaries.count(s) > 1})
|
||
if repeated:
|
||
raise CommandError("повторяется тема ensure_issue: " + ", ".join(f"«{s}»" for s in repeated)
|
||
+ " — в Jira получится одна задача на два временных ключа")
|
||
return out
|
||
|
||
|
||
def substitute(cmd_type, params, mapping: dict) -> dict:
|
||
"""Подставить настоящие ключи вместо временных (temp_key самой ensure_issue не трогается)."""
|
||
out = dict(params)
|
||
for field in SPEC[cmd_type]["keys"]:
|
||
if field == "temp_key":
|
||
continue
|
||
value = out.get(field)
|
||
if isinstance(value, list):
|
||
out[field] = [mapping.get(v, v) for v in value]
|
||
elif isinstance(value, str):
|
||
out[field] = mapping.get(value, value)
|
||
return out
|
||
|
||
|
||
def _norm_text(value):
|
||
"""Значение поля для сравнения с зеркалом: переводы строки CRLF и LF не различаются (описание, правленное в UI Jira,
|
||
приходит с CRLF, O-9 #16); null и пустая строка равны (решение 11)."""
|
||
if isinstance(value, str):
|
||
value = value.replace("\r\n", "\n")
|
||
return value or None
|
||
|
||
|
||
def check_confirmed(conn, cmd, link_types: dict) -> tuple[bool, dict | None]:
|
||
"""Виден ли желаемый результат в зеркале. ensure_issue, ensure_comment, ensure_attachment и refresh по состоянию не
|
||
подтверждаются: потерянный отчёт ensure_issue закрывается довыдачей — агент находит свою задачу поиском дубля
|
||
(bridge-reliability, решение 4); совпадение темы с задачей неизвестного автора не доказательство; вложений в
|
||
зеркале нет (bug-and-attachments); команды спринта и ранга в зеркале не видны (design #18).
|
||
story_points сравнивается точно: null равен только null, 0 не равен null (R-16)."""
|
||
ctype, p = cmd["type"], cmd["params"]
|
||
if ctype in ("ensure_issue", "ensure_comment", "ensure_attachment", "fetch_attachment", "refresh", "ensure_rank",
|
||
*SPRINT_TYPES):
|
||
return False, None
|
||
issue = mirror.get_issue(conn, p.get("key") or p.get("source"))
|
||
if issue is None:
|
||
return False, None
|
||
if ctype == "ensure_fields":
|
||
if "story_points" in p and issue.get("story_points") != p["story_points"]:
|
||
return False, None
|
||
for field in FIELD_PARAMS:
|
||
if field in p and _norm_text(issue.get(field)) != _norm_text(p[field]):
|
||
return False, None
|
||
if any(l not in issue["labels"] for l in p.get("labels_add", [])):
|
||
return False, None
|
||
if any(l in issue["labels"] for l in p.get("labels_remove", [])):
|
||
return False, None
|
||
return True, None
|
||
if ctype == "ensure_status":
|
||
target = p["target"]
|
||
final = target[-1] if isinstance(target, list) else target
|
||
return issue["status"] == final, None
|
||
if ctype == "ensure_link":
|
||
# связь есть И конфликтов нет: другой тип между теми же задачами или обратное направление любого типа
|
||
jira_type = link_types.get(p["link_type"], p["link_type"])
|
||
all_types = set(link_types.values())
|
||
present = any(l["target_key"] == p["target"] and l["link_type"] == jira_type for l in issue["links_out"])
|
||
other_type = any(l["target_key"] == p["target"] and l["link_type"] in all_types - {jira_type}
|
||
for l in issue["links_out"])
|
||
reverse = any(l["source_key"] == p["target"] and l["link_type"] in all_types for l in issue["links_in"])
|
||
return present and not other_type and not reverse, None
|
||
if ctype == "ensure_sprint":
|
||
want = None if p["sprint"] == "backlog" else p["sprint"]
|
||
return (issue.get("sprint") or None) == want, None
|
||
return False, None
|