redmine/redmine_ticket_meta.py

#!/usr/bin/env python3
"""Resolve human-friendly Redmine names to IDs for set-fields / mutations.

Caches project metadata under <cache-dir>/meta/<project>/ so agents can use
--status Closed instead of --status-id 5 without hammering the API every run.
"""

from __future__ import annotations

import json
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable


def now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def meta_dir(cache_dir: Path, project: str) -> Path:
    path = Path(cache_dir) / "meta" / project
    path.mkdir(parents=True, exist_ok=True)
    return path


def _load_json(path: Path) -> dict[str, Any] | None:
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    return data if isinstance(data, dict) else None


def _save_json(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    tmp.replace(path)


def _fresh(path: Path, max_age_seconds: int) -> bool:
    if not path.is_file():
        return False
    age = time.time() - path.stat().st_mtime
    return age <= max_age_seconds


def normalize_name(value: str) -> str:
    return " ".join(str(value).strip().lower().split())


def fetch_issue_statuses(client: Any) -> list[dict[str, Any]]:
    data = client.request("GET", "/issue_statuses.json")
    statuses = data.get("issue_statuses") if isinstance(data, dict) else None
    return list(statuses) if isinstance(statuses, list) else []


def fetch_issue_priorities(client: Any) -> list[dict[str, Any]]:
    data = client.request("GET", "/enumerations/issue_priorities.json")
    priorities = data.get("issue_priorities") if isinstance(data, dict) else None
    return list(priorities) if isinstance(priorities, list) else []


def fetch_project_categories(client: Any, project: str) -> list[dict[str, Any]]:
    data = client.request("GET", f"/projects/{project}/issue_categories.json")
    cats = data.get("issue_categories") if isinstance(data, dict) else None
    return list(cats) if isinstance(cats, list) else []


def fetch_project_memberships(client: Any, project: str) -> list[dict[str, Any]]:
    """Memberships include nested user objects for assignee resolution."""
    members: list[dict[str, Any]] = []
    offset = 0
    total = None
    while total is None or offset < total:
        data = client.request(
            "GET",
            f"/projects/{project}/memberships.json",
            {"limit": 100, "offset": offset},
        )
        page = data.get("memberships") if isinstance(data, dict) else []
        if not isinstance(page, list):
            page = []
        members.extend(page)
        total = int(data.get("total_count", len(members))) if isinstance(data, dict) else len(members)
        if not page:
            break
        offset += len(page)
    return members


def users_from_memberships(memberships: list[dict[str, Any]]) -> list[dict[str, Any]]:
    users: list[dict[str, Any]] = []
    seen: set[int] = set()
    for m in memberships:
        user = m.get("user") if isinstance(m, dict) else None
        if not isinstance(user, dict) or user.get("id") is None:
            continue
        uid = int(user["id"])
        if uid in seen:
            continue
        seen.add(uid)
        users.append(
            {
                "id": uid,
                "name": user.get("name") or f"{user.get('firstname', '')} {user.get('lastname', '')}".strip(),
                "login": user.get("login"),
            }
        )
    return users


def refresh_meta_cache(
    client: Any,
    cache_dir: Path,
    project: str,
    *,
    force: bool = False,
    max_age_seconds: int = 6 * 3600,
) -> dict[str, Path]:
    root = meta_dir(cache_dir, project)
    paths = {
        "statuses": root / "issue_statuses.json",
        "priorities": root / "issue_priorities.json",
        "categories": root / "issue_categories.json",
        "users": root / "project_users.json",
    }
    if not force and all(_fresh(p, max_age_seconds) for p in paths.values()):
        return paths

    statuses = fetch_issue_statuses(client)
    priorities = fetch_issue_priorities(client)
    categories = fetch_project_categories(client, project)
    memberships = fetch_project_memberships(client, project)
    users = users_from_memberships(memberships)

    _save_json(paths["statuses"], {"fetched_at": now_iso(), "issue_statuses": statuses})
    _save_json(paths["priorities"], {"fetched_at": now_iso(), "issue_priorities": priorities})
    _save_json(paths["categories"], {"fetched_at": now_iso(), "issue_categories": categories})
    _save_json(paths["users"], {"fetched_at": now_iso(), "users": users})
    return paths


def _items_from_file(path: Path, key: str) -> list[dict[str, Any]]:
    data = _load_json(path) or {}
    items = data.get(key)
    return list(items) if isinstance(items, list) else []


def resolve_id_by_name(
    items: list[dict[str, Any]],
    name: str,
    *,
    id_key: str = "id",
    name_keys: tuple[str, ...] = ("name",),
) -> int:
    target = normalize_name(name)
    if target.isdigit():
        return int(target)
    matches: list[dict[str, Any]] = []
    for item in items:
        if not isinstance(item, dict):
            continue
        candidates = []
        for nk in name_keys:
            val = item.get(nk)
            if val is not None:
                candidates.append(normalize_name(str(val)))
        # also full first+last style
        if item.get("firstname") or item.get("lastname"):
            candidates.append(
                normalize_name(f"{item.get('firstname', '')} {item.get('lastname', '')}")
            )
        if any(c == target for c in candidates if c):
            matches.append(item)
            continue
        # substring match as fallback
        if any(target in c for c in candidates if c):
            matches.append(item)
    if not matches:
        available = []
        for item in items[:30]:
            if isinstance(item, dict):
                available.append(str(item.get("name") or item.get("login") or item.get(id_key)))
        raise LookupError(
            f"No match for {name!r}. Examples: {', '.join(available) or '(empty meta cache)'}"
        )
    if len(matches) > 1:
        # prefer exact name over substring
        exact = [
            m
            for m in matches
            if any(normalize_name(str(m.get(k) or "")) == target for k in name_keys)
            or normalize_name(f"{m.get('firstname', '')} {m.get('lastname', '')}") == target
        ]
        if len(exact) == 1:
            return int(exact[0][id_key])
        labels = [str(m.get("name") or m.get("login") or m.get(id_key)) for m in matches[:8]]
        raise LookupError(f"Ambiguous match for {name!r}: {', '.join(labels)}")
    return int(matches[0][id_key])


def load_meta_lists(cache_dir: Path, project: str) -> dict[str, list[dict[str, Any]]]:
    root = meta_dir(cache_dir, project)
    return {
        "statuses": _items_from_file(root / "issue_statuses.json", "issue_statuses"),
        "priorities": _items_from_file(root / "issue_priorities.json", "issue_priorities"),
        "categories": _items_from_file(root / "issue_categories.json", "issue_categories"),
        "users": _items_from_file(root / "project_users.json", "users"),
    }


def resolve_field_names(
    cache_dir: Path,
    project: str,
    *,
    status_name: str | None = None,
    assignee_name: str | None = None,
    priority_name: str | None = None,
    category_name: str | None = None,
    client: Any | None = None,
    refresh: bool = False,
) -> dict[str, int]:
    """Return numeric id fields from optional name strings.

    If meta cache is empty and client is provided, refresh automatically.
    """
    lists = load_meta_lists(cache_dir, project)
    need_refresh = refresh or (
        client is not None
        and (
            (status_name and not lists["statuses"])
            or (assignee_name and not lists["users"])
            or (priority_name and not lists["priorities"])
            or (category_name and not lists["categories"])
        )
    )
    if need_refresh and client is not None:
        refresh_meta_cache(client, cache_dir, project, force=True)
        lists = load_meta_lists(cache_dir, project)

    out: dict[str, int] = {}
    if status_name:
        out["status_id"] = resolve_id_by_name(lists["statuses"], status_name)
    if priority_name:
        out["priority_id"] = resolve_id_by_name(lists["priorities"], priority_name)
    if category_name:
        if normalize_name(category_name) in {"", "none", "clear", "null"}:
            out["category_id"] = 0
        else:
            out["category_id"] = resolve_id_by_name(lists["categories"], category_name)
    if assignee_name:
        if normalize_name(assignee_name) in {"", "none", "unassigned", "nobody", "null", "0"}:
            out["assigned_to_id"] = 0
        else:
            out["assigned_to_id"] = resolve_id_by_name(
                lists["users"],
                assignee_name,
                name_keys=("name", "login"),
            )
    return out