import os
import logging
import json
import re
import zipfile
from datetime import date, datetime
from functools import lru_cache
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, quote_plus

import uvicorn
from fastapi import FastAPI, File, Request, UploadFile
from fastapi.encoders import jsonable_encoder
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from openpyxl import load_workbook
from sqlalchemy import create_engine, event, text

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

BASE_DIR = Path(__file__).resolve().parent
STATIC_DIR = BASE_DIR / "static"
TEMPLATES_DIR = BASE_DIR / "templates"
DB_PATH = BASE_DIR / "data.db"

STATIC_DIR.mkdir(exist_ok=True)
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")

engine = create_engine(
    f"sqlite:///{DB_PATH}",
    connect_args={"check_same_thread": False},
)


@event.listens_for(engine, "connect")
def configure_sqlite_connection(dbapi_connection: Any, _: Any) -> None:
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA journal_mode=WAL")
    cursor.execute("PRAGMA synchronous=NORMAL")
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.execute("PRAGMA busy_timeout=5000")
    cursor.execute("PRAGMA temp_store=MEMORY")
    cursor.close()

EXCLUDED_SUPPORT_CODES = {"ZZZZZZ"}
EXCLUDED_SUPPORT_NAMES = {"공통", "경영지원부", "기술개발센터", "임원실", "기술개발부", "총괄기획실"}
SUPPORT_DEPARTMENT_NAMES = ("경영지원부", "임원실", "총괄기획실", "기술개발센터", "기술개발부", "공통")
VOUCHER_PATTERN = re.compile(r"^11-(\d{4})(\d{2})(\d{2})-[^-]+-[^-]+-[^-]+$")

REVENUE_SQL = "(account_code LIKE '401101%' OR account_code LIKE '401102%')"
SUPPORT_COST_DEPT_SQL = (
    "cost_dept_name IN ('경영지원부', '임원실', '총괄기획실', '기술개발센터', '기술개발부', '공통')"
)
FIELD_COST_DEPT_SQL = (
    "COALESCE(cost_dept_name, '') NOT IN ('경영지원부', '임원실', '총괄기획실', '기술개발센터', '기술개발부', '공통')"
)

FIELD_LABELS = {
    "approval_status": "결재상태",
    "voucher_number": "가전표번호",
    "account_code": "계정코드",
    "account_name": "계정명칭",
    "debit_supply": "차변공급가",
    "debit_vat": "차변부가세",
    "credit_supply": "대변공급가",
    "credit_vat": "대변부가세",
    "issuing_dept_code": "발의부서코드",
    "issuing_dept_name": "발의부서명",
    "confirmed_voucher_number": "확정전표번호",
    "support_dept_code": "지원부서코드",
    "support_dept_name": "지원부서명",
    "cost_dept_code": "원가부서코드",
    "cost_dept_name": "원가부서명",
    "memo1": "적요1",
    "memo2": "적요2",
    "partner_code": "거래처코드",
    "partner_name": "거래처명칭",
    "tax_code": "세무코드",
    "posting_date": "증빙일자",
    "voucher_type": "전표종류",
    "management_item": "관리항목",
}

FORM_FIELDS = list(FIELD_LABELS.keys())

DIRECT_HEADER_MAP = {
    "결재상태": "approval_status",
    "가전표번호": "voucher_number",
    "계정코드": "account_code",
    "계정명칭": "account_name",
    "차변공급가": "debit_supply",
    "차변부가세": "debit_vat",
    "대변공급가": "credit_supply",
    "대변부가세": "credit_vat",
    "발의부서코드": "issuing_dept_code",
    "발의부서명": "issuing_dept_name",
    "발의부서명칭": "issuing_dept_name",
    "지원부서코드": "support_dept_code",
    "지원부서명": "support_dept_name",
    "지원부서명칭": "support_dept_name",
    "원가부서코드": "cost_dept_code",
    "원가부서명": "cost_dept_name",
    "원가부서명칭": "cost_dept_name",
    "적요1": "memo1",
    "적요2": "memo2",
    "거래처코드": "partner_code",
    "거래처명칭": "partner_name",
    "세무코드": "tax_code",
    "증빙일자": "posting_date",
    "전표종류": "voucher_type",
    "관리항목": "management_item",
}

DEFAULT_APP_OPTION_ITEMS = {
    "labor_grades": [
        ("president", "사장", "사장"),
        ("vice_president", "부사장", "부사장"),
        ("executive_vice_president", "전무", "전무"),
        ("managing_director", "상무", "상무"),
        ("director", "이사", "이사"),
        ("general_manager", "부장", "부장"),
        ("deputy_general_manager", "차장", "차장"),
        ("manager", "과장", "과장"),
        ("assistant_manager", "대리", "대리"),
        ("staff", "사원", "사원"),
        ("principal", "수석", "수석"),
        ("senior_manager", "책임", "책임"),
        ("senior", "선임", "선임"),
        ("researcher", "연구원", "연구원"),
    ],
    "expected_as_rates": [
        ("as_0", "0%", "0"),
        ("as_2", "2%", "2"),
        ("as_5", "5%", "5"),
        ("as_10", "10%", "10"),
    ],
    "expected_sga_rates": [
        ("sga_13", "13%", "13"),
        ("sga_15", "15%", "15"),
        ("sga_20", "20%", "20"),
        ("sga_25", "25%", "25"),
    ],
    "uncontracted_categories": [
        ("general", "일반 미계약", "general"),
        ("precontract", "사전 사업 코드", "precontract"),
        ("corporate_rnd", "기업 연구개발", "corporate_rnd"),
        ("external_research", "외부 연구과제", "external_research"),
    ],
    "project_rules": [
        ("legacy_variant_cutoff_year", "이전 연도 변경/차수 제외 기준", "23"),
        ("detail_visible_min_year", "세부내역 반영 시작 연도", "2023"),
    ],
    "dashboard_revenue_metrics": [
        ("design_revenue", "설계", "#4f7cff"),
        ("design_other_revenue", "설계 외", "#67c7c9"),
        ("supervision_revenue", "감리", "#233a5a"),
        ("inspection_revenue", "점검", "#ffb54a"),
    ],
    "dashboard_expense_metrics": [
        ("cost_sum", "원가", "#4f7cff"),
        ("sga_sum", "판관비", "#67c7c9"),
        ("labor_sum", "원가인건비", "#233a5a"),
        ("outsourcing_sum", "원가외주비", "#ffb54a"),
    ],
    "annual_metric_cards": [
        ("revenue_sum", "수금", "수금"),
        ("project_cost_sum", "원가(프로젝트)", "원가(프로젝트)"),
        ("support_cost_sum", "원가(지원부서)", "원가(지원부서)"),
        ("support_sga_sum", "판관비(지원부서)", "판관비(지원부서)"),
        ("field_sga_sum", "판관비(현업부서)", "판관비(현업부서)"),
        ("labor_sum", "원가인건비", "원가인건비"),
        ("outsourcing_sum", "원가외주비", "원가외주비"),
        ("total_expense", "비용합계", "비용합계"),
        ("operating_balance", "영업수지", "영업수지"),
    ],
    "annual_expense_chart_metrics": [
        ("labor_sum", "원가인건비", "#8b5cf6"),
        ("outsourcing_sum", "원가외주비", "#ec4899"),
        ("project_cost_sum", "원가(프로젝트)", "#0ea5a4"),
        ("support_cost_sum", "원가(지원부서)", "#67b7dc"),
        ("support_sga_sum", "판관비(지원부서)", "#f59e0b"),
        ("field_sga_sum", "판관비(현업부서)", "#f97316"),
    ],
    "annual_balance_chart_metrics": [
        ("revenue_sum", "수금", "#0f766e"),
        ("total_expense", "비용합계", "#1d4ed8"),
        ("operating_balance", "영업수지", "#dc2626"),
    ],
}

DEFAULT_APP_KEYWORD_RULES = {
    "special_x_classification": [
        ("external_research", "과제"),
        ("external_research", "연구과제"),
        ("external_research", "연구소"),
        ("external_research", "연구용역"),
        ("external_research", "연구"),
        ("corporate_rnd", "신규노선개발"),
        ("corporate_rnd", "프로그램 개발"),
        ("corporate_rnd", "프로그램개발"),
        ("corporate_rnd", "BIM"),
        ("corporate_rnd", "시스템"),
        ("corporate_rnd", "혁신"),
    ]
}


def ensure_default_app_config(conn: Any) -> None:
    for group_key, items in DEFAULT_APP_OPTION_ITEMS.items():
        for sort_order, (item_key, label, value_text) in enumerate(items):
            conn.execute(
                text(
                    """
                    INSERT INTO app_option_items (
                        group_key, item_key, label, value_text, sort_order, is_active, meta_json
                    ) VALUES (
                        :group_key, :item_key, :label, :value_text, :sort_order, 1, '{}'
                    )
                    ON CONFLICT(group_key, item_key) DO UPDATE SET
                        label = excluded.label,
                        value_text = excluded.value_text,
                        sort_order = excluded.sort_order
                    """
                ),
                {
                    "group_key": group_key,
                    "item_key": item_key,
                    "label": label,
                    "value_text": value_text,
                    "sort_order": sort_order,
                },
            )

    for rule_group, items in DEFAULT_APP_KEYWORD_RULES.items():
        for sort_order, (category_key, keyword) in enumerate(items):
            conn.execute(
                text(
                    """
                    INSERT INTO app_keyword_rules (
                        rule_group, category_key, keyword, sort_order, is_active
                    ) VALUES (
                        :rule_group, :category_key, :keyword, :sort_order, 1
                    )
                    ON CONFLICT(rule_group, category_key, keyword) DO UPDATE SET
                        sort_order = excluded.sort_order
                    """
                ),
                {
                    "rule_group": rule_group,
                    "category_key": category_key,
                    "keyword": keyword,
                    "sort_order": sort_order,
                },
            )

    load_app_config.cache_clear()


def init_db() -> None:
    with engine.begin() as conn:
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS transactions (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    approval_status TEXT,
                    voucher_number TEXT,
                    account_code TEXT,
                    account_name TEXT,
                    debit_supply REAL DEFAULT 0,
                    debit_vat REAL DEFAULT 0,
                    credit_supply REAL DEFAULT 0,
                    credit_vat REAL DEFAULT 0,
                    issuing_dept_code TEXT,
                    issuing_dept_name TEXT,
                    confirmed_voucher_number TEXT,
                    support_dept_code TEXT,
                    support_dept_name TEXT,
                    cost_dept_code TEXT,
                    cost_dept_name TEXT,
                    memo1 TEXT,
                    memo2 TEXT,
                    partner_code TEXT,
                    partner_name TEXT,
                    tax_code TEXT,
                    posting_date TEXT,
                    voucher_type TEXT,
                    management_item TEXT,
                    accounting_category TEXT,
                    amount REAL DEFAULT 0,
                    year INTEGER,
                    month INTEGER,
                    day INTEGER,
                    source_file TEXT,
                    last_editor_session_id TEXT DEFAULT '',
                    last_client_submitted_at TEXT DEFAULT '',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_transactions_year_month
                ON transactions (year, month)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_transactions_support
                ON transactions (support_dept_code, support_dept_name)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_transactions_support_category
                ON transactions (support_dept_code, accounting_category)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_transactions_support_account
                ON transactions (support_dept_code, account_code)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_transactions_source_file
                ON transactions (source_file)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_transactions_updated_at
                ON transactions (updated_at)
                """
            )
        )
        transaction_columns = {
            row[1]
            for row in conn.execute(text("PRAGMA table_info(transactions)")).fetchall()
        }
        required_transaction_columns = {
            "last_editor_session_id": "TEXT DEFAULT ''",
            "last_client_submitted_at": "TEXT DEFAULT ''",
        }
        for column_name, column_type in required_transaction_columns.items():
            if column_name not in transaction_columns:
                conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {column_name} {column_type}"))
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_status (
                    support_dept_code TEXT PRIMARY KEY,
                    support_dept_name TEXT NOT NULL,
                    progress_rate REAL DEFAULT 0,
                    contract_amount REAL DEFAULT 0,
                    collection_amount REAL DEFAULT 0,
                    collection_entries_json TEXT DEFAULT '[]',
                    change_round TEXT DEFAULT '',
                    item_investment REAL DEFAULT 0,
                    task_plan_department_budget REAL DEFAULT 0,
                    task_plan_outsource_budget REAL DEFAULT 0,
                    task_plan_outsource_detail TEXT DEFAULT '',
                    task_plan_joint_operating_cost REAL DEFAULT 0,
                    task_plan_entries_json TEXT DEFAULT '[]',
                    exec_budget_labor_by_grade REAL DEFAULT 0,
                    exec_labor_rates_json TEXT DEFAULT '{}',
                    exec_budget_outsource REAL DEFAULT 0,
                    exec_budget_cost_plan REAL DEFAULT 0,
                    exec_budget_entries_json TEXT DEFAULT '[]',
                    actual_input_entries_json TEXT DEFAULT '[]',
                    project_type TEXT DEFAULT '',
                    expected_as_rate REAL DEFAULT 0,
                    expected_sga_rate REAL DEFAULT 0,
                    expected_as_cost REAL DEFAULT 0,
                    expected_sga_budget REAL DEFAULT 0,
                    last_editor_session_id TEXT DEFAULT '',
                    last_client_submitted_at TEXT DEFAULT '',
                    project_start_date TEXT DEFAULT '',
                    project_end_date TEXT DEFAULT '',
                    completion_status TEXT DEFAULT '',
                    notes TEXT DEFAULT '',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_basic_info (
                    support_dept_code TEXT PRIMARY KEY,
                    support_dept_name TEXT NOT NULL DEFAULT '',
                    contract_amount REAL DEFAULT 0,
                    project_type TEXT DEFAULT '',
                    expected_as_rate REAL DEFAULT 0,
                    expected_sga_rate REAL DEFAULT 0,
                    expected_as_cost REAL DEFAULT 0,
                    expected_sga_budget REAL DEFAULT 0,
                    exec_labor_rates_json TEXT DEFAULT '{}',
                    change_round TEXT DEFAULT '',
                    project_start_date TEXT DEFAULT '',
                    project_end_date TEXT DEFAULT '',
                    completion_status TEXT DEFAULT '',
                    notes TEXT DEFAULT '',
                    last_editor_session_id TEXT DEFAULT '',
                    last_client_submitted_at TEXT DEFAULT '',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_basic_info_updated_at
                ON project_basic_info (updated_at)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_page_state (
                    page_key TEXT PRIMARY KEY,
                    selected_code TEXT DEFAULT '',
                    selected_year TEXT DEFAULT '',
                    analysis_open INTEGER DEFAULT 0,
                    uncontracted_year_start TEXT DEFAULT '',
                    uncontracted_year_end TEXT DEFAULT '',
                    related_project_selections_json TEXT DEFAULT '{}',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS app_option_items (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    group_key TEXT NOT NULL,
                    item_key TEXT NOT NULL,
                    label TEXT NOT NULL,
                    value_text TEXT NOT NULL DEFAULT '',
                    sort_order INTEGER NOT NULL DEFAULT 0,
                    is_active INTEGER NOT NULL DEFAULT 1,
                    meta_json TEXT DEFAULT '{}',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    UNIQUE(group_key, item_key)
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_app_option_items_group
                ON app_option_items (group_key, sort_order, id)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS app_keyword_rules (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    rule_group TEXT NOT NULL,
                    category_key TEXT NOT NULL,
                    keyword TEXT NOT NULL,
                    sort_order INTEGER NOT NULL DEFAULT 0,
                    is_active INTEGER NOT NULL DEFAULT 1,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    UNIQUE(rule_group, category_key, keyword)
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_app_keyword_rules_group
                ON app_keyword_rules (rule_group, category_key, sort_order, id)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_collection_entries (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    support_dept_code TEXT NOT NULL,
                    position INTEGER NOT NULL DEFAULT 0,
                    vendor TEXT DEFAULT '',
                    progress_type TEXT DEFAULT '',
                    billing_round TEXT DEFAULT '',
                    billing_type TEXT DEFAULT '',
                    billing_date TEXT DEFAULT '',
                    billed_amount REAL DEFAULT 0,
                    round TEXT DEFAULT '',
                    date TEXT DEFAULT '',
                    due_date TEXT DEFAULT '',
                    amount REAL DEFAULT 0,
                    balance_amount REAL DEFAULT 0,
                    collection_rate REAL DEFAULT 0,
                    note TEXT DEFAULT '',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_collection_entries_code_pos
                ON project_collection_entries (support_dept_code, position)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_task_plan_entries (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    support_dept_code TEXT NOT NULL,
                    position INTEGER NOT NULL DEFAULT 0,
                    group_name TEXT DEFAULT '',
                    dept_name TEXT DEFAULT '',
                    work_name TEXT DEFAULT '',
                    amount REAL DEFAULT 0,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_task_plan_entries_code_pos
                ON project_task_plan_entries (support_dept_code, position)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_exec_budget_entries (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    support_dept_code TEXT NOT NULL,
                    position INTEGER NOT NULL DEFAULT 0,
                    group_name TEXT DEFAULT '',
                    grade TEXT DEFAULT '',
                    hours TEXT DEFAULT '',
                    dept_name TEXT DEFAULT '',
                    work_name TEXT DEFAULT '',
                    account_code TEXT DEFAULT '',
                    account_name TEXT DEFAULT '',
                    amount REAL DEFAULT 0,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_exec_budget_entries_code_pos
                ON project_exec_budget_entries (support_dept_code, position)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_actual_input_entries (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    support_dept_code TEXT NOT NULL,
                    position INTEGER NOT NULL DEFAULT 0,
                    group_name TEXT DEFAULT '',
                    grade TEXT DEFAULT '',
                    minutes TEXT DEFAULT '',
                    label TEXT DEFAULT '',
                    reference TEXT DEFAULT '',
                    note TEXT DEFAULT '',
                    amount REAL DEFAULT 0,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_actual_input_entries_code_pos
                ON project_actual_input_entries (support_dept_code, position)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_comparison_notes (
                    support_dept_code TEXT NOT NULL,
                    item_key TEXT NOT NULL,
                    note TEXT DEFAULT '',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (support_dept_code, item_key)
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_comparison_notes_code
                ON project_comparison_notes (support_dept_code, item_key)
                """
            )
        )
        page_state_columns_before = {
            row[1]
            for row in conn.execute(text("PRAGMA table_info(project_page_state)")).fetchall()
        }
        if "session_id" not in page_state_columns_before:
            legacy_rows = conn.execute(
                text(
                    """
                    SELECT page_key, selected_code, selected_year, analysis_open,
                           related_project_selections_json, updated_at
                    FROM project_page_state
                    """
                )
            ).mappings().all()
            conn.execute(text("ALTER TABLE project_page_state RENAME TO project_page_state_legacy"))
            conn.execute(
                text(
                    """
                    CREATE TABLE project_page_state (
                        page_key TEXT NOT NULL,
                        session_id TEXT NOT NULL DEFAULT '',
                        selected_code TEXT DEFAULT '',
                        selected_year TEXT DEFAULT '',
                        analysis_open INTEGER DEFAULT 0,
                        uncontracted_year_start TEXT DEFAULT '',
                        uncontracted_year_end TEXT DEFAULT '',
                        related_project_selections_json TEXT DEFAULT '{}',
                        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                        PRIMARY KEY (page_key, session_id)
                    )
                    """
                )
            )
            for row in legacy_rows:
                conn.execute(
                    text(
                        """
                        INSERT INTO project_page_state (
                            page_key, session_id, selected_code, selected_year,
                            analysis_open, uncontracted_year_start, uncontracted_year_end,
                            related_project_selections_json, updated_at
                        ) VALUES (
                            :page_key, '', :selected_code, :selected_year,
                            :analysis_open, '', '', :related_project_selections_json, :updated_at
                        )
                        """
                    ),
                    dict(row),
                )
            conn.execute(text("DROP TABLE project_page_state_legacy"))
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_page_state_page_session
                ON project_page_state (page_key, session_id)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_related_links (
                    base_support_dept_code TEXT NOT NULL,
                    related_support_dept_code TEXT NOT NULL,
                    link_source TEXT DEFAULT 'manual',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (base_support_dept_code, related_support_dept_code)
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_uncontracted_classification (
                    support_dept_code TEXT PRIMARY KEY,
                    category TEXT DEFAULT '',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_uncontracted_classification_category
                ON project_uncontracted_classification (category)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_quick_links (
                    page_key TEXT NOT NULL,
                    support_dept_code TEXT NOT NULL,
                    sort_order INTEGER NOT NULL DEFAULT 0,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (page_key, support_dept_code)
                )
                """
            )
        )
        quick_link_columns_before = {
            row[1]
            for row in conn.execute(text("PRAGMA table_info(project_quick_links)")).fetchall()
        }
        if "session_id" in quick_link_columns_before:
            legacy_rows = conn.execute(
                text(
                    """
                    SELECT page_key, support_dept_code, MIN(sort_order) AS sort_order, MAX(updated_at) AS updated_at
                    FROM project_quick_links
                    GROUP BY page_key, support_dept_code
                    """
                )
            ).mappings().all()
            conn.execute(text("ALTER TABLE project_quick_links RENAME TO project_quick_links_legacy"))
            conn.execute(
                text(
                    """
                    CREATE TABLE project_quick_links (
                        page_key TEXT NOT NULL,
                        support_dept_code TEXT NOT NULL,
                        sort_order INTEGER NOT NULL DEFAULT 0,
                        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                        PRIMARY KEY (page_key, support_dept_code)
                    )
                    """
                )
            )
            for row in legacy_rows:
                conn.execute(
                    text(
                        """
                        INSERT INTO project_quick_links (
                            page_key, support_dept_code, sort_order, updated_at
                        ) VALUES (
                            :page_key, :support_dept_code, :sort_order, :updated_at
                        )
                        """
                    ),
                    dict(row),
                )
            conn.execute(text("DROP TABLE project_quick_links_legacy"))
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_quick_links_page_sort
                ON project_quick_links (page_key, sort_order, updated_at)
                """
            )
        )
        ensure_default_app_config(conn)
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_related_links_base
                ON project_related_links (base_support_dept_code)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_related_links_related
                ON project_related_links (related_support_dept_code)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_contract_info (
                    support_dept_code TEXT PRIMARY KEY,
                    raw_contract_code TEXT DEFAULT '',
                    business_division TEXT DEFAULT '',
                    order_method TEXT DEFAULT '',
                    owner_department TEXT DEFAULT '',
                    client_name TEXT DEFAULT '',
                    support_dept_name TEXT DEFAULT '',
                    work_category TEXT DEFAULT '',
                    order_date TEXT DEFAULT '',
                    contract_date TEXT DEFAULT '',
                    project_start_date TEXT DEFAULT '',
                    project_end_date TEXT DEFAULT '',
                    contract_status TEXT DEFAULT '',
                    joint_contract TEXT DEFAULT '',
                    pm_name TEXT DEFAULT '',
                    progress_status TEXT DEFAULT '',
                    total_contract_amount REAL DEFAULT 0,
                    hanmac_contract_amount REAL DEFAULT 0,
                    review_tag TEXT DEFAULT '',
                    review_note TEXT DEFAULT '',
                    source_file TEXT DEFAULT '',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_contract_info_updated_at
                ON project_contract_info (updated_at)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_contract_change_summary (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    raw_summary_code TEXT DEFAULT '',
                    normalized_title TEXT DEFAULT '',
                    owner_department TEXT DEFAULT '',
                    business_division TEXT DEFAULT '',
                    support_dept_name TEXT DEFAULT '',
                    change_date TEXT DEFAULT '',
                    client_name TEXT DEFAULT '',
                    original_contract_period TEXT DEFAULT '',
                    changed_project_end_date TEXT DEFAULT '',
                    initial_contract_amount REAL DEFAULT 0,
                    previous_contract_amount REAL DEFAULT 0,
                    changed_contract_amount REAL DEFAULT 0,
                    delta_amount REAL DEFAULT 0,
                    source_file TEXT DEFAULT '',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_contract_change_summary_code
                ON project_contract_change_summary (raw_summary_code, change_date)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_contract_change_summary_title
                ON project_contract_change_summary (normalized_title)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_contract_change_round (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    support_dept_code TEXT DEFAULT '',
                    raw_round_code TEXT DEFAULT '',
                    normalized_title TEXT DEFAULT '',
                    owner_department TEXT DEFAULT '',
                    business_division TEXT DEFAULT '',
                    support_dept_name TEXT DEFAULT '',
                    change_date TEXT DEFAULT '',
                    client_name TEXT DEFAULT '',
                    original_contract_period TEXT DEFAULT '',
                    changed_project_end_date TEXT DEFAULT '',
                    initial_contract_amount REAL DEFAULT 0,
                    changed_contract_amount REAL DEFAULT 0,
                    delta_amount REAL DEFAULT 0,
                    source_file TEXT DEFAULT '',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_contract_change_round_code
                ON project_contract_change_round (support_dept_code, change_date)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_contract_change_round_title
                ON project_contract_change_round (normalized_title)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE TABLE IF NOT EXISTS project_billing_entries (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    support_dept_code TEXT,
                    raw_project_code TEXT DEFAULT '',
                    round_code TEXT DEFAULT '',
                    support_department TEXT DEFAULT '',
                    business_division TEXT DEFAULT '',
                    support_dept_name TEXT DEFAULT '',
                    contract_amount REAL DEFAULT 0,
                    client_name TEXT DEFAULT '',
                    billing_type TEXT DEFAULT '',
                    progress_round TEXT DEFAULT '',
                    billing_date TEXT DEFAULT '',
                    tax_invoice_date TEXT DEFAULT '',
                    expected_collection_date TEXT DEFAULT '',
                    billed_amount REAL DEFAULT 0,
                    collected_amount REAL DEFAULT 0,
                    balance_amount REAL DEFAULT 0,
                    collection_rate REAL DEFAULT 0,
                    note TEXT DEFAULT '',
                    source_file TEXT DEFAULT '',
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_billing_entries_code
                ON project_billing_entries (support_dept_code, billing_date)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_billing_entries_raw_code
                ON project_billing_entries (raw_project_code)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_billing_entries_round_code
                ON project_billing_entries (round_code)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_billing_entries_source_file
                ON project_billing_entries (source_file)
                """
            )
        )
        conn.execute(
            text(
                """
                CREATE INDEX IF NOT EXISTS idx_project_billing_entries_updated_at
                ON project_billing_entries (updated_at)
                """
            )
        )
        existing_columns = {
            row[1]
            for row in conn.execute(text("PRAGMA table_info(project_status)")).fetchall()
        }
        required_columns = {
            "last_editor_session_id": "TEXT DEFAULT ''",
            "last_client_submitted_at": "TEXT DEFAULT ''",
            "contract_amount": "REAL DEFAULT 0",
            "collection_entries_json": "TEXT DEFAULT '[]'",
            "task_plan_department_budget": "REAL DEFAULT 0",
            "task_plan_outsource_budget": "REAL DEFAULT 0",
            "task_plan_outsource_detail": "TEXT DEFAULT ''",
            "task_plan_joint_operating_cost": "REAL DEFAULT 0",
            "task_plan_entries_json": "TEXT DEFAULT '[]'",
            "exec_budget_labor_by_grade": "REAL DEFAULT 0",
            "exec_labor_rates_json": "TEXT DEFAULT '{}'",
            "exec_budget_outsource": "REAL DEFAULT 0",
            "exec_budget_cost_plan": "REAL DEFAULT 0",
            "exec_budget_entries_json": "TEXT DEFAULT '[]'",
            "actual_input_entries_json": "TEXT DEFAULT '[]'",
            "project_type": "TEXT DEFAULT ''",
            "expected_as_rate": "REAL DEFAULT 0",
            "expected_sga_rate": "REAL DEFAULT 0",
            "expected_as_cost": "REAL DEFAULT 0",
            "expected_sga_budget": "REAL DEFAULT 0",
            "project_start_date": "TEXT DEFAULT ''",
            "project_end_date": "TEXT DEFAULT ''",
            "completion_status": "TEXT DEFAULT ''",
        }
        for column_name, column_type in required_columns.items():
            if column_name not in existing_columns:
                conn.execute(text(f"ALTER TABLE project_status ADD COLUMN {column_name} {column_type}"))
        page_state_columns = {
            row[1]
            for row in conn.execute(text("PRAGMA table_info(project_page_state)")).fetchall()
        }
        required_page_state_columns = {
            "session_id": "TEXT DEFAULT ''",
            "selected_code": "TEXT DEFAULT ''",
            "selected_year": "TEXT DEFAULT ''",
            "analysis_open": "INTEGER DEFAULT 0",
            "uncontracted_year_start": "TEXT DEFAULT ''",
            "uncontracted_year_end": "TEXT DEFAULT ''",
            "related_project_selections_json": "TEXT DEFAULT '{}'",
        }
        for column_name, column_type in required_page_state_columns.items():
            if column_name not in page_state_columns:
                conn.execute(text(f"ALTER TABLE project_page_state ADD COLUMN {column_name} {column_type}"))
        related_link_columns = {
            row[1]
            for row in conn.execute(text("PRAGMA table_info(project_related_links)")).fetchall()
        }
        if "link_source" not in related_link_columns:
            conn.execute(text("ALTER TABLE project_related_links ADD COLUMN link_source TEXT DEFAULT 'manual'"))
        migrate_project_status_entries(conn)
        migrate_project_basic_info(conn)
        ensure_default_app_config(conn)
        conn.execute(text("ANALYZE"))


@lru_cache(maxsize=1)
def load_app_config() -> dict[str, Any]:
    with engine.connect() as conn:
        option_rows = conn.execute(
            text(
                """
                SELECT group_key, item_key, label, value_text, sort_order
                FROM app_option_items
                WHERE is_active = 1
                ORDER BY group_key, sort_order, id
                """
            )
        ).mappings().all()
        keyword_rows = conn.execute(
            text(
                """
                SELECT rule_group, category_key, keyword, sort_order
                FROM app_keyword_rules
                WHERE is_active = 1
                ORDER BY rule_group, category_key, sort_order, id
                """
            )
        ).mappings().all()
    options: dict[str, list[dict[str, Any]]] = {}
    for row in option_rows:
        options.setdefault(row["group_key"], []).append(
            {
                "item_key": row["item_key"],
                "label": row["label"],
                "value": row["value_text"],
                "sort_order": row["sort_order"],
            }
        )
    keyword_rules: dict[str, dict[str, list[str]]] = {}
    for row in keyword_rows:
        keyword_rules.setdefault(row["rule_group"], {}).setdefault(row["category_key"], []).append(row["keyword"])
    return {"options": options, "keyword_rules": keyword_rules}


def get_option_items(group_key: str) -> list[dict[str, Any]]:
    return list(load_app_config().get("options", {}).get(group_key, []))


def get_keyword_rule_groups(rule_group: str) -> dict[str, list[str]]:
    return dict(load_app_config().get("keyword_rules", {}).get(rule_group, {}))


def get_labor_grade_options() -> list[str]:
    return [item["label"] for item in get_option_items("labor_grades")]


def get_expected_as_rate_options() -> list[dict[str, Any]]:
    return get_option_items("expected_as_rates")


def get_expected_sga_rate_options() -> list[dict[str, Any]]:
    return get_option_items("expected_sga_rates")


def get_collection_progress_type_options() -> list[str]:
    preferred_order = {"선급금": 0, "기성금": 1, "준공금": 2}
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT DISTINCT billing_type
                FROM project_billing_entries
                WHERE COALESCE(billing_type, '') <> ''
                ORDER BY billing_type
                """
            )
        ).fetchall()
    values = {
        normalized
        for (value,) in rows
        for normalized in [normalize_collection_progress_type(value)]
        if normalized
    }
    if not values:
        values = {"선급금", "기성금", "준공금"}
    return sorted(values, key=lambda item: (preferred_order.get(item, 999), item))


def get_collection_billing_type_options() -> list[str]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT DISTINCT billing_type
                FROM project_collection_entries
                WHERE COALESCE(billing_type, '') <> ''
                ORDER BY billing_type
                """
            )
        ).fetchall()
    values = {
        normalized
        for (value,) in rows
        for normalized in [normalize_collection_billing_type(value)]
        if normalized
    }
    values.add("기타")
    if not values:
        values = {"계약분", "기타"}
    return list(sorted(values))


def get_uncontracted_category_options() -> list[dict[str, Any]]:
    return get_option_items("uncontracted_categories")


def get_project_runtime_settings() -> dict[str, str]:
    return {item["item_key"]: str(item.get("value", "")) for item in get_option_items("project_rules")}


def get_special_x_classification_rules() -> dict[str, list[str]]:
    return get_keyword_rule_groups("special_x_classification")


def save_project_runtime_setting(item_key: Any, value_text: Any) -> None:
    normalized_item_key = normalize_text(item_key)
    if not normalized_item_key:
        raise ValueError("설정 키가 올바르지 않습니다.")
    with engine.begin() as conn:
        existing = conn.execute(
            text(
                """
                SELECT group_key, item_key
                FROM app_option_items
                WHERE group_key = 'project_rules' AND item_key = :item_key
                """
            ),
            {"item_key": normalized_item_key},
        ).mappings().first()
        if not existing:
            raise ValueError("존재하지 않는 런타임 설정입니다.")
        conn.execute(
            text(
                """
                UPDATE app_option_items
                SET value_text = :value_text,
                    updated_at = CURRENT_TIMESTAMP
                WHERE group_key = 'project_rules' AND item_key = :item_key
                """
            ),
            {
                "item_key": normalized_item_key,
                "value_text": normalize_text(value_text),
            },
        )
    load_app_config.cache_clear()


def count_transactions() -> int:
    with engine.begin() as conn:
        return conn.execute(text("SELECT COUNT(*) FROM transactions")).scalar_one()


def existing_source_files() -> set[str]:
    with engine.begin() as conn:
        rows = conn.execute(
            text("SELECT DISTINCT source_file FROM transactions WHERE COALESCE(source_file, '') <> ''")
        ).fetchall()
    return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}


def existing_contract_source_files() -> set[str]:
    with engine.begin() as conn:
        rows = conn.execute(
            text("SELECT DISTINCT source_file FROM project_contract_info WHERE COALESCE(source_file, '') <> ''")
        ).fetchall()
    return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}


def existing_billing_source_files() -> set[str]:
    with engine.begin() as conn:
        rows = conn.execute(
            text("SELECT DISTINCT source_file FROM project_billing_entries WHERE COALESCE(source_file, '') <> ''")
        ).fetchall()
    return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}


def existing_change_contract_summary_source_files() -> set[str]:
    with engine.begin() as conn:
        rows = conn.execute(
            text("SELECT DISTINCT source_file FROM project_contract_change_summary WHERE COALESCE(source_file, '') <> ''")
        ).fetchall()
    return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}


def existing_change_contract_round_source_files() -> set[str]:
    with engine.begin() as conn:
        rows = conn.execute(
            text("SELECT DISTINCT source_file FROM project_contract_change_round WHERE COALESCE(source_file, '') <> ''")
        ).fetchall()
    return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}


def workbook_row_values(sheet: Any, row_number: int) -> list[str]:
    return [normalize_text(sheet.cell(row_number, column).value) for column in range(1, sheet.max_column + 1)]


def detect_excel_import_kind(workbook: Any, filename: str = "") -> str:
    sheet = workbook.active
    row1 = workbook_row_values(sheet, 1)
    row5 = workbook_row_values(sheet, 5) if sheet.max_row >= 5 else []
    filename = normalize_text(filename)
    if {"총괄코드", "총 계약금액", "한맥계약금액"}.issubset(set(row1)):
        return "contract_status"
    if {"총괄코드", "최초계약금액", "이전계약금액", "변경계약금액", "증감액"}.issubset(set(row1)):
        return "change_contract_summary"
    if {"차수코드", "당초계약금액", "변경계약금액", "증감액"}.issubset(set(row1)):
        return "change_contract_round"
    if {"차수코드", "차수사업명", "청구금액", "수금금액"}.issubset(set(row5)):
        return "billing_status"
    if "계약현황" in filename:
        return "contract_status"
    if "변경계약금액현황" in filename and "총괄" in filename:
        return "change_contract_summary"
    if "변경계약금액현황" in filename and "차수" in filename:
        return "change_contract_round"
    if "기성청구현황" in filename:
        return "billing_status"
    return "transactions"


def import_contract_status_workbook(workbook: Any, source_file: str) -> int:
    sheet = workbook.active
    with engine.begin() as conn:
        conn.execute(
            text("DELETE FROM project_contract_info WHERE source_file = :source_file"),
            {"source_file": source_file},
        )
        inserted = 0
        for row in sheet.iter_rows(min_row=2, values_only=True):
            support_dept_code = normalize_project_code(row[1] if len(row) > 1 else "")
            if not support_dept_code:
                continue
            payload = {
                "support_dept_code": support_dept_code,
                "raw_contract_code": normalize_text(row[1] if len(row) > 1 else ""),
                "business_division": normalize_text(row[0] if len(row) > 0 else ""),
                "order_method": normalize_text(row[2] if len(row) > 2 else ""),
                "owner_department": normalize_text(row[3] if len(row) > 3 else ""),
                "client_name": normalize_text(row[4] if len(row) > 4 else ""),
                "support_dept_name": normalize_text(row[5] if len(row) > 5 else ""),
                "work_category": normalize_text(row[6] if len(row) > 6 else ""),
                "order_date": normalize_date_text(row[7] if len(row) > 7 else ""),
                "contract_date": normalize_date_text(row[8] if len(row) > 8 else ""),
                "project_start_date": normalize_date_text(row[9] if len(row) > 9 else ""),
                "project_end_date": normalize_date_text(row[10] if len(row) > 10 else ""),
                "contract_status": normalize_text(row[11] if len(row) > 11 else ""),
                "joint_contract": normalize_text(row[12] if len(row) > 12 else ""),
                "pm_name": normalize_text(row[13] if len(row) > 13 else ""),
                "progress_status": normalize_text(row[14] if len(row) > 14 else ""),
                "total_contract_amount": normalize_amount(row[15] if len(row) > 15 else 0),
                "hanmac_contract_amount": normalize_amount(row[16] if len(row) > 16 else 0),
                "review_tag": "",
                "review_note": "",
                "source_file": source_file,
            }
            conn.execute(
                text(
                    """
                    INSERT INTO project_contract_info (
                        support_dept_code, raw_contract_code, business_division, order_method,
                        owner_department, client_name, support_dept_name, work_category,
                        order_date, contract_date, project_start_date, project_end_date,
                        contract_status, joint_contract, pm_name, progress_status,
                        total_contract_amount, hanmac_contract_amount, review_tag, review_note,
                        source_file, updated_at
                    ) VALUES (
                        :support_dept_code, :raw_contract_code, :business_division, :order_method,
                        :owner_department, :client_name, :support_dept_name, :work_category,
                        :order_date, :contract_date, :project_start_date, :project_end_date,
                        :contract_status, :joint_contract, :pm_name, :progress_status,
                        :total_contract_amount, :hanmac_contract_amount, :review_tag, :review_note,
                        :source_file, CURRENT_TIMESTAMP
                    )
                    ON CONFLICT(support_dept_code) DO UPDATE SET
                        raw_contract_code = excluded.raw_contract_code,
                        business_division = excluded.business_division,
                        order_method = excluded.order_method,
                        owner_department = excluded.owner_department,
                        client_name = excluded.client_name,
                        support_dept_name = excluded.support_dept_name,
                        work_category = excluded.work_category,
                        order_date = excluded.order_date,
                        contract_date = excluded.contract_date,
                        project_start_date = excluded.project_start_date,
                        project_end_date = excluded.project_end_date,
                        contract_status = excluded.contract_status,
                        joint_contract = excluded.joint_contract,
                        pm_name = excluded.pm_name,
                        progress_status = excluded.progress_status,
                        total_contract_amount = excluded.total_contract_amount,
                        hanmac_contract_amount = excluded.hanmac_contract_amount,
                        source_file = excluded.source_file,
                        updated_at = CURRENT_TIMESTAMP
                    """
                ),
                payload,
            )
            inserted += 1
    refresh_contract_review_tags()
    sync_auto_project_related_links()
    return inserted


def import_billing_status_workbook(workbook: Any, source_file: str) -> int:
    sheet = workbook.active
    with engine.begin() as conn:
        conn.execute(
            text("DELETE FROM project_billing_entries WHERE source_file = :source_file"),
            {"source_file": source_file},
        )
        inserted = 0
        current: dict[str, Any] = {}
        for row in sheet.iter_rows(min_row=6, values_only=True):
            values = list(row)
            if values and all(value in (None, "") for value in values):
                continue
            if values[0] is not None:
                current["support_department"] = normalize_text(values[0])
            if len(values) > 1 and values[1] is not None:
                current["business_division"] = normalize_text(values[1])
            if len(values) > 2 and values[2] is not None:
                current["raw_project_code"] = normalize_text(values[2])
            if len(values) > 3 and values[3] is not None:
                current["round_code"] = normalize_text(values[3])
            if len(values) > 4 and values[4] is not None:
                current["support_dept_name"] = normalize_text(values[4])
            if len(values) > 5 and values[5] is not None:
                current["contract_amount"] = normalize_amount(values[5])
            if len(values) > 6 and values[6] is not None:
                current["client_name"] = normalize_text(values[6])

            round_code_text = normalize_text(current.get("round_code"))
            round_prefix = next((character.upper() for character in round_code_text if character.isalpha()), "Y")
            support_dept_code = normalize_project_code(
                current.get("raw_project_code") or current.get("round_code"),
                default_prefix=round_prefix,
            )
            if not support_dept_code:
                continue

            summary_row = normalize_text(values[11] if len(values) > 11 else "") == "합계" or normalize_text(values[10] if len(values) > 10 else "").startswith("수금 :")
            department_summary = "합계" in normalize_text(values[4] if len(values) > 4 else "")
            if summary_row or department_summary:
                continue

            payload = {
                "support_dept_code": support_dept_code,
                "raw_project_code": normalize_text(current.get("raw_project_code")),
                "round_code": normalize_text(current.get("round_code")),
                "support_department": normalize_text(current.get("support_department")),
                "business_division": normalize_text(current.get("business_division")),
                "support_dept_name": normalize_text(current.get("support_dept_name")),
                "contract_amount": normalize_amount(current.get("contract_amount")),
                "client_name": normalize_text(current.get("client_name")),
                "billing_type": normalize_text(values[7] if len(values) > 7 else ""),
                "progress_round": normalize_round_value(values[8] if len(values) > 8 else ""),
                "billing_date": normalize_date_text(values[9] if len(values) > 9 else ""),
                "tax_invoice_date": normalize_date_text(values[10] if len(values) > 10 else ""),
                "expected_collection_date": normalize_date_text(values[11] if len(values) > 11 else ""),
                "billed_amount": normalize_amount(values[12] if len(values) > 12 else 0),
                "collected_amount": normalize_amount(values[13] if len(values) > 13 else 0),
                "balance_amount": normalize_amount(values[14] if len(values) > 14 else 0),
                "collection_rate": normalize_amount(values[15] if len(values) > 15 else 0),
                "note": normalize_text(values[16] if len(values) > 16 else ""),
                "source_file": source_file,
            }
            conn.execute(
                text(
                    """
                    INSERT INTO project_billing_entries (
                        support_dept_code, raw_project_code, round_code, support_department,
                        business_division, support_dept_name, contract_amount, client_name,
                        billing_type, progress_round, billing_date, tax_invoice_date,
                        expected_collection_date, billed_amount, collected_amount,
                        balance_amount, collection_rate, note, source_file, updated_at
                    ) VALUES (
                        :support_dept_code, :raw_project_code, :round_code, :support_department,
                        :business_division, :support_dept_name, :contract_amount, :client_name,
                        :billing_type, :progress_round, :billing_date, :tax_invoice_date,
                        :expected_collection_date, :billed_amount, :collected_amount,
                        :balance_amount, :collection_rate, :note, :source_file, CURRENT_TIMESTAMP
                    )
                    """
                ),
                payload,
            )
            inserted += 1
    refresh_contract_review_tags()
    return inserted


def import_change_contract_summary_workbook(workbook: Any, source_file: str) -> int:
    sheet = workbook.active
    with engine.begin() as conn:
        conn.execute(
            text("DELETE FROM project_contract_change_summary WHERE source_file = :source_file"),
            {"source_file": source_file},
        )
        inserted = 0
        current_department = ""
        for row in sheet.iter_rows(min_row=2, values_only=True):
            values = list(row)
            if not values or all(value in (None, "") for value in values):
                continue
            first_value = normalize_text(values[0] if len(values) > 0 else "")
            if "소 계" in first_value or first_value.startswith("<"):
                continue
            if first_value:
                current_department = first_value
            raw_summary_code = normalize_text(values[2] if len(values) > 2 else "").replace("\xa0", "")
            support_dept_name = normalize_text(values[3] if len(values) > 3 else "")
            if not raw_summary_code or not support_dept_name:
                continue
            payload = {
                "raw_summary_code": "".join(character for character in raw_summary_code if character.isdigit()),
                "normalized_title": normalize_project_title_for_linking(support_dept_name),
                "owner_department": current_department,
                "business_division": normalize_text(values[1] if len(values) > 1 else ""),
                "support_dept_name": support_dept_name,
                "change_date": normalize_date_text(values[4] if len(values) > 4 else ""),
                "client_name": normalize_text(values[5] if len(values) > 5 else ""),
                "original_contract_period": normalize_text(values[6] if len(values) > 6 else ""),
                "changed_project_end_date": normalize_date_text(values[7] if len(values) > 7 else ""),
                "initial_contract_amount": normalize_amount(values[8] if len(values) > 8 else 0),
                "previous_contract_amount": normalize_amount(values[9] if len(values) > 9 else 0),
                "changed_contract_amount": normalize_amount(values[10] if len(values) > 10 else 0),
                "delta_amount": normalize_amount(values[11] if len(values) > 11 else 0),
                "source_file": source_file,
            }
            conn.execute(
                text(
                    """
                    INSERT INTO project_contract_change_summary (
                        raw_summary_code, normalized_title, owner_department, business_division,
                        support_dept_name, change_date, client_name, original_contract_period,
                        changed_project_end_date, initial_contract_amount, previous_contract_amount,
                        changed_contract_amount, delta_amount, source_file, updated_at
                    ) VALUES (
                        :raw_summary_code, :normalized_title, :owner_department, :business_division,
                        :support_dept_name, :change_date, :client_name, :original_contract_period,
                        :changed_project_end_date, :initial_contract_amount, :previous_contract_amount,
                        :changed_contract_amount, :delta_amount, :source_file, CURRENT_TIMESTAMP
                    )
                    """
                ),
                payload,
            )
            inserted += 1
    sync_auto_project_related_links()
    return inserted


def import_change_contract_round_workbook(workbook: Any, source_file: str) -> int:
    sheet = workbook.active
    with engine.begin() as conn:
        conn.execute(
            text("DELETE FROM project_contract_change_round WHERE source_file = :source_file"),
            {"source_file": source_file},
        )
        inserted = 0
        current_department = ""
        for row in sheet.iter_rows(min_row=2, values_only=True):
            values = list(row)
            if not values or all(value in (None, "") for value in values):
                continue
            first_value = normalize_text(values[0] if len(values) > 0 else "")
            if "소 계" in first_value or first_value.startswith("<"):
                continue
            if first_value:
                current_department = first_value
            raw_round_code = normalize_text(values[2] if len(values) > 2 else "").replace("\xa0", "")
            support_dept_name = normalize_text(values[3] if len(values) > 3 else "")
            support_dept_code = normalize_project_code(raw_round_code)
            if not support_dept_code or not support_dept_name:
                continue
            payload = {
                "support_dept_code": support_dept_code,
                "raw_round_code": raw_round_code,
                "normalized_title": normalize_project_title_for_linking(support_dept_name),
                "owner_department": current_department,
                "business_division": normalize_text(values[1] if len(values) > 1 else ""),
                "support_dept_name": support_dept_name,
                "change_date": normalize_date_text(values[4] if len(values) > 4 else ""),
                "client_name": normalize_text(values[5] if len(values) > 5 else ""),
                "original_contract_period": normalize_text(values[6] if len(values) > 6 else ""),
                "changed_project_end_date": normalize_date_text(values[7] if len(values) > 7 else ""),
                "initial_contract_amount": normalize_amount(values[8] if len(values) > 8 else 0),
                "changed_contract_amount": normalize_amount(values[9] if len(values) > 9 else 0),
                "delta_amount": normalize_amount(values[10] if len(values) > 10 else 0),
                "source_file": source_file,
            }
            conn.execute(
                text(
                    """
                    INSERT INTO project_contract_change_round (
                        support_dept_code, raw_round_code, normalized_title, owner_department,
                        business_division, support_dept_name, change_date, client_name,
                        original_contract_period, changed_project_end_date, initial_contract_amount,
                        changed_contract_amount, delta_amount, source_file, updated_at
                    ) VALUES (
                        :support_dept_code, :raw_round_code, :normalized_title, :owner_department,
                        :business_division, :support_dept_name, :change_date, :client_name,
                        :original_contract_period, :changed_project_end_date, :initial_contract_amount,
                        :changed_contract_amount, :delta_amount, :source_file, CURRENT_TIMESTAMP
                    )
                    """
                ),
                payload,
            )
            inserted += 1
    sync_auto_project_related_links()
    return inserted


def select_latest_contract_change_entry(rows: list[dict[str, Any]]) -> dict[str, Any]:
    def sort_key(row: dict[str, Any]) -> tuple[str, int, float, float]:
        return (
            normalize_text(row.get("change_date")),
            1 if normalize_amount(row.get("changed_contract_amount")) > 0 else 0,
            abs(normalize_amount(row.get("delta_amount"))),
            normalize_amount(row.get("changed_contract_amount")),
        )

    return max(rows, key=sort_key) if rows else {}


def get_project_contract_change_maps() -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], dict[str, str], dict[str, str]]:
    with engine.begin() as conn:
        summary_rows = conn.execute(
            text("SELECT * FROM project_contract_change_summary ORDER BY raw_summary_code, change_date, id")
        ).mappings().all()
        round_rows = conn.execute(
            text("SELECT * FROM project_contract_change_round ORDER BY support_dept_code, change_date, id")
        ).mappings().all()
        existing_codes = {
            normalize_text(row[0])
            for row in conn.execute(
                text(
                    """
                    SELECT DISTINCT support_dept_code FROM transactions WHERE COALESCE(support_dept_code, '') <> ''
                    UNION
                    SELECT DISTINCT support_dept_code FROM project_status WHERE COALESCE(support_dept_code, '') <> ''
                    UNION
                    SELECT DISTINCT support_dept_code FROM project_contract_info WHERE COALESCE(support_dept_code, '') <> ''
                    UNION
                    SELECT DISTINCT support_dept_code FROM project_billing_entries WHERE COALESCE(support_dept_code, '') <> ''
                    UNION
                    SELECT DISTINCT support_dept_code FROM project_contract_change_round WHERE COALESCE(support_dept_code, '') <> ''
                    """
                )
            ).fetchall()
            if normalize_text(row[0])
        }
        contracted_codes = {
            normalize_text(row[0])
            for row in conn.execute(
                text(
                    """
                    SELECT support_dept_code
                    FROM project_contract_info
                    WHERE COALESCE(hanmac_contract_amount, 0) > 0
                    UNION
                    SELECT support_dept_code
                    FROM project_billing_entries
                    GROUP BY support_dept_code
                    HAVING SUM(COALESCE(collected_amount, 0)) > 0 OR MAX(COALESCE(contract_amount, 0)) > 0
                    """
                )
            ).fetchall()
            if normalize_text(row[0])
        }

    summary_by_title: dict[str, list[dict[str, Any]]] = {}
    for row in summary_rows:
        title_key = normalize_text(row["normalized_title"])
        if title_key:
            summary_by_title.setdefault(title_key, []).append(dict(row))
    latest_summary_by_title = {
        title_key: select_latest_contract_change_entry(rows)
        for title_key, rows in summary_by_title.items()
        if rows
    }

    round_by_code: dict[str, list[dict[str, Any]]] = {}
    round_codes_by_title: dict[str, set[str]] = {}
    for row in round_rows:
        code = normalize_text(row["support_dept_code"])
        title_key = normalize_text(row["normalized_title"])
        if code:
            round_by_code.setdefault(code, []).append(dict(row))
        if title_key and code:
            round_codes_by_title.setdefault(title_key, set()).add(code)
    latest_round_by_code = {
        code: select_latest_contract_change_entry(rows)
        for code, rows in round_by_code.items()
        if rows
    }

    representative_by_title: dict[str, str] = {}
    for title_key, codes in round_codes_by_title.items():
        sorted_codes = sorted(codes)
        if not sorted_codes:
            continue

        def representative_rank(code: str) -> tuple[int, str]:
            if code in contracted_codes:
                return (0, code)
            if code.startswith("Y") and code in existing_codes:
                return (1, code)
            if code.startswith("Z"):
                return (2, code)
            if code.startswith("X"):
                return (3, code)
            return (4, code)

        representative_by_title[title_key] = min(sorted_codes, key=representative_rank)

    title_by_code = {
        code: title_key
        for title_key, codes in round_codes_by_title.items()
        for code in codes
    }
    return latest_summary_by_title, latest_round_by_code, representative_by_title, title_by_code


def refresh_contract_review_tags() -> None:
    latest_summary_by_title, latest_round_by_code, _, title_by_code = get_project_contract_change_maps()
    with engine.begin() as conn:
        billing_rows = conn.execute(
            text(
                """
                SELECT support_dept_code, MAX(contract_amount) AS billing_contract_amount
                FROM project_billing_entries
                GROUP BY support_dept_code
                """
            )
        ).mappings().all()
        billing_map = {
            normalize_text(row["support_dept_code"]): normalize_amount(row["billing_contract_amount"])
            for row in billing_rows
            if normalize_text(row["support_dept_code"])
        }
        contract_rows = conn.execute(
            text("SELECT support_dept_code, support_dept_name, hanmac_contract_amount FROM project_contract_info")
        ).mappings().all()
        for row in contract_rows:
            support_dept_code = normalize_text(row["support_dept_code"])
            hanmac_contract_amount = normalize_amount(row["hanmac_contract_amount"])
            billing_contract_amount = normalize_amount(billing_map.get(support_dept_code))
            title_key = title_by_code.get(support_dept_code) or normalize_project_title_for_linking(row.get("support_dept_name"))
            latest_changed_contract_amount = (
                normalize_amount((latest_summary_by_title.get(title_key) or {}).get("changed_contract_amount"))
                or normalize_amount((latest_round_by_code.get(support_dept_code) or {}).get("changed_contract_amount"))
            )
            comparison_contract_amount = latest_changed_contract_amount or hanmac_contract_amount
            review_tag = ""
            review_note = ""
            if billing_contract_amount and abs(comparison_contract_amount - billing_contract_amount) > 0.5:
                review_tag = "변경계약 검토 필요"
                if latest_changed_contract_amount:
                    review_note = (
                        f"변경계약금액 {latest_changed_contract_amount:,.0f}원 / "
                        f"기성청구현황 계약금액 {billing_contract_amount:,.0f}원"
                    )
                else:
                    review_note = (
                        f"계약현황 한맥계약금액 {hanmac_contract_amount:,.0f}원 / "
                        f"기성청구현황 계약금액 {billing_contract_amount:,.0f}원"
                    )
            conn.execute(
                text(
                    """
                    UPDATE project_contract_info
                    SET review_tag = :review_tag,
                        review_note = :review_note,
                        updated_at = CURRENT_TIMESTAMP
                    WHERE support_dept_code = :support_dept_code
                    """
                ),
                {
                    "support_dept_code": support_dept_code,
                    "review_tag": review_tag,
                    "review_note": review_note,
                },
            )


def sync_auto_project_related_links() -> None:
    with engine.begin() as conn:
        billing_rows = conn.execute(
            text(
                """
                SELECT support_dept_code, raw_project_code, round_code
                FROM project_billing_entries
                WHERE COALESCE(support_dept_code, '') <> ''
                """
            )
        ).mappings().all()
        change_round_rows = conn.execute(
            text(
                """
                SELECT support_dept_code, normalized_title, support_dept_name
                FROM project_contract_change_round
                WHERE COALESCE(support_dept_code, '') <> ''
                """
            )
        ).mappings().all()
        existing_codes = {
            normalize_text(row[0])
            for row in conn.execute(
                text(
                    """
                    SELECT DISTINCT support_dept_code FROM transactions
                    WHERE COALESCE(support_dept_code, '') <> ''
                    UNION
                    SELECT DISTINCT support_dept_code FROM project_status
                    WHERE COALESCE(support_dept_code, '') <> ''
                    UNION
                    SELECT DISTINCT support_dept_code FROM project_contract_info
                    WHERE COALESCE(support_dept_code, '') <> ''
                    UNION
                    SELECT DISTINCT support_dept_code FROM project_billing_entries
                    WHERE COALESCE(support_dept_code, '') <> ''
                    UNION
                    SELECT DISTINCT support_dept_code FROM project_contract_change_round
                    WHERE COALESCE(support_dept_code, '') <> ''
                    """
                )
            ).fetchall()
            if normalize_text(row[0])
        }
        project_names = {
            normalize_text(row[0]): normalize_text(row[1])
            for row in conn.execute(
                text(
                    """
                    WITH project_names AS (
                        SELECT support_dept_code, MAX(support_dept_name) AS support_dept_name
                        FROM transactions
                        WHERE COALESCE(support_dept_code, '') <> ''
                        GROUP BY support_dept_code
                        UNION
                        SELECT support_dept_code, support_dept_name
                        FROM project_contract_info
                        WHERE COALESCE(support_dept_code, '') <> ''
                        UNION
                        SELECT support_dept_code, support_dept_name
                        FROM project_billing_entries
                        WHERE COALESCE(support_dept_code, '') <> ''
                        UNION
                        SELECT support_dept_code, support_dept_name
                        FROM project_contract_change_round
                        WHERE COALESCE(support_dept_code, '') <> ''
                    )
                    SELECT support_dept_code, support_dept_name
                    FROM project_names
                    WHERE COALESCE(support_dept_code, '') <> ''
                    """
                )
            ).fetchall()
            if normalize_text(row[0])
        }
        contracted_codes = {
            normalize_text(row[0])
            for row in conn.execute(
                text(
                    """
                    SELECT support_dept_code
                    FROM project_contract_info
                    WHERE COALESCE(hanmac_contract_amount, 0) > 0
                    UNION
                    SELECT support_dept_code
                    FROM project_billing_entries
                    GROUP BY support_dept_code
                    HAVING SUM(COALESCE(collected_amount, 0)) > 0 OR MAX(COALESCE(contract_amount, 0)) > 0
                    """
                )
            ).fetchall()
            if normalize_text(row[0])
        }

        cluster_map: dict[str, set[str]] = {}
        for row in billing_rows:
            base_code = normalize_text(row["support_dept_code"])
            raw_project_code = normalize_text(row["raw_project_code"])
            round_code = normalize_project_code(row["round_code"], default_prefix=base_code[:1] or "Y")
            if not base_code:
                continue
            cluster_key = raw_project_code or base_code
            cluster = cluster_map.setdefault(cluster_key, set())
            if base_code in existing_codes:
                cluster.add(base_code)
            if round_code and round_code in existing_codes:
                cluster.add(round_code)

        change_round_title_groups: dict[str, set[str]] = {}
        change_contract_codes: set[str] = set()
        for row in change_round_rows:
            code = normalize_text(row["support_dept_code"])
            title_key = normalize_text(row["normalized_title"]) or normalize_project_title_for_linking(row["support_dept_name"])
            if not code or not title_key:
                continue
            change_contract_codes.add(code)
            change_round_title_groups.setdefault(title_key, set()).add(code)

        project_codes_by_title: dict[str, set[str]] = {}
        for code, name in project_names.items():
            title_key = normalize_project_title_for_linking(name)
            if title_key:
                project_codes_by_title.setdefault(title_key, set()).add(code)

        for title_key, round_codes in change_round_title_groups.items():
            cluster_codes = set(round_codes)
            cluster_codes.update(project_codes_by_title.get(title_key, set()))
            cluster_map.setdefault(f"change_round::{title_key}", set()).update(
                code for code in cluster_codes if code in existing_codes
            )

        title_groups: dict[str, set[str]] = {}
        for code, name in project_names.items():
            normalized_title = normalize_project_title_for_linking(name)
            if len(normalized_title) < 8:
                continue
            title_groups.setdefault(normalized_title, set()).add(code)

        def choose_representative(codes: set[str]) -> str:
            sorted_codes = sorted(codes)
            contracted_non_special = [
                code
                for code in sorted_codes
                if code in contracted_codes and not code.startswith(("X", "Z"))
            ]
            if contracted_non_special:
                return contracted_non_special[0]
            plain_non_special = [
                code
                for code in sorted_codes
                if not code.startswith(("X", "Z"))
                and not has_project_variant_marker(project_names.get(code, ""))
            ]
            if plain_non_special:
                return plain_non_special[0]
            contracted_any = [code for code in sorted_codes if code in contracted_codes]
            if contracted_any:
                return contracted_any[0]
            return sorted_codes[0] if sorted_codes else ""

        for normalized_title, codes in title_groups.items():
            if len(codes) < 2:
                continue
            representative_code = choose_representative(codes)
            if not representative_code:
                continue
            variant_codes = {
                code
                for code in codes
                if code != representative_code and (
                    code.startswith(("X", "Z"))
                    or has_project_variant_marker(project_names.get(code, ""))
                )
            }
            if not variant_codes:
                continue
            cluster_map.setdefault(f"title::{normalized_title}", set()).update({representative_code, *variant_codes})

        title_items = sorted(
            ((title_key, set(codes)) for title_key, codes in title_groups.items() if len(title_key) >= 8),
            key=lambda item: len(item[0]),
        )
        for index, (base_title, base_codes) in enumerate(title_items):
            for related_title, related_codes in title_items[index + 1:]:
                if base_title not in related_title and related_title not in base_title:
                    continue
                merged_codes = set(base_codes) | set(related_codes)
                representative_code = choose_representative(merged_codes)
                if not representative_code:
                    continue
                variant_codes = {
                    code
                    for code in merged_codes
                    if code != representative_code and (
                        code.startswith(("X", "Z"))
                        or has_project_variant_marker(project_names.get(code, ""))
                    )
                }
                if not variant_codes:
                    continue
                cluster_map.setdefault(
                    f"title_fuzzy::{base_title if len(base_title) <= len(related_title) else related_title}",
                    set(),
                ).update({representative_code, *variant_codes})

        conn.execute(text("DELETE FROM project_related_links WHERE COALESCE(link_source, 'manual') LIKE 'auto%'"))
        for cluster_key, cluster_codes in cluster_map.items():
            normalized_cluster = sorted(cluster_codes)
            if len(normalized_cluster) < 2:
                continue
            if str(cluster_key).startswith("change_round::"):
                link_source = "auto_change_contract"
            elif str(cluster_key).startswith("title_fuzzy::"):
                link_source = "auto_title_fuzzy"
            elif str(cluster_key).startswith("title::"):
                link_source = "auto_title"
            else:
                link_source = "auto_round"
            for base_code in normalized_cluster:
                for related_code in normalized_cluster:
                    if base_code == related_code:
                        continue
                    conn.execute(
                        text(
                            """
                            INSERT INTO project_related_links (
                                base_support_dept_code,
                                related_support_dept_code,
                                link_source,
                                updated_at
                            ) VALUES (
                                :base_support_dept_code,
                                :related_support_dept_code,
                                :link_source,
                                CURRENT_TIMESTAMP
                            )
                            ON CONFLICT(base_support_dept_code, related_support_dept_code) DO UPDATE SET
                                link_source = excluded.link_source,
                                updated_at = CURRENT_TIMESTAMP
                            """
                        ),
                        {
                            "base_support_dept_code": base_code,
                            "related_support_dept_code": related_code,
                            "link_source": link_source,
                        },
                    )

        for base_code, related_code, link_source in conn.execute(
            text(
                """
                SELECT base_support_dept_code, related_support_dept_code, COALESCE(link_source, 'manual') AS link_source
                FROM project_related_links
                WHERE COALESCE(link_source, 'manual') LIKE 'auto_title%'
                """
            )
        ).fetchall():
            normalized_base = normalize_text(base_code)
            normalized_related = normalize_text(related_code)
            if (
                normalized_base in change_contract_codes
                and normalized_related not in change_contract_codes
                and not normalized_related.startswith("X")
            ) or (
                normalized_related in change_contract_codes
                and normalized_base not in change_contract_codes
                and not normalized_base.startswith("X")
            ):
                conn.execute(
                    text(
                        """
                        DELETE FROM project_related_links
                        WHERE base_support_dept_code = :base_support_dept_code
                          AND related_support_dept_code = :related_support_dept_code
                        """
                    ),
                    {
                        "base_support_dept_code": normalized_base,
                        "related_support_dept_code": normalized_related,
                    },
                )

        x_codes_linked_to_change: set[str] = set()
        for base_code, related_code in conn.execute(
            text(
                """
                SELECT base_support_dept_code, related_support_dept_code
                FROM project_related_links
                WHERE (
                    base_support_dept_code LIKE 'X%' AND related_support_dept_code <> ''
                ) OR (
                    related_support_dept_code LIKE 'X%' AND base_support_dept_code <> ''
                )
                """
            )
        ).fetchall():
            normalized_base = normalize_text(base_code)
            normalized_related = normalize_text(related_code)
            if normalized_base.startswith("X") and normalized_related in change_contract_codes:
                x_codes_linked_to_change.add(normalized_base)
            if normalized_related.startswith("X") and normalized_base in change_contract_codes:
                x_codes_linked_to_change.add(normalized_related)

        for base_code, related_code in conn.execute(
            text(
                """
                SELECT base_support_dept_code, related_support_dept_code
                FROM project_related_links
                WHERE COALESCE(link_source, 'manual') LIKE 'auto_title%'
                """
            )
        ).fetchall():
            normalized_base = normalize_text(base_code)
            normalized_related = normalize_text(related_code)
            should_delete = (
                normalized_base in x_codes_linked_to_change
                and normalized_related not in change_contract_codes
                and not normalized_related.startswith("X")
            ) or (
                normalized_related in x_codes_linked_to_change
                and normalized_base not in change_contract_codes
                and not normalized_base.startswith("X")
            )
            if should_delete:
                conn.execute(
                    text(
                        """
                        DELETE FROM project_related_links
                        WHERE base_support_dept_code = :base_support_dept_code
                          AND related_support_dept_code = :related_support_dept_code
                        """
                    ),
                    {
                        "base_support_dept_code": normalized_base,
                        "related_support_dept_code": normalized_related,
                    },
                )


@app.on_event("startup")
def on_startup() -> None:
    init_db()
    auto_import_project_excels()
    sync_auto_project_related_links()
    normalize_all_collection_entry_storage()
    logger.info("DB ready at %s", DB_PATH)


def normalize_text(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    return str(value).strip()


def normalize_amount(value: Any) -> float:
    if value in (None, ""):
        return 0.0
    if isinstance(value, (int, float)):
        return float(value)
    cleaned = (
        str(value)
        .strip()
        .replace(",", "")
        .replace("원", "")
        .replace("(", "-")
        .replace(")", "")
    )
    if not cleaned:
        return 0.0
    try:
        return float(cleaned)
    except ValueError:
        return 0.0


def normalize_date_text(value: Any) -> str:
    if value in (None, ""):
        return ""
    if isinstance(value, datetime):
        return value.date().isoformat()
    if isinstance(value, date):
        return value.isoformat()
    text_value = normalize_text(value)
    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y%m%d"):
        try:
            return datetime.strptime(text_value, fmt).date().isoformat()
        except ValueError:
            continue
    return text_value


def normalize_collection_progress_type(value: Any) -> str:
    text_value = normalize_text(value)
    if not text_value:
        return ""
    if "준공" in text_value:
        return "준공금"
    if any(keyword in text_value for keyword in ("선수", "선급")):
        return "선급금"
    if "기성" in text_value:
        return "기성금"
    return ""


def normalize_collection_billing_type(value: Any) -> str:
    text_value = normalize_text(value)
    if not text_value:
        return ""
    if "계약" in text_value:
        return "계약분"
    if "기타" in text_value:
        return "기타"
    return ""


def normalize_collection_entry_fields(row: dict[str, Any]) -> dict[str, Any]:
    normalized = dict(row)
    progress_type = normalize_collection_progress_type(normalized.get("progress_type"))
    raw_billing_type = normalize_text(normalized.get("billing_type"))
    billing_type = normalize_collection_billing_type(raw_billing_type)
    if not progress_type and raw_billing_type in {"선수금", "선급금", "기성금", "준공금"}:
        progress_type = normalize_collection_progress_type(raw_billing_type)
    normalized["progress_type"] = progress_type
    normalized["billing_type"] = billing_type or ("계약분" if progress_type else "")
    return normalized


def normalize_project_code(value: Any, default_prefix: str = "Y") -> str:
    text_value = normalize_text(value).replace("\u3164", "").replace("\xa0", "")
    prefix = ""
    for character in text_value:
        if character.isalpha():
            prefix = character.upper()
            break
    digits = "".join(character for character in text_value if character.isdigit())
    if not digits:
        return ""
    return f"{prefix or default_prefix}{int(digits)}"


def normalize_round_value(value: Any) -> str:
    text_value = normalize_text(value)
    if not text_value:
        return ""
    digits = "".join(character for character in text_value if character.isdigit())
    if digits:
        return str(int(digits))
    return text_value


PROJECT_TITLE_LINK_STRIP_PATTERNS = (
    r"\((?:\d+\s*차|[가-힣A-Za-z0-9\s]*변경[가-힣A-Za-z0-9\s]*|[가-힣A-Za-z0-9\s]*보완[가-힣A-Za-z0-9\s]*|[가-힣A-Za-z0-9\s]*입찰[가-힣A-Za-z0-9\s]*|가칭)\)",
    r"\d+\s*차",
    r"변경",
    r"보완",
    r"입찰",
    r"기술제안",
    r"실시설계",
    r"조사[·ㆍ]설계",
    r"조사설계",
    r"기본\s*및\s*실시설계",
    r"기본및실시설계",
    r"기본설계",
    r"설계",
    r"용역",
    r"시공단계",
    r"건설사업관리",
    r"가칭",
)


def normalize_project_title_for_linking(value: Any) -> str:
    text_value = normalize_text(value)
    if not text_value:
        return ""
    normalized = text_value
    for pattern in PROJECT_TITLE_LINK_STRIP_PATTERNS:
        normalized = re.sub(pattern, "", normalized, flags=re.IGNORECASE)
    normalized = re.sub(r"[^가-힣A-Za-z0-9]", "", normalized)
    return normalized


def has_project_variant_marker(value: Any) -> bool:
    text_value = normalize_text(value)
    if not text_value:
        return False
    return bool(re.search(r"(\d+\s*차|[nN]\s*차|변경|보완|지연보상금|가칭|연차|년분)", text_value))


def classify_special_x_project(name: Any) -> str:
    normalized_name = normalize_text(name)
    if not normalized_name:
        return "precontract"
    rules = get_special_x_classification_rules()
    for category_key, keywords in rules.items():
        if any(keyword in normalized_name for keyword in keywords):
            return category_key
    return "precontract"


def decode_json_rows(value: Any) -> list[dict[str, Any]]:
    text_value = normalize_text(value)
    if not text_value:
        return []
    try:
        rows = json.loads(text_value)
    except json.JSONDecodeError:
        return []
    return [row for row in rows if isinstance(row, dict)]


def encode_json_rows(rows: list[dict[str, Any]]) -> str:
    return json.dumps(rows, ensure_ascii=False)


def normalize_collection_entry_row(row: dict[str, Any]) -> dict[str, Any]:
    normalized = {
        "vendor": clean_row_text(row.get("vendor")),
        "progress_type": clean_row_text(row.get("progress_type")),
        "billing_round": normalize_round_value(row.get("billing_round")),
        "billing_type": clean_row_text(row.get("billing_type")),
        "billing_date": normalize_date_text(row.get("billing_date")),
        "billed_amount": normalize_amount(row.get("billed_amount")),
        "round": normalize_round_value(row.get("round")),
        "date": normalize_date_text(row.get("date")),
        "due_date": normalize_date_text(row.get("due_date")),
        "amount": normalize_amount(row.get("amount")),
        "balance_amount": normalize_amount(row.get("balance_amount")),
        "collection_rate": normalize_amount(row.get("collection_rate")),
        "note": clean_row_text(row.get("note")),
    }
    return normalize_collection_entry_fields(normalized)


def normalize_task_plan_entry_row(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "group": clean_row_text(row.get("group")),
        "dept_name": clean_row_text(row.get("dept_name")),
        "work_name": clean_row_text(row.get("work_name")),
        "amount": normalize_amount(row.get("amount")),
    }


def normalize_exec_budget_entry_row(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "group": clean_row_text(row.get("group")),
        "grade": clean_row_text(row.get("grade")),
        "hours": clean_row_text(row.get("hours")),
        "dept_name": clean_row_text(row.get("dept_name")),
        "work_name": clean_row_text(row.get("work_name")),
        "account_code": clean_row_text(row.get("account_code")),
        "account_name": clean_row_text(row.get("account_name")),
        "amount": normalize_amount(row.get("amount")),
    }


def normalize_actual_input_entry_row(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "group": clean_row_text(row.get("group")),
        "grade": clean_row_text(row.get("grade")),
        "minutes": clean_row_text(row.get("minutes")),
        "label": clean_row_text(row.get("label")),
        "reference": clean_row_text(row.get("reference")),
        "note": clean_row_text(row.get("note")),
        "amount": normalize_amount(row.get("amount")),
    }


def extract_project_status_entry_sets(row: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
    collection_entries = [normalize_collection_entry_row(item) for item in decode_json_rows(row.get("collection_entries_json"))]
    task_plan_entries = [normalize_task_plan_entry_row(item) for item in decode_json_rows(row.get("task_plan_entries_json"))]
    exec_budget_entries = [normalize_exec_budget_entry_row(item) for item in decode_json_rows(row.get("exec_budget_entries_json"))]
    actual_input_entries = [normalize_actual_input_entry_row(item) for item in decode_json_rows(row.get("actual_input_entries_json"))]

    if not collection_entries and normalize_amount(row.get("collection_amount")):
        collection_entries = [
            normalize_collection_entry_row(
                {
                    "vendor": "",
                    "round": "",
                    "amount": row.get("collection_amount", ""),
                    "date": "",
                    "due_date": "",
                    "note": "기존 수기 입력값",
                }
            )
        ]

    if not task_plan_entries:
        fallback_task_rows = []
        if normalize_amount(row.get("task_plan_department_budget")):
            fallback_task_rows.append(
                {
                    "group": "department",
                    "dept_name": "기존 부서별 배분",
                    "work_name": "",
                    "amount": row.get("task_plan_department_budget", ""),
                }
            )
        if normalize_amount(row.get("task_plan_outsource_budget")):
            fallback_task_rows.append(
                {
                    "group": "outsource",
                    "dept_name": "기존 외주비",
                    "work_name": row.get("task_plan_outsource_detail", ""),
                    "amount": row.get("task_plan_outsource_budget", ""),
                }
            )
        if normalize_amount(row.get("task_plan_joint_operating_cost")):
            fallback_task_rows.append(
                {
                    "group": "joint",
                    "dept_name": "기존 합사운영비",
                    "work_name": "",
                    "amount": row.get("task_plan_joint_operating_cost", ""),
                }
            )
        task_plan_entries = [normalize_task_plan_entry_row(item) for item in fallback_task_rows]

    if not exec_budget_entries:
        fallback_exec_rows = []
        if normalize_amount(row.get("exec_budget_labor_by_grade")):
            fallback_exec_rows.append(
                {
                    "group": "labor",
                    "grade": "기존 인건비",
                    "hours": "",
                    "amount": row.get("exec_budget_labor_by_grade", ""),
                }
            )
        if normalize_amount(row.get("exec_budget_outsource")):
            fallback_exec_rows.append(
                {
                    "group": "outsource",
                    "dept_name": "기존 외주비",
                    "work_name": "",
                    "amount": row.get("exec_budget_outsource", ""),
                }
            )
        if normalize_amount(row.get("exec_budget_cost_plan")):
            fallback_exec_rows.append(
                {
                    "group": "cost_plan",
                    "account_code": "기존",
                    "account_name": "비용계획",
                    "amount": row.get("exec_budget_cost_plan", ""),
                }
            )
        exec_budget_entries = [normalize_exec_budget_entry_row(item) for item in fallback_exec_rows]

    if not actual_input_entries and normalize_amount(row.get("item_investment")):
        actual_input_entries = [
            normalize_actual_input_entry_row(
                {
                    "reference": "",
                    "amount": row.get("item_investment", ""),
                    "note": "기존 항목별투입액",
                }
            )
        ]

    return {
        "collection_entries": collection_entries,
        "task_plan_entries": task_plan_entries,
        "exec_budget_entries": exec_budget_entries,
        "actual_input_entries": actual_input_entries,
    }


def replace_project_status_child_entries(
    conn: Any,
    support_dept_code: str,
    collection_entries: list[dict[str, Any]],
    task_plan_entries: list[dict[str, Any]],
    exec_budget_entries: list[dict[str, Any]],
    actual_input_entries: list[dict[str, Any]],
) -> None:
    conn.execute(
        text("DELETE FROM project_collection_entries WHERE support_dept_code = :support_dept_code"),
        {"support_dept_code": support_dept_code},
    )
    conn.execute(
        text("DELETE FROM project_task_plan_entries WHERE support_dept_code = :support_dept_code"),
        {"support_dept_code": support_dept_code},
    )
    conn.execute(
        text("DELETE FROM project_exec_budget_entries WHERE support_dept_code = :support_dept_code"),
        {"support_dept_code": support_dept_code},
    )
    conn.execute(
        text("DELETE FROM project_actual_input_entries WHERE support_dept_code = :support_dept_code"),
        {"support_dept_code": support_dept_code},
    )

    for position, row in enumerate(collection_entries):
        normalized = normalize_collection_entry_row(row)
        conn.execute(
            text(
                """
                INSERT INTO project_collection_entries (
                    support_dept_code, position, vendor, progress_type, billing_round,
                    billing_type, billing_date, billed_amount, round, date, due_date,
                    amount, balance_amount, collection_rate, note, updated_at
                ) VALUES (
                    :support_dept_code, :position, :vendor, :progress_type, :billing_round,
                    :billing_type, :billing_date, :billed_amount, :round, :date, :due_date,
                    :amount, :balance_amount, :collection_rate, :note, CURRENT_TIMESTAMP
                )
                """
            ),
            {"support_dept_code": support_dept_code, "position": position, **normalized},
        )

    for position, row in enumerate(task_plan_entries):
        normalized = normalize_task_plan_entry_row(row)
        conn.execute(
            text(
                """
                INSERT INTO project_task_plan_entries (
                    support_dept_code, position, group_name, dept_name, work_name, amount, updated_at
                ) VALUES (
                    :support_dept_code, :position, :group_name, :dept_name, :work_name, :amount, CURRENT_TIMESTAMP
                )
                """
            ),
            {
                "support_dept_code": support_dept_code,
                "position": position,
                "group_name": normalized["group"],
                "dept_name": normalized["dept_name"],
                "work_name": normalized["work_name"],
                "amount": normalized["amount"],
            },
        )

    for position, row in enumerate(exec_budget_entries):
        normalized = normalize_exec_budget_entry_row(row)
        conn.execute(
            text(
                """
                INSERT INTO project_exec_budget_entries (
                    support_dept_code, position, group_name, grade, hours, dept_name,
                    work_name, account_code, account_name, amount, updated_at
                ) VALUES (
                    :support_dept_code, :position, :group_name, :grade, :hours, :dept_name,
                    :work_name, :account_code, :account_name, :amount, CURRENT_TIMESTAMP
                )
                """
            ),
            {
                "support_dept_code": support_dept_code,
                "position": position,
                "group_name": normalized["group"],
                "grade": normalized["grade"],
                "hours": normalized["hours"],
                "dept_name": normalized["dept_name"],
                "work_name": normalized["work_name"],
                "account_code": normalized["account_code"],
                "account_name": normalized["account_name"],
                "amount": normalized["amount"],
            },
        )

    for position, row in enumerate(actual_input_entries):
        normalized = normalize_actual_input_entry_row(row)
        conn.execute(
            text(
                """
                INSERT INTO project_actual_input_entries (
                    support_dept_code, position, group_name, grade, minutes, label,
                    reference, note, amount, updated_at
                ) VALUES (
                    :support_dept_code, :position, :group_name, :grade, :minutes, :label,
                    :reference, :note, :amount, CURRENT_TIMESTAMP
                )
                """
            ),
            {
                "support_dept_code": support_dept_code,
                "position": position,
                "group_name": normalized["group"],
                "grade": normalized["grade"],
                "minutes": normalized["minutes"],
                "label": normalized["label"],
                "reference": normalized["reference"],
                "note": normalized["note"],
                "amount": normalized["amount"],
            },
        )


def load_project_status_entry_maps(conn: Any) -> dict[str, dict[str, list[dict[str, Any]]]]:
    result: dict[str, dict[str, list[dict[str, Any]]]] = {}

    collection_rows = conn.execute(
        text(
            """
            SELECT support_dept_code, position, vendor, progress_type, billing_round,
                   billing_type, billing_date, billed_amount, round, date, due_date,
                   amount, balance_amount, collection_rate, note
            FROM project_collection_entries
            ORDER BY support_dept_code, position, id
            """
        )
    ).mappings().all()
    for row in collection_rows:
        code = normalize_text(row["support_dept_code"])
        result.setdefault(code, {})["collection_entries"] = result.setdefault(code, {}).get("collection_entries", [])
        result[code]["collection_entries"].append(
            normalize_collection_entry_row({key: row[key] for key in row.keys() if key not in {"support_dept_code", "position"}})
        )

    task_rows = conn.execute(
        text(
            """
            SELECT support_dept_code, position, group_name, dept_name, work_name, amount
            FROM project_task_plan_entries
            ORDER BY support_dept_code, position, id
            """
        )
    ).mappings().all()
    for row in task_rows:
        code = normalize_text(row["support_dept_code"])
        result.setdefault(code, {})["task_plan_entries"] = result.setdefault(code, {}).get("task_plan_entries", [])
        result[code]["task_plan_entries"].append(
            normalize_task_plan_entry_row(
                {
                    "group": row["group_name"],
                    "dept_name": row["dept_name"],
                    "work_name": row["work_name"],
                    "amount": row["amount"],
                }
            )
        )

    exec_rows = conn.execute(
        text(
            """
            SELECT support_dept_code, position, group_name, grade, hours, dept_name,
                   work_name, account_code, account_name, amount
            FROM project_exec_budget_entries
            ORDER BY support_dept_code, position, id
            """
        )
    ).mappings().all()
    for row in exec_rows:
        code = normalize_text(row["support_dept_code"])
        result.setdefault(code, {})["exec_budget_entries"] = result.setdefault(code, {}).get("exec_budget_entries", [])
        result[code]["exec_budget_entries"].append(
            normalize_exec_budget_entry_row(
                {
                    "group": row["group_name"],
                    "grade": row["grade"],
                    "hours": row["hours"],
                    "dept_name": row["dept_name"],
                    "work_name": row["work_name"],
                    "account_code": row["account_code"],
                    "account_name": row["account_name"],
                    "amount": row["amount"],
                }
            )
        )

    actual_rows = conn.execute(
        text(
            """
            SELECT support_dept_code, position, group_name, grade, minutes, label,
                   reference, note, amount
            FROM project_actual_input_entries
            ORDER BY support_dept_code, position, id
            """
        )
    ).mappings().all()
    for row in actual_rows:
        code = normalize_text(row["support_dept_code"])
        result.setdefault(code, {})["actual_input_entries"] = result.setdefault(code, {}).get("actual_input_entries", [])
        result[code]["actual_input_entries"].append(
            normalize_actual_input_entry_row(
                {
                    "group": row["group_name"],
                    "grade": row["grade"],
                    "minutes": row["minutes"],
                    "label": row["label"],
                    "reference": row["reference"],
                    "note": row["note"],
                    "amount": row["amount"],
                }
            )
        )

    return result


def ensure_project_entry_set(entry_set: dict[str, list[dict[str, Any]]] | None) -> dict[str, list[dict[str, Any]]]:
    source = entry_set or {}
    return {
        "collection_entries": list(source.get("collection_entries", [])),
        "task_plan_entries": list(source.get("task_plan_entries", [])),
        "exec_budget_entries": list(source.get("exec_budget_entries", [])),
        "actual_input_entries": list(source.get("actual_input_entries", [])),
    }


def migrate_project_status_entries(conn: Any) -> None:
    migrated_codes = {
        normalize_text(row[0])
        for row in conn.execute(
            text(
                """
                SELECT DISTINCT support_dept_code
                FROM (
                    SELECT support_dept_code FROM project_collection_entries
                    UNION ALL
                    SELECT support_dept_code FROM project_task_plan_entries
                    UNION ALL
                    SELECT support_dept_code FROM project_exec_budget_entries
                    UNION ALL
                    SELECT support_dept_code FROM project_actual_input_entries
                )
                WHERE COALESCE(support_dept_code, '') <> ''
                """
            )
        ).fetchall()
        if normalize_text(row[0])
    }
    source_rows = conn.execute(
        text(
            """
            SELECT support_dept_code,
                   collection_entries_json,
                   task_plan_entries_json,
                   exec_budget_entries_json,
                   actual_input_entries_json,
                   collection_amount,
                   task_plan_department_budget,
                   task_plan_outsource_budget,
                   task_plan_outsource_detail,
                   task_plan_joint_operating_cost,
                   exec_budget_labor_by_grade,
                   exec_budget_outsource,
                   exec_budget_cost_plan,
                   item_investment
            FROM project_status
            WHERE COALESCE(support_dept_code, '') <> ''
            """
        )
    ).mappings().all()
    for row in source_rows:
        support_dept_code = normalize_text(row["support_dept_code"])
        if not support_dept_code or support_dept_code in migrated_codes:
            continue
        entry_sets = extract_project_status_entry_sets(dict(row))
        replace_project_status_child_entries(
            conn,
            support_dept_code,
            entry_sets["collection_entries"],
            entry_sets["task_plan_entries"],
            entry_sets["exec_budget_entries"],
            entry_sets["actual_input_entries"],
        )


def migrate_project_basic_info(conn: Any) -> None:
    existing_codes = {
        normalize_text(row[0])
        for row in conn.execute(
            text("SELECT support_dept_code FROM project_basic_info WHERE COALESCE(support_dept_code, '') <> ''")
        ).fetchall()
        if normalize_text(row[0])
    }
    source_rows = conn.execute(
        text(
            """
            SELECT support_dept_code, support_dept_name, contract_amount, project_type,
                   expected_as_rate, expected_sga_rate, expected_as_cost, expected_sga_budget,
                   exec_labor_rates_json, change_round, project_start_date, project_end_date,
                   completion_status, notes, last_editor_session_id, last_client_submitted_at
            FROM project_status
            WHERE COALESCE(support_dept_code, '') <> ''
            """
        )
    ).mappings().all()
    for row in source_rows:
        support_dept_code = normalize_text(row["support_dept_code"])
        if not support_dept_code or support_dept_code in existing_codes:
            continue
        conn.execute(
            text(
                """
                INSERT INTO project_basic_info (
                    support_dept_code, support_dept_name, contract_amount, project_type,
                    expected_as_rate, expected_sga_rate, expected_as_cost, expected_sga_budget,
                    exec_labor_rates_json, change_round, project_start_date, project_end_date,
                    completion_status, notes, last_editor_session_id, last_client_submitted_at,
                    updated_at
                ) VALUES (
                    :support_dept_code, :support_dept_name, :contract_amount, :project_type,
                    :expected_as_rate, :expected_sga_rate, :expected_as_cost, :expected_sga_budget,
                    :exec_labor_rates_json, :change_round, :project_start_date, :project_end_date,
                    :completion_status, :notes, :last_editor_session_id, :last_client_submitted_at,
                    CURRENT_TIMESTAMP
                )
                """
            ),
            dict(row),
        )


def save_project_basic_info_section(conn: Any, payload: dict[str, Any]) -> None:
    conn.execute(
        text(
            """
            INSERT INTO project_basic_info (
                support_dept_code, support_dept_name, contract_amount, project_type,
                expected_as_rate, expected_sga_rate, expected_as_cost, expected_sga_budget,
                exec_labor_rates_json, change_round, project_start_date, project_end_date,
                completion_status, notes, last_editor_session_id, last_client_submitted_at,
                updated_at
            ) VALUES (
                :support_dept_code, :support_dept_name, :contract_amount, :project_type,
                :expected_as_rate, :expected_sga_rate, :expected_as_cost, :expected_sga_budget,
                :exec_labor_rates_json, :change_round, :project_start_date, :project_end_date,
                :completion_status, :notes, :last_editor_session_id, :last_client_submitted_at,
                CURRENT_TIMESTAMP
            )
            ON CONFLICT(support_dept_code) DO UPDATE SET
                support_dept_name = excluded.support_dept_name,
                contract_amount = excluded.contract_amount,
                project_type = excluded.project_type,
                expected_as_rate = excluded.expected_as_rate,
                expected_sga_rate = excluded.expected_sga_rate,
                expected_as_cost = excluded.expected_as_cost,
                expected_sga_budget = excluded.expected_sga_budget,
                exec_labor_rates_json = excluded.exec_labor_rates_json,
                change_round = excluded.change_round,
                project_start_date = excluded.project_start_date,
                project_end_date = excluded.project_end_date,
                completion_status = excluded.completion_status,
                notes = excluded.notes,
                last_editor_session_id = excluded.last_editor_session_id,
                last_client_submitted_at = excluded.last_client_submitted_at,
                updated_at = CURRENT_TIMESTAMP
            """
        ),
        payload,
    )


def load_project_status_entries_for_code(conn: Any, support_dept_code: str) -> dict[str, list[dict[str, Any]]]:
    code = normalize_text(support_dept_code)
    entry_map = ensure_project_entry_set(None)
    if not code:
        return entry_map

    collection_rows = conn.execute(
        text(
            """
            SELECT vendor, progress_type, billing_round, billing_type, billing_date,
                   billed_amount, round, date, due_date, amount, balance_amount,
                   collection_rate, note
            FROM project_collection_entries
            WHERE support_dept_code = :support_dept_code
            ORDER BY position, id
            """
        ),
        {"support_dept_code": code},
    ).mappings().all()
    entry_map["collection_entries"] = [
        normalize_collection_entry_row(dict(row))
        for row in collection_rows
    ]

    task_rows = conn.execute(
        text(
            """
            SELECT group_name, dept_name, work_name, amount
            FROM project_task_plan_entries
            WHERE support_dept_code = :support_dept_code
            ORDER BY position, id
            """
        ),
        {"support_dept_code": code},
    ).mappings().all()
    entry_map["task_plan_entries"] = [
        normalize_task_plan_entry_row(
            {
                "group": row["group_name"],
                "dept_name": row["dept_name"],
                "work_name": row["work_name"],
                "amount": row["amount"],
            }
        )
        for row in task_rows
    ]

    exec_rows = conn.execute(
        text(
            """
            SELECT group_name, grade, hours, dept_name, work_name, account_code, account_name, amount
            FROM project_exec_budget_entries
            WHERE support_dept_code = :support_dept_code
            ORDER BY position, id
            """
        ),
        {"support_dept_code": code},
    ).mappings().all()
    entry_map["exec_budget_entries"] = [
        normalize_exec_budget_entry_row(
            {
                "group": row["group_name"],
                "grade": row["grade"],
                "hours": row["hours"],
                "dept_name": row["dept_name"],
                "work_name": row["work_name"],
                "account_code": row["account_code"],
                "account_name": row["account_name"],
                "amount": row["amount"],
            }
        )
        for row in exec_rows
    ]

    actual_rows = conn.execute(
        text(
            """
            SELECT group_name, grade, minutes, label, reference, note, amount
            FROM project_actual_input_entries
            WHERE support_dept_code = :support_dept_code
            ORDER BY position, id
            """
        ),
        {"support_dept_code": code},
    ).mappings().all()
    entry_map["actual_input_entries"] = [
        normalize_actual_input_entry_row(
            {
                "group": row["group_name"],
                "grade": row["grade"],
                "minutes": row["minutes"],
                "label": row["label"],
                "reference": row["reference"],
                "note": row["note"],
                "amount": row["amount"],
            }
        )
        for row in actual_rows
    ]
    return entry_map


def sync_project_status_cache_row(conn: Any, support_dept_code: str) -> None:
    code = normalize_text(support_dept_code)
    if not code:
        return
    basic_info = conn.execute(
        text("SELECT * FROM project_basic_info WHERE support_dept_code = :support_dept_code"),
        {"support_dept_code": code},
    ).mappings().first()
    entry_set = load_project_status_entries_for_code(conn, code)
    collection_entries = entry_set["collection_entries"]
    task_plan_entries = entry_set["task_plan_entries"]
    exec_budget_entries = entry_set["exec_budget_entries"]
    actual_input_entries = entry_set["actual_input_entries"]

    contract_amount = normalize_amount((basic_info or {}).get("contract_amount"))
    collection_amount = sum_row_amounts(collection_entries)
    progress_rate = (collection_amount / contract_amount * 100) if contract_amount else 0.0

    task_plan_department_rows = [row for row in task_plan_entries if normalize_text(row.get("group")) == "department"]
    task_plan_outsource_rows = [row for row in task_plan_entries if normalize_text(row.get("group")) == "outsource"]
    task_plan_joint_rows = [row for row in task_plan_entries if normalize_text(row.get("group")) == "joint"]
    exec_labor_rows = [row for row in exec_budget_entries if normalize_text(row.get("group")) == "labor"]
    exec_outsource_rows = [row for row in exec_budget_entries if normalize_text(row.get("group")) == "outsource"]
    exec_cost_plan_rows = [row for row in exec_budget_entries if normalize_text(row.get("group")) == "cost_plan"]
    item_investment = sum_row_amounts(actual_input_entries)

    support_name = normalize_text((basic_info or {}).get("support_dept_name"))
    if not support_name:
        support_name = normalize_text(
            conn.execute(
                text(
                    """
                    SELECT support_dept_name
                    FROM (
                        SELECT support_dept_name, 1 AS priority FROM project_contract_info WHERE support_dept_code = :support_dept_code
                        UNION ALL
                        SELECT support_dept_name, 2 AS priority FROM project_billing_entries WHERE support_dept_code = :support_dept_code
                        UNION ALL
                        SELECT support_dept_name, 3 AS priority FROM transactions WHERE support_dept_code = :support_dept_code
                    )
                    WHERE COALESCE(support_dept_name, '') <> ''
                    ORDER BY priority
                    LIMIT 1
                    """
                ),
                {"support_dept_code": code},
            ).scalar()
        )

    payload = {
        "support_dept_code": code,
        "support_dept_name": support_name,
        "progress_rate": progress_rate,
        "contract_amount": contract_amount,
        "collection_amount": collection_amount,
        "collection_entries_json": encode_json_rows(collection_entries),
        "change_round": normalize_text((basic_info or {}).get("change_round")),
        "item_investment": item_investment,
        "task_plan_department_budget": sum_row_amounts(task_plan_department_rows),
        "task_plan_outsource_budget": sum_row_amounts(task_plan_outsource_rows),
        "task_plan_outsource_detail": "\n".join(
            f"{normalize_text(row.get('dept_name'))} / {normalize_text(row.get('work_name'))}: {format_amount_for_text(row.get('amount'))}".strip(" /:")
            for row in task_plan_outsource_rows
        ),
        "task_plan_joint_operating_cost": sum_row_amounts(task_plan_joint_rows),
        "task_plan_entries_json": encode_json_rows(task_plan_entries),
        "exec_budget_labor_by_grade": sum_row_amounts(exec_labor_rows),
        "exec_labor_rates_json": normalize_text((basic_info or {}).get("exec_labor_rates_json")) or "{}",
        "exec_budget_outsource": sum_row_amounts(exec_outsource_rows),
        "exec_budget_cost_plan": sum_row_amounts(exec_cost_plan_rows),
        "exec_budget_entries_json": encode_json_rows(exec_budget_entries),
        "actual_input_entries_json": encode_json_rows(actual_input_entries),
        "project_type": normalize_text((basic_info or {}).get("project_type")),
        "expected_as_rate": normalize_amount((basic_info or {}).get("expected_as_rate")),
        "expected_sga_rate": normalize_amount((basic_info or {}).get("expected_sga_rate")),
        "expected_as_cost": normalize_amount((basic_info or {}).get("expected_as_cost")),
        "expected_sga_budget": normalize_amount((basic_info or {}).get("expected_sga_budget")),
        "last_editor_session_id": normalize_text((basic_info or {}).get("last_editor_session_id")),
        "last_client_submitted_at": normalize_text((basic_info or {}).get("last_client_submitted_at")),
        "project_start_date": normalize_text((basic_info or {}).get("project_start_date")),
        "project_end_date": normalize_text((basic_info or {}).get("project_end_date")),
        "completion_status": normalize_text((basic_info or {}).get("completion_status")),
        "notes": normalize_text((basic_info or {}).get("notes")),
    }
    conn.execute(
        text(
            """
            INSERT INTO project_status (
                support_dept_code, support_dept_name, progress_rate, contract_amount,
                collection_amount, collection_entries_json, change_round, item_investment,
                task_plan_department_budget, task_plan_outsource_budget, task_plan_outsource_detail,
                task_plan_joint_operating_cost, task_plan_entries_json, exec_budget_labor_by_grade,
                exec_labor_rates_json, exec_budget_outsource, exec_budget_cost_plan,
                exec_budget_entries_json, actual_input_entries_json, project_type,
                expected_as_rate, expected_sga_rate, expected_as_cost, expected_sga_budget,
                last_editor_session_id, last_client_submitted_at, project_start_date,
                project_end_date, completion_status, notes, updated_at
            ) VALUES (
                :support_dept_code, :support_dept_name, :progress_rate, :contract_amount,
                :collection_amount, :collection_entries_json, :change_round, :item_investment,
                :task_plan_department_budget, :task_plan_outsource_budget, :task_plan_outsource_detail,
                :task_plan_joint_operating_cost, :task_plan_entries_json, :exec_budget_labor_by_grade,
                :exec_labor_rates_json, :exec_budget_outsource, :exec_budget_cost_plan,
                :exec_budget_entries_json, :actual_input_entries_json, :project_type,
                :expected_as_rate, :expected_sga_rate, :expected_as_cost, :expected_sga_budget,
                :last_editor_session_id, :last_client_submitted_at, :project_start_date,
                :project_end_date, :completion_status, :notes, CURRENT_TIMESTAMP
            )
            ON CONFLICT(support_dept_code) DO UPDATE SET
                support_dept_name = excluded.support_dept_name,
                progress_rate = excluded.progress_rate,
                contract_amount = excluded.contract_amount,
                collection_amount = excluded.collection_amount,
                collection_entries_json = excluded.collection_entries_json,
                change_round = excluded.change_round,
                item_investment = excluded.item_investment,
                task_plan_department_budget = excluded.task_plan_department_budget,
                task_plan_outsource_budget = excluded.task_plan_outsource_budget,
                task_plan_outsource_detail = excluded.task_plan_outsource_detail,
                task_plan_joint_operating_cost = excluded.task_plan_joint_operating_cost,
                task_plan_entries_json = excluded.task_plan_entries_json,
                exec_budget_labor_by_grade = excluded.exec_budget_labor_by_grade,
                exec_labor_rates_json = excluded.exec_labor_rates_json,
                exec_budget_outsource = excluded.exec_budget_outsource,
                exec_budget_cost_plan = excluded.exec_budget_cost_plan,
                exec_budget_entries_json = excluded.exec_budget_entries_json,
                actual_input_entries_json = excluded.actual_input_entries_json,
                project_type = excluded.project_type,
                expected_as_rate = excluded.expected_as_rate,
                expected_sga_rate = excluded.expected_sga_rate,
                expected_as_cost = excluded.expected_as_cost,
                expected_sga_budget = excluded.expected_sga_budget,
                last_editor_session_id = excluded.last_editor_session_id,
                last_client_submitted_at = excluded.last_client_submitted_at,
                project_start_date = excluded.project_start_date,
                project_end_date = excluded.project_end_date,
                completion_status = excluded.completion_status,
                notes = excluded.notes,
                updated_at = CURRENT_TIMESTAMP
            """
        ),
        payload,
    )


def clean_row_text(value: Any) -> str:
    return normalize_text(value)


def filter_amount_rows(rows: list[dict[str, Any]], amount_key: str = "amount") -> list[dict[str, Any]]:
    cleaned_rows: list[dict[str, Any]] = []
    for row in rows:
        normalized_row = {key: clean_row_text(value) for key, value in row.items()}
        amount = normalize_amount(normalized_row.get(amount_key))
        if amount or any(value for key, value in normalized_row.items() if key != amount_key):
            normalized_row[amount_key] = amount
            cleaned_rows.append(normalized_row)
    return cleaned_rows


def sum_row_amounts(rows: list[dict[str, Any]], amount_key: str = "amount") -> float:
    return sum(normalize_amount(row.get(amount_key)) for row in rows)


def get_support_department_options() -> list[dict[str, str]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT DISTINCT support_dept_code, support_dept_name
                FROM (
                    SELECT support_dept_code, support_dept_name
                    FROM transactions
                    UNION ALL
                    SELECT support_dept_code, support_dept_name
                    FROM project_contract_info
                    UNION ALL
                    SELECT support_dept_code, support_dept_name
                    FROM project_billing_entries
                ) AS merged
                WHERE COALESCE(support_dept_code, '') <> ''
                  AND COALESCE(support_dept_name, '') <> ''
                  AND support_dept_code NOT IN ('ZZZZZZ')
                  AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
                ORDER BY support_dept_code, support_dept_name
                """
            )
        ).mappings().all()
    return [
        {
            "support_dept_code": normalize_text(row["support_dept_code"]),
            "support_dept_name": normalize_text(row["support_dept_name"]),
        }
        for row in rows
    ]


def get_cost_department_options() -> list[dict[str, str]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT DISTINCT cost_dept_name
                FROM transactions
                WHERE COALESCE(cost_dept_name, '') <> ''
                ORDER BY cost_dept_name
                """
            )
        ).mappings().all()
    return [
        {"cost_dept_name": normalize_text(row["cost_dept_name"])}
        for row in rows
        if normalize_text(row["cost_dept_name"])
    ]


def get_cost_account_options() -> list[dict[str, str]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT DISTINCT account_code, account_name
                FROM transactions
                WHERE accounting_category = '원가'
                  AND COALESCE(account_code, '') <> ''
                  AND COALESCE(account_name, '') <> ''
                ORDER BY account_code, account_name
                """
            )
        ).mappings().all()
    deduped: dict[tuple[str, str], dict[str, str]] = {}
    for row in rows:
        account_code = normalize_text(row["account_code"])
        account_name = normalize_text(row["account_name"])
        if not account_code or not account_name:
            continue
        normalized_code, normalized_name, _ = normalize_account_display(account_code, account_name)
        key = (normalized_code, normalized_name)
        deduped[key] = {
            "account_code": normalized_code,
            "account_name": normalized_name,
        }
    return sorted(deduped.values(), key=lambda item: (item["account_code"], item["account_name"]))


def get_import_sync_summary() -> dict[str, Any]:
    with engine.begin() as conn:
        row = conn.execute(
            text(
                """
                SELECT
                    (SELECT COUNT(*) FROM project_contract_info) AS contract_project_count,
                    (SELECT COUNT(*) FROM project_billing_entries) AS billing_entry_count,
                    (SELECT COUNT(DISTINCT support_dept_code) FROM project_billing_entries) AS billing_project_count,
                    (SELECT COUNT(*) FROM project_contract_info WHERE COALESCE(review_tag, '') <> '') AS review_needed_count,
                    (SELECT SUM(hanmac_contract_amount) FROM project_contract_info) AS total_hanmac_contract_amount,
                    (SELECT SUM(collected_amount) FROM project_billing_entries) AS total_collected_amount,
                    (SELECT MAX(updated_at) FROM project_contract_info) AS latest_contract_sync,
                    (SELECT MAX(updated_at) FROM project_billing_entries) AS latest_billing_sync
                """
            )
        ).mappings().first()
    return dict(row) if row else {}


def get_project_contract_info_map() -> dict[str, dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text("SELECT * FROM project_contract_info ORDER BY support_dept_code")
        ).mappings().all()
    return {normalize_text(row["support_dept_code"]): dict(row) for row in rows}


def get_project_billing_summary_map() -> dict[str, dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT support_dept_code,
                       MAX(support_dept_name) AS support_dept_name,
                       MAX(contract_amount) AS contract_amount,
                       MAX(client_name) AS client_name,
                       MAX(support_department) AS support_department,
                       MAX(business_division) AS business_division,
                       SUM(billed_amount) AS billed_amount,
                       SUM(collected_amount) AS collected_amount,
                       SUM(balance_amount) AS balance_amount,
                       MAX(billing_date) AS latest_billing_date
                FROM project_billing_entries
                GROUP BY support_dept_code
                ORDER BY support_dept_code
                """
            )
        ).mappings().all()
        entry_rows = conn.execute(
            text(
                """
                SELECT support_dept_code,
                       billing_type,
                       progress_round,
                       billing_date,
                       tax_invoice_date,
                       expected_collection_date,
                       billed_amount,
                       collected_amount,
                       balance_amount,
                       collection_rate,
                       note
                FROM project_billing_entries
                ORDER BY support_dept_code, billing_date, progress_round, id
                """
            )
        ).mappings().all()
    result = {normalize_text(row["support_dept_code"]): dict(row) for row in rows}
    for item in result.values():
        item["entries"] = []
    for row in entry_rows:
        support_dept_code = normalize_text(row["support_dept_code"])
        if support_dept_code not in result:
            continue
        result[support_dept_code]["entries"].append(
            normalize_collection_entry_row(
                {
                "progress_type": "",
                "billing_round": normalize_round_value(row["progress_round"]),
                "billing_type": normalize_text(row["billing_type"]),
                "billing_date": normalize_date_text(row["billing_date"]),
                "billed_amount": normalize_amount(row["billed_amount"]),
                "round": normalize_round_value(row["progress_round"]),
                "date": normalize_date_text(row["tax_invoice_date"]) or normalize_date_text(row["expected_collection_date"]),
                "amount": normalize_amount(row["collected_amount"]),
                "balance_amount": normalize_amount(row["balance_amount"]),
                "collection_rate": normalize_amount(row["collection_rate"]),
                "note": normalize_text(row["note"]),
                }
            )
        )
    return result


def merge_project_external_fields(
    item: dict[str, Any],
    contract_info: dict[str, Any] | None,
    billing_summary: dict[str, Any] | None,
    latest_summary_change: dict[str, Any] | None = None,
    latest_round_change: dict[str, Any] | None = None,
    change_representative_code: str = "",
    change_title_key: str = "",
) -> dict[str, Any]:
    contract_info = contract_info or {}
    billing_summary = billing_summary or {}
    support_dept_name = normalize_text(item.get("support_dept_name")) or normalize_text(contract_info.get("support_dept_name")) or normalize_text(billing_summary.get("support_dept_name"))
    contract_amount = normalize_amount(item.get("contract_amount"))
    latest_summary_change = latest_summary_change or {}
    latest_round_change = latest_round_change or {}
    latest_changed_contract_amount = (
        normalize_amount(latest_summary_change.get("changed_contract_amount"))
        or normalize_amount(latest_round_change.get("changed_contract_amount"))
    )
    current_code = normalize_text(item.get("support_dept_code"))
    if change_representative_code and current_code == change_representative_code and latest_changed_contract_amount:
        contract_amount = latest_changed_contract_amount
    elif not contract_amount:
        contract_amount = (
            normalize_amount(contract_info.get("hanmac_contract_amount"))
            or normalize_amount(billing_summary.get("contract_amount"))
        )
        if not contract_amount and latest_changed_contract_amount and (not change_representative_code or current_code == change_representative_code):
            contract_amount = latest_changed_contract_amount
    collection_amount = normalize_amount(item.get("collection_amount"))
    if not collection_amount:
        collection_amount = normalize_amount(billing_summary.get("collected_amount"))
    collection_entries = item.get("collection_entries")
    if not collection_entries:
        collection_entries = billing_summary.get("entries", [])
    project_start_date = normalize_text(item.get("project_start_date")) or normalize_text(contract_info.get("project_start_date"))
    project_end_date = (
        normalize_text(item.get("project_end_date"))
        or normalize_text(contract_info.get("project_end_date"))
        or normalize_text(latest_summary_change.get("changed_project_end_date"))
        or normalize_text(latest_round_change.get("changed_project_end_date"))
    )
    completion_status = normalize_text(item.get("completion_status")) or normalize_text(contract_info.get("progress_status"))
    project_type = (
        normalize_text(item.get("project_type"))
        or normalize_text(contract_info.get("business_division"))
        or normalize_text(billing_summary.get("business_division"))
        or normalize_text(latest_summary_change.get("business_division"))
        or normalize_text(latest_round_change.get("business_division"))
    )
    progress_rate = normalize_amount(item.get("progress_rate"))
    if not progress_rate and contract_amount:
        progress_rate = (collection_amount / contract_amount * 100) if contract_amount else 0.0

    item["support_dept_name"] = support_dept_name
    item["contract_amount"] = contract_amount
    item["collection_amount"] = collection_amount
    item["collection_entries"] = collection_entries or []
    item["project_start_date"] = project_start_date
    item["project_end_date"] = project_end_date
    item["completion_status"] = completion_status
    item["project_type"] = project_type
    item["progress_rate"] = progress_rate
    item["client_name"] = (
        normalize_text(contract_info.get("client_name"))
        or normalize_text(billing_summary.get("client_name"))
        or normalize_text(latest_summary_change.get("client_name"))
        or normalize_text(latest_round_change.get("client_name"))
    )
    item["order_method"] = normalize_text(contract_info.get("order_method"))
    item["joint_contract"] = normalize_text(contract_info.get("joint_contract"))
    item["pm_name"] = normalize_text(contract_info.get("pm_name"))
    item["contract_status"] = normalize_text(contract_info.get("contract_status"))
    item["progress_status"] = normalize_text(contract_info.get("progress_status"))
    item["work_category"] = normalize_text(contract_info.get("work_category"))
    item["review_tag"] = normalize_text(contract_info.get("review_tag"))
    item["review_note"] = normalize_text(contract_info.get("review_note"))
    item["total_contract_amount"] = normalize_amount(contract_info.get("total_contract_amount"))
    item["hanmac_contract_amount"] = normalize_amount(contract_info.get("hanmac_contract_amount"))
    item["billing_contract_amount"] = normalize_amount(billing_summary.get("contract_amount"))
    item["billed_amount"] = normalize_amount(billing_summary.get("billed_amount"))
    item["collection_balance_amount"] = normalize_amount(billing_summary.get("balance_amount"))
    item["latest_billing_date"] = normalize_text(billing_summary.get("latest_billing_date"))
    item["changed_contract_amount"] = latest_changed_contract_amount
    item["changed_contract_date"] = normalize_text(latest_summary_change.get("change_date")) or normalize_text(latest_round_change.get("change_date"))
    item["changed_project_end_date"] = normalize_text(latest_summary_change.get("changed_project_end_date")) or normalize_text(latest_round_change.get("changed_project_end_date"))
    item["change_contract_representative_code"] = change_representative_code
    item["change_contract_title_key"] = change_title_key
    return item


def normalize_account_display(account_code: Any, account_name: Any) -> tuple[str, str, str]:
    normalized_code = normalize_text(account_code)[:6]
    normalized_name = re.sub(r"\s*\(.*$", "", normalize_text(account_name)).strip()
    if normalized_code and normalized_name:
        label = f"{normalized_code} · {normalized_name}"
    else:
        label = normalized_name or normalized_code or "미분류"
    return normalized_code, normalized_name, label


def get_data_version() -> str:
    with engine.begin() as conn:
        transaction_updated = conn.execute(text("SELECT MAX(updated_at) FROM transactions")).scalar()
        project_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_status")).scalar()
        contract_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_contract_info")).scalar()
        billing_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_billing_entries")).scalar()
        change_summary_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_contract_change_summary")).scalar()
        change_round_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_contract_change_round")).scalar()
    versions = [
        normalize_text(transaction_updated),
        normalize_text(project_updated),
        normalize_text(contract_updated),
        normalize_text(billing_updated),
        normalize_text(change_summary_updated),
        normalize_text(change_round_updated),
    ]
    return max((version for version in versions if version), default="")


def build_health_payload() -> dict[str, str]:
    return {
        "status": "ok",
        "server_time": datetime.now().isoformat(timespec="seconds"),
        "data_version": get_data_version(),
    }


def check_record_revision(conn: Any, table_name: str, key_column: str, key_value: Any, edit_revision: str) -> None:
    if not key_value or not edit_revision:
        return
    current_revision = conn.execute(
        text(f"SELECT updated_at FROM {table_name} WHERE {key_column} = :key_value"),
        {"key_value": key_value},
    ).scalar()
    current_revision_text = normalize_text(current_revision)
    if current_revision_text and current_revision_text != normalize_text(edit_revision):
        raise ValueError("다른 사용자가 먼저 수정했습니다. 최신 화면으로 다시 확인한 뒤 저장해주세요.")


def date_diff_days(start_date: str, end_date: str) -> int | None:
    if not start_date or not end_date:
        return None
    try:
        start = datetime.strptime(start_date, "%Y-%m-%d").date()
        end = datetime.strptime(end_date, "%Y-%m-%d").date()
        return (end - start).days
    except ValueError:
        return None


def detect_category(account_code: str) -> str:
    if account_code.startswith("5"):
        return "원가"
    if account_code.startswith("4"):
        return "수입/매출액"
    if account_code.startswith("6"):
        return "판관비"
    return "기타"


def extract_period(voucher_number: str, posting_date: str) -> tuple[int | None, int | None, int | None]:
    voucher_match = VOUCHER_PATTERN.match(voucher_number)
    if voucher_match:
        year_text, month_text, day_text = voucher_match.groups()
        return int(year_text), int(month_text), int(day_text)

    parsed_date = normalize_date_text(posting_date)
    if re.match(r"^\d{4}-\d{2}-\d{2}$", parsed_date):
        parsed = datetime.strptime(parsed_date, "%Y-%m-%d")
        return parsed.year, parsed.month, parsed.day
    return None, None, None


def choose_amount(debit_supply: float, credit_supply: float) -> float:
    if debit_supply:
        return abs(debit_supply)
    if credit_supply:
        return abs(credit_supply)
    return 0.0


def canonical_header_name(value: Any) -> str | None:
    normalized = normalize_text(value).replace(" ", "")
    if not normalized:
        return None
    if normalized in DIRECT_HEADER_MAP:
        return DIRECT_HEADER_MAP[normalized]
    if "확정전표" in normalized:
        return "confirmed_voucher_number"
    return None


def empty_record() -> dict[str, str]:
    record = {field: "" for field in FORM_FIELDS}
    record["id"] = ""
    return record


def build_transaction_payload(raw: dict[str, Any], source_file: str = "") -> dict[str, Any]:
    payload: dict[str, Any] = {}
    for field in FORM_FIELDS:
        if field in {"debit_supply", "debit_vat", "credit_supply", "credit_vat"}:
            payload[field] = normalize_amount(raw.get(field))
        elif field == "posting_date":
            payload[field] = normalize_date_text(raw.get(field))
        else:
            payload[field] = normalize_text(raw.get(field))

    payload["accounting_category"] = detect_category(payload["account_code"])
    payload["amount"] = choose_amount(payload["debit_supply"], payload["credit_supply"])
    year, month, day = extract_period(payload["voucher_number"], payload["posting_date"])
    payload["year"] = year
    payload["month"] = month
    payload["day"] = day
    payload["source_file"] = source_file
    return payload


def save_transaction(payload: dict[str, Any], record_id: int | None = None) -> None:
    init_db()
    params = {
        **payload,
        "record_id": record_id,
        "last_editor_session_id": normalize_text(payload.get("client_session_id")),
        "last_client_submitted_at": normalize_text(payload.get("client_submitted_at")),
    }
    with engine.begin() as conn:
        if record_id:
            check_record_revision(conn, "transactions", "id", record_id, normalize_text(payload.get("edit_revision")))
            conn.execute(
                text(
                    """
                    UPDATE transactions
                    SET approval_status = :approval_status,
                        voucher_number = :voucher_number,
                        account_code = :account_code,
                        account_name = :account_name,
                        debit_supply = :debit_supply,
                        debit_vat = :debit_vat,
                        credit_supply = :credit_supply,
                        credit_vat = :credit_vat,
                        issuing_dept_code = :issuing_dept_code,
                        issuing_dept_name = :issuing_dept_name,
                        confirmed_voucher_number = :confirmed_voucher_number,
                        support_dept_code = :support_dept_code,
                        support_dept_name = :support_dept_name,
                        cost_dept_code = :cost_dept_code,
                        cost_dept_name = :cost_dept_name,
                        memo1 = :memo1,
                        memo2 = :memo2,
                        partner_code = :partner_code,
                        partner_name = :partner_name,
                        tax_code = :tax_code,
                        posting_date = :posting_date,
                        voucher_type = :voucher_type,
                        management_item = :management_item,
                        accounting_category = :accounting_category,
                        amount = :amount,
                        year = :year,
                        month = :month,
                        day = :day,
                        source_file = COALESCE(NULLIF(:source_file, ''), source_file),
                        last_editor_session_id = :last_editor_session_id,
                        last_client_submitted_at = :last_client_submitted_at,
                        updated_at = CURRENT_TIMESTAMP
                    WHERE id = :record_id
                    """
                ),
                params,
            )
            return

        conn.execute(
            text(
                """
                INSERT INTO transactions (
                    approval_status,
                    voucher_number,
                    account_code,
                    account_name,
                    debit_supply,
                    debit_vat,
                    credit_supply,
                    credit_vat,
                    issuing_dept_code,
                    issuing_dept_name,
                    confirmed_voucher_number,
                    support_dept_code,
                    support_dept_name,
                    cost_dept_code,
                    cost_dept_name,
                    memo1,
                    memo2,
                    partner_code,
                    partner_name,
                    tax_code,
                    posting_date,
                    voucher_type,
                    management_item,
                    accounting_category,
                    amount,
                    year,
                    month,
                    day,
                    source_file,
                    last_editor_session_id,
                    last_client_submitted_at
                ) VALUES (
                    :approval_status,
                    :voucher_number,
                    :account_code,
                    :account_name,
                    :debit_supply,
                    :debit_vat,
                    :credit_supply,
                    :credit_vat,
                    :issuing_dept_code,
                    :issuing_dept_name,
                    :confirmed_voucher_number,
                    :support_dept_code,
                    :support_dept_name,
                    :cost_dept_code,
                    :cost_dept_name,
                    :memo1,
                    :memo2,
                    :partner_code,
                    :partner_name,
                    :tax_code,
                    :posting_date,
                    :voucher_type,
                    :management_item,
                    :accounting_category,
                    :amount,
                    :year,
                    :month,
                    :day,
                    :source_file,
                    :last_editor_session_id,
                    :last_client_submitted_at
                )
                """
            ),
            params,
        )


def get_record_for_edit(record_id: int | None) -> dict[str, Any]:
    if not record_id:
        return empty_record()
    with engine.begin() as conn:
        row = conn.execute(
            text("SELECT * FROM transactions WHERE id = :record_id"),
            {"record_id": record_id},
        ).mappings().first()
    if not row:
        return empty_record()
    data = dict(row)
    for key, value in list(data.items()):
        if value is None:
            data[key] = ""
    return data


def get_support_businesses() -> list[dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT support_dept_code, support_dept_name, COUNT(*) AS row_count
                FROM transactions
                WHERE COALESCE(support_dept_code, '') <> ''
                  AND support_dept_code NOT IN ('ZZZZZZ')
                  AND COALESCE(support_dept_name, '') <> ''
                  AND support_dept_name NOT IN (
                      '공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
                  )
                GROUP BY support_dept_code, support_dept_name
                ORDER BY support_dept_code, support_dept_name
                """
            )
        ).mappings().all()
    result = [dict(row) for row in rows]
    for item in result:
        item["project_duration_days"] = date_diff_days(item.get("project_start_date", ""), item.get("project_end_date", ""))
        item["planned_total"] = (
            (item.get("task_plan_department_budget") or 0)
            + (item.get("task_plan_outsource_budget") or 0)
            + (item.get("task_plan_joint_operating_cost") or 0)
            + (item.get("exec_budget_labor_by_grade") or 0)
            + (item.get("exec_budget_outsource") or 0)
            + (item.get("exec_budget_cost_plan") or 0)
            + (item.get("expected_as_cost") or 0)
            + (item.get("expected_sga_budget") or 0)
        )
        item["actual_total_expense"] = (
            (item.get("total_cost") or 0)
            + (item.get("total_sga") or 0)
        )
    return result


def get_monthly_summary() -> list[dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT year,
                       month,
                       SUM(
                           CASE
                               WHEN accounting_category = '원가'
                                    AND account_code NOT LIKE '5012%'
                                    AND account_code NOT LIKE '5017%'
                               THEN amount
                               ELSE 0
                           END
                       ) AS cost_sum,
                       SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_sum,
                       SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_sum,
                       SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_sum
                FROM transactions
                WHERE year IS NOT NULL
                  AND month IS NOT NULL
                GROUP BY year, month
                ORDER BY year, month
                """
            )
        ).mappings().all()
    return [dict(row) for row in rows]


def get_yearly_summary() -> list[dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT year,
                       SUM(
                           CASE
                               WHEN accounting_category = '원가'
                                    AND account_code NOT LIKE '5012%'
                                    AND account_code NOT LIKE '5017%'
                               THEN amount
                               ELSE 0
                           END
                       ) AS cost_sum,
                       SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_sum,
                       SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_sum,
                       SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_sum
                FROM transactions
                WHERE year IS NOT NULL
                GROUP BY year
                ORDER BY year
                """
            )
        ).mappings().all()
    return [dict(row) for row in rows]


def get_business_monthly_summary() -> list[dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT year,
                       month,
                       support_dept_code,
                       support_dept_name,
                       SUM(CASE WHEN accounting_category = '원가' THEN amount ELSE 0 END) AS cost_sum,
                       SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_sum
                FROM transactions
                WHERE year IS NOT NULL
                  AND month IS NOT NULL
                  AND COALESCE(support_dept_code, '') <> ''
                  AND support_dept_code NOT IN ('ZZZZZZ')
                  AND COALESCE(support_dept_name, '') <> ''
                  AND support_dept_name NOT IN (
                      '공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
                  )
                GROUP BY year, month, support_dept_code, support_dept_name
                ORDER BY year, month, support_dept_code, support_dept_name
                """
            )
        ).mappings().all()
    return [dict(row) for row in rows]


def get_project_status_rows() -> list[dict[str, Any]]:
    contract_info_map = get_project_contract_info_map()
    billing_summary_map = get_project_billing_summary_map()
    latest_summary_by_title, latest_round_by_code, representative_by_title, title_by_code = get_project_contract_change_maps()
    with engine.begin() as conn:
        entry_maps = load_project_status_entry_maps(conn)
        rows = conn.execute(
            text(
                f"""
                SELECT b.support_dept_code,
                       b.support_dept_name,
                       b.row_count,
                       COALESCE(ps.progress_rate, 0) AS progress_rate,
                       COALESCE(ps.contract_amount, 0) AS contract_amount,
                       COALESCE(ps.collection_amount, 0) AS collection_amount,
                       COALESCE(ps.collection_entries_json, '[]') AS collection_entries_json,
                       COALESCE(ps.change_round, '') AS change_round,
                       COALESCE(ps.item_investment, 0) AS item_investment,
                       COALESCE(ps.task_plan_department_budget, 0) AS task_plan_department_budget,
                       COALESCE(ps.task_plan_outsource_budget, 0) AS task_plan_outsource_budget,
                       COALESCE(ps.task_plan_outsource_detail, '') AS task_plan_outsource_detail,
                       COALESCE(ps.task_plan_joint_operating_cost, 0) AS task_plan_joint_operating_cost,
                       COALESCE(ps.task_plan_entries_json, '[]') AS task_plan_entries_json,
                       COALESCE(ps.exec_budget_labor_by_grade, 0) AS exec_budget_labor_by_grade,
                       COALESCE(ps.exec_budget_outsource, 0) AS exec_budget_outsource,
                       COALESCE(ps.exec_budget_cost_plan, 0) AS exec_budget_cost_plan,
                       COALESCE(ps.exec_budget_entries_json, '[]') AS exec_budget_entries_json,
                       COALESCE(ps.actual_input_entries_json, '[]') AS actual_input_entries_json,
                       COALESCE(ps.expected_as_cost, 0) AS expected_as_cost,
                       COALESCE(ps.expected_sga_budget, 0) AS expected_sga_budget,
                       COALESCE(ps.project_start_date, '') AS project_start_date,
                       COALESCE(ps.project_end_date, '') AS project_end_date,
                       COALESCE(ps.completion_status, '') AS completion_status,
                       COALESCE(ps.notes, '') AS notes,
                       COALESCE(agg.total_cost, 0) AS total_cost,
                       COALESCE(agg.total_sga, 0) AS total_sga,
                       COALESCE(agg.total_revenue, 0) AS total_revenue,
                       COALESCE(agg.actual_labor, 0) AS actual_labor,
                       COALESCE(agg.actual_outsource, 0) AS actual_outsource,
                       COALESCE(agg.latest_year, 0) AS latest_year,
                       COALESCE(agg.latest_month, 0) AS latest_month
                FROM (
                    SELECT support_dept_code, support_dept_name, COUNT(*) AS row_count
                    FROM transactions
                    WHERE COALESCE(support_dept_code, '') <> ''
                      AND support_dept_code NOT IN ('ZZZZZZ')
                      AND COALESCE(support_dept_name, '') <> ''
                      AND support_dept_name NOT IN (
                          '공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
                      )
                    GROUP BY support_dept_code, support_dept_name
                ) AS b
                LEFT JOIN project_status AS ps
                    ON ps.support_dept_code = b.support_dept_code
                LEFT JOIN (
                    SELECT support_dept_code,
                           SUM(CASE WHEN accounting_category = '원가' THEN amount ELSE 0 END) AS total_cost,
                           SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS total_sga,
                           SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS total_revenue,
                           SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS actual_labor,
                           SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS actual_outsource,
                           MAX(year) AS latest_year,
                           MAX(month) AS latest_month
                    FROM transactions
                    WHERE COALESCE(support_dept_code, '') <> ''
                    GROUP BY support_dept_code
                ) AS agg
                    ON agg.support_dept_code = b.support_dept_code
                ORDER BY b.support_dept_code, b.support_dept_name
                """
            )
        ).mappings().all()
    result = []
    seen_codes: set[str] = set()
    for row in rows:
        item = dict(row)
        support_dept_code = normalize_text(item.get("support_dept_code"))
        entry_set = ensure_project_entry_set(entry_maps.get(support_dept_code)) if support_dept_code in entry_maps else extract_project_status_entry_sets(item)
        item["collection_entries"] = entry_set["collection_entries"]
        item["task_plan_entries"] = entry_set["task_plan_entries"]
        item["exec_budget_entries"] = entry_set["exec_budget_entries"]
        item["actual_input_entries"] = entry_set["actual_input_entries"]
        item.pop("collection_entries_json", None)
        item.pop("task_plan_entries_json", None)
        item.pop("exec_budget_entries_json", None)
        item.pop("actual_input_entries_json", None)
        item = merge_project_external_fields(
            item,
            contract_info_map.get(normalize_text(item.get("support_dept_code"))),
            billing_summary_map.get(normalize_text(item.get("support_dept_code"))),
            latest_summary_by_title.get(title_by_code.get(normalize_text(item.get("support_dept_code"))) or normalize_project_title_for_linking(item.get("support_dept_name"))),
            latest_round_by_code.get(normalize_text(item.get("support_dept_code"))),
            representative_by_title.get(title_by_code.get(normalize_text(item.get("support_dept_code"))) or normalize_project_title_for_linking(item.get("support_dept_name")), ""),
            title_by_code.get(normalize_text(item.get("support_dept_code"))) or normalize_project_title_for_linking(item.get("support_dept_name")),
        )
        seen_codes.add(normalize_text(item.get("support_dept_code")))
        result.append(item)
    for support_dept_code in sorted((set(contract_info_map) | set(billing_summary_map)) - seen_codes):
        fallback_title_key = (
            title_by_code.get(support_dept_code)
            or normalize_project_title_for_linking(
                contract_info_map.get(support_dept_code, {}).get("support_dept_name")
                or billing_summary_map.get(support_dept_code, {}).get("support_dept_name")
            )
        )
        result.append(
            merge_project_external_fields(
                {
                    "support_dept_code": support_dept_code,
                    "support_dept_name": "",
                    "row_count": 0,
                    "progress_rate": 0,
                    "contract_amount": 0,
                    "collection_amount": 0,
                    "collection_entries": [],
                    "change_round": "",
                    "item_investment": 0,
                    "task_plan_department_budget": 0,
                    "task_plan_outsource_budget": 0,
                    "task_plan_outsource_detail": "",
                    "task_plan_joint_operating_cost": 0,
                    "task_plan_entries": [],
                    "exec_budget_labor_by_grade": 0,
                    "exec_budget_outsource": 0,
                    "exec_budget_cost_plan": 0,
                    "exec_budget_entries": [],
                    "actual_input_entries": [],
                    "expected_as_cost": 0,
                    "expected_sga_budget": 0,
                    "project_start_date": "",
                    "project_end_date": "",
                    "completion_status": "",
                    "notes": "",
                    "total_cost": 0,
                    "total_sga": 0,
                    "total_revenue": 0,
                    "actual_labor": 0,
                    "actual_outsource": 0,
                    "latest_year": 0,
                    "latest_month": 0,
                    "project_type": "",
                },
                contract_info_map.get(support_dept_code),
                billing_summary_map.get(support_dept_code),
                latest_summary_by_title.get(fallback_title_key),
                latest_round_by_code.get(support_dept_code),
                representative_by_title.get(fallback_title_key, ""),
                fallback_title_key,
            )
        )
    return result


def get_project_status_row_for_code(support_dept_code: str | None) -> dict[str, Any] | None:
    normalized_code = normalize_text(support_dept_code)
    if not normalized_code:
        return None
    for item in get_project_status_rows():
        if normalize_text(item.get("support_dept_code")) == normalized_code:
            return item
    return None


def get_project_comparison_notes_map() -> dict[str, dict[str, str]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT support_dept_code, item_key, COALESCE(note, '') AS note
                FROM project_comparison_notes
                WHERE COALESCE(support_dept_code, '') <> ''
                """
            )
        ).mappings().all()
    result: dict[str, dict[str, str]] = {}
    for row in rows:
        code = normalize_text(row.get("support_dept_code"))
        item_key = normalize_text(row.get("item_key"))
        if not code or not item_key:
            continue
        result.setdefault(code, {})[item_key] = normalize_text(row.get("note"))
    return result


def save_project_comparison_note(support_dept_code: str | None, item_key: str | None, note: str | None) -> None:
    code = normalize_text(support_dept_code)
    normalized_item_key = normalize_text(item_key)
    if not code or not normalized_item_key:
        return
    normalized_note = normalize_text(note)
    with engine.begin() as conn:
        if normalized_note:
            conn.execute(
                text(
                    """
                    INSERT INTO project_comparison_notes (
                        support_dept_code, item_key, note, updated_at
                    ) VALUES (
                        :support_dept_code, :item_key, :note, CURRENT_TIMESTAMP
                    )
                    ON CONFLICT(support_dept_code, item_key) DO UPDATE SET
                        note = excluded.note,
                        updated_at = CURRENT_TIMESTAMP
                    """
                ),
                {
                    "support_dept_code": code,
                    "item_key": normalized_item_key,
                    "note": normalized_note,
                },
            )
        else:
            conn.execute(
                text(
                    """
                    DELETE FROM project_comparison_notes
                    WHERE support_dept_code = :support_dept_code
                      AND item_key = :item_key
                    """
                ),
                {
                    "support_dept_code": code,
                    "item_key": normalized_item_key,
                },
            )


def get_project_status_for_edit(support_dept_code: str | None) -> dict[str, Any]:
    contract_info_map = get_project_contract_info_map()
    billing_summary_map = get_project_billing_summary_map()
    latest_summary_by_title, latest_round_by_code, representative_by_title, title_by_code = get_project_contract_change_maps()
    if not support_dept_code:
        return {
            "support_dept_code": "",
            "support_dept_name": "",
            "progress_rate": "",
            "contract_amount": "",
            "collection_amount": "",
            "collection_entries": [],
            "change_round": "",
            "item_investment": "",
            "task_plan_department_budget": "",
            "task_plan_outsource_budget": "",
            "task_plan_outsource_detail": "",
            "task_plan_joint_operating_cost": "",
            "task_plan_entries": [],
            "exec_budget_labor_by_grade": "",
            "exec_labor_rates": {},
            "exec_budget_outsource": "",
            "exec_budget_cost_plan": "",
            "exec_budget_entries": [],
            "actual_input_entries": [],
            "project_type": "",
            "expected_as_rate": "",
            "expected_sga_rate": "",
            "expected_as_cost": "",
            "expected_sga_budget": "",
            "project_start_date": "",
            "project_end_date": "",
            "completion_status": "",
            "notes": "",
            "client_name": "",
            "order_method": "",
            "joint_contract": "",
            "pm_name": "",
            "contract_status": "",
            "progress_status": "",
            "work_category": "",
            "review_tag": "",
            "review_note": "",
            "total_contract_amount": 0,
            "hanmac_contract_amount": 0,
            "billing_contract_amount": 0,
            "billed_amount": 0,
            "collection_balance_amount": 0,
            "latest_billing_date": "",
            "updated_at": "",
        }

    with engine.begin() as conn:
        entry_maps = load_project_status_entry_maps(conn)
        row = conn.execute(
            text(
                """
                SELECT b.support_dept_code,
                       b.support_dept_name,
                       COALESCE(ps.progress_rate, '') AS progress_rate,
                       COALESCE(ps.contract_amount, '') AS contract_amount,
                       COALESCE(ps.collection_amount, '') AS collection_amount,
                       COALESCE(ps.collection_entries_json, '[]') AS collection_entries_json,
                       COALESCE(ps.change_round, '') AS change_round,
                       COALESCE(ps.item_investment, '') AS item_investment,
                       COALESCE(ps.task_plan_department_budget, '') AS task_plan_department_budget,
                       COALESCE(ps.task_plan_outsource_budget, '') AS task_plan_outsource_budget,
                       COALESCE(ps.task_plan_outsource_detail, '') AS task_plan_outsource_detail,
                       COALESCE(ps.task_plan_joint_operating_cost, '') AS task_plan_joint_operating_cost,
                       COALESCE(ps.task_plan_entries_json, '[]') AS task_plan_entries_json,
                       COALESCE(ps.exec_budget_labor_by_grade, '') AS exec_budget_labor_by_grade,
                       COALESCE(ps.exec_labor_rates_json, '{}') AS exec_labor_rates_json,
                       COALESCE(ps.exec_budget_outsource, '') AS exec_budget_outsource,
                       COALESCE(ps.exec_budget_cost_plan, '') AS exec_budget_cost_plan,
                       COALESCE(ps.exec_budget_entries_json, '[]') AS exec_budget_entries_json,
                       COALESCE(ps.actual_input_entries_json, '[]') AS actual_input_entries_json,
                       COALESCE(ps.project_type, '') AS project_type,
                       COALESCE(ps.expected_as_rate, '') AS expected_as_rate,
                       COALESCE(ps.expected_sga_rate, '') AS expected_sga_rate,
                       COALESCE(ps.expected_as_cost, '') AS expected_as_cost,
                       COALESCE(ps.expected_sga_budget, '') AS expected_sga_budget,
                       COALESCE(ps.last_editor_session_id, '') AS last_editor_session_id,
                       COALESCE(ps.last_client_submitted_at, '') AS last_client_submitted_at,
                       COALESCE(ps.project_start_date, '') AS project_start_date,
                       COALESCE(ps.project_end_date, '') AS project_end_date,
                       COALESCE(ps.completion_status, '') AS completion_status,
                       COALESCE(ps.notes, '') AS notes,
                       COALESCE(ps.updated_at, '') AS updated_at
                FROM (
                    SELECT DISTINCT support_dept_code, support_dept_name
                    FROM transactions
                    WHERE support_dept_code = :support_dept_code
                ) AS b
                LEFT JOIN project_status AS ps
                    ON ps.support_dept_code = b.support_dept_code
                """
            ),
            {"support_dept_code": support_dept_code},
        ).mappings().first()
    if not row:
        return {
            "support_dept_code": support_dept_code,
            "support_dept_name": "",
            "progress_rate": "",
            "contract_amount": "",
            "collection_amount": "",
            "collection_entries": [],
            "change_round": "",
            "item_investment": "",
            "task_plan_department_budget": "",
            "task_plan_outsource_budget": "",
            "task_plan_outsource_detail": "",
            "task_plan_joint_operating_cost": "",
            "task_plan_entries": [],
            "exec_budget_labor_by_grade": "",
            "exec_labor_rates": {},
            "exec_budget_outsource": "",
            "exec_budget_cost_plan": "",
            "exec_budget_entries": [],
            "actual_input_entries": [],
            "project_type": "",
            "expected_as_rate": "",
            "expected_sga_rate": "",
            "expected_as_cost": "",
            "expected_sga_budget": "",
            "project_start_date": "",
            "project_end_date": "",
            "completion_status": "",
            "notes": "",
            "client_name": "",
            "order_method": "",
            "joint_contract": "",
            "pm_name": "",
            "contract_status": "",
            "progress_status": "",
            "work_category": "",
            "review_tag": "",
            "review_note": "",
            "total_contract_amount": 0,
            "hanmac_contract_amount": 0,
            "billing_contract_amount": 0,
            "billed_amount": 0,
            "collection_balance_amount": 0,
            "latest_billing_date": "",
            "updated_at": "",
        }
    result = dict(row)
    support_dept_code = normalize_text(result.get("support_dept_code"))
    entry_set = ensure_project_entry_set(entry_maps.get(support_dept_code)) if support_dept_code in entry_maps else extract_project_status_entry_sets(result)
    result["collection_entries"] = entry_set["collection_entries"]
    result["task_plan_entries"] = entry_set["task_plan_entries"]
    result["exec_budget_entries"] = entry_set["exec_budget_entries"]
    result["actual_input_entries"] = entry_set["actual_input_entries"]
    result.pop("collection_entries_json", None)
    result.pop("task_plan_entries_json", None)
    result.pop("exec_budget_entries_json", None)
    result.pop("actual_input_entries_json", None)
    try:
        result["exec_labor_rates"] = json.loads(normalize_text(result.pop("exec_labor_rates_json", "{}")) or "{}")
    except json.JSONDecodeError:
        result["exec_labor_rates"] = {}
    return merge_project_external_fields(
        result,
        contract_info_map.get(normalize_text(result.get("support_dept_code"))),
        billing_summary_map.get(normalize_text(result.get("support_dept_code"))),
        latest_summary_by_title.get(title_by_code.get(normalize_text(result.get("support_dept_code"))) or normalize_project_title_for_linking(result.get("support_dept_name"))),
        latest_round_by_code.get(normalize_text(result.get("support_dept_code"))),
        representative_by_title.get(title_by_code.get(normalize_text(result.get("support_dept_code"))) or normalize_project_title_for_linking(result.get("support_dept_name")), ""),
        title_by_code.get(normalize_text(result.get("support_dept_code"))) or normalize_project_title_for_linking(result.get("support_dept_name")),
    )


def get_project_page_state(session_id: str | None = None) -> dict[str, Any]:
    normalized_session_id = normalize_text(session_id)
    with engine.begin() as conn:
        row = conn.execute(
            text(
                """
                SELECT COALESCE(selected_code, '') AS selected_code,
                       COALESCE(selected_year, '') AS selected_year,
                       COALESCE(analysis_open, 0) AS analysis_open,
                       COALESCE(uncontracted_year_start, '') AS uncontracted_year_start,
                       COALESCE(uncontracted_year_end, '') AS uncontracted_year_end,
                       COALESCE(related_project_selections_json, '{}') AS related_project_selections_json
                FROM project_page_state
                WHERE page_key = 'projects'
                  AND session_id = :session_id
                """
            ),
            {"session_id": normalized_session_id},
        ).mappings().first()
    if not row and normalized_session_id:
        with engine.begin() as conn:
            row = conn.execute(
                text(
                    """
                    SELECT COALESCE(selected_code, '') AS selected_code,
                           COALESCE(selected_year, '') AS selected_year,
                           COALESCE(analysis_open, 0) AS analysis_open,
                           COALESCE(uncontracted_year_start, '') AS uncontracted_year_start,
                           COALESCE(uncontracted_year_end, '') AS uncontracted_year_end,
                           COALESCE(related_project_selections_json, '{}') AS related_project_selections_json
                    FROM project_page_state
                    WHERE page_key = 'projects'
                      AND session_id = ''
                    """
                )
            ).mappings().first()
    if not row:
        return {
            "selected_code": "",
            "selected_year": "",
            "analysis_open": False,
            "uncontracted_year_start": "",
            "uncontracted_year_end": "",
            "related_project_selections": {},
        }
    try:
        related_project_selections_raw = json.loads(normalize_text(row["related_project_selections_json"]) or "{}")
    except json.JSONDecodeError:
        related_project_selections_raw = {}
    related_project_selections = {}
    if isinstance(related_project_selections_raw, dict):
        related_project_selections = {
            normalize_text(key): [
                normalize_text(value)
                for value in values
                if normalize_text(value)
            ]
            for key, values in related_project_selections_raw.items()
            if normalize_text(key) and isinstance(values, list)
        }
    return {
        "selected_code": normalize_text(row["selected_code"]),
        "selected_year": normalize_text(row["selected_year"]),
        "analysis_open": bool(row["analysis_open"]),
        "uncontracted_year_start": normalize_text(row["uncontracted_year_start"]),
        "uncontracted_year_end": normalize_text(row["uncontracted_year_end"]),
        "related_project_selections": related_project_selections,
    }


def save_project_page_state(payload: dict[str, Any]) -> None:
    session_id = normalize_text(payload.get("session_id"))
    selected_code = normalize_text(payload.get("selected_code"))
    selected_year = normalize_text(payload.get("selected_year"))
    analysis_open = 1 if payload.get("analysis_open") else 0
    uncontracted_year_start = normalize_text(payload.get("uncontracted_year_start"))
    uncontracted_year_end = normalize_text(payload.get("uncontracted_year_end"))
    raw_related = payload.get("related_project_selections") or {}
    related_project_selections = {}
    if isinstance(raw_related, dict):
        related_project_selections = {
            normalize_text(key): [
                normalize_text(value)
                for value in values
                if normalize_text(value)
            ]
            for key, values in raw_related.items()
            if normalize_text(key) and isinstance(values, list)
        }
    with engine.begin() as conn:
        conn.execute(
            text(
                """
                INSERT INTO project_page_state (
                    page_key,
                    session_id,
                    selected_code,
                    selected_year,
                    analysis_open,
                    uncontracted_year_start,
                    uncontracted_year_end,
                    related_project_selections_json,
                    updated_at
                ) VALUES (
                    'projects',
                    :session_id,
                    :selected_code,
                    :selected_year,
                    :analysis_open,
                    :uncontracted_year_start,
                    :uncontracted_year_end,
                    :related_project_selections_json,
                    CURRENT_TIMESTAMP
                )
                ON CONFLICT(page_key, session_id) DO UPDATE SET
                    selected_code = excluded.selected_code,
                    selected_year = excluded.selected_year,
                    analysis_open = excluded.analysis_open,
                    uncontracted_year_start = excluded.uncontracted_year_start,
                    uncontracted_year_end = excluded.uncontracted_year_end,
                    related_project_selections_json = excluded.related_project_selections_json,
                    updated_at = CURRENT_TIMESTAMP
                """
            ),
            {
                "session_id": session_id,
                "selected_code": selected_code,
                "selected_year": selected_year,
                "analysis_open": analysis_open,
                "uncontracted_year_start": uncontracted_year_start,
                "uncontracted_year_end": uncontracted_year_end,
                "related_project_selections_json": json.dumps(related_project_selections, ensure_ascii=False),
            },
        )
    for base_code, related_codes in related_project_selections.items():
        save_project_related_links(base_code, related_codes)


def get_project_related_links_map() -> dict[str, list[str]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT base_support_dept_code, related_support_dept_code
                FROM project_related_links
                ORDER BY base_support_dept_code, related_support_dept_code
                """
            )
        ).mappings().all()
    related_map: dict[str, list[str]] = {}
    for row in rows:
        base_code = normalize_text(row["base_support_dept_code"])
        related_code = normalize_text(row["related_support_dept_code"])
        if not base_code or not related_code:
            continue
        related_map.setdefault(base_code, []).append(related_code)
    return related_map


def get_project_quick_links(session_id: str | None = None) -> list[str]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT support_dept_code
                FROM project_quick_links
                WHERE page_key = 'projects'
                ORDER BY sort_order, updated_at DESC, support_dept_code
                """
            )
        ).mappings().all()
    return [normalize_text(row["support_dept_code"]) for row in rows if normalize_text(row["support_dept_code"])]


def save_project_quick_links(session_id: str | None, codes: list[str]) -> None:
    normalized_codes: list[str] = []
    for code in codes:
        normalized_code = normalize_text(code)
        if normalized_code and normalized_code not in normalized_codes:
            normalized_codes.append(normalized_code)
    with engine.begin() as conn:
        conn.execute(
            text(
                """
                DELETE FROM project_quick_links
                WHERE page_key = 'projects'
                """
            )
        )
        for sort_order, support_dept_code in enumerate(normalized_codes):
            conn.execute(
                text(
                    """
                    INSERT INTO project_quick_links (
                        page_key, support_dept_code, sort_order, updated_at
                    ) VALUES (
                        'projects', :support_dept_code, :sort_order, CURRENT_TIMESTAMP
                    )
                    """
                ),
                {
                    "support_dept_code": support_dept_code,
                    "sort_order": sort_order,
                },
            )


def get_project_uncontracted_classification_map() -> dict[str, str]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT support_dept_code, category
                FROM project_uncontracted_classification
                WHERE COALESCE(support_dept_code, '') <> ''
                """
            )
        ).fetchall()
    return {
        normalize_text(row[0]): normalize_text(row[1])
        for row in rows
        if normalize_text(row[0])
    }


def save_project_uncontracted_classification(support_dept_code: Any, category: Any) -> None:
    normalized_code = normalize_text(support_dept_code)
    normalized_category = normalize_text(category)
    allowed_categories = {"general", "precontract", "corporate_rnd", "external_research"}
    if not normalized_code:
        raise ValueError("프로젝트 코드가 필요합니다.")
    if normalized_category not in allowed_categories:
        raise ValueError("허용되지 않는 미계약 분류입니다.")
    with engine.begin() as conn:
        conn.execute(
            text(
                """
                INSERT INTO project_uncontracted_classification (
                    support_dept_code,
                    category,
                    updated_at
                ) VALUES (
                    :support_dept_code,
                    :category,
                    CURRENT_TIMESTAMP
                )
                ON CONFLICT(support_dept_code) DO UPDATE SET
                    category = excluded.category,
                    updated_at = CURRENT_TIMESTAMP
                """
            ),
            {
                "support_dept_code": normalized_code,
                "category": normalized_category,
            },
        )


def save_project_related_links(base_support_dept_code: str, related_codes: list[Any]) -> None:
    base_code = normalize_text(base_support_dept_code)
    if not base_code:
        return
    normalized_codes = sorted(
        {
            normalize_text(code)
            for code in related_codes
            if normalize_text(code) and normalize_text(code) != base_code
        }
    )
    with engine.begin() as conn:
        conn.execute(
            text(
                """
                DELETE FROM project_related_links
                WHERE base_support_dept_code = :base_support_dept_code
                  AND COALESCE(link_source, 'manual') = 'manual'
                """
            ),
            {"base_support_dept_code": base_code},
        )
        for related_code in normalized_codes:
            conn.execute(
                text(
                    """
                    INSERT INTO project_related_links (
                        base_support_dept_code,
                        related_support_dept_code,
                        link_source,
                        updated_at
                    ) VALUES (
                        :base_support_dept_code,
                        :related_support_dept_code,
                        'manual',
                        CURRENT_TIMESTAMP
                    )
                    ON CONFLICT(base_support_dept_code, related_support_dept_code) DO UPDATE SET
                        link_source = excluded.link_source,
                        updated_at = CURRENT_TIMESTAMP
                    """
                ),
                {
                    "base_support_dept_code": base_code,
                    "related_support_dept_code": related_code,
                },
            )


def get_project_year_options() -> list[int]:
    return get_available_years()


def resolve_selected_year(selected_year: int | None) -> int | None:
    return selected_year


def parse_optional_year(value: Any) -> int | None:
    text = normalize_text(value)
    if not text:
        return None
    return int(text) if text.isdigit() else None


def get_recent_10_start_year() -> int | None:
    available_years = get_available_years()
    if not available_years:
        return None
    return max(available_years) - 9


def get_project_dashboard_summary(selected_year: int | None) -> dict[str, Any]:
    selected_year = resolve_selected_year(selected_year)
    project_year_clause = ""
    collection_year_clause = ""
    params: dict[str, Any] = {}
    if selected_year:
        project_year_clause = "AND p.year = :selected_year"
        collection_year_clause = "AND year = :selected_year"
        params["selected_year"] = selected_year
    else:
        recent_10_start_year = get_recent_10_start_year()
        if recent_10_start_year is not None:
            project_year_clause = "AND p.year >= :recent_10_start_year"
            collection_year_clause = "AND year >= :recent_10_start_year"
            params["recent_10_start_year"] = recent_10_start_year

    with engine.begin() as conn:
        project_row = conn.execute(
            text(
                f"""
                SELECT COUNT(*) AS related_projects,
                       SUM(COALESCE(p.expense_amount, 0)) AS expense_amount,
                       COUNT(CASE WHEN COALESCE(ps.completion_status, '') IN ('종료', '완료', 'Y', 'YES') THEN 1 END) AS completed_projects
                FROM (
                    SELECT year,
                           support_dept_code,
                           support_dept_name,
                           SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount
                    FROM transactions
                    WHERE COALESCE(support_dept_code, '') <> ''
                      AND support_dept_code NOT IN ('ZZZZZZ')
                      AND COALESCE(support_dept_name, '') <> ''
                      AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
                    GROUP BY year, support_dept_code, support_dept_name
                ) AS p
                LEFT JOIN project_status AS ps
                    ON ps.support_dept_code = p.support_dept_code
                WHERE 1=1
                  {project_year_clause}
                """
            ),
            params,
        ).mappings().first()

        collection_row = conn.execute(
            text(
                f"""
                SELECT COUNT(*) AS collection_transaction_count,
                       COUNT(DISTINCT support_dept_code) AS collection_project_count,
                       SUM(COALESCE(credit_supply, 0)) AS collection_amount
                FROM transactions
                WHERE COALESCE(credit_supply, 0) <> 0
                  AND ({REVENUE_SQL})
                  AND COALESCE(support_dept_code, '') <> ''
                  AND support_dept_code NOT IN ('ZZZZZZ')
                  AND COALESCE(support_dept_name, '') <> ''
                  {collection_year_clause}
                """
            ),
            params,
        ).mappings().first()

        input_row = conn.execute(
            text(
                """
                SELECT COUNT(CASE WHEN COALESCE(collection_amount, 0) <> 0 THEN 1 END) AS collection_input_projects,
                       COUNT(CASE WHEN COALESCE(completion_status, '') <> '' THEN 1 END) AS completion_input_projects
                FROM project_status
                """
            )
        ).mappings().first()

    return {
        **(dict(project_row) if project_row else {}),
        **(dict(collection_row) if collection_row else {}),
        **(dict(input_row) if input_row else {}),
    }


def get_uncontracted_project_dashboard(selected_year: int | None) -> dict[str, Any]:
    selected_year = resolve_selected_year(selected_year)
    transaction_year_clause = ""
    params: dict[str, Any] = {}
    if selected_year:
        transaction_year_clause = "AND t.year = :selected_year"
        params["selected_year"] = selected_year
    else:
        recent_10_start_year = get_recent_10_start_year()
        if recent_10_start_year is not None:
            transaction_year_clause = "AND t.year >= :recent_10_start_year"
            params["recent_10_start_year"] = recent_10_start_year

    with engine.begin() as conn:
        summary = conn.execute(
            text(
                f"""
                WITH project_universe AS (
                    SELECT DISTINCT support_dept_code, support_dept_name
                    FROM transactions
                    WHERE COALESCE(support_dept_code, '') <> ''
                      AND support_dept_code NOT IN ('ZZZZZZ')
                      AND COALESCE(support_dept_name, '') <> ''
                      AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
                    UNION
                    SELECT support_dept_code, support_dept_name
                    FROM project_contract_info
                    WHERE COALESCE(support_dept_code, '') <> ''
                    UNION
                    SELECT support_dept_code, support_dept_name
                    FROM project_billing_entries
                    WHERE COALESCE(support_dept_code, '') <> ''
                ),
                contract_flags AS (
                    SELECT support_dept_code,
                           COALESCE(hanmac_contract_amount, 0) AS hanmac_contract_amount,
                           COALESCE(review_tag, '') AS review_tag
                    FROM project_contract_info
                ),
                project_amounts AS (
                    SELECT t.support_dept_code,
                           SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
                           SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount
                    FROM transactions AS t
                    WHERE COALESCE(t.support_dept_code, '') <> ''
                      AND t.support_dept_code NOT IN ('ZZZZZZ')
                      {transaction_year_clause}
                    GROUP BY t.support_dept_code
                )
                SELECT
                    SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 THEN 1 ELSE 0 END) AS uncontracted_projects,
                    SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 AND COALESCE(project_amounts.expense_amount, 0) > 0 THEN 1 ELSE 0 END) AS cost_incurred_projects,
                    SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 THEN COALESCE(project_amounts.expense_amount, 0) ELSE 0 END) AS expense_amount,
                    SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 THEN COALESCE(project_amounts.revenue_amount, 0) ELSE 0 END) AS revenue_amount,
                    SUM(CASE WHEN COALESCE(contract_flags.review_tag, '') <> '' THEN 1 ELSE 0 END) AS review_needed_projects
                FROM project_universe
                LEFT JOIN contract_flags
                    ON contract_flags.support_dept_code = project_universe.support_dept_code
                LEFT JOIN project_amounts
                    ON project_amounts.support_dept_code = project_universe.support_dept_code
                """
            ),
            params,
        ).mappings().first()

        yearly_rows = conn.execute(
            text(
                f"""
                WITH yearly_costs AS (
                    SELECT t.year,
                           t.support_dept_code,
                           SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
                           SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount
                    FROM transactions AS t
                    WHERE COALESCE(t.support_dept_code, '') <> ''
                      AND t.support_dept_code NOT IN ('ZZZZZZ')
                      AND t.year IS NOT NULL
                      {transaction_year_clause}
                    GROUP BY t.year, t.support_dept_code
                )
                SELECT yearly_costs.year,
                       COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN yearly_costs.support_dept_code END) AS uncontracted_projects,
                       COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 AND COALESCE(yearly_costs.expense_amount, 0) > 0 THEN yearly_costs.support_dept_code END) AS cost_incurred_projects,
                       SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(yearly_costs.expense_amount, 0) ELSE 0 END) AS expense_amount,
                       SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(yearly_costs.revenue_amount, 0) ELSE 0 END) AS revenue_amount
                FROM yearly_costs
                LEFT JOIN project_contract_info AS c
                    ON c.support_dept_code = yearly_costs.support_dept_code
                GROUP BY yearly_costs.year
                ORDER BY yearly_costs.year
                """
            ),
            params,
        ).mappings().all()

        monthly_focus_year = selected_year
        if monthly_focus_year is None:
            monthly_focus_year = conn.execute(
                text(
                    """
                    WITH monthly_candidates AS (
                        SELECT MAX(t.year) AS latest_year
                        FROM transactions AS t
                        LEFT JOIN project_contract_info AS c
                            ON c.support_dept_code = t.support_dept_code
                        WHERE COALESCE(t.support_dept_code, '') <> ''
                          AND t.support_dept_code NOT IN ('ZZZZZZ')
                          AND COALESCE(c.hanmac_contract_amount, 0) <= 0
                          AND (t.account_code LIKE '5%' OR t.account_code LIKE '6%')
                    )
                    SELECT latest_year FROM monthly_candidates
                    """
                )
            ).scalar()

        monthly_rows: list[dict[str, Any]] = []
        if monthly_focus_year:
            monthly_rows = conn.execute(
                text(
                    f"""
                    WITH monthly_costs AS (
                        SELECT t.month,
                               t.support_dept_code,
                               SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
                               SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount
                        FROM transactions AS t
                        WHERE COALESCE(t.support_dept_code, '') <> ''
                          AND t.support_dept_code NOT IN ('ZZZZZZ')
                          AND t.year = :monthly_focus_year
                          AND t.month IS NOT NULL
                        GROUP BY t.month, t.support_dept_code
                    )
                    SELECT monthly_costs.month,
                           COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN monthly_costs.support_dept_code END) AS uncontracted_projects,
                           COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 AND COALESCE(monthly_costs.expense_amount, 0) > 0 THEN monthly_costs.support_dept_code END) AS cost_incurred_projects,
                           SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(monthly_costs.expense_amount, 0) ELSE 0 END) AS expense_amount,
                           SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(monthly_costs.revenue_amount, 0) ELSE 0 END) AS revenue_amount
                    FROM monthly_costs
                    LEFT JOIN project_contract_info AS c
                        ON c.support_dept_code = monthly_costs.support_dept_code
                    GROUP BY monthly_costs.month
                    ORDER BY monthly_costs.month
                    """
                ),
                {"monthly_focus_year": monthly_focus_year},
            ).mappings().all()

        top_rows = conn.execute(
            text(
                f"""
                WITH project_costs AS (
                    SELECT t.support_dept_code,
                           MAX(t.support_dept_name) AS support_dept_name,
                           SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
                           SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount,
                           MAX(t.year) AS latest_year,
                           MAX(t.month) AS latest_month
                    FROM transactions AS t
                    WHERE COALESCE(t.support_dept_code, '') <> ''
                      AND t.support_dept_code NOT IN ('ZZZZZZ')
                      {transaction_year_clause}
                    GROUP BY t.support_dept_code
                )
                SELECT project_costs.support_dept_code,
                       project_costs.support_dept_name,
                       project_costs.expense_amount,
                       project_costs.revenue_amount,
                       project_costs.latest_year,
                       project_costs.latest_month,
                       COALESCE(c.review_tag, '') AS review_tag
                FROM project_costs
                LEFT JOIN project_contract_info AS c
                    ON c.support_dept_code = project_costs.support_dept_code
                WHERE COALESCE(c.hanmac_contract_amount, 0) <= 0
                  AND COALESCE(project_costs.expense_amount, 0) > 0
                ORDER BY project_costs.expense_amount DESC, project_costs.support_dept_code
                LIMIT 12
                """
            ),
            params,
        ).mappings().all()

    return {
        "summary": dict(summary) if summary else {},
        "yearly_rows": [dict(row) for row in yearly_rows],
        "monthly_rows": [dict(row) for row in monthly_rows],
        "monthly_focus_year": int(monthly_focus_year) if monthly_focus_year else None,
        "top_rows": [dict(row) for row in top_rows],
    }


def get_project_revenue_mix(selected_year: int | None = None) -> list[dict[str, Any]]:
    params: dict[str, Any] = {}
    if selected_year:
        with engine.begin() as conn:
            rows = conn.execute(
                text(
                    f"""
                    SELECT
                        year,
                        month,
                        SUM(CASE WHEN account_code LIKE '40110101%' AND memo1 LIKE '%설계%' THEN amount ELSE 0 END) AS design_revenue,
                        SUM(CASE WHEN account_code LIKE '40110101%' AND (memo1 NOT LIKE '%설계%' OR COALESCE(memo1, '') = '') THEN amount ELSE 0 END) AS design_other_revenue,
                        SUM(CASE WHEN account_code LIKE '40110102%' THEN amount ELSE 0 END) AS supervision_revenue,
                        SUM(CASE WHEN account_code LIKE '40110103%' THEN amount ELSE 0 END) AS inspection_revenue
                    FROM transactions
                    WHERE year = :selected_year
                      AND month IS NOT NULL
                    GROUP BY year, month
                    ORDER BY year, month
                    """
                ),
                {"selected_year": selected_year},
            ).mappings().all()
        result = [dict(row) for row in rows]
        for item in result:
            item["label"] = f"{int(item['month'])}월" if item.get("month") is not None else str(item.get("year", ""))
        return result

    with engine.begin() as conn:
        rows = conn.execute(
            text(
                f"""
                SELECT
                    year,
                    SUM(CASE WHEN account_code LIKE '40110101%' AND memo1 LIKE '%설계%' THEN amount ELSE 0 END) AS design_revenue,
                    SUM(CASE WHEN account_code LIKE '40110101%' AND (memo1 NOT LIKE '%설계%' OR COALESCE(memo1, '') = '') THEN amount ELSE 0 END) AS design_other_revenue,
                    SUM(CASE WHEN account_code LIKE '40110102%' THEN amount ELSE 0 END) AS supervision_revenue,
                    SUM(CASE WHEN account_code LIKE '40110103%' THEN amount ELSE 0 END) AS inspection_revenue
                FROM transactions
                WHERE year IS NOT NULL
                GROUP BY year
                ORDER BY year
                """
            )
        ).mappings().all()
    result = [dict(row) for row in rows]
    for item in result:
        item["label"] = str(item.get("year", ""))
    return result[-10:]


def get_project_revenue_mix_monthly() -> list[dict[str, Any]]:
    recent_10_start_year = get_recent_10_start_year()
    params: dict[str, Any] = {}
    year_clause = ""
    if recent_10_start_year is not None:
        year_clause = "AND year >= :recent_10_start_year"
        params["recent_10_start_year"] = recent_10_start_year

    with engine.begin() as conn:
        rows = conn.execute(
            text(
                f"""
                SELECT
                    year,
                    month,
                    SUM(CASE WHEN account_code LIKE '40110101%' AND memo1 LIKE '%설계%' THEN amount ELSE 0 END) AS design_revenue,
                    SUM(CASE WHEN account_code LIKE '40110101%' AND (memo1 NOT LIKE '%설계%' OR COALESCE(memo1, '') = '') THEN amount ELSE 0 END) AS design_other_revenue,
                    SUM(CASE WHEN account_code LIKE '40110102%' THEN amount ELSE 0 END) AS supervision_revenue,
                    SUM(CASE WHEN account_code LIKE '40110103%' THEN amount ELSE 0 END) AS inspection_revenue
                FROM transactions
                WHERE month IS NOT NULL
                  {year_clause}
                GROUP BY year, month
                ORDER BY year, month
                """
            ),
            params,
        ).mappings().all()
    result = [dict(row) for row in rows]
    for item in result:
        item["label"] = f"{int(item['month'])}월" if item.get("month") is not None else str(item.get("year", ""))
    return result


def get_project_cost_by_year(selected_year: int | None) -> list[dict[str, Any]]:
    year_clause = ""
    params: dict[str, Any] = {}
    if selected_year:
        year_clause = "AND year = :selected_year"
        params["selected_year"] = selected_year
    else:
        recent_10_start_year = get_recent_10_start_year()
        if recent_10_start_year is not None:
            year_clause = "AND year >= :recent_10_start_year"
            params["recent_10_start_year"] = recent_10_start_year

    contract_info_map = get_project_contract_info_map()
    billing_summary_map = get_project_billing_summary_map()
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                f"""
                SELECT year,
                       support_dept_code,
                       support_dept_name,
                       SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount,
                       SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS revenue_amount
                FROM transactions
                WHERE COALESCE(support_dept_code, '') <> ''
                  AND support_dept_code NOT IN ('ZZZZZZ')
                  AND COALESCE(support_dept_name, '') <> ''
                  AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
                  {year_clause}
                GROUP BY year, support_dept_code, support_dept_name
                ORDER BY year DESC, expense_amount DESC, support_dept_code
                """
            ),
            params,
        ).mappings().all()
    result = [dict(row) for row in rows]
    existing_codes = {normalize_text(row["support_dept_code"]) for row in result}
    candidate_codes = sorted((set(contract_info_map) | set(billing_summary_map)) - existing_codes)
    recent_10_start_year = get_recent_10_start_year()
    for support_dept_code in candidate_codes:
        contract_info = contract_info_map.get(support_dept_code, {})
        billing_summary = billing_summary_map.get(support_dept_code, {})
        fallback_date = (
            normalize_text(contract_info.get("project_start_date"))
            or normalize_text(contract_info.get("contract_date"))
            or normalize_text(billing_summary.get("latest_billing_date"))
        )
        fallback_year = 0
        if re.match(r"^\d{4}-\d{2}-\d{2}$", fallback_date):
            fallback_year = int(fallback_date[:4])
        if selected_year and fallback_year and fallback_year != selected_year:
            continue
        if not selected_year and recent_10_start_year is not None and fallback_year and fallback_year < recent_10_start_year:
            continue
        result.append(
            {
                "year": fallback_year,
                "support_dept_code": support_dept_code,
                "support_dept_name": normalize_text(contract_info.get("support_dept_name")) or normalize_text(billing_summary.get("support_dept_name")),
                "expense_amount": 0,
                "revenue_amount": normalize_amount(billing_summary.get("collected_amount")),
            }
        )
    result.sort(key=lambda item: (-(int(item.get("year") or 0)), -normalize_amount(item.get("expense_amount")), normalize_text(item.get("support_dept_code"))))
    return result


def get_project_account_breakdowns(selected_year: int | None) -> dict[str, dict[str, list[dict[str, Any]]]]:
    year_clause = ""
    params: dict[str, Any] = {}
    if selected_year:
        year_clause = "AND year = :selected_year"
        params["selected_year"] = selected_year
    else:
        recent_10_start_year = get_recent_10_start_year()
        if recent_10_start_year is not None:
            year_clause = "AND year >= :recent_10_start_year"
            params["recent_10_start_year"] = recent_10_start_year

    with engine.begin() as conn:
        rows = conn.execute(
            text(
                f"""
                SELECT support_dept_code,
                       CASE
                           WHEN {REVENUE_SQL} THEN 'revenue'
                           WHEN accounting_category = '원가' THEN 'cost'
                           WHEN accounting_category = '판관비' THEN 'sga'
                           ELSE 'other'
                       END AS breakdown_kind,
                       COALESCE(account_code, '') AS account_code,
                       COALESCE(account_name, '') AS account_name,
                       SUM(COALESCE(amount, 0)) AS total_amount
                FROM transactions
                WHERE COALESCE(support_dept_code, '') <> ''
                  AND support_dept_code NOT IN ('ZZZZZZ')
                  AND COALESCE(support_dept_name, '') <> ''
                  AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
                  AND ({REVENUE_SQL} OR accounting_category IN ('원가', '판관비'))
                  {year_clause}
                GROUP BY support_dept_code, breakdown_kind, account_code, account_name
                ORDER BY support_dept_code, breakdown_kind, total_amount DESC, account_code, account_name
                """
            ),
            params,
        ).mappings().all()

    result: dict[str, dict[str, dict[str, float]]] = {}
    for row in rows:
        code = row["support_dept_code"]
        kind = row["breakdown_kind"]
        if kind == "other":
            continue
        _, _, label = normalize_account_display(row["account_code"], row["account_name"])
        result.setdefault(code, {"revenue": {}, "cost": {}, "sga": {}})
        result[code][kind][label] = result[code][kind].get(label, 0.0) + float(row["total_amount"] or 0)

    normalized_result: dict[str, dict[str, list[dict[str, Any]]]] = {}
    for code, buckets in result.items():
        normalized_result[code] = {}
        for kind, entries in buckets.items():
            normalized_result[code][kind] = [
                {"label": label, "amount": amount}
                for label, amount in sorted(entries.items(), key=lambda item: item[1], reverse=True)
            ]
    return normalized_result


def get_recent_transactions(limit: int = 50) -> list[dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT id,
                       year,
                       month,
                       voucher_number,
                       account_code,
                       account_name,
                       support_dept_code,
                       support_dept_name,
                       cost_dept_code,
                       cost_dept_name,
                       accounting_category,
                       amount,
                       memo1,
                       management_item,
                       source_file,
                       updated_at
                FROM transactions
                ORDER BY COALESCE(year, 0) DESC,
                         COALESCE(month, 0) DESC,
                         id DESC
                LIMIT :limit_count
                """
            ),
            {"limit_count": limit},
        ).mappings().all()
    return [dict(row) for row in rows]


def get_overview_stats(selected_year: int | None = None) -> dict[str, Any]:
    year_clause = ""
    params: dict[str, Any] = {}
    if selected_year:
        year_clause = "WHERE year = :selected_year"
        params["selected_year"] = selected_year
    else:
        recent_10_start_year = get_recent_10_start_year()
        if recent_10_start_year is not None:
            year_clause = "WHERE year >= :recent_10_start_year"
            params["recent_10_start_year"] = recent_10_start_year

    with engine.begin() as conn:
        row = conn.execute(
            text(
                f"""
                SELECT COUNT(*) AS total_rows,
                       COUNT(DISTINCT source_file) AS source_files,
                       COUNT(DISTINCT CASE
                           WHEN COALESCE(support_dept_code, '') <> ''
                                AND support_dept_code NOT IN ('ZZZZZZ')
                                AND COALESCE(support_dept_name, '') <> ''
                                AND support_dept_name NOT IN (
                                    '공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
                                )
                           THEN support_dept_code || '|' || support_dept_name
                       END) AS business_count,
                       SUM(CASE WHEN accounting_category = '원가' THEN amount ELSE 0 END) AS total_cost,
                       SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS total_sga,
                       SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS total_revenue
                FROM transactions
                {year_clause}
                """
            ),
            params,
        ).mappings().first()
    return dict(row) if row else {}


def get_available_years() -> list[int]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT DISTINCT year
                FROM transactions
                WHERE year IS NOT NULL
                ORDER BY year
                """
            )
        ).fetchall()
    return [int(row[0]) for row in rows if row[0] is not None]


def get_financial_series(granularity: str) -> list[dict[str, Any]]:
    group_fields = "year" if granularity == "yearly" else "year, month"
    order_fields = "year" if granularity == "yearly" else "year, month"
    month_where = "" if granularity == "yearly" else "AND month IS NOT NULL"
    project_cost_sql = (
        "accounting_category = '원가' "
        "AND account_code NOT LIKE '5012%' "
        "AND account_code NOT LIKE '5017%' "
        "AND support_dept_code <> 'ZZZZZZ' "
        f"AND {FIELD_COST_DEPT_SQL}"
    )
    support_cost_sql = (
        "accounting_category = '원가' "
        "AND account_code NOT LIKE '5012%' "
        "AND account_code NOT LIKE '5017%' "
        "AND support_dept_code = 'ZZZZZZ' "
        f"AND {FIELD_COST_DEPT_SQL}"
    )
    support_sga_sql = f"accounting_category = '판관비' AND {SUPPORT_COST_DEPT_SQL}"
    field_sga_sql = f"accounting_category = '판관비' AND {FIELD_COST_DEPT_SQL}"

    with engine.begin() as conn:
        rows = conn.execute(
            text(
                f"""
                SELECT {group_fields},
                       SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS revenue_sum,
                       SUM(CASE WHEN {project_cost_sql} THEN amount ELSE 0 END) AS project_cost_sum,
                       SUM(CASE WHEN {support_cost_sql} THEN amount ELSE 0 END) AS support_cost_sum,
                       SUM(CASE WHEN {support_sga_sql} THEN amount ELSE 0 END) AS support_sga_sum,
                       SUM(CASE WHEN {field_sga_sql} THEN amount ELSE 0 END) AS field_sga_sum,
                       SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_sum,
                       SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_sum
                FROM transactions
                WHERE year IS NOT NULL
                  {month_where}
                GROUP BY {group_fields}
                ORDER BY {order_fields}
                """
            )
        ).mappings().all()

    result: list[dict[str, Any]] = []
    for row in rows:
        item = dict(row)
        item["total_expense"] = (
            (item.get("project_cost_sum") or 0)
            + (item.get("support_cost_sum") or 0)
            + (item.get("support_sga_sum") or 0)
            + (item.get("field_sga_sum") or 0)
            + (item.get("labor_sum") or 0)
            + (item.get("outsourcing_sum") or 0)
        )
        item["operating_balance"] = (item.get("revenue_sum") or 0) - item["total_expense"]
        item["label"] = str(item["year"]) if granularity == "yearly" else f"{item['year']}-{int(item['month']):02d}"
        result.append(item)
    return result


def get_source_files_summary() -> list[dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(
            text(
                """
                SELECT source_file, COUNT(*) AS row_count
                FROM transactions
                WHERE COALESCE(source_file, '') <> ''
                GROUP BY source_file
                ORDER BY row_count DESC, source_file
                """
            )
        ).mappings().all()
    return [dict(row) for row in rows]


def parse_excel_upload(upload_file: UploadFile) -> int:
    init_db()
    workbook = load_workbook(upload_file.file, data_only=True)
    import_kind = detect_excel_import_kind(workbook, upload_file.filename or "")
    if import_kind == "contract_status":
        return import_contract_status_workbook(workbook, upload_file.filename or "")
    if import_kind == "change_contract_summary":
        return import_change_contract_summary_workbook(workbook, upload_file.filename or "")
    if import_kind == "change_contract_round":
        return import_change_contract_round_workbook(workbook, upload_file.filename or "")
    if import_kind == "billing_status":
        return import_billing_status_workbook(workbook, upload_file.filename or "")

    sheet = workbook.active
    headers = [canonical_header_name(cell.value) for cell in next(sheet.iter_rows(min_row=1, max_row=1))]
    inserted = 0

    for row in sheet.iter_rows(min_row=2, values_only=True):
        raw: dict[str, Any] = {}
        has_value = False
        for index, value in enumerate(row):
            field_name = headers[index] if index < len(headers) else None
            if field_name:
                raw[field_name] = value
            if normalize_text(value):
                has_value = True

        if not has_value:
            continue

        payload = build_transaction_payload(raw, source_file=upload_file.filename or "")
        if not payload["voucher_number"] and not payload["account_code"] and not payload["account_name"]:
            continue
        save_transaction(payload)
        inserted += 1

    return inserted


def import_excel_path(path: Path) -> int:
    with path.open("rb") as excel_file:
        upload = UploadFile(filename=path.name, file=excel_file)
        return parse_excel_upload(upload)


def auto_import_project_excels() -> None:
    init_db()
    excel_files = [
        path
        for path in sorted(BASE_DIR.glob("*.xlsx"))
        if not path.name.startswith("~$")
    ]
    if not excel_files:
        return

    known_files = existing_source_files()
    known_contract_files = existing_contract_source_files()
    known_billing_files = existing_billing_source_files()
    known_change_summary_files = existing_change_contract_summary_source_files()
    known_change_round_files = existing_change_contract_round_source_files()
    if count_transactions() > 0 and all(file.name in known_files for file in excel_files):
        if all(
            file.name in known_contract_files
            or file.name in known_billing_files
            or file.name in known_change_summary_files
            or file.name in known_change_round_files
            for file in excel_files
        ):
            return

    for excel_path in excel_files:
        if not zipfile.is_zipfile(excel_path):
            logger.warning("Skipping non-Excel or temporary workbook during auto-import: %s", excel_path.name)
            continue
        try:
            workbook = load_workbook(excel_path, data_only=True)
        except zipfile.BadZipFile:
            logger.warning("Skipping invalid workbook during auto-import: %s", excel_path.name)
            continue
        except Exception:
            logger.exception("Failed to inspect workbook during auto-import: %s", excel_path.name)
            continue
        import_kind = detect_excel_import_kind(workbook, excel_path.name)
        if import_kind == "contract_status" and excel_path.name in known_contract_files:
            continue
        if import_kind == "change_contract_summary" and excel_path.name in known_change_summary_files:
            continue
        if import_kind == "change_contract_round" and excel_path.name in known_change_round_files:
            continue
        if import_kind == "billing_status" and excel_path.name in known_billing_files:
            continue
        if import_kind == "transactions" and excel_path.name in known_files:
            continue
        try:
            with excel_path.open("rb") as excel_file:
                upload = UploadFile(filename=excel_path.name, file=excel_file)
                inserted = parse_excel_upload(upload)
            logger.info("Auto-imported %s rows from %s", inserted, excel_path.name)
        except Exception:
            logger.exception("Failed to auto-import workbook: %s", excel_path.name)


def normalize_all_collection_entry_storage() -> None:
    with engine.begin() as conn:
        entry_rows = conn.execute(
            text(
                """
                SELECT id, support_dept_code, vendor, progress_type, billing_round, billing_type,
                       billing_date, billed_amount, round, date, due_date, amount,
                       balance_amount, collection_rate, note
                FROM project_collection_entries
                ORDER BY support_dept_code, position, id
                """
            )
        ).mappings().all()
        grouped_entries: dict[str, list[dict[str, Any]]] = {}
        for row in entry_rows:
            row_dict = dict(row)
            entry_id = row_dict.pop("id", None)
            support_dept_code = normalize_text(row_dict.pop("support_dept_code", ""))
            normalized = normalize_collection_entry_row(row_dict)
            grouped_entries.setdefault(support_dept_code, []).append(normalized)
            if (
                normalize_text(row.get("progress_type")) != normalized["progress_type"]
                or normalize_text(row.get("billing_type")) != normalized["billing_type"]
            ):
                conn.execute(
                    text(
                        """
                        UPDATE project_collection_entries
                        SET progress_type = :progress_type,
                            billing_type = :billing_type,
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = :id
                        """
                    ),
                    {
                        "id": entry_id,
                        "progress_type": normalized["progress_type"],
                        "billing_type": normalized["billing_type"],
                    },
                )

        cached_rows = conn.execute(
            text(
                """
                SELECT support_dept_code, collection_entries_json
                FROM project_status
                WHERE collection_entries_json IS NOT NULL AND collection_entries_json <> ''
                """
            )
        ).mappings().all()
        for row in cached_rows:
            support_dept_code = normalize_text(row["support_dept_code"])
            normalized_entries = grouped_entries.get(support_dept_code)
            if normalized_entries is None:
                raw_entries = decode_json_rows(row["collection_entries_json"])
                normalized_entries = [normalize_collection_entry_row(item) for item in raw_entries]
            conn.execute(
                text(
                    """
                    UPDATE project_status
                    SET collection_entries_json = :collection_entries_json,
                        updated_at = CURRENT_TIMESTAMP
                    WHERE support_dept_code = :support_dept_code
                    """
                ),
                {
                    "support_dept_code": support_dept_code,
                    "collection_entries_json": encode_json_rows(normalized_entries),
                },
            )


def parse_manual_form(raw_body: bytes) -> dict[str, Any]:
    parsed = parse_qs(raw_body.decode("utf-8"))
    payload = {key: values[0] if values else "" for key, values in parsed.items()}
    return payload


def parse_project_form(raw_body: bytes) -> dict[str, Any]:
    parsed = parse_qs(raw_body.decode("utf-8"))
    payload: dict[str, Any] = {}
    for key, values in parsed.items():
        payload[key] = values if key.endswith("[]") else (values[0] if values else "")
    return payload


def build_named_amount_rows(
    labels: list[Any],
    amounts: list[Any],
    *,
    label_key: str = "label",
    amount_key: str = "amount",
) -> list[dict[str, Any]]:
    rows = []
    for index, label in enumerate(labels):
        rows.append(
            {
                label_key: label,
                amount_key: amounts[index] if index < len(amounts) else "",
            }
        )
    return filter_amount_rows(rows, amount_key=amount_key)


def build_triplet_amount_rows(
    first_values: list[Any],
    second_values: list[Any],
    amounts: list[Any],
    *,
    first_key: str,
    second_key: str,
    amount_key: str = "amount",
) -> list[dict[str, Any]]:
    max_length = max(len(first_values), len(second_values), len(amounts))
    rows = []
    for index in range(max_length):
        rows.append(
            {
                first_key: first_values[index] if index < len(first_values) else "",
                second_key: second_values[index] if index < len(second_values) else "",
                amount_key: amounts[index] if index < len(amounts) else "",
            }
        )
    return filter_amount_rows(rows, amount_key=amount_key)


def build_collection_rows(payload: dict[str, Any]) -> list[dict[str, Any]]:
    progress_types = payload.get("collection_progress_type[]", [])
    billing_rounds = payload.get("collection_billing_round[]", [])
    billing_types = payload.get("collection_billing_type[]", [])
    billing_dates = payload.get("collection_billing_date[]", [])
    billed_amounts = payload.get("collection_billed_amount[]", [])
    collection_rounds = payload.get("collection_round[]", [])
    collection_dates = payload.get("collection_date[]", [])
    collection_amounts = payload.get("collection_amount_row[]", [])
    rows = []
    total_rows = max(
        len(progress_types),
        len(billing_rounds),
        len(billing_types),
        len(billing_dates),
        len(billed_amounts),
        len(collection_rounds),
        len(collection_dates),
        len(collection_amounts),
    )
    for index in range(total_rows):
        rows.append(
            {
                "progress_type": progress_types[index] if index < len(progress_types) else "",
                "billing_round": billing_rounds[index] if index < len(billing_rounds) else "",
                "billing_type": billing_types[index] if index < len(billing_types) else "",
                "billing_date": billing_dates[index] if index < len(billing_dates) else "",
                "billed_amount": billed_amounts[index] if index < len(billed_amounts) else "",
                "round": collection_rounds[index] if index < len(collection_rounds) else "",
                "date": collection_dates[index] if index < len(collection_dates) else "",
                "amount": collection_amounts[index] if index < len(collection_amounts) else "",
            }
        )
    filtered_rows: list[dict[str, Any]] = []
    for row in rows:
        normalized_row = {key: clean_row_text(value) for key, value in row.items()}
        amount = normalize_amount(normalized_row.get("amount"))
        billed_amount = normalize_amount(normalized_row.get("billed_amount"))
        has_other_value = any(
            value for key, value in normalized_row.items()
            if key not in {"amount", "billed_amount"}
        )
        if amount or billed_amount or has_other_value:
            normalized_row["amount"] = amount
            normalized_row["billed_amount"] = billed_amount
            filtered_rows.append(normalized_row)
    for row in filtered_rows:
        normalized_fields = normalize_collection_entry_fields(row)
        row["progress_type"] = normalized_fields["progress_type"]
        row["billing_type"] = normalized_fields["billing_type"]
        row["billing_round"] = normalize_round_value(row.get("billing_round"))
        row["round"] = normalize_round_value(row.get("round"))
        row["billing_date"] = normalize_date_text(row.get("billing_date"))
        row["date"] = normalize_date_text(row.get("date"))
        row["billed_amount"] = normalize_amount(row.get("billed_amount"))
    return filtered_rows


def build_project_status_payload(payload: dict[str, Any]) -> dict[str, Any]:
    contract_amount = normalize_amount(payload.get("contract_amount"))
    collection_rows = build_collection_rows(payload)
    collection_amount = sum_row_amounts(collection_rows)
    progress_rate = (collection_amount / contract_amount * 100) if contract_amount else 0.0

    task_plan_department_rows = build_triplet_amount_rows(
        payload.get("task_plan_department_dept[]", []),
        payload.get("task_plan_department_work[]", []),
        payload.get("task_plan_department_amount[]", []),
        first_key="dept_name",
        second_key="work_name",
    )
    for row in task_plan_department_rows:
        row["group"] = "department"
    task_plan_outsource_rows = build_triplet_amount_rows(
        payload.get("task_plan_outsource_dept[]", []),
        payload.get("task_plan_outsource_work[]", []),
        payload.get("task_plan_outsource_amount[]", []),
        first_key="dept_name",
        second_key="work_name",
    )
    for row in task_plan_outsource_rows:
        row["group"] = "outsource"
    task_plan_joint_rows = build_triplet_amount_rows(
        payload.get("task_plan_joint_dept[]", []),
        payload.get("task_plan_joint_work[]", []),
        payload.get("task_plan_joint_amount[]", []),
        first_key="dept_name",
        second_key="work_name",
    )
    for row in task_plan_joint_rows:
        row["group"] = "joint"
    task_plan_rows = task_plan_department_rows + task_plan_outsource_rows + task_plan_joint_rows

    exec_labor_rows = build_triplet_amount_rows(
        payload.get("exec_labor_grade[]", []),
        payload.get("exec_labor_hours[]", []),
        payload.get("exec_labor_amount[]", []),
        first_key="grade",
        second_key="hours",
    )
    for row in exec_labor_rows:
        row["group"] = "labor"
    exec_outsource_rows = build_triplet_amount_rows(
        payload.get("exec_outsource_dept[]", []),
        payload.get("exec_outsource_work[]", []),
        payload.get("exec_outsource_amount[]", []),
        first_key="dept_name",
        second_key="work_name",
    )
    for row in exec_outsource_rows:
        row["group"] = "outsource"
    exec_cost_plan_rows = build_triplet_amount_rows(
        payload.get("exec_cost_plan_code[]", []),
        payload.get("exec_cost_plan_name[]", []),
        payload.get("exec_cost_plan_amount[]", []),
        first_key="account_code",
        second_key="account_name",
    )
    for row in exec_cost_plan_rows:
        row["group"] = "cost_plan"
    exec_budget_rows = exec_labor_rows + exec_outsource_rows + exec_cost_plan_rows

    actual_labor_grades = payload.get("actual_labor_grade[]", [])
    actual_labor_minutes = payload.get("actual_labor_minutes[]", [])
    actual_labor_amounts = payload.get("actual_labor_amount[]", [])
    actual_labor_rows: list[dict[str, Any]] = []
    actual_labor_max_length = max(
        len(actual_labor_grades),
        len(actual_labor_minutes),
        len(actual_labor_amounts),
    )
    for index in range(actual_labor_max_length):
        row = {
            "grade": actual_labor_grades[index] if index < len(actual_labor_grades) else "",
            "minutes": actual_labor_minutes[index] if index < len(actual_labor_minutes) else "",
            "amount": actual_labor_amounts[index] if index < len(actual_labor_amounts) else "",
        }
        normalized_row = {key: clean_row_text(value) for key, value in row.items()}
        amount = normalize_amount(normalized_row.get("amount"))
        has_other_value = any(
            value for key, value in normalized_row.items()
            if key != "amount"
        )
        if amount or has_other_value:
            normalized_row["amount"] = amount
            actual_labor_rows.append(normalized_row)
    for row in actual_labor_rows:
        row["group"] = "labor"
    actual_labor_adjustment_total = normalize_amount(payload.get("actual_labor_adjustment_total"))
    actual_labor_adjustment_rows = []
    if actual_labor_adjustment_total:
        actual_labor_adjustment_rows.append(
            {
                "group": "labor_adjustment",
                "label": "인건비 조정",
                "amount": actual_labor_adjustment_total,
            }
        )

    actual_as_rows = build_named_amount_rows(
        payload.get("actual_as_label[]", []),
        payload.get("actual_as_amount[]", []),
        label_key="label",
        amount_key="amount",
    )
    for row in actual_as_rows:
        row["group"] = "as"

    actual_labor_joint_rows = build_named_amount_rows(
        payload.get("actual_labor_joint_label[]", []),
        payload.get("actual_labor_joint_amount[]", []),
        label_key="label",
        amount_key="amount",
    )
    for row in actual_labor_joint_rows:
        row["group"] = "labor_joint"

    actual_sga_rows = build_named_amount_rows(
        payload.get("actual_sga_label[]", []),
        payload.get("actual_sga_amount[]", []),
        label_key="label",
        amount_key="amount",
    )
    for row in actual_sga_rows:
        row["group"] = "sga"

    actual_input_rows = actual_labor_rows + actual_labor_adjustment_rows + actual_labor_joint_rows + actual_as_rows + actual_sga_rows

    if not actual_input_rows:
        legacy_refs = payload.get("actual_input_ref[]", [])
        legacy_amounts = payload.get("actual_input_amount[]", [])
        legacy_notes = payload.get("actual_input_note[]", [])
        for index, ref in enumerate(legacy_refs):
            actual_input_rows.append(
                {
                    "reference": ref,
                    "amount": legacy_amounts[index] if index < len(legacy_amounts) else "",
                    "note": legacy_notes[index] if index < len(legacy_notes) else "",
                }
            )
        actual_input_rows = filter_amount_rows(actual_input_rows, amount_key="amount")

    expected_as_rate = normalize_amount(payload.get("expected_as_rate"))
    expected_sga_rate = normalize_amount(payload.get("expected_sga_rate"))
    expected_as_cost = normalize_amount(payload.get("expected_as_cost"))
    expected_sga_budget = normalize_amount(payload.get("expected_sga_budget"))
    if not expected_as_cost and contract_amount and expected_as_rate:
        expected_as_cost = contract_amount * expected_as_rate / 100
    if not expected_sga_budget and contract_amount and expected_sga_rate:
        expected_sga_budget = contract_amount * expected_sga_rate / 100
    exec_labor_rates = normalize_text(payload.get("exec_labor_rates_json")) or "{}"

    return {
        "support_dept_code": normalize_text(payload.get("support_dept_code")),
        "support_dept_name": normalize_text(payload.get("support_dept_name")),
        "progress_rate": progress_rate,
        "contract_amount": contract_amount,
        "collection_amount": collection_amount,
        "collection_entries_json": encode_json_rows(collection_rows),
        "change_round": normalize_text(payload.get("change_round")),
        "item_investment": sum_row_amounts(actual_input_rows),
        "task_plan_department_budget": sum_row_amounts(task_plan_department_rows),
        "task_plan_outsource_budget": sum_row_amounts(task_plan_outsource_rows),
        "task_plan_outsource_detail": "\n".join(
            f"{normalize_text(row.get('dept_name'))} / {normalize_text(row.get('work_name'))}: {format_amount_for_text(row.get('amount'))}".strip(" /:")
            for row in task_plan_outsource_rows
        ),
        "task_plan_joint_operating_cost": sum_row_amounts(task_plan_joint_rows),
        "task_plan_entries_json": encode_json_rows(task_plan_rows),
        "exec_budget_labor_by_grade": sum_row_amounts(exec_labor_rows),
        "exec_labor_rates_json": exec_labor_rates,
        "exec_budget_outsource": sum_row_amounts(exec_outsource_rows),
        "exec_budget_cost_plan": sum_row_amounts(exec_cost_plan_rows),
        "exec_budget_entries_json": encode_json_rows(exec_budget_rows),
        "actual_input_entries_json": encode_json_rows(actual_input_rows),
        "project_type": normalize_text(payload.get("project_type")),
        "expected_as_rate": expected_as_rate,
        "expected_sga_rate": expected_sga_rate,
        "expected_as_cost": expected_as_cost,
        "expected_sga_budget": expected_sga_budget,
        "last_editor_session_id": normalize_text(payload.get("client_session_id")),
        "last_client_submitted_at": normalize_text(payload.get("client_submitted_at")),
        "project_start_date": normalize_date_text(payload.get("project_start_date")),
        "project_end_date": normalize_date_text(payload.get("project_end_date")),
        "completion_status": normalize_text(payload.get("completion_status")),
        "notes": normalize_text(payload.get("notes")),
        "_collection_rows": collection_rows,
        "_task_plan_rows": task_plan_rows,
        "_exec_budget_rows": exec_budget_rows,
        "_actual_input_rows": actual_input_rows,
    }


def project_status_payload_has_meaningful_data(payload: dict[str, Any]) -> bool:
    if normalize_amount(payload.get("contract_amount")):
        return True
    if normalize_amount(payload.get("collection_amount")):
        return True
    if normalize_amount(payload.get("task_plan_department_budget")):
        return True
    if normalize_amount(payload.get("task_plan_outsource_budget")):
        return True
    if normalize_amount(payload.get("task_plan_joint_operating_cost")):
        return True
    if normalize_amount(payload.get("exec_budget_labor_by_grade")):
        return True
    if normalize_amount(payload.get("exec_budget_outsource")):
        return True
    if normalize_amount(payload.get("exec_budget_cost_plan")):
        return True
    if normalize_amount(payload.get("item_investment")):
        return True
    for key in (
        "project_type",
        "project_start_date",
        "project_end_date",
        "completion_status",
        "notes",
        "change_round",
        "support_dept_name",
        "task_plan_outsource_detail",
    ):
        if normalize_text(payload.get(key)):
            return True
    for key in (
        "collection_entries_json",
        "task_plan_entries_json",
        "exec_budget_entries_json",
        "actual_input_entries_json",
    ):
        if decode_json_rows(payload.get(key)):
            return True
    return False


def format_amount_for_text(value: Any) -> str:
    amount = normalize_amount(value)
    return f"{amount:,.0f}"


def save_project_status(payload: dict[str, Any]) -> None:
    normalized_payload = build_project_status_payload(payload)
    support_dept_code = normalize_text(normalized_payload.get("support_dept_code"))
    if not support_dept_code:
        return
    collection_rows = normalized_payload.pop("_collection_rows", [])
    task_plan_rows = normalized_payload.pop("_task_plan_rows", [])
    exec_budget_rows = normalized_payload.pop("_exec_budget_rows", [])
    actual_input_rows = normalized_payload.pop("_actual_input_rows", [])

    basic_info_field_keys = (
        "support_dept_name",
        "contract_amount",
        "project_type",
        "expected_as_rate",
        "expected_sga_rate",
        "expected_as_cost",
        "expected_sga_budget",
        "exec_labor_rates_json",
        "change_round",
        "project_start_date",
        "project_end_date",
        "completion_status",
        "notes",
        "last_editor_session_id",
        "last_client_submitted_at",
    )
    collection_field_keys = (
        "collection_progress_type[]",
        "collection_billing_round[]",
        "collection_billing_type[]",
        "collection_billing_date[]",
        "collection_billed_amount[]",
        "collection_round[]",
        "collection_date[]",
        "collection_amount_row[]",
    )
    task_plan_field_keys = (
        "task_plan_department_dept[]",
        "task_plan_department_work[]",
        "task_plan_department_amount[]",
        "task_plan_outsource_dept[]",
        "task_plan_outsource_work[]",
        "task_plan_outsource_amount[]",
        "task_plan_joint_dept[]",
        "task_plan_joint_work[]",
        "task_plan_joint_amount[]",
    )
    exec_budget_field_keys = (
        "exec_labor_grade[]",
        "exec_labor_hours[]",
        "exec_labor_amount[]",
        "exec_outsource_dept[]",
        "exec_outsource_work[]",
        "exec_outsource_amount[]",
        "exec_cost_plan_code[]",
        "exec_cost_plan_name[]",
        "exec_cost_plan_amount[]",
    )
    actual_input_field_keys = (
        "actual_labor_grade[]",
        "actual_labor_minutes[]",
        "actual_labor_amount[]",
        "actual_labor_adjustment_total",
        "actual_labor_joint_label[]",
        "actual_labor_joint_amount[]",
        "actual_as_label[]",
        "actual_as_amount[]",
        "actual_sga_label[]",
        "actual_sga_amount[]",
    )

    def payload_has_any(keys: tuple[str, ...]) -> bool:
        return any(key in payload for key in keys)

    with engine.begin() as conn:
        existing_row = conn.execute(
            text("SELECT * FROM project_status WHERE support_dept_code = :support_dept_code"),
            {"support_dept_code": support_dept_code},
        ).mappings().first()
        if existing_row:
            existing_payload = dict(existing_row)
            if (
                project_status_payload_has_meaningful_data(existing_payload)
                and not project_status_payload_has_meaningful_data(normalized_payload)
            ):
                raise ValueError("기존 입력값을 불러오지 않은 빈 상태로는 저장할 수 없습니다.")
        existing_basic_info = conn.execute(
            text("SELECT * FROM project_basic_info WHERE support_dept_code = :support_dept_code"),
            {"support_dept_code": support_dept_code},
        ).mappings().first()
        existing_entry_set = load_project_status_entries_for_code(conn, support_dept_code)

        if existing_basic_info:
            for key in basic_info_field_keys:
                if key not in payload:
                    normalized_payload[key] = existing_basic_info.get(key)
        if not payload_has_any(collection_field_keys):
            collection_rows = existing_entry_set["collection_entries"]
        if not payload_has_any(task_plan_field_keys):
            task_plan_rows = existing_entry_set["task_plan_entries"]
        else:
            existing_task_rows = existing_entry_set["task_plan_entries"]
            if not payload_has_any(("task_plan_department_dept[]", "task_plan_department_work[]", "task_plan_department_amount[]")):
                task_plan_rows.extend(row for row in existing_task_rows if normalize_text(row.get("group")) == "department")
            if not payload_has_any(("task_plan_outsource_dept[]", "task_plan_outsource_work[]", "task_plan_outsource_amount[]")):
                task_plan_rows.extend(row for row in existing_task_rows if normalize_text(row.get("group")) == "outsource")
            if not payload_has_any(("task_plan_joint_dept[]", "task_plan_joint_work[]", "task_plan_joint_amount[]")):
                task_plan_rows.extend(row for row in existing_task_rows if normalize_text(row.get("group")) == "joint")
        if not payload_has_any(exec_budget_field_keys):
            exec_budget_rows = existing_entry_set["exec_budget_entries"]
        else:
            existing_exec_rows = existing_entry_set["exec_budget_entries"]
            if not payload_has_any(("exec_labor_grade[]", "exec_labor_hours[]", "exec_labor_amount[]")):
                exec_budget_rows.extend(row for row in existing_exec_rows if normalize_text(row.get("group")) == "labor")
            if not payload_has_any(("exec_outsource_dept[]", "exec_outsource_work[]", "exec_outsource_amount[]")):
                exec_budget_rows.extend(row for row in existing_exec_rows if normalize_text(row.get("group")) == "outsource")
            if not payload_has_any(("exec_cost_plan_code[]", "exec_cost_plan_name[]", "exec_cost_plan_amount[]")):
                exec_budget_rows.extend(row for row in existing_exec_rows if normalize_text(row.get("group")) == "cost_plan")
        if not payload_has_any(actual_input_field_keys):
            actual_input_rows = existing_entry_set["actual_input_entries"]
        else:
            existing_actual_rows = existing_entry_set["actual_input_entries"]
            if not payload_has_any(("actual_labor_grade[]", "actual_labor_minutes[]", "actual_labor_amount[]")):
                actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "labor")
            if "actual_labor_adjustment_total" not in payload:
                actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "labor_adjustment")
            if not payload_has_any(("actual_labor_joint_label[]", "actual_labor_joint_amount[]")):
                actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "labor_joint")
            if not payload_has_any(("actual_as_label[]", "actual_as_amount[]")):
                actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "as")
            if not payload_has_any(("actual_sga_label[]", "actual_sga_amount[]")):
                actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "sga")

        normalized_payload["collection_amount"] = sum_row_amounts(collection_rows)
        normalized_payload["progress_rate"] = (
            normalized_payload["collection_amount"] / normalize_amount(normalized_payload.get("contract_amount")) * 100
            if normalize_amount(normalized_payload.get("contract_amount"))
            else 0.0
        )
        task_plan_department_rows = [row for row in task_plan_rows if normalize_text(row.get("group")) == "department"]
        task_plan_outsource_rows = [row for row in task_plan_rows if normalize_text(row.get("group")) == "outsource"]
        task_plan_joint_rows = [row for row in task_plan_rows if normalize_text(row.get("group")) == "joint"]
        exec_labor_rows = [row for row in exec_budget_rows if normalize_text(row.get("group")) == "labor"]
        exec_outsource_rows = [row for row in exec_budget_rows if normalize_text(row.get("group")) == "outsource"]
        exec_cost_plan_rows = [row for row in exec_budget_rows if normalize_text(row.get("group")) == "cost_plan"]
        normalized_payload["collection_entries_json"] = encode_json_rows(collection_rows)
        normalized_payload["task_plan_department_budget"] = sum_row_amounts(task_plan_department_rows)
        normalized_payload["task_plan_outsource_budget"] = sum_row_amounts(task_plan_outsource_rows)
        normalized_payload["task_plan_outsource_detail"] = "\n".join(
            f"{normalize_text(row.get('dept_name'))} / {normalize_text(row.get('work_name'))}: {format_amount_for_text(row.get('amount'))}".strip(" /:")
            for row in task_plan_outsource_rows
        )
        normalized_payload["task_plan_joint_operating_cost"] = sum_row_amounts(task_plan_joint_rows)
        normalized_payload["task_plan_entries_json"] = encode_json_rows(task_plan_rows)
        normalized_payload["exec_budget_labor_by_grade"] = sum_row_amounts(exec_labor_rows)
        normalized_payload["exec_budget_outsource"] = sum_row_amounts(exec_outsource_rows)
        normalized_payload["exec_budget_cost_plan"] = sum_row_amounts(exec_cost_plan_rows)
        normalized_payload["exec_budget_entries_json"] = encode_json_rows(exec_budget_rows)
        normalized_payload["item_investment"] = sum_row_amounts(actual_input_rows)
        normalized_payload["actual_input_entries_json"] = encode_json_rows(actual_input_rows)
        check_record_revision(
            conn,
            "project_status",
            "support_dept_code",
            support_dept_code,
            normalize_text(payload.get("edit_revision")),
        )
        save_project_basic_info_section(conn, normalized_payload)
        replace_project_status_child_entries(
            conn,
            support_dept_code,
            collection_rows,
            task_plan_rows,
            exec_budget_rows,
            actual_input_rows,
        )
        sync_project_status_cache_row(conn, support_dept_code)


def base_context(request: Request, message: str = "") -> dict[str, Any]:
    health_payload = build_health_payload()
    return {
        "request": request,
        "message": message,
        "data_version": health_payload["data_version"],
        "server_time": health_payload["server_time"],
        "import_sync_summary": get_import_sync_summary(),
    }


def render_home(
    request: Request,
    edit_id: int | None = None,
    message: str = "",
    overview_year: int | None = None,
) -> HTMLResponse:
    init_db()
    available_years = get_available_years()
    context = {
        **base_context(request, message),
        "overview": get_overview_stats(overview_year),
        "overview_selected_year": overview_year,
        "project_dashboard": get_project_dashboard_summary(overview_year),
        "project_revenue_mix_yearly": get_project_revenue_mix(),
        "project_revenue_mix_monthly": get_project_revenue_mix_monthly(),
        "yearly_summary": get_yearly_summary(),
        "monthly_summary": get_monthly_summary(),
        "available_years": available_years,
        "dashboard_revenue_metric_options": get_option_items("dashboard_revenue_metrics"),
        "dashboard_expense_metric_options": get_option_items("dashboard_expense_metrics"),
    }
    return templates.TemplateResponse(request, "dashboard.html", context)


def render_projects_page(
    request: Request,
    edit_code: str | None = None,
    focus_code: str | None = None,
    selected_year: int | None = None,
    message: str = "",
) -> HTMLResponse:
    init_db()
    selected_year = resolve_selected_year(selected_year)
    context = {
        **base_context(request, message),
        "project_year_options": get_project_year_options(),
        "selected_year": selected_year,
        "project_dashboard": get_project_dashboard_summary(selected_year),
        "project_revenue_mix": get_project_revenue_mix(selected_year),
        "project_cost_by_year": get_project_cost_by_year(selected_year),
        "project_monthly_cost_rows": get_business_monthly_summary(),
        "project_account_breakdowns": get_project_account_breakdowns(selected_year),
        "project_status_rows": get_project_status_rows(),
        "project_comparison_notes": get_project_comparison_notes_map(),
        "project_edit": get_project_status_for_edit(edit_code),
        "project_focus_code": normalize_text(focus_code),
        "project_page_state": get_project_page_state(),
        "project_related_links": get_project_related_links_map(),
        "project_uncontracted_classifications": get_project_uncontracted_classification_map(),
        "support_department_options": get_support_department_options(),
        "cost_department_options": get_cost_department_options(),
        "cost_account_options": get_cost_account_options(),
        "labor_grade_options": get_labor_grade_options(),
        "expected_as_rate_options": get_expected_as_rate_options(),
        "expected_sga_rate_options": get_expected_sga_rate_options(),
        "collection_progress_type_options": get_collection_progress_type_options(),
        "collection_billing_type_options": get_collection_billing_type_options(),
        "uncontracted_category_options": get_uncontracted_category_options(),
        "special_x_classification_rules": get_special_x_classification_rules(),
        "project_runtime_settings": get_project_runtime_settings(),
    }
    return templates.TemplateResponse(request, "projects.html", context)


def render_annual_summary_page(request: Request, message: str = "") -> HTMLResponse:
    init_db()
    context = {
        **base_context(request, message),
        "available_years": get_available_years(),
        "yearly_financial_series": get_financial_series("yearly"),
        "monthly_financial_series": get_financial_series("monthly"),
        "annual_metric_cards": get_option_items("annual_metric_cards"),
        "annual_expense_chart_metrics": get_option_items("annual_expense_chart_metrics"),
        "annual_balance_chart_metrics": get_option_items("annual_balance_chart_metrics"),
    }
    return templates.TemplateResponse(request, "annual_summary.html", context)


@app.get("/health")
async def health() -> dict[str, str]:
    return build_health_payload()


@app.get("/")
async def home(request: Request, edit_id: int | None = None, overview_year: int | None = None):
    try:
        return render_home(request, edit_id=edit_id, overview_year=overview_year)
    except Exception as exc:
        logger.exception("홈페이지 에러: %s", exc)
        return HTMLResponse("<h1>서버 오류</h1><p>로그를 확인해주세요.</p>", status_code=500)


@app.get("/projects")
async def projects(
    request: Request,
    edit_code: str | None = None,
    focus_code: str | None = None,
    year: str | None = None,
):
    try:
        return render_projects_page(
            request,
            edit_code=edit_code,
            focus_code=focus_code,
            selected_year=parse_optional_year(year),
        )
    except Exception as exc:
        logger.exception("사업현황 페이지 에러: %s", exc)
        return HTMLResponse("<h1>서버 오류</h1><p>로그를 확인해주세요.</p>", status_code=500)


@app.get("/projects/edit-data")
async def project_edit_data(code: str | None = None):
    try:
        return JSONResponse(content=jsonable_encoder(get_project_status_for_edit(code)))
    except Exception as exc:
        logger.exception("사업현황 편집 데이터 조회 에러: %s", exc)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


@app.post("/projects/page-state")
async def project_page_state_save(request: Request):
    try:
        payload = await request.json()
        if not isinstance(payload, dict):
            raise ValueError("잘못된 페이지 상태 형식입니다.")
        save_project_page_state(payload)
        return JSONResponse(content={"status": "ok"})
    except Exception as exc:
        logger.exception("사업현황 페이지 상태 저장 에러: %s", exc)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


@app.get("/projects/page-state")
async def project_page_state_load(session_id: str = ""):
    try:
        return JSONResponse(content=jsonable_encoder(get_project_page_state(session_id)))
    except Exception as exc:
        logger.exception("사업현황 페이지 상태 조회 에러: %s", exc)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


@app.post("/projects/related-links")
async def project_related_links_save(request: Request):
    try:
        payload = await request.json()
        if not isinstance(payload, dict):
            raise ValueError("잘못된 연관 프로젝트 형식입니다.")
        base_code = normalize_text(payload.get("base_code"))
        related_codes = payload.get("related_codes") or []
        if not isinstance(related_codes, list):
            raise ValueError("연관 프로젝트 목록 형식이 올바르지 않습니다.")
        save_project_related_links(base_code, related_codes)
        return JSONResponse(content={"status": "ok"})
    except Exception as exc:
        logger.exception("연관 프로젝트 저장 에러: %s", exc)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


@app.get("/projects/quick-links")
async def project_quick_links_load(session_id: str = ""):
    try:
        return JSONResponse(content={"codes": get_project_quick_links(session_id)})
    except Exception as exc:
        logger.exception("프로젝트 바로가기 조회 에러: %s", exc)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


@app.post("/projects/quick-links")
async def project_quick_links_save(request: Request):
    try:
        payload = await request.json()
        if not isinstance(payload, dict):
            raise ValueError("잘못된 바로가기 형식입니다.")
        codes = payload.get("codes") or []
        if not isinstance(codes, list):
            raise ValueError("바로가기 목록 형식이 잘못되었습니다.")
        session_id = normalize_text(payload.get("session_id"))
        save_project_quick_links(session_id, [normalize_text(code) for code in codes if isinstance(code, str)])
        return JSONResponse(content={"status": "ok", "codes": get_project_quick_links(session_id)})
    except Exception as exc:
        logger.exception("프로젝트 바로가기 저장 에러: %s", exc)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


@app.post("/projects/uncontracted-category")
async def project_uncontracted_category_save(request: Request):
    try:
        payload = await request.json()
        if not isinstance(payload, dict):
            raise ValueError("잘못된 미계약 분류 형식입니다.")
        save_project_uncontracted_classification(
            payload.get("support_dept_code"),
            payload.get("category"),
        )
        return JSONResponse(content={"status": "ok"})
    except Exception as exc:
        logger.exception("미계약 분류 저장 에러: %s", exc)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


@app.post("/projects/runtime-setting")
async def project_runtime_setting_save(request: Request):
    try:
        payload = await request.json()
        if not isinstance(payload, dict):
            raise ValueError("잘못된 설정 형식입니다.")
        save_project_runtime_setting(payload.get("item_key"), payload.get("value"))
        return JSONResponse(content={"status": "ok", "settings": get_project_runtime_settings()})
    except Exception as exc:
        logger.exception("프로젝트 런타임 설정 저장 에러: %s", exc)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


@app.post("/projects/comparison-note")
async def project_comparison_note_save(request: Request):
    try:
        payload = await request.json()
        if not isinstance(payload, dict):
            raise ValueError("잘못된 비고 형식입니다.")
        save_project_comparison_note(
            payload.get("support_dept_code"),
            payload.get("item_key"),
            payload.get("note"),
        )
        return JSONResponse(content={"status": "ok"})
    except Exception as exc:
        logger.exception("계획 대비 실제 비교 비고 저장 에러: %s", exc)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


@app.get("/annual-summary")
async def annual_summary(request: Request):
    try:
        return render_annual_summary_page(request)
    except Exception as exc:
        logger.exception("연도별 수익 비용 정리 페이지 에러: %s", exc)
        return HTMLResponse("<h1>서버 오류</h1><p>로그를 확인해주세요.</p>", status_code=500)


@app.post("/upload")
async def upload_excel(request: Request, excel_file: UploadFile = File(...)):
    try:
        inserted = parse_excel_upload(excel_file)
        return render_home(request, message=f"{inserted}건의 엑셀 데이터를 DB에 저장했습니다.")
    except Exception as exc:
        logger.exception("엑셀 업로드 에러: %s", exc)
        return render_home(request, message=f"엑셀 업로드 중 오류가 발생했습니다: {exc}")


@app.post("/records/save")
async def save_record(request: Request):
    form_data: dict[str, Any] = {}
    try:
        form_data = parse_manual_form(await request.body())
        record_id = normalize_text(form_data.get("id"))
        payload = build_transaction_payload(form_data)
        save_transaction(payload, int(record_id) if record_id else None)
        return RedirectResponse("/", status_code=303)
    except Exception as exc:
        logger.exception("데이터 저장 에러: %s", exc)
        record_id = normalize_text(form_data.get("id"))
        return render_home(
            request,
            edit_id=int(record_id) if record_id else None,
            message=f"데이터 저장 중 오류가 발생했습니다: {exc}",
        )


@app.post("/projects/save")
async def save_project(request: Request):
    form_data: dict[str, Any] = {}
    try:
        form_data = parse_project_form(await request.body())
        save_project_status(form_data)
        code = normalize_text(form_data.get("support_dept_code"))
        selected_year = normalize_text(form_data.get("selected_year"))
        redirect_url = "/projects"
        query_parts = []
        if code:
            query_parts.append(f"focus_code={quote_plus(code)}")
        if selected_year:
            query_parts.append(f"year={quote_plus(selected_year)}")
        if query_parts:
            redirect_url += "?" + "&".join(query_parts)
        return RedirectResponse(redirect_url, status_code=303)
    except Exception as exc:
        logger.exception("사업현황 저장 에러: %s", exc)
        code = normalize_text(form_data.get("support_dept_code"))
        selected_year_text = normalize_text(form_data.get("selected_year"))
        selected_year = int(selected_year_text) if selected_year_text.isdigit() else None
        return render_projects_page(
            request,
            edit_code=code or None,
            focus_code=code or None,
            selected_year=selected_year,
            message=f"사업현황 저장 중 오류가 발생했습니다: {exc}",
        )


@app.post("/projects/save-json")
async def save_project_json(request: Request):
    form_data: dict[str, Any] = {}
    try:
        form_data = parse_project_form(await request.body())
        save_project_status(form_data)
        code = normalize_text(form_data.get("support_dept_code"))
        health_payload = build_health_payload()
        return JSONResponse(
            content=jsonable_encoder(
                {
                    "ok": True,
                    "support_dept_code": code,
                    "project_edit": get_project_status_for_edit(code),
                    "project_row": get_project_status_row_for_code(code),
                    "data_version": health_payload.get("data_version", ""),
                    "server_time": health_payload.get("server_time", ""),
                }
            )
        )
    except Exception as exc:
        logger.exception("사업현황 JSON 저장 에러: %s", exc)
        return JSONResponse(
            status_code=500,
            content={
                "ok": False,
                "error": str(exc) or "사업현황 저장 중 오류가 발생했습니다.",
            },
        )


if __name__ == "__main__":
    auto_reload = os.getenv("INTRANET_AUTO_RELOAD", "1").lower() not in {"0", "false", "no"}
    uvicorn.run("main:app", host="0.0.0.0", port=8010, reload=auto_reload, reload_dirs=[str(BASE_DIR)])
