sdlc-hub/src/jira_hub/projects.py

"""Проекты-метки моста: таблица project, проверка метки, базовая JQL из меток, доски, миграция v3 → v4."""
import logging
import re

from . import db, settings

log = logging.getLogger(__name__)

LEGACY_LABEL = "AWPCA"
LEGACY_NAME = "АРМ КА"
MAX_LEN = 100
BOARD_TYPES = ("scrum", "kanban")
_FORBIDDEN = ',"\'()'
_FIELDS = "label, name, board_id, board_type, created_at"


class ProjectError(ValueError):
    """Ошибка данных проекта; field — label | name | board_id."""

    def __init__(self, field: str, reason: str):
        super().__init__(f"{field}: {reason}")
        self.field = field


class ProjectExists(ProjectError):
    def __init__(self, label: str):
        super().__init__("label", f"проект с такой меткой уже есть: {label}")


def _label_error(label) -> str | None:
    if not isinstance(label, str) or not label:
        return "ожидается непустая строка"
    if len(label) > MAX_LEN:
        return f"длиннее {MAX_LEN} символов"
    if any(c.isspace() or c in _FORBIDDEN for c in label):
        return "пробелы и символы , \" ' ( ) недопустимы"
    return None


def validate(label, name, board_id) -> tuple[str, str, int | None]:
    """Проверить поля проекта; вернуть (метка, название без пробелов по краям, доска)."""
    error = _label_error(label)
    if error:
        raise ProjectError("label", error)
    if not isinstance(name, str) or not name.strip():
        raise ProjectError("name", "ожидается непустая строка")
    if len(name.strip()) > MAX_LEN:
        raise ProjectError("name", f"длиннее {MAX_LEN} символов")
    if board_id is not None and (not isinstance(board_id, int) or isinstance(board_id, bool) or board_id <= 0):
        raise ProjectError("board_id", "ожидается целое > 0 или null")
    return label, name.strip(), board_id


def get(conn, label: str) -> dict | None:
    row = conn.execute(f"SELECT {_FIELDS} FROM project WHERE label = ?", (label,)).fetchone()
    return None if row is None else dict(row)


def create(conn, label: str, name: str, board_id: int | None = None) -> dict:
    label, name, board_id = validate(label, name, board_id)
    with db.transaction(conn):
        if label.casefold() in {x.casefold() for x in labels(conn)}:
            raise ProjectExists(label)
        conn.execute("INSERT INTO project(label, name, board_id, board_type, created_at) VALUES (?, ?, ?, NULL, ?)",
                     (label, name, board_id, db.now()))
    return get(conn, label)


def list_projects(conn) -> list[dict]:
    rows = conn.execute(
        f"SELECT {_FIELDS}, (SELECT COUNT(*) FROM issue i WHERE EXISTS "
        "(SELECT 1 FROM json_each(i.labels) WHERE value = project.label)) AS issues_count "
        "FROM project ORDER BY created_at, label").fetchall()
    return [dict(r) for r in rows]


def light_list(conn) -> list[dict]:
    """Проекты без issues_count (label, board_id, board_type), порядок как у list_projects."""
    return [dict(r) for r in conn.execute("SELECT label, board_id, board_type FROM project ORDER BY created_at, label")]


def labels(conn) -> list[str]:
    return [r["label"] for r in conn.execute("SELECT label FROM project ORDER BY created_at, label")]


def base_jql(conn) -> str | None:
    """Общая JQL обновления: все метки проектов; без проектов — None."""
    found = labels(conn)
    if not found:
        return None
    key = settings.get(conn, "project_key") or "AURORA"
    return f"project = {key} AND labels in ({', '.join(found)})"


def boards(conn) -> list[dict]:
    """Доски проектов без повторов (порядок первого проекта). type — None, если тип неизвестен хотя бы у одного
    проекта доски (агент тогда запросит его, record_board_types проставит всем); иначе первый известный."""
    out: dict[int, str | None] = {}
    unknown: set[int] = set()
    for r in conn.execute("SELECT board_id, board_type FROM project WHERE board_id IS NOT NULL ORDER BY created_at, label"):
        out.setdefault(r["board_id"], r["board_type"])
        if r["board_type"] is None:
            unknown.add(r["board_id"])
    return [{"id": i, "type": None if i in unknown else t} for i, t in out.items()]


def record_board_types(conn, found) -> int:
    """Записать типы досок {"<id>": "scrum" | "kanban"} всем проектам с этой доской; прочее пропустить."""
    if not isinstance(found, dict):
        return 0
    updated = 0
    for raw, kind in found.items():
        if kind not in BOARD_TYPES:
            continue
        try:
            board = int(raw)
        except (TypeError, ValueError):
            continue
        updated += conn.execute("UPDATE project SET board_type = ? WHERE board_id = ?", (kind, board)).rowcount
    return updated


_LEGACY = re.compile(r"\blabels\s*(?:=\s*(?P<one>\S+)|in\s*\((?P<many>[^()]*)\))\s*$", re.IGNORECASE)


def parse_legacy_labels(jql) -> list[str] | None:
    """Метки из прежней base_jql (`labels = X` / `labels in (X, Y)` в конце); не разобрана — None."""
    m = _LEGACY.search(jql) if isinstance(jql, str) else None
    if not m:
        return None
    out: list[str] = []
    for raw in ([m["one"]] if m["one"] is not None else m["many"].split(",")):
        label = raw.strip().strip('"')
        if _label_error(label):
            return None
        if label.casefold() not in {x.casefold() for x in out}:
            out.append(label)
    return out or None


def ensure_legacy(conn, board_id: int | None) -> bool:
    """Завести AWPCA «АРМ КА» с доской, если проекта с меткой awpca (без учёта регистра) нет; True — заведён.
    Неверная доска из конфига — проект без доски."""
    if LEGACY_LABEL.casefold() in {x.casefold() for x in labels(conn)}:
        return False
    try:
        validate(LEGACY_LABEL, LEGACY_NAME, board_id)
    except ProjectError:
        board_id = None
    create(conn, LEGACY_LABEL, LEGACY_NAME, board_id)
    return True


def migrate_v3(conn) -> None:
    """Миграция v3 → v4: проекты из прежней base_jql, доска — единственному проекту, очередь и журнал — на новую JQL."""
    old = settings.get(conn, "base_jql")
    found = parse_legacy_labels(old)
    if found is None:
        log.warning("миграция v4: base_jql не разобрана (%r) — проектов нет, заведите их POST /projects", old)
        return
    board = settings.get(conn, "board_id")
    board = board if isinstance(board, int) and not isinstance(board, bool) and board > 0 else None
    if board is not None and len(found) > 1:
        log.warning("миграция v4: доска %s не отдана ни одному проекту — меток несколько: %s", board, ", ".join(found))
        board = None
    for label in found:
        create(conn, label, LEGACY_NAME if label == LEGACY_LABEL else label, board)
    new = base_jql(conn)
    conn.execute("UPDATE sync_run SET query = ? WHERE query = ?", (new, old))
    conn.execute("UPDATE command SET params = json_set(params, '$.jql', ?) WHERE type = 'refresh' "
                 "AND status IN ('queued', 'claimed') AND json_extract(params, '$.jql') = ?", (new, old))