"""D04 확정 · 업체 지정 DB (Raw SQL · 표는 db_management/020_project_groups.sql).""" from __future__ import annotations import json from typing import Any import aiomysql from config.config_db import get_db_pool async def get_plan_project(project_id: str) -> dict[str, Any] | None: async with ( get_db_pool().acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor, ): await cursor.execute( "SELECT id, name, region, project_kind, route_start_m FROM projects" " WHERE id = %s AND deleted_at IS NULL", (project_id,), ) return await cursor.fetchone() async def list_year_routes(project_id: str) -> list[dict[str, Any]]: """연도 구간(오름차순) — `route_json` 은 파싱해서 돌려줌(없으면 None).""" async with ( get_db_pool().acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor, ): await cursor.execute( "SELECT year, epsg, route_json, fingerprint, confirmed_at FROM plan_route_years" " WHERE project_id = %s ORDER BY year", (project_id,), ) rows = await cursor.fetchall() for row in rows: raw = row["route_json"] row["route_json"] = json.loads(raw) if isinstance(raw, str | bytes) else raw return rows async def list_assignees(project_id: str) -> list[dict[str, Any]]: async with ( get_db_pool().acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor, ): await cursor.execute( "SELECT id, year, company_id, company_name, email, notified_at, linked_project_id" " FROM plan_route_assignees WHERE project_id = %s ORDER BY year, id", (project_id,), ) return await cursor.fetchall() async def get_company_contact(company_id: int) -> dict[str, Any] | None: """등록 회사 이름 + 대표 메일(회사 마스터 사용자).""" async with ( get_db_pool().acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor, ): await cursor.execute( "SELECT c.id, c.name, u.email FROM companies c" " LEFT JOIN users u ON u.id = c.master_user_id AND u.deleted_at IS NULL" " WHERE c.id = %s AND c.deleted_at IS NULL", (company_id,), ) return await cursor.fetchone() async def apply_assignee_changes( project_id: str, delete_ids: list[int], updates: list[dict], inserts: list[dict] ) -> None: async with get_db_pool().acquire() as connection, connection.cursor() as cursor: await connection.begin() try: for assignee_id in delete_ids: await cursor.execute( "DELETE FROM plan_route_assignees WHERE id = %s AND project_id = %s", (assignee_id, project_id), ) for row in updates: await cursor.execute( "UPDATE plan_route_assignees SET company_name = %s, email = %s" " WHERE id = %s AND project_id = %s", (row["company_name"], row["email"], row["id"], project_id), ) for row in inserts: await cursor.execute( "INSERT INTO plan_route_assignees" " (project_id, year, company_id, company_name, email) VALUES (%s,%s,%s,%s,%s)", (project_id, row["year"], row["company_id"], row["company_name"], row["email"]), ) await connection.commit() except Exception: await connection.rollback() raise async def mark_confirmed(project_id: str, years: list[int]) -> None: if not years: return marks = ",".join(["%s"] * len(years)) async with get_db_pool().acquire() as connection, connection.cursor() as cursor: await cursor.execute( f"UPDATE plan_route_years SET confirmed_at = NOW() WHERE project_id = %s" f" AND year IN ({marks})", (project_id, *years), ) await connection.commit() async def mark_notified(assignee_id: int) -> None: async with get_db_pool().acquire() as connection, connection.cursor() as cursor: await cursor.execute( "UPDATE plan_route_assignees SET notified_at = NOW() WHERE id = %s", (assignee_id,) ) await connection.commit()