import os import logging import json import re from datetime import date, datetime 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, 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}, ) 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", } 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) """ ) ) 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_page_state ( page_key TEXT PRIMARY KEY, selected_code TEXT DEFAULT '', selected_year TEXT DEFAULT '', analysis_open INTEGER DEFAULT 0, related_project_selections_json TEXT DEFAULT '{}', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """ ) ) 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 INDEX IF NOT EXISTS idx_project_related_links_base ON project_related_links (base_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 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) """ ) ) 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 = { "selected_code": "TEXT DEFAULT ''", "selected_year": "TEXT DEFAULT ''", "analysis_open": "INTEGER DEFAULT 0", "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'")) 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 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(row5)): return "billing_status" if "계약현황" in filename: return "contract_status" 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 refresh_contract_review_tags() -> None: 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, 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)) review_tag = "" review_note = "" if billing_contract_amount and abs(hanmac_contract_amount - billing_contract_amount) > 0.5: review_tag = "변경계약 검토 필요" 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() 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, '') <> '' """ ) ).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) conn.execute(text("DELETE FROM project_related_links WHERE COALESCE(link_source, 'manual') = 'auto'")) for cluster_codes in cluster_map.values(): normalized_cluster = sorted(cluster_codes) if len(normalized_cluster) < 2: continue 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, 'auto', 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, }, ) @app.on_event("startup") def on_startup() -> None: init_db() auto_import_project_excels() sync_auto_project_related_links() 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_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 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 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( { "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, ) -> 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")) if not contract_amount: contract_amount = normalize_amount(contract_info.get("hanmac_contract_amount")) or normalize_amount(billing_summary.get("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")) 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")) 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")) 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")) 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() versions = [ normalize_text(transaction_updated), normalize_text(project_updated), normalize_text(contract_updated), normalize_text(billing_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() with engine.begin() as 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) item["collection_entries"] = decode_json_rows(item.pop("collection_entries_json", "[]")) item["task_plan_entries"] = decode_json_rows(item.pop("task_plan_entries_json", "[]")) item["exec_budget_entries"] = decode_json_rows(item.pop("exec_budget_entries_json", "[]")) item["actual_input_entries"] = decode_json_rows(item.pop("actual_input_entries_json", "[]")) 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"))), ) 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): 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), ) ) return result 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() 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: 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) result["collection_entries"] = decode_json_rows(result.pop("collection_entries_json", "[]")) result["task_plan_entries"] = decode_json_rows(result.pop("task_plan_entries_json", "[]")) result["exec_budget_entries"] = decode_json_rows(result.pop("exec_budget_entries_json", "[]")) result["actual_input_entries"] = decode_json_rows(result.pop("actual_input_entries_json", "[]")) try: result["exec_labor_rates"] = json.loads(normalize_text(result.pop("exec_labor_rates_json", "{}")) or "{}") except json.JSONDecodeError: result["exec_labor_rates"] = {} if not result["collection_entries"] and normalize_amount(result.get("collection_amount")): result["collection_entries"] = [ { "vendor": "", "round": "", "amount": result.get("collection_amount", ""), "date": "", "due_date": "", "note": "기존 수기 입력값", } ] if not result["task_plan_entries"]: fallback_task_rows = [] if normalize_amount(result.get("task_plan_department_budget")): fallback_task_rows.append( { "group": "department", "dept_name": "기존 부서별 배분", "work_name": "", "amount": result.get("task_plan_department_budget", ""), } ) if normalize_amount(result.get("task_plan_outsource_budget")): fallback_task_rows.append( { "group": "outsource", "dept_name": "기존 외주비", "work_name": result.get("task_plan_outsource_detail", ""), "amount": result.get("task_plan_outsource_budget", ""), } ) if normalize_amount(result.get("task_plan_joint_operating_cost")): fallback_task_rows.append( { "group": "joint", "dept_name": "기존 합사운영비", "work_name": "", "amount": result.get("task_plan_joint_operating_cost", ""), } ) result["task_plan_entries"] = fallback_task_rows if not result["exec_budget_entries"]: fallback_exec_rows = [] if normalize_amount(result.get("exec_budget_labor_by_grade")): fallback_exec_rows.append( { "group": "labor", "grade": "기존 인건비", "hours": "", "amount": result.get("exec_budget_labor_by_grade", ""), } ) if normalize_amount(result.get("exec_budget_outsource")): fallback_exec_rows.append( { "group": "outsource", "dept_name": "기존 외주비", "work_name": "", "amount": result.get("exec_budget_outsource", ""), } ) if normalize_amount(result.get("exec_budget_cost_plan")): fallback_exec_rows.append( { "group": "cost_plan", "account_code": "기존", "account_name": "비용계획", "amount": result.get("exec_budget_cost_plan", ""), } ) result["exec_budget_entries"] = fallback_exec_rows if not result["actual_input_entries"] and normalize_amount(result.get("item_investment")): result["actual_input_entries"] = [ { "reference": "", "amount": result.get("item_investment", ""), "note": "기존 항목별투입액", } ] 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"))), ) def get_project_page_state() -> dict[str, Any]: 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(related_project_selections_json, '{}') AS related_project_selections_json FROM project_page_state WHERE page_key = 'projects' """ ) ).mappings().first() if not row: return { "selected_code": "", "selected_year": "", "analysis_open": False, "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"]), "related_project_selections": related_project_selections, } def save_project_page_state(payload: dict[str, Any]) -> None: 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 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, selected_code, selected_year, analysis_open, related_project_selections_json, updated_at ) VALUES ( 'projects', :selected_code, :selected_year, :analysis_open, :related_project_selections_json, CURRENT_TIMESTAMP ) ON CONFLICT(page_key) DO UPDATE SET selected_code = excluded.selected_code, selected_year = excluded.selected_year, analysis_open = excluded.analysis_open, related_project_selections_json = excluded.related_project_selections_json, updated_at = CURRENT_TIMESTAMP """ ), { "selected_code": selected_code, "selected_year": selected_year, "analysis_open": analysis_open, "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 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 == "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 = sorted(BASE_DIR.glob("*.xlsx")) if not excel_files: return known_files = existing_source_files() known_contract_files = existing_contract_source_files() known_billing_files = existing_billing_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 for file in excel_files): return for excel_path in excel_files: workbook = load_workbook(excel_path, data_only=True) 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 == "billing_status" and excel_path.name in known_billing_files: continue if import_kind == "transactions" and excel_path.name in known_files: continue 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) 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: 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 = contract_amount * expected_as_rate / 100 if contract_amount else 0.0 expected_sga_budget = contract_amount * expected_sga_rate / 100 if contract_amount else 0.0 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")), } 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 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("기존 입력값을 불러오지 않은 빈 상태로는 저장할 수 없습니다.") check_record_revision( conn, "project_status", "support_dept_code", support_dept_code, normalize_text(payload.get("edit_revision")), ) 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 """ ), normalized_payload, ) 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, } return templates.TemplateResponse(request, "dashboard.html", context) def render_projects_page( request: Request, edit_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_edit": get_project_status_for_edit(edit_code), "project_page_state": get_project_page_state(), "project_related_links": get_project_related_links_map(), "support_department_options": get_support_department_options(), "cost_department_options": get_cost_department_options(), "cost_account_options": get_cost_account_options(), } 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"), } 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("

서버 오류

로그를 확인해주세요.

", status_code=500) @app.get("/projects") async def projects(request: Request, edit_code: str | None = None, year: str | None = None): try: return render_projects_page(request, edit_code=edit_code, selected_year=parse_optional_year(year)) except Exception as exc: logger.exception("사업현황 페이지 에러: %s", exc) return HTMLResponse("

서버 오류

로그를 확인해주세요.

", 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.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("/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("

서버 오류

로그를 확인해주세요.

", 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"edit_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, selected_year=selected_year, message=f"사업현황 저장 중 오류가 발생했습니다: {exc}", ) 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)])