"""Контракт команд: каждая — желаемый результат на одной задаче Jira (specs/commands/spec.md). Здесь валидация, работа с ключами (в том числе временными new:), оценка длительности и проверка «результат уже виден в зеркале» для подтверждения по состоянию.""" import re from . import mirror TEMP_KEY = re.compile(r"^new:\d+$") FIELD_PARAMS = ["summary", "description", "priority", "duedate", "planned_start", "planned_end", "epic_key", "assignee"] 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"]}, "ensure_fields": {"required": ["key"], "optional": FIELD_PARAMS + ["labels_add", "labels_remove"], "keys": ["key", "epic_key"]}, "ensure_comment": {"required": ["key", "marker", "body"], "optional": [], "keys": ["key"]}, "ensure_status": {"required": ["key", "target"], "optional": [], "keys": ["key"]}, "ensure_link": {"required": ["source", "target", "link_type"], "optional": ["replace"], "keys": ["source", "target"]}, "ensure_sprint": {"required": ["key", "sprint"], "optional": [], "keys": ["key"]}, "refresh": {"required": [], "optional": ["keys", "jql"], "keys": ["keys"]}, } WRITE_TYPES = [t for t in SPEC if t != "refresh"] LIST_PARAMS = {"labels", "labels_add", "labels_remove", "keys"} BOOL_PARAMS = {"replace"} class CommandError(ValueError): """Ошибка валидации пакета — наружу уходит как 422.""" 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 _type_error(ctype, name, value) -> str | None: """Проверка типа значения параметра (M-3 #12): иначе .strip() и подстановка ключей падают с 500. null допустим везде — обязательность проверяется отдельно.""" if value is None: return None if name in LIST_PARAMS: ok, want = isinstance(value, list) and all(isinstance(v, str) for v in value), "список строк" elif name in BOOL_PARAMS: ok, want = isinstance(value, bool), "true или false" elif ctype == "ensure_status" and name == "target": ok = isinstance(value, str) or (isinstance(value, list) and all(isinstance(v, str) for v in value)) want = "строка или список строк" else: ok, want = isinstance(value, str), "строка" return None if ok else f"{name} — {want}" 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)}") unknown = set(params) - set(spec["required"]) - set(spec["optional"]) if unknown: raise CommandError(f"команда {i} ({ctype}): лишние параметры {', '.join(sorted(unknown))}") 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)) 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:") if params["temp_key"] in defined: raise CommandError(f"команда {i}: temp_key {params['temp_key']} уже объявлен") if params["issue_type"] not in ("task", "epic"): raise CommandError(f"команда {i}: issue_type — task или epic") 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 estimate(types, pacing: dict) -> int: return sum(int(pacing.get(t, 0)) for t in types) def check_confirmed(conn, cmd, link_types: dict) -> tuple[bool, dict | None]: """Виден ли желаемый результат в зеркале. ensure_issue, ensure_comment и refresh по состоянию не подтверждаются: потерянный отчёт ensure_issue закрывается довыдачей — агент находит свою задачу поиском дубля (bridge-reliability, решение 4); совпадение темы с задачей неизвестного автора не доказательство.""" ctype, p = cmd["type"], cmd["params"] if ctype in ("ensure_issue", "ensure_comment", "refresh"): 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": for field in FIELD_PARAMS: if field in p and (issue.get(field) or None) != (p[field] or None): 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