Aislo/D04_PlanConfirm/D04_PlanConfirm_Repository.py

"""D04 확정 · 업체 지정 DB (Raw SQL · 표는 db_management/020_project_groups.sql · 023).

노선 · 조각 · 연도 = 설계 내용 파일(`D03_PlanAnalysis_Store`) · 여기는 업체 줄 · 연도 상태 줄만.
업체 줄 `year` 칸은 읽지 않음(연도 = 파일 조각 연도 · 590-3 에서 지움)."""

from __future__ import annotations

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()


def line_names(lines: list[dict[str, Any]]) -> list[str]:
    """선 이름 — 붙인 이름(594-1) · 없으면 차례 이름: 갈래 없는 선 = 본선(둘 이상이면
    본선 1 · 2 …) · 갈래 = 갈래 1 · 2 …(선 차례 · 이름 붙은 선도 수를 셈).

    ⚠ 짝 = TS `D03_PlanAnalysis/logic/D03_PlanAnalysis_Plan.ts` `lineNames` — 거울 시험 530-3."""
    roots = sum(1 for line in lines if not line.get("branch"))
    out, root, branch = [], 0, 0
    for line in lines:
        if line.get("branch"):
            branch += 1
            auto = f"갈래 {branch}"
        else:
            root += 1
            auto = f"본선 {root}" if roots > 1 else "본선"
        out.append((line.get("name") or "").strip() or auto)
    return out


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, piece_id, company_id, company_name, email, notified_at, linked_project_id"
            " FROM plan_route_assignees WHERE project_id = %s ORDER BY 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 piece_id = %s, company_name = %s, email = %s"
                    " WHERE id = %s AND project_id = %s",  # 옛 조각 id → 세트 열쇠(638-1)
                    (row["piece_id"], row["company_name"], row["email"], row["id"], project_id),
                )
            for row in inserts:
                await cursor.execute(
                    "INSERT INTO plan_route_assignees"
                    " (project_id, piece_id, company_id, company_name, email)"
                    " VALUES (%s,%s,%s,%s,%s)",
                    (
                        project_id,
                        row["piece_id"],
                        row["company_id"],
                        row["company_name"],
                        row["email"],
                    ),
                )
            await connection.commit()
        except Exception:
            await connection.rollback()
            raise


async def mark_confirmed(project_id: str, fingerprints: dict[int, str]) -> None:
    """연도 확정 — {연도: 파일 지문} · 상태 줄이 없거나 지문이 낡았으면 새로 맞춤."""
    if not fingerprints:
        return
    async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
        for year, fingerprint in fingerprints.items():
            await cursor.execute(
                "INSERT INTO plan_route_years (project_id, year, fingerprint, confirmed_at)"
                " VALUES (%s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE"
                " fingerprint = VALUES(fingerprint), confirmed_at = NOW()",
                (project_id, year, fingerprint),
            )
        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()