"""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], )