Aislo/Z01_MasterData/Z01_MasterData_Repository_WorkItems.py

"""Z01 공종 저장소 — `master_work_item`·`master_work_item_row`·`master_revision` 읽기·쓰기(SQL 만).

무엇을 쓸지는 `Z01_MasterData_WorkItems` 순수 함수(`plan_rows`·`reset_plan`)가 정함 ·
여기는 받은 대로 씀.
표 모양 = `db_management/020_master_cost.sql`(PLAN 1-0 계약). 트랜잭션은 부르는 쪽(라우터)이 엶.
"""

from __future__ import annotations

import json
from typing import Any

import aiomysql

HEAD_COLUMNS = (
    "id, axis, work_item_key, code, number, name, path_name, parent_key, parent_mode, "
    "basis_quantity, basis_unit, status, notes, source, seed, edition, sort_order, "
    "updated_by, updated_at"
)
ROW_COLUMNS = (
    "id, variant, resource_kind, resource_key, resource_name, resource_spec, amount, "
    "amount_unit, alternative_amount, group_ratio_pct, raw_cell, pum_table_id, seed, "
    "sort_order, is_deleted"
)
_JSON = ("notes", "source", "seed")


def _parsed(row: dict[str, Any]) -> dict[str, Any]:
    """JSON 칸은 MariaDB 가 글자로 돌려줌 — 풀어서 씀."""
    for key in _JSON:
        if isinstance(row.get(key), (str, bytes)):
            row[key] = json.loads(row[key])
    return row


def _like(q: str) -> str:
    return "%" + q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%"


async def list_items(
    connection: aiomysql.Connection, *, axis: str, status: str, q: str, page: int, size: int
) -> dict[str, Any]:
    """목록 한 쪽 + 상태별 개수(상태 거르기 전 · 탭 숫자용). total 은 개수에서 셈(질의 하나 덜)."""
    where, params = ["1"], []
    if axis:
        where.append("w.axis = %s")
        params.append(axis)
    if q:
        where.append("(w.work_item_key LIKE %s OR w.number LIKE %s OR w.path_name LIKE %s)")
        params += [_like(q)] * 3
    picked, picked_params = list(where), list(params)
    if status:
        picked.append("w.status = %s")
        picked_params.append(status)
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            "SELECT w.status, COUNT(*) AS n FROM master_work_item w "
            f"WHERE {' AND '.join(where)} GROUP BY w.status",
            params,
        )
        counts = {row["status"]: int(row["n"]) for row in await cursor.fetchall()}
        await cursor.execute(
            "SELECT w.work_item_key, w.axis, w.number, w.name, w.path_name, w.status, "
            "(SELECT COUNT(*) FROM master_work_item_row r "
            " WHERE r.work_item_id = w.id AND r.is_deleted = 0) AS row_count "
            f"FROM master_work_item w WHERE {' AND '.join(picked)} "
            "ORDER BY w.axis, w.sort_order, w.id LIMIT %s OFFSET %s",
            [*picked_params, size, (page - 1) * size],
        )
        items = list(await cursor.fetchall())
    total = counts.get(status, 0) if status else sum(counts.values())
    return {"items": items, "total": total, "counts": counts}


async def fetch_item(
    connection: aiomysql.Connection, key: str, *, lock: bool = False
) -> dict[str, Any] | None:
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            f"SELECT {HEAD_COLUMNS} FROM master_work_item WHERE work_item_key = %s"
            + (" FOR UPDATE" if lock else ""),
            (key,),
        )
        row = await cursor.fetchone()
    return _parsed(row) if row else None


async def fetch_rows(
    connection: aiomysql.Connection, item_id: int, *, with_deleted: bool = False
) -> list[dict[str, Any]]:
    """구성 줄 — 기본은 살아 있는 줄만 · [초기값으로] 는 숨긴 줄까지."""
    async with connection.cursor(aiomysql.DictCursor) as cursor:
        await cursor.execute(
            f"SELECT {ROW_COLUMNS} FROM master_work_item_row WHERE work_item_id = %s"
            + ("" if with_deleted else " AND is_deleted = 0")
            + " ORDER BY sort_order, id",
            (item_id,),
        )
        return [_parsed(row) for row in await cursor.fetchall()]


async def revision(connection: aiomysql.Connection, scope: str, *, lock: bool = False) -> int:
    """지금 revision(없으면 0). lock 이면 줄을 만들어 잡음 — 같은 공종 저장이 줄 서게."""
    async with connection.cursor() as cursor:
        if lock:
            await cursor.execute(
                "INSERT IGNORE INTO master_revision (scope, revision) VALUES (%s, 0)", (scope,)
            )
        await cursor.execute(
            "SELECT revision FROM master_revision WHERE scope = %s"
            + (" FOR UPDATE" if lock else ""),
            (scope,),
        )
        row = await cursor.fetchone()
    return int(row[0]) if row else 0


async def bump_revision(connection: aiomysql.Connection, scope: str) -> None:
    async with connection.cursor() as cursor:
        await cursor.execute(
            "INSERT INTO master_revision (scope, revision) VALUES (%s, 1) "
            "ON DUPLICATE KEY UPDATE revision = revision + 1",
            (scope,),
        )


async def known_resource_keys(connection: aiomysql.Connection, keys: set[str]) -> set[str]:
    """그중 기초단가(`master_base_price.row_key`)에 실재하는 열쇠."""
    if not keys:
        return set()
    marks = ", ".join(["%s"] * len(keys))
    async with connection.cursor() as cursor:
        await cursor.execute(
            "SELECT DISTINCT row_key FROM master_base_price "
            f"WHERE is_deleted = 0 AND row_key IN ({marks})",
            list(keys),
        )
        return {row[0] for row in await cursor.fetchall()}


async def write(
    connection: aiomysql.Connection,
    item_id: int,
    head: dict[str, Any],
    plan: dict[str, list[Any]],
    by: Any,
) -> None:
    """머리 칸 + 줄 쓰기(`plan_rows`·`reset_plan` 모양).

    칸 이름은 순수 함수의 고정 목록에서만 옴(요청 글자가 SQL 에 안 들어감).
    """
    async with connection.cursor() as cursor:
        sets = "".join(f"{field} = %s, " for field in head)
        await cursor.execute(
            f"UPDATE master_work_item SET {sets}updated_by = %s, updated_at = NOW() WHERE id = %s",
            [*head.values(), by, item_id],
        )
        for row in plan["updates"]:  # 줄마다 칸이 다를 수 있음(seed 에 있는 칸만) — 한 공종 수십 줄
            values = {k: v for k, v in row.items() if k != "id"}
            sets = "".join(f"{field} = %s, " for field in values)
            await cursor.execute(
                f"UPDATE master_work_item_row SET {sets}is_deleted = 0 "
                "WHERE id = %s AND work_item_id = %s",
                [*values.values(), row["id"], item_id],
            )
        if plan["inserts"]:
            fields = list(plan["inserts"][0])
            await cursor.executemany(
                f"INSERT INTO master_work_item_row (work_item_id, {', '.join(fields)}) "
                f"VALUES (%s, {', '.join(['%s'] * len(fields))})",
                [[item_id, *(row[f] for f in fields)] for row in plan["inserts"]],
            )
        for ids, sql in (
            (plan["hide"], "UPDATE master_work_item_row SET is_deleted = 1"),
            (plan["drop"], "DELETE FROM master_work_item_row"),
        ):
            if ids:
                await cursor.execute(
                    f"{sql} WHERE work_item_id = %s AND id IN ({', '.join(['%s'] * len(ids))})",
                    [item_id, *ids],
                )