YJKIM/HM_project_Viewer_Board
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
HM_project_Viewer_Board/backup_snapshots/20260410_0900/main.py
T

6155 lines
264 KiB
Python
Raw Permalink Blame History

import os
import logging
import json
import re
import zipfile
from datetime import date, datetime
from functools import lru_cache
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, quote_plus
import uvicorn
from fastapi import FastAPI, File, Request, UploadFile
from fastapi.encoders import jsonable_encoder
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from openpyxl import load_workbook
from sqlalchemy import create_engine, event, text
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
BASE_DIR = Path(__file__).resolve().parent
STATIC_DIR = BASE_DIR / "static"
TEMPLATES_DIR = BASE_DIR / "templates"
DB_PATH = BASE_DIR / "data.db"
STATIC_DIR.mkdir(exist_ok=True)
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
engine = create_engine(
f"sqlite:///{DB_PATH}",
connect_args={"check_same_thread": False},
)
@event.listens_for(engine, "connect")
def configure_sqlite_connection(dbapi_connection: Any, _: Any) -> None:
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA synchronous=NORMAL")
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute("PRAGMA busy_timeout=5000")
cursor.execute("PRAGMA temp_store=MEMORY")
cursor.close()
EXCLUDED_SUPPORT_CODES = {"ZZZZZZ"}
EXCLUDED_SUPPORT_NAMES = {"공통", "경영지원부", "기술개발센터", "임원실", "기술개발부", "총괄기획실"}
SUPPORT_DEPARTMENT_NAMES = ("경영지원부", "임원실", "총괄기획실", "기술개발센터", "기술개발부", "공통")
VOUCHER_PATTERN = re.compile(r"^11-(\d{4})(\d{2})(\d{2})-[^-]+-[^-]+-[^-]+$")
REVENUE_SQL = "(account_code LIKE '401101%' OR account_code LIKE '401102%')"
SUPPORT_COST_DEPT_SQL = (
"cost_dept_name IN ('경영지원부', '임원실', '총괄기획실', '기술개발센터', '기술개발부', '공통')"
)
FIELD_COST_DEPT_SQL = (
"COALESCE(cost_dept_name, '') NOT IN ('경영지원부', '임원실', '총괄기획실', '기술개발센터', '기술개발부', '공통')"
)
FIELD_LABELS = {
"approval_status": "결재상태",
"voucher_number": "가전표번호",
"account_code": "계정코드",
"account_name": "계정명칭",
"debit_supply": "차변공급가",
"debit_vat": "차변부가세",
"credit_supply": "대변공급가",
"credit_vat": "대변부가세",
"issuing_dept_code": "발의부서코드",
"issuing_dept_name": "발의부서명",
"confirmed_voucher_number": "확정전표번호",
"support_dept_code": "지원부서코드",
"support_dept_name": "지원부서명",
"cost_dept_code": "원가부서코드",
"cost_dept_name": "원가부서명",
"memo1": "적요1",
"memo2": "적요2",
"partner_code": "거래처코드",
"partner_name": "거래처명칭",
"tax_code": "세무코드",
"posting_date": "증빙일자",
"voucher_type": "전표종류",
"management_item": "관리항목",
}
FORM_FIELDS = list(FIELD_LABELS.keys())
DIRECT_HEADER_MAP = {
"결재상태": "approval_status",
"가전표번호": "voucher_number",
"계정코드": "account_code",
"계정명칭": "account_name",
"차변공급가": "debit_supply",
"차변부가세": "debit_vat",
"대변공급가": "credit_supply",
"대변부가세": "credit_vat",
"발의부서코드": "issuing_dept_code",
"발의부서명": "issuing_dept_name",
"발의부서명칭": "issuing_dept_name",
"지원부서코드": "support_dept_code",
"지원부서명": "support_dept_name",
"지원부서명칭": "support_dept_name",
"원가부서코드": "cost_dept_code",
"원가부서명": "cost_dept_name",
"원가부서명칭": "cost_dept_name",
"적요1": "memo1",
"적요2": "memo2",
"거래처코드": "partner_code",
"거래처명칭": "partner_name",
"세무코드": "tax_code",
"증빙일자": "posting_date",
"전표종류": "voucher_type",
"관리항목": "management_item",
}
DEFAULT_APP_OPTION_ITEMS = {
"labor_grades": [
("president", "사장", "사장"),
("vice_president", "부사장", "부사장"),
("executive_vice_president", "전무", "전무"),
("managing_director", "상무", "상무"),
("director", "이사", "이사"),
("general_manager", "부장", "부장"),
("deputy_general_manager", "차장", "차장"),
("manager", "과장", "과장"),
("assistant_manager", "대리", "대리"),
("staff", "사원", "사원"),
("principal", "수석", "수석"),
("senior_manager", "책임", "책임"),
("senior", "선임", "선임"),
("researcher", "연구원", "연구원"),
],
"expected_as_rates": [
("as_0", "0%", "0"),
("as_2", "2%", "2"),
("as_5", "5%", "5"),
("as_10", "10%", "10"),
],
"expected_sga_rates": [
("sga_13", "13%", "13"),
("sga_15", "15%", "15"),
("sga_20", "20%", "20"),
("sga_25", "25%", "25"),
],
"uncontracted_categories": [
("general", "일반 미계약", "general"),
("precontract", "사전 사업 코드", "precontract"),
("corporate_rnd", "기업 연구개발", "corporate_rnd"),
("external_research", "외부 연구과제", "external_research"),
],
"project_rules": [
("legacy_variant_cutoff_year", "이전 연도 변경/차수 제외 기준", "23"),
("detail_visible_min_year", "세부내역 반영 시작 연도", "2023"),
],
"dashboard_revenue_metrics": [
("design_revenue", "설계", "#4f7cff"),
("design_other_revenue", "설계 외", "#67c7c9"),
("supervision_revenue", "감리", "#233a5a"),
("inspection_revenue", "점검", "#ffb54a"),
],
"dashboard_expense_metrics": [
("cost_sum", "원가", "#4f7cff"),
("sga_sum", "판관비", "#67c7c9"),
("labor_sum", "원가인건비", "#233a5a"),
("outsourcing_sum", "원가외주비", "#ffb54a"),
],
"annual_metric_cards": [
("revenue_sum", "수금", "수금"),
("project_cost_sum", "원가(프로젝트)", "원가(프로젝트)"),
("support_cost_sum", "원가(지원부서)", "원가(지원부서)"),
("support_sga_sum", "판관비(지원부서)", "판관비(지원부서)"),
("field_sga_sum", "판관비(현업부서)", "판관비(현업부서)"),
("labor_sum", "원가인건비", "원가인건비"),
("outsourcing_sum", "원가외주비", "원가외주비"),
("total_expense", "비용합계", "비용합계"),
("operating_balance", "영업수지", "영업수지"),
],
"annual_expense_chart_metrics": [
("labor_sum", "원가인건비", "#8b5cf6"),
("outsourcing_sum", "원가외주비", "#ec4899"),
("project_cost_sum", "원가(프로젝트)", "#0ea5a4"),
("support_cost_sum", "원가(지원부서)", "#67b7dc"),
("support_sga_sum", "판관비(지원부서)", "#f59e0b"),
("field_sga_sum", "판관비(현업부서)", "#f97316"),
],
"annual_balance_chart_metrics": [
("revenue_sum", "수금", "#0f766e"),
("total_expense", "비용합계", "#1d4ed8"),
("operating_balance", "영업수지", "#dc2626"),
],
}
DEFAULT_APP_KEYWORD_RULES = {
"special_x_classification": [
("external_research", "과제"),
("external_research", "연구과제"),
("external_research", "연구소"),
("external_research", "연구용역"),
("external_research", "연구"),
("corporate_rnd", "신규노선개발"),
("corporate_rnd", "프로그램 개발"),
("corporate_rnd", "프로그램개발"),
("corporate_rnd", "BIM"),
("corporate_rnd", "시스템"),
("corporate_rnd", "혁신"),
]
}
def ensure_default_app_config(conn: Any) -> None:
for group_key, items in DEFAULT_APP_OPTION_ITEMS.items():
for sort_order, (item_key, label, value_text) in enumerate(items):
conn.execute(
text(
"""
INSERT INTO app_option_items (
group_key, item_key, label, value_text, sort_order, is_active, meta_json
) VALUES (
:group_key, :item_key, :label, :value_text, :sort_order, 1, '{}'
)
ON CONFLICT(group_key, item_key) DO UPDATE SET
label = excluded.label,
value_text = excluded.value_text,
sort_order = excluded.sort_order
"""
),
{
"group_key": group_key,
"item_key": item_key,
"label": label,
"value_text": value_text,
"sort_order": sort_order,
},
)
for rule_group, items in DEFAULT_APP_KEYWORD_RULES.items():
for sort_order, (category_key, keyword) in enumerate(items):
conn.execute(
text(
"""
INSERT INTO app_keyword_rules (
rule_group, category_key, keyword, sort_order, is_active
) VALUES (
:rule_group, :category_key, :keyword, :sort_order, 1
)
ON CONFLICT(rule_group, category_key, keyword) DO UPDATE SET
sort_order = excluded.sort_order
"""
),
{
"rule_group": rule_group,
"category_key": category_key,
"keyword": keyword,
"sort_order": sort_order,
},
)
load_app_config.cache_clear()
def init_db() -> None:
with engine.begin() as conn:
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
approval_status TEXT,
voucher_number TEXT,
account_code TEXT,
account_name TEXT,
debit_supply REAL DEFAULT 0,
debit_vat REAL DEFAULT 0,
credit_supply REAL DEFAULT 0,
credit_vat REAL DEFAULT 0,
issuing_dept_code TEXT,
issuing_dept_name TEXT,
confirmed_voucher_number TEXT,
support_dept_code TEXT,
support_dept_name TEXT,
cost_dept_code TEXT,
cost_dept_name TEXT,
memo1 TEXT,
memo2 TEXT,
partner_code TEXT,
partner_name TEXT,
tax_code TEXT,
posting_date TEXT,
voucher_type TEXT,
management_item TEXT,
accounting_category TEXT,
amount REAL DEFAULT 0,
year INTEGER,
month INTEGER,
day INTEGER,
source_file TEXT,
last_editor_session_id TEXT DEFAULT '',
last_client_submitted_at TEXT DEFAULT '',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_transactions_year_month
ON transactions (year, month)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_transactions_support
ON transactions (support_dept_code, support_dept_name)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_transactions_support_category
ON transactions (support_dept_code, accounting_category)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_transactions_support_account
ON transactions (support_dept_code, account_code)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_transactions_source_file
ON transactions (source_file)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_transactions_updated_at
ON transactions (updated_at)
"""
)
)
transaction_columns = {
row[1]
for row in conn.execute(text("PRAGMA table_info(transactions)")).fetchall()
}
required_transaction_columns = {
"last_editor_session_id": "TEXT DEFAULT ''",
"last_client_submitted_at": "TEXT DEFAULT ''",
}
for column_name, column_type in required_transaction_columns.items():
if column_name not in transaction_columns:
conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {column_name} {column_type}"))
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_status (
support_dept_code TEXT PRIMARY KEY,
support_dept_name TEXT NOT NULL,
progress_rate REAL DEFAULT 0,
contract_amount REAL DEFAULT 0,
collection_amount REAL DEFAULT 0,
collection_entries_json TEXT DEFAULT '[]',
change_round TEXT DEFAULT '',
item_investment REAL DEFAULT 0,
task_plan_department_budget REAL DEFAULT 0,
task_plan_outsource_budget REAL DEFAULT 0,
task_plan_outsource_detail TEXT DEFAULT '',
task_plan_joint_operating_cost REAL DEFAULT 0,
task_plan_entries_json TEXT DEFAULT '[]',
exec_budget_labor_by_grade REAL DEFAULT 0,
exec_labor_rates_json TEXT DEFAULT '{}',
exec_budget_outsource REAL DEFAULT 0,
exec_budget_cost_plan REAL DEFAULT 0,
exec_budget_entries_json TEXT DEFAULT '[]',
actual_input_entries_json TEXT DEFAULT '[]',
project_type TEXT DEFAULT '',
expected_as_rate REAL DEFAULT 0,
expected_sga_rate REAL DEFAULT 0,
expected_as_cost REAL DEFAULT 0,
expected_sga_budget REAL DEFAULT 0,
last_editor_session_id TEXT DEFAULT '',
last_client_submitted_at TEXT DEFAULT '',
project_start_date TEXT DEFAULT '',
project_end_date TEXT DEFAULT '',
completion_status TEXT DEFAULT '',
notes TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_basic_info (
support_dept_code TEXT PRIMARY KEY,
support_dept_name TEXT NOT NULL DEFAULT '',
contract_amount REAL DEFAULT 0,
project_type TEXT DEFAULT '',
expected_as_rate REAL DEFAULT 0,
expected_sga_rate REAL DEFAULT 0,
expected_as_cost REAL DEFAULT 0,
expected_sga_budget REAL DEFAULT 0,
exec_labor_rates_json TEXT DEFAULT '{}',
change_round TEXT DEFAULT '',
project_start_date TEXT DEFAULT '',
project_end_date TEXT DEFAULT '',
completion_status TEXT DEFAULT '',
notes TEXT DEFAULT '',
last_editor_session_id TEXT DEFAULT '',
last_client_submitted_at TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_basic_info_updated_at
ON project_basic_info (updated_at)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_page_state (
page_key TEXT PRIMARY KEY,
selected_code TEXT DEFAULT '',
selected_year TEXT DEFAULT '',
analysis_open INTEGER DEFAULT 0,
uncontracted_year_start TEXT DEFAULT '',
uncontracted_year_end TEXT DEFAULT '',
related_project_selections_json TEXT DEFAULT '{}',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS app_option_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
group_key TEXT NOT NULL,
item_key TEXT NOT NULL,
label TEXT NOT NULL,
value_text TEXT NOT NULL DEFAULT '',
sort_order INTEGER NOT NULL DEFAULT 0,
is_active INTEGER NOT NULL DEFAULT 1,
meta_json TEXT DEFAULT '{}',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(group_key, item_key)
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_app_option_items_group
ON app_option_items (group_key, sort_order, id)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS app_keyword_rules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
rule_group TEXT NOT NULL,
category_key TEXT NOT NULL,
keyword TEXT NOT NULL,
sort_order INTEGER NOT NULL DEFAULT 0,
is_active INTEGER NOT NULL DEFAULT 1,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(rule_group, category_key, keyword)
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_app_keyword_rules_group
ON app_keyword_rules (rule_group, category_key, sort_order, id)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_collection_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
support_dept_code TEXT NOT NULL,
position INTEGER NOT NULL DEFAULT 0,
vendor TEXT DEFAULT '',
progress_type TEXT DEFAULT '',
billing_round TEXT DEFAULT '',
billing_type TEXT DEFAULT '',
billing_date TEXT DEFAULT '',
billed_amount REAL DEFAULT 0,
round TEXT DEFAULT '',
date TEXT DEFAULT '',
due_date TEXT DEFAULT '',
amount REAL DEFAULT 0,
balance_amount REAL DEFAULT 0,
collection_rate REAL DEFAULT 0,
note TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_collection_entries_code_pos
ON project_collection_entries (support_dept_code, position)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_task_plan_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
support_dept_code TEXT NOT NULL,
position INTEGER NOT NULL DEFAULT 0,
group_name TEXT DEFAULT '',
dept_name TEXT DEFAULT '',
work_name TEXT DEFAULT '',
amount REAL DEFAULT 0,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_task_plan_entries_code_pos
ON project_task_plan_entries (support_dept_code, position)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_exec_budget_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
support_dept_code TEXT NOT NULL,
position INTEGER NOT NULL DEFAULT 0,
group_name TEXT DEFAULT '',
grade TEXT DEFAULT '',
hours TEXT DEFAULT '',
dept_name TEXT DEFAULT '',
work_name TEXT DEFAULT '',
account_code TEXT DEFAULT '',
account_name TEXT DEFAULT '',
amount REAL DEFAULT 0,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_exec_budget_entries_code_pos
ON project_exec_budget_entries (support_dept_code, position)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_actual_input_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
support_dept_code TEXT NOT NULL,
position INTEGER NOT NULL DEFAULT 0,
group_name TEXT DEFAULT '',
grade TEXT DEFAULT '',
minutes TEXT DEFAULT '',
label TEXT DEFAULT '',
reference TEXT DEFAULT '',
note TEXT DEFAULT '',
amount REAL DEFAULT 0,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_actual_input_entries_code_pos
ON project_actual_input_entries (support_dept_code, position)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_comparison_notes (
support_dept_code TEXT NOT NULL,
item_key TEXT NOT NULL,
note TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (support_dept_code, item_key)
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_comparison_notes_code
ON project_comparison_notes (support_dept_code, item_key)
"""
)
)
page_state_columns_before = {
row[1]
for row in conn.execute(text("PRAGMA table_info(project_page_state)")).fetchall()
}
if "session_id" not in page_state_columns_before:
legacy_rows = conn.execute(
text(
"""
SELECT page_key, selected_code, selected_year, analysis_open,
related_project_selections_json, updated_at
FROM project_page_state
"""
)
).mappings().all()
conn.execute(text("ALTER TABLE project_page_state RENAME TO project_page_state_legacy"))
conn.execute(
text(
"""
CREATE TABLE project_page_state (
page_key TEXT NOT NULL,
session_id TEXT NOT NULL DEFAULT '',
selected_code TEXT DEFAULT '',
selected_year TEXT DEFAULT '',
analysis_open INTEGER DEFAULT 0,
uncontracted_year_start TEXT DEFAULT '',
uncontracted_year_end TEXT DEFAULT '',
related_project_selections_json TEXT DEFAULT '{}',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (page_key, session_id)
)
"""
)
)
for row in legacy_rows:
conn.execute(
text(
"""
INSERT INTO project_page_state (
page_key, session_id, selected_code, selected_year,
analysis_open, uncontracted_year_start, uncontracted_year_end,
related_project_selections_json, updated_at
) VALUES (
:page_key, '', :selected_code, :selected_year,
:analysis_open, '', '', :related_project_selections_json, :updated_at
)
"""
),
dict(row),
)
conn.execute(text("DROP TABLE project_page_state_legacy"))
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_page_state_page_session
ON project_page_state (page_key, session_id)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_related_links (
base_support_dept_code TEXT NOT NULL,
related_support_dept_code TEXT NOT NULL,
link_source TEXT DEFAULT 'manual',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (base_support_dept_code, related_support_dept_code)
)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_uncontracted_classification (
support_dept_code TEXT PRIMARY KEY,
category TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_uncontracted_classification_category
ON project_uncontracted_classification (category)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_quick_links (
page_key TEXT NOT NULL,
support_dept_code TEXT NOT NULL,
sort_order INTEGER NOT NULL DEFAULT 0,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (page_key, support_dept_code)
)
"""
)
)
quick_link_columns_before = {
row[1]
for row in conn.execute(text("PRAGMA table_info(project_quick_links)")).fetchall()
}
if "session_id" in quick_link_columns_before:
legacy_rows = conn.execute(
text(
"""
SELECT page_key, support_dept_code, MIN(sort_order) AS sort_order, MAX(updated_at) AS updated_at
FROM project_quick_links
GROUP BY page_key, support_dept_code
"""
)
).mappings().all()
conn.execute(text("ALTER TABLE project_quick_links RENAME TO project_quick_links_legacy"))
conn.execute(
text(
"""
CREATE TABLE project_quick_links (
page_key TEXT NOT NULL,
support_dept_code TEXT NOT NULL,
sort_order INTEGER NOT NULL DEFAULT 0,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (page_key, support_dept_code)
)
"""
)
)
for row in legacy_rows:
conn.execute(
text(
"""
INSERT INTO project_quick_links (
page_key, support_dept_code, sort_order, updated_at
) VALUES (
:page_key, :support_dept_code, :sort_order, :updated_at
)
"""
),
dict(row),
)
conn.execute(text("DROP TABLE project_quick_links_legacy"))
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_quick_links_page_sort
ON project_quick_links (page_key, sort_order, updated_at)
"""
)
)
ensure_default_app_config(conn)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_related_links_base
ON project_related_links (base_support_dept_code)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_related_links_related
ON project_related_links (related_support_dept_code)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_contract_info (
support_dept_code TEXT PRIMARY KEY,
raw_contract_code TEXT DEFAULT '',
business_division TEXT DEFAULT '',
order_method TEXT DEFAULT '',
owner_department TEXT DEFAULT '',
client_name TEXT DEFAULT '',
support_dept_name TEXT DEFAULT '',
work_category TEXT DEFAULT '',
order_date TEXT DEFAULT '',
contract_date TEXT DEFAULT '',
project_start_date TEXT DEFAULT '',
project_end_date TEXT DEFAULT '',
contract_status TEXT DEFAULT '',
joint_contract TEXT DEFAULT '',
pm_name TEXT DEFAULT '',
progress_status TEXT DEFAULT '',
total_contract_amount REAL DEFAULT 0,
hanmac_contract_amount REAL DEFAULT 0,
review_tag TEXT DEFAULT '',
review_note TEXT DEFAULT '',
source_file TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_contract_info_updated_at
ON project_contract_info (updated_at)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_contract_change_summary (
id INTEGER PRIMARY KEY AUTOINCREMENT,
raw_summary_code TEXT DEFAULT '',
normalized_title TEXT DEFAULT '',
owner_department TEXT DEFAULT '',
business_division TEXT DEFAULT '',
support_dept_name TEXT DEFAULT '',
change_date TEXT DEFAULT '',
client_name TEXT DEFAULT '',
original_contract_period TEXT DEFAULT '',
changed_project_end_date TEXT DEFAULT '',
initial_contract_amount REAL DEFAULT 0,
previous_contract_amount REAL DEFAULT 0,
changed_contract_amount REAL DEFAULT 0,
delta_amount REAL DEFAULT 0,
source_file TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_contract_change_summary_code
ON project_contract_change_summary (raw_summary_code, change_date)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_contract_change_summary_title
ON project_contract_change_summary (normalized_title)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_contract_change_round (
id INTEGER PRIMARY KEY AUTOINCREMENT,
support_dept_code TEXT DEFAULT '',
raw_round_code TEXT DEFAULT '',
normalized_title TEXT DEFAULT '',
owner_department TEXT DEFAULT '',
business_division TEXT DEFAULT '',
support_dept_name TEXT DEFAULT '',
change_date TEXT DEFAULT '',
client_name TEXT DEFAULT '',
original_contract_period TEXT DEFAULT '',
changed_project_end_date TEXT DEFAULT '',
initial_contract_amount REAL DEFAULT 0,
changed_contract_amount REAL DEFAULT 0,
delta_amount REAL DEFAULT 0,
source_file TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_contract_change_round_code
ON project_contract_change_round (support_dept_code, change_date)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_contract_change_round_title
ON project_contract_change_round (normalized_title)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_billing_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
support_dept_code TEXT,
raw_project_code TEXT DEFAULT '',
round_code TEXT DEFAULT '',
support_department TEXT DEFAULT '',
business_division TEXT DEFAULT '',
support_dept_name TEXT DEFAULT '',
contract_amount REAL DEFAULT 0,
client_name TEXT DEFAULT '',
billing_type TEXT DEFAULT '',
progress_round TEXT DEFAULT '',
billing_date TEXT DEFAULT '',
tax_invoice_date TEXT DEFAULT '',
expected_collection_date TEXT DEFAULT '',
billed_amount REAL DEFAULT 0,
collected_amount REAL DEFAULT 0,
balance_amount REAL DEFAULT 0,
collection_rate REAL DEFAULT 0,
note TEXT DEFAULT '',
source_file TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_billing_entries_code
ON project_billing_entries (support_dept_code, billing_date)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_billing_entries_raw_code
ON project_billing_entries (raw_project_code)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_billing_entries_round_code
ON project_billing_entries (round_code)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_billing_entries_source_file
ON project_billing_entries (source_file)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_billing_entries_updated_at
ON project_billing_entries (updated_at)
"""
)
)
existing_columns = {
row[1]
for row in conn.execute(text("PRAGMA table_info(project_status)")).fetchall()
}
required_columns = {
"last_editor_session_id": "TEXT DEFAULT ''",
"last_client_submitted_at": "TEXT DEFAULT ''",
"contract_amount": "REAL DEFAULT 0",
"collection_entries_json": "TEXT DEFAULT '[]'",
"task_plan_department_budget": "REAL DEFAULT 0",
"task_plan_outsource_budget": "REAL DEFAULT 0",
"task_plan_outsource_detail": "TEXT DEFAULT ''",
"task_plan_joint_operating_cost": "REAL DEFAULT 0",
"task_plan_entries_json": "TEXT DEFAULT '[]'",
"exec_budget_labor_by_grade": "REAL DEFAULT 0",
"exec_labor_rates_json": "TEXT DEFAULT '{}'",
"exec_budget_outsource": "REAL DEFAULT 0",
"exec_budget_cost_plan": "REAL DEFAULT 0",
"exec_budget_entries_json": "TEXT DEFAULT '[]'",
"actual_input_entries_json": "TEXT DEFAULT '[]'",
"project_type": "TEXT DEFAULT ''",
"expected_as_rate": "REAL DEFAULT 0",
"expected_sga_rate": "REAL DEFAULT 0",
"expected_as_cost": "REAL DEFAULT 0",
"expected_sga_budget": "REAL DEFAULT 0",
"project_start_date": "TEXT DEFAULT ''",
"project_end_date": "TEXT DEFAULT ''",
"completion_status": "TEXT DEFAULT ''",
}
for column_name, column_type in required_columns.items():
if column_name not in existing_columns:
conn.execute(text(f"ALTER TABLE project_status ADD COLUMN {column_name} {column_type}"))
page_state_columns = {
row[1]
for row in conn.execute(text("PRAGMA table_info(project_page_state)")).fetchall()
}
required_page_state_columns = {
"session_id": "TEXT DEFAULT ''",
"selected_code": "TEXT DEFAULT ''",
"selected_year": "TEXT DEFAULT ''",
"analysis_open": "INTEGER DEFAULT 0",
"uncontracted_year_start": "TEXT DEFAULT ''",
"uncontracted_year_end": "TEXT DEFAULT ''",
"related_project_selections_json": "TEXT DEFAULT '{}'",
}
for column_name, column_type in required_page_state_columns.items():
if column_name not in page_state_columns:
conn.execute(text(f"ALTER TABLE project_page_state ADD COLUMN {column_name} {column_type}"))
related_link_columns = {
row[1]
for row in conn.execute(text("PRAGMA table_info(project_related_links)")).fetchall()
}
if "link_source" not in related_link_columns:
conn.execute(text("ALTER TABLE project_related_links ADD COLUMN link_source TEXT DEFAULT 'manual'"))
migrate_project_status_entries(conn)
migrate_project_basic_info(conn)
ensure_default_app_config(conn)
conn.execute(text("ANALYZE"))
@lru_cache(maxsize=1)
def load_app_config() -> dict[str, Any]:
with engine.connect() as conn:
option_rows = conn.execute(
text(
"""
SELECT group_key, item_key, label, value_text, sort_order
FROM app_option_items
WHERE is_active = 1
ORDER BY group_key, sort_order, id
"""
)
).mappings().all()
keyword_rows = conn.execute(
text(
"""
SELECT rule_group, category_key, keyword, sort_order
FROM app_keyword_rules
WHERE is_active = 1
ORDER BY rule_group, category_key, sort_order, id
"""
)
).mappings().all()
options: dict[str, list[dict[str, Any]]] = {}
for row in option_rows:
options.setdefault(row["group_key"], []).append(
{
"item_key": row["item_key"],
"label": row["label"],
"value": row["value_text"],
"sort_order": row["sort_order"],
}
)
keyword_rules: dict[str, dict[str, list[str]]] = {}
for row in keyword_rows:
keyword_rules.setdefault(row["rule_group"], {}).setdefault(row["category_key"], []).append(row["keyword"])
return {"options": options, "keyword_rules": keyword_rules}
def get_option_items(group_key: str) -> list[dict[str, Any]]:
return list(load_app_config().get("options", {}).get(group_key, []))
def get_keyword_rule_groups(rule_group: str) -> dict[str, list[str]]:
return dict(load_app_config().get("keyword_rules", {}).get(rule_group, {}))
def get_labor_grade_options() -> list[str]:
return [item["label"] for item in get_option_items("labor_grades")]
def get_expected_as_rate_options() -> list[dict[str, Any]]:
return get_option_items("expected_as_rates")
def get_expected_sga_rate_options() -> list[dict[str, Any]]:
return get_option_items("expected_sga_rates")
def get_collection_progress_type_options() -> list[str]:
preferred_order = {"선급금": 0, "기성금": 1, "준공금": 2}
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT billing_type
FROM project_billing_entries
WHERE COALESCE(billing_type, '') <> ''
ORDER BY billing_type
"""
)
).fetchall()
values = {
normalized
for (value,) in rows
for normalized in [normalize_collection_progress_type(value)]
if normalized
}
if not values:
values = {"선급금", "기성금", "준공금"}
return sorted(values, key=lambda item: (preferred_order.get(item, 999), item))
def get_collection_billing_type_options() -> list[str]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT billing_type
FROM project_collection_entries
WHERE COALESCE(billing_type, '') <> ''
ORDER BY billing_type
"""
)
).fetchall()
values = {
normalized
for (value,) in rows
for normalized in [normalize_collection_billing_type(value)]
if normalized
}
values.add("기타")
if not values:
values = {"계약분", "기타"}
return list(sorted(values))
def get_uncontracted_category_options() -> list[dict[str, Any]]:
return get_option_items("uncontracted_categories")
def get_project_runtime_settings() -> dict[str, str]:
return {item["item_key"]: str(item.get("value", "")) for item in get_option_items("project_rules")}
def get_special_x_classification_rules() -> dict[str, list[str]]:
return get_keyword_rule_groups("special_x_classification")
def save_project_runtime_setting(item_key: Any, value_text: Any) -> None:
normalized_item_key = normalize_text(item_key)
if not normalized_item_key:
raise ValueError("설정 키가 올바르지 않습니다.")
with engine.begin() as conn:
existing = conn.execute(
text(
"""
SELECT group_key, item_key
FROM app_option_items
WHERE group_key = 'project_rules' AND item_key = :item_key
"""
),
{"item_key": normalized_item_key},
).mappings().first()
if not existing:
raise ValueError("존재하지 않는 런타임 설정입니다.")
conn.execute(
text(
"""
UPDATE app_option_items
SET value_text = :value_text,
updated_at = CURRENT_TIMESTAMP
WHERE group_key = 'project_rules' AND item_key = :item_key
"""
),
{
"item_key": normalized_item_key,
"value_text": normalize_text(value_text),
},
)
load_app_config.cache_clear()
def count_transactions() -> int:
with engine.begin() as conn:
return conn.execute(text("SELECT COUNT(*) FROM transactions")).scalar_one()
def existing_source_files() -> set[str]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT DISTINCT source_file FROM transactions WHERE COALESCE(source_file, '') <> ''")
).fetchall()
return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}
def existing_contract_source_files() -> set[str]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT DISTINCT source_file FROM project_contract_info WHERE COALESCE(source_file, '') <> ''")
).fetchall()
return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}
def existing_billing_source_files() -> set[str]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT DISTINCT source_file FROM project_billing_entries WHERE COALESCE(source_file, '') <> ''")
).fetchall()
return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}
def existing_change_contract_summary_source_files() -> set[str]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT DISTINCT source_file FROM project_contract_change_summary WHERE COALESCE(source_file, '') <> ''")
).fetchall()
return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}
def existing_change_contract_round_source_files() -> set[str]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT DISTINCT source_file FROM project_contract_change_round WHERE COALESCE(source_file, '') <> ''")
).fetchall()
return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}
def workbook_row_values(sheet: Any, row_number: int) -> list[str]:
return [normalize_text(sheet.cell(row_number, column).value) for column in range(1, sheet.max_column + 1)]
def detect_excel_import_kind(workbook: Any, filename: str = "") -> str:
sheet = workbook.active
row1 = workbook_row_values(sheet, 1)
row5 = workbook_row_values(sheet, 5) if sheet.max_row >= 5 else []
filename = normalize_text(filename)
if {"총괄코드", "총 계약금액", "한맥계약금액"}.issubset(set(row1)):
return "contract_status"
if {"총괄코드", "최초계약금액", "이전계약금액", "변경계약금액", "증감액"}.issubset(set(row1)):
return "change_contract_summary"
if {"차수코드", "당초계약금액", "변경계약금액", "증감액"}.issubset(set(row1)):
return "change_contract_round"
if {"차수코드", "차수사업명", "청구금액", "수금금액"}.issubset(set(row5)):
return "billing_status"
if "계약현황" in filename:
return "contract_status"
if "변경계약금액현황" in filename and "총괄" in filename:
return "change_contract_summary"
if "변경계약금액현황" in filename and "차수" in filename:
return "change_contract_round"
if "기성청구현황" in filename:
return "billing_status"
return "transactions"
def import_contract_status_workbook(workbook: Any, source_file: str) -> int:
sheet = workbook.active
with engine.begin() as conn:
conn.execute(
text("DELETE FROM project_contract_info WHERE source_file = :source_file"),
{"source_file": source_file},
)
inserted = 0
for row in sheet.iter_rows(min_row=2, values_only=True):
support_dept_code = normalize_project_code(row[1] if len(row) > 1 else "")
if not support_dept_code:
continue
payload = {
"support_dept_code": support_dept_code,
"raw_contract_code": normalize_text(row[1] if len(row) > 1 else ""),
"business_division": normalize_text(row[0] if len(row) > 0 else ""),
"order_method": normalize_text(row[2] if len(row) > 2 else ""),
"owner_department": normalize_text(row[3] if len(row) > 3 else ""),
"client_name": normalize_text(row[4] if len(row) > 4 else ""),
"support_dept_name": normalize_text(row[5] if len(row) > 5 else ""),
"work_category": normalize_text(row[6] if len(row) > 6 else ""),
"order_date": normalize_date_text(row[7] if len(row) > 7 else ""),
"contract_date": normalize_date_text(row[8] if len(row) > 8 else ""),
"project_start_date": normalize_date_text(row[9] if len(row) > 9 else ""),
"project_end_date": normalize_date_text(row[10] if len(row) > 10 else ""),
"contract_status": normalize_text(row[11] if len(row) > 11 else ""),
"joint_contract": normalize_text(row[12] if len(row) > 12 else ""),
"pm_name": normalize_text(row[13] if len(row) > 13 else ""),
"progress_status": normalize_text(row[14] if len(row) > 14 else ""),
"total_contract_amount": normalize_amount(row[15] if len(row) > 15 else 0),
"hanmac_contract_amount": normalize_amount(row[16] if len(row) > 16 else 0),
"review_tag": "",
"review_note": "",
"source_file": source_file,
}
conn.execute(
text(
"""
INSERT INTO project_contract_info (
support_dept_code, raw_contract_code, business_division, order_method,
owner_department, client_name, support_dept_name, work_category,
order_date, contract_date, project_start_date, project_end_date,
contract_status, joint_contract, pm_name, progress_status,
total_contract_amount, hanmac_contract_amount, review_tag, review_note,
source_file, updated_at
) VALUES (
:support_dept_code, :raw_contract_code, :business_division, :order_method,
:owner_department, :client_name, :support_dept_name, :work_category,
:order_date, :contract_date, :project_start_date, :project_end_date,
:contract_status, :joint_contract, :pm_name, :progress_status,
:total_contract_amount, :hanmac_contract_amount, :review_tag, :review_note,
:source_file, CURRENT_TIMESTAMP
)
ON CONFLICT(support_dept_code) DO UPDATE SET
raw_contract_code = excluded.raw_contract_code,
business_division = excluded.business_division,
order_method = excluded.order_method,
owner_department = excluded.owner_department,
client_name = excluded.client_name,
support_dept_name = excluded.support_dept_name,
work_category = excluded.work_category,
order_date = excluded.order_date,
contract_date = excluded.contract_date,
project_start_date = excluded.project_start_date,
project_end_date = excluded.project_end_date,
contract_status = excluded.contract_status,
joint_contract = excluded.joint_contract,
pm_name = excluded.pm_name,
progress_status = excluded.progress_status,
total_contract_amount = excluded.total_contract_amount,
hanmac_contract_amount = excluded.hanmac_contract_amount,
source_file = excluded.source_file,
updated_at = CURRENT_TIMESTAMP
"""
),
payload,
)
inserted += 1
refresh_contract_review_tags()
sync_auto_project_related_links()
return inserted
def import_billing_status_workbook(workbook: Any, source_file: str) -> int:
sheet = workbook.active
with engine.begin() as conn:
conn.execute(
text("DELETE FROM project_billing_entries WHERE source_file = :source_file"),
{"source_file": source_file},
)
inserted = 0
current: dict[str, Any] = {}
for row in sheet.iter_rows(min_row=6, values_only=True):
values = list(row)
if values and all(value in (None, "") for value in values):
continue
if values[0] is not None:
current["support_department"] = normalize_text(values[0])
if len(values) > 1 and values[1] is not None:
current["business_division"] = normalize_text(values[1])
if len(values) > 2 and values[2] is not None:
current["raw_project_code"] = normalize_text(values[2])
if len(values) > 3 and values[3] is not None:
current["round_code"] = normalize_text(values[3])
if len(values) > 4 and values[4] is not None:
current["support_dept_name"] = normalize_text(values[4])
if len(values) > 5 and values[5] is not None:
current["contract_amount"] = normalize_amount(values[5])
if len(values) > 6 and values[6] is not None:
current["client_name"] = normalize_text(values[6])
round_code_text = normalize_text(current.get("round_code"))
round_prefix = next((character.upper() for character in round_code_text if character.isalpha()), "Y")
support_dept_code = normalize_project_code(
current.get("raw_project_code") or current.get("round_code"),
default_prefix=round_prefix,
)
if not support_dept_code:
continue
summary_row = normalize_text(values[11] if len(values) > 11 else "") == "합계" or normalize_text(values[10] if len(values) > 10 else "").startswith("수금 :")
department_summary = "합계" in normalize_text(values[4] if len(values) > 4 else "")
if summary_row or department_summary:
continue
payload = {
"support_dept_code": support_dept_code,
"raw_project_code": normalize_text(current.get("raw_project_code")),
"round_code": normalize_text(current.get("round_code")),
"support_department": normalize_text(current.get("support_department")),
"business_division": normalize_text(current.get("business_division")),
"support_dept_name": normalize_text(current.get("support_dept_name")),
"contract_amount": normalize_amount(current.get("contract_amount")),
"client_name": normalize_text(current.get("client_name")),
"billing_type": normalize_text(values[7] if len(values) > 7 else ""),
"progress_round": normalize_round_value(values[8] if len(values) > 8 else ""),
"billing_date": normalize_date_text(values[9] if len(values) > 9 else ""),
"tax_invoice_date": normalize_date_text(values[10] if len(values) > 10 else ""),
"expected_collection_date": normalize_date_text(values[11] if len(values) > 11 else ""),
"billed_amount": normalize_amount(values[12] if len(values) > 12 else 0),
"collected_amount": normalize_amount(values[13] if len(values) > 13 else 0),
"balance_amount": normalize_amount(values[14] if len(values) > 14 else 0),
"collection_rate": normalize_amount(values[15] if len(values) > 15 else 0),
"note": normalize_text(values[16] if len(values) > 16 else ""),
"source_file": source_file,
}
conn.execute(
text(
"""
INSERT INTO project_billing_entries (
support_dept_code, raw_project_code, round_code, support_department,
business_division, support_dept_name, contract_amount, client_name,
billing_type, progress_round, billing_date, tax_invoice_date,
expected_collection_date, billed_amount, collected_amount,
balance_amount, collection_rate, note, source_file, updated_at
) VALUES (
:support_dept_code, :raw_project_code, :round_code, :support_department,
:business_division, :support_dept_name, :contract_amount, :client_name,
:billing_type, :progress_round, :billing_date, :tax_invoice_date,
:expected_collection_date, :billed_amount, :collected_amount,
:balance_amount, :collection_rate, :note, :source_file, CURRENT_TIMESTAMP
)
"""
),
payload,
)
inserted += 1
refresh_contract_review_tags()
return inserted
def import_change_contract_summary_workbook(workbook: Any, source_file: str) -> int:
sheet = workbook.active
with engine.begin() as conn:
conn.execute(
text("DELETE FROM project_contract_change_summary WHERE source_file = :source_file"),
{"source_file": source_file},
)
inserted = 0
current_department = ""
for row in sheet.iter_rows(min_row=2, values_only=True):
values = list(row)
if not values or all(value in (None, "") for value in values):
continue
first_value = normalize_text(values[0] if len(values) > 0 else "")
if "소 계" in first_value or first_value.startswith("<"):
continue
if first_value:
current_department = first_value
raw_summary_code = normalize_text(values[2] if len(values) > 2 else "").replace("\xa0", "")
support_dept_name = normalize_text(values[3] if len(values) > 3 else "")
if not raw_summary_code or not support_dept_name:
continue
payload = {
"raw_summary_code": "".join(character for character in raw_summary_code if character.isdigit()),
"normalized_title": normalize_project_title_for_linking(support_dept_name),
"owner_department": current_department,
"business_division": normalize_text(values[1] if len(values) > 1 else ""),
"support_dept_name": support_dept_name,
"change_date": normalize_date_text(values[4] if len(values) > 4 else ""),
"client_name": normalize_text(values[5] if len(values) > 5 else ""),
"original_contract_period": normalize_text(values[6] if len(values) > 6 else ""),
"changed_project_end_date": normalize_date_text(values[7] if len(values) > 7 else ""),
"initial_contract_amount": normalize_amount(values[8] if len(values) > 8 else 0),
"previous_contract_amount": normalize_amount(values[9] if len(values) > 9 else 0),
"changed_contract_amount": normalize_amount(values[10] if len(values) > 10 else 0),
"delta_amount": normalize_amount(values[11] if len(values) > 11 else 0),
"source_file": source_file,
}
conn.execute(
text(
"""
INSERT INTO project_contract_change_summary (
raw_summary_code, normalized_title, owner_department, business_division,
support_dept_name, change_date, client_name, original_contract_period,
changed_project_end_date, initial_contract_amount, previous_contract_amount,
changed_contract_amount, delta_amount, source_file, updated_at
) VALUES (
:raw_summary_code, :normalized_title, :owner_department, :business_division,
:support_dept_name, :change_date, :client_name, :original_contract_period,
:changed_project_end_date, :initial_contract_amount, :previous_contract_amount,
:changed_contract_amount, :delta_amount, :source_file, CURRENT_TIMESTAMP
)
"""
),
payload,
)
inserted += 1
sync_auto_project_related_links()
return inserted
def import_change_contract_round_workbook(workbook: Any, source_file: str) -> int:
sheet = workbook.active
with engine.begin() as conn:
conn.execute(
text("DELETE FROM project_contract_change_round WHERE source_file = :source_file"),
{"source_file": source_file},
)
inserted = 0
current_department = ""
for row in sheet.iter_rows(min_row=2, values_only=True):
values = list(row)
if not values or all(value in (None, "") for value in values):
continue
first_value = normalize_text(values[0] if len(values) > 0 else "")
if "소 계" in first_value or first_value.startswith("<"):
continue
if first_value:
current_department = first_value
raw_round_code = normalize_text(values[2] if len(values) > 2 else "").replace("\xa0", "")
support_dept_name = normalize_text(values[3] if len(values) > 3 else "")
support_dept_code = normalize_project_code(raw_round_code)
if not support_dept_code or not support_dept_name:
continue
payload = {
"support_dept_code": support_dept_code,
"raw_round_code": raw_round_code,
"normalized_title": normalize_project_title_for_linking(support_dept_name),
"owner_department": current_department,
"business_division": normalize_text(values[1] if len(values) > 1 else ""),
"support_dept_name": support_dept_name,
"change_date": normalize_date_text(values[4] if len(values) > 4 else ""),
"client_name": normalize_text(values[5] if len(values) > 5 else ""),
"original_contract_period": normalize_text(values[6] if len(values) > 6 else ""),
"changed_project_end_date": normalize_date_text(values[7] if len(values) > 7 else ""),
"initial_contract_amount": normalize_amount(values[8] if len(values) > 8 else 0),
"changed_contract_amount": normalize_amount(values[9] if len(values) > 9 else 0),
"delta_amount": normalize_amount(values[10] if len(values) > 10 else 0),
"source_file": source_file,
}
conn.execute(
text(
"""
INSERT INTO project_contract_change_round (
support_dept_code, raw_round_code, normalized_title, owner_department,
business_division, support_dept_name, change_date, client_name,
original_contract_period, changed_project_end_date, initial_contract_amount,
changed_contract_amount, delta_amount, source_file, updated_at
) VALUES (
:support_dept_code, :raw_round_code, :normalized_title, :owner_department,
:business_division, :support_dept_name, :change_date, :client_name,
:original_contract_period, :changed_project_end_date, :initial_contract_amount,
:changed_contract_amount, :delta_amount, :source_file, CURRENT_TIMESTAMP
)
"""
),
payload,
)
inserted += 1
sync_auto_project_related_links()
return inserted
def select_latest_contract_change_entry(rows: list[dict[str, Any]]) -> dict[str, Any]:
def sort_key(row: dict[str, Any]) -> tuple[str, int, float, float]:
return (
normalize_text(row.get("change_date")),
1 if normalize_amount(row.get("changed_contract_amount")) > 0 else 0,
abs(normalize_amount(row.get("delta_amount"))),
normalize_amount(row.get("changed_contract_amount")),
)
return max(rows, key=sort_key) if rows else {}
def get_project_contract_change_maps() -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], dict[str, str], dict[str, str]]:
with engine.begin() as conn:
summary_rows = conn.execute(
text("SELECT * FROM project_contract_change_summary ORDER BY raw_summary_code, change_date, id")
).mappings().all()
round_rows = conn.execute(
text("SELECT * FROM project_contract_change_round ORDER BY support_dept_code, change_date, id")
).mappings().all()
existing_codes = {
normalize_text(row[0])
for row in conn.execute(
text(
"""
SELECT DISTINCT support_dept_code FROM transactions WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_status WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_contract_info WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_billing_entries WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_contract_change_round WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).fetchall()
if normalize_text(row[0])
}
contracted_codes = {
normalize_text(row[0])
for row in conn.execute(
text(
"""
SELECT support_dept_code
FROM project_contract_info
WHERE COALESCE(hanmac_contract_amount, 0) > 0
UNION
SELECT support_dept_code
FROM project_billing_entries
GROUP BY support_dept_code
HAVING SUM(COALESCE(collected_amount, 0)) > 0 OR MAX(COALESCE(contract_amount, 0)) > 0
"""
)
).fetchall()
if normalize_text(row[0])
}
summary_by_title: dict[str, list[dict[str, Any]]] = {}
for row in summary_rows:
title_key = normalize_text(row["normalized_title"])
if title_key:
summary_by_title.setdefault(title_key, []).append(dict(row))
latest_summary_by_title = {
title_key: select_latest_contract_change_entry(rows)
for title_key, rows in summary_by_title.items()
if rows
}
round_by_code: dict[str, list[dict[str, Any]]] = {}
round_codes_by_title: dict[str, set[str]] = {}
for row in round_rows:
code = normalize_text(row["support_dept_code"])
title_key = normalize_text(row["normalized_title"])
if code:
round_by_code.setdefault(code, []).append(dict(row))
if title_key and code:
round_codes_by_title.setdefault(title_key, set()).add(code)
latest_round_by_code = {
code: select_latest_contract_change_entry(rows)
for code, rows in round_by_code.items()
if rows
}
representative_by_title: dict[str, str] = {}
for title_key, codes in round_codes_by_title.items():
sorted_codes = sorted(codes)
if not sorted_codes:
continue
def representative_rank(code: str) -> tuple[int, str]:
if code in contracted_codes:
return (0, code)
if code.startswith("Y") and code in existing_codes:
return (1, code)
if code.startswith("Z"):
return (2, code)
if code.startswith("X"):
return (3, code)
return (4, code)
representative_by_title[title_key] = min(sorted_codes, key=representative_rank)
title_by_code = {
code: title_key
for title_key, codes in round_codes_by_title.items()
for code in codes
}
return latest_summary_by_title, latest_round_by_code, representative_by_title, title_by_code
def refresh_contract_review_tags() -> None:
latest_summary_by_title, latest_round_by_code, _, title_by_code = get_project_contract_change_maps()
with engine.begin() as conn:
billing_rows = conn.execute(
text(
"""
SELECT support_dept_code, MAX(contract_amount) AS billing_contract_amount
FROM project_billing_entries
GROUP BY support_dept_code
"""
)
).mappings().all()
billing_map = {
normalize_text(row["support_dept_code"]): normalize_amount(row["billing_contract_amount"])
for row in billing_rows
if normalize_text(row["support_dept_code"])
}
contract_rows = conn.execute(
text("SELECT support_dept_code, support_dept_name, hanmac_contract_amount FROM project_contract_info")
).mappings().all()
for row in contract_rows:
support_dept_code = normalize_text(row["support_dept_code"])
hanmac_contract_amount = normalize_amount(row["hanmac_contract_amount"])
billing_contract_amount = normalize_amount(billing_map.get(support_dept_code))
title_key = title_by_code.get(support_dept_code) or normalize_project_title_for_linking(row.get("support_dept_name"))
latest_changed_contract_amount = (
normalize_amount((latest_summary_by_title.get(title_key) or {}).get("changed_contract_amount"))
or normalize_amount((latest_round_by_code.get(support_dept_code) or {}).get("changed_contract_amount"))
)
comparison_contract_amount = latest_changed_contract_amount or hanmac_contract_amount
review_tag = ""
review_note = ""
if billing_contract_amount and abs(comparison_contract_amount - billing_contract_amount) > 0.5:
review_tag = "변경계약 검토 필요"
if latest_changed_contract_amount:
review_note = (
f"변경계약금액 {latest_changed_contract_amount:,.0f}원 / "
f"기성청구현황 계약금액 {billing_contract_amount:,.0f}원"
)
else:
review_note = (
f"계약현황 한맥계약금액 {hanmac_contract_amount:,.0f}원 / "
f"기성청구현황 계약금액 {billing_contract_amount:,.0f}원"
)
conn.execute(
text(
"""
UPDATE project_contract_info
SET review_tag = :review_tag,
review_note = :review_note,
updated_at = CURRENT_TIMESTAMP
WHERE support_dept_code = :support_dept_code
"""
),
{
"support_dept_code": support_dept_code,
"review_tag": review_tag,
"review_note": review_note,
},
)
def sync_auto_project_related_links() -> None:
with engine.begin() as conn:
billing_rows = conn.execute(
text(
"""
SELECT support_dept_code, raw_project_code, round_code
FROM project_billing_entries
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).mappings().all()
change_round_rows = conn.execute(
text(
"""
SELECT support_dept_code, normalized_title, support_dept_name
FROM project_contract_change_round
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).mappings().all()
existing_codes = {
normalize_text(row[0])
for row in conn.execute(
text(
"""
SELECT DISTINCT support_dept_code FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_status
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_contract_info
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_billing_entries
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_contract_change_round
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).fetchall()
if normalize_text(row[0])
}
project_names = {
normalize_text(row[0]): normalize_text(row[1])
for row in conn.execute(
text(
"""
WITH project_names AS (
SELECT support_dept_code, MAX(support_dept_name) AS support_dept_name
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
GROUP BY support_dept_code
UNION
SELECT support_dept_code, support_dept_name
FROM project_contract_info
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT support_dept_code, support_dept_name
FROM project_billing_entries
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT support_dept_code, support_dept_name
FROM project_contract_change_round
WHERE COALESCE(support_dept_code, '') <> ''
)
SELECT support_dept_code, support_dept_name
FROM project_names
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).fetchall()
if normalize_text(row[0])
}
contracted_codes = {
normalize_text(row[0])
for row in conn.execute(
text(
"""
SELECT support_dept_code
FROM project_contract_info
WHERE COALESCE(hanmac_contract_amount, 0) > 0
UNION
SELECT support_dept_code
FROM project_billing_entries
GROUP BY support_dept_code
HAVING SUM(COALESCE(collected_amount, 0)) > 0 OR MAX(COALESCE(contract_amount, 0)) > 0
"""
)
).fetchall()
if normalize_text(row[0])
}
cluster_map: dict[str, set[str]] = {}
for row in billing_rows:
base_code = normalize_text(row["support_dept_code"])
raw_project_code = normalize_text(row["raw_project_code"])
round_code = normalize_project_code(row["round_code"], default_prefix=base_code[:1] or "Y")
if not base_code:
continue
cluster_key = raw_project_code or base_code
cluster = cluster_map.setdefault(cluster_key, set())
if base_code in existing_codes:
cluster.add(base_code)
if round_code and round_code in existing_codes:
cluster.add(round_code)
change_round_title_groups: dict[str, set[str]] = {}
change_contract_codes: set[str] = set()
for row in change_round_rows:
code = normalize_text(row["support_dept_code"])
title_key = normalize_text(row["normalized_title"]) or normalize_project_title_for_linking(row["support_dept_name"])
if not code or not title_key:
continue
change_contract_codes.add(code)
change_round_title_groups.setdefault(title_key, set()).add(code)
project_codes_by_title: dict[str, set[str]] = {}
for code, name in project_names.items():
title_key = normalize_project_title_for_linking(name)
if title_key:
project_codes_by_title.setdefault(title_key, set()).add(code)
for title_key, round_codes in change_round_title_groups.items():
cluster_codes = set(round_codes)
cluster_codes.update(project_codes_by_title.get(title_key, set()))
cluster_map.setdefault(f"change_round::{title_key}", set()).update(
code for code in cluster_codes if code in existing_codes
)
title_groups: dict[str, set[str]] = {}
for code, name in project_names.items():
normalized_title = normalize_project_title_for_linking(name)
if len(normalized_title) < 8:
continue
title_groups.setdefault(normalized_title, set()).add(code)
def choose_representative(codes: set[str]) -> str:
sorted_codes = sorted(codes)
contracted_non_special = [
code
for code in sorted_codes
if code in contracted_codes and not code.startswith(("X", "Z"))
]
if contracted_non_special:
return contracted_non_special[0]
plain_non_special = [
code
for code in sorted_codes
if not code.startswith(("X", "Z"))
and not has_project_variant_marker(project_names.get(code, ""))
]
if plain_non_special:
return plain_non_special[0]
contracted_any = [code for code in sorted_codes if code in contracted_codes]
if contracted_any:
return contracted_any[0]
return sorted_codes[0] if sorted_codes else ""
for normalized_title, codes in title_groups.items():
if len(codes) < 2:
continue
representative_code = choose_representative(codes)
if not representative_code:
continue
variant_codes = {
code
for code in codes
if code != representative_code and (
code.startswith(("X", "Z"))
or has_project_variant_marker(project_names.get(code, ""))
)
}
if not variant_codes:
continue
cluster_map.setdefault(f"title::{normalized_title}", set()).update({representative_code, *variant_codes})
title_items = sorted(
((title_key, set(codes)) for title_key, codes in title_groups.items() if len(title_key) >= 8),
key=lambda item: len(item[0]),
)
for index, (base_title, base_codes) in enumerate(title_items):
for related_title, related_codes in title_items[index + 1:]:
if base_title not in related_title and related_title not in base_title:
continue
merged_codes = set(base_codes) | set(related_codes)
representative_code = choose_representative(merged_codes)
if not representative_code:
continue
variant_codes = {
code
for code in merged_codes
if code != representative_code and (
code.startswith(("X", "Z"))
or has_project_variant_marker(project_names.get(code, ""))
)
}
if not variant_codes:
continue
cluster_map.setdefault(
f"title_fuzzy::{base_title if len(base_title) <= len(related_title) else related_title}",
set(),
).update({representative_code, *variant_codes})
conn.execute(text("DELETE FROM project_related_links WHERE COALESCE(link_source, 'manual') LIKE 'auto%'"))
for cluster_key, cluster_codes in cluster_map.items():
normalized_cluster = sorted(cluster_codes)
if len(normalized_cluster) < 2:
continue
if str(cluster_key).startswith("change_round::"):
link_source = "auto_change_contract"
elif str(cluster_key).startswith("title_fuzzy::"):
link_source = "auto_title_fuzzy"
elif str(cluster_key).startswith("title::"):
link_source = "auto_title"
else:
link_source = "auto_round"
for base_code in normalized_cluster:
for related_code in normalized_cluster:
if base_code == related_code:
continue
conn.execute(
text(
"""
INSERT INTO project_related_links (
base_support_dept_code,
related_support_dept_code,
link_source,
updated_at
) VALUES (
:base_support_dept_code,
:related_support_dept_code,
:link_source,
CURRENT_TIMESTAMP
)
ON CONFLICT(base_support_dept_code, related_support_dept_code) DO UPDATE SET
link_source = excluded.link_source,
updated_at = CURRENT_TIMESTAMP
"""
),
{
"base_support_dept_code": base_code,
"related_support_dept_code": related_code,
"link_source": link_source,
},
)
for base_code, related_code, link_source in conn.execute(
text(
"""
SELECT base_support_dept_code, related_support_dept_code, COALESCE(link_source, 'manual') AS link_source
FROM project_related_links
WHERE COALESCE(link_source, 'manual') LIKE 'auto_title%'
"""
)
).fetchall():
normalized_base = normalize_text(base_code)
normalized_related = normalize_text(related_code)
if (
normalized_base in change_contract_codes
and normalized_related not in change_contract_codes
and not normalized_related.startswith("X")
) or (
normalized_related in change_contract_codes
and normalized_base not in change_contract_codes
and not normalized_base.startswith("X")
):
conn.execute(
text(
"""
DELETE FROM project_related_links
WHERE base_support_dept_code = :base_support_dept_code
AND related_support_dept_code = :related_support_dept_code
"""
),
{
"base_support_dept_code": normalized_base,
"related_support_dept_code": normalized_related,
},
)
x_codes_linked_to_change: set[str] = set()
for base_code, related_code in conn.execute(
text(
"""
SELECT base_support_dept_code, related_support_dept_code
FROM project_related_links
WHERE (
base_support_dept_code LIKE 'X%' AND related_support_dept_code <> ''
) OR (
related_support_dept_code LIKE 'X%' AND base_support_dept_code <> ''
)
"""
)
).fetchall():
normalized_base = normalize_text(base_code)
normalized_related = normalize_text(related_code)
if normalized_base.startswith("X") and normalized_related in change_contract_codes:
x_codes_linked_to_change.add(normalized_base)
if normalized_related.startswith("X") and normalized_base in change_contract_codes:
x_codes_linked_to_change.add(normalized_related)
for base_code, related_code in conn.execute(
text(
"""
SELECT base_support_dept_code, related_support_dept_code
FROM project_related_links
WHERE COALESCE(link_source, 'manual') LIKE 'auto_title%'
"""
)
).fetchall():
normalized_base = normalize_text(base_code)
normalized_related = normalize_text(related_code)
should_delete = (
normalized_base in x_codes_linked_to_change
and normalized_related not in change_contract_codes
and not normalized_related.startswith("X")
) or (
normalized_related in x_codes_linked_to_change
and normalized_base not in change_contract_codes
and not normalized_base.startswith("X")
)
if should_delete:
conn.execute(
text(
"""
DELETE FROM project_related_links
WHERE base_support_dept_code = :base_support_dept_code
AND related_support_dept_code = :related_support_dept_code
"""
),
{
"base_support_dept_code": normalized_base,
"related_support_dept_code": normalized_related,
},
)
@app.on_event("startup")
def on_startup() -> None:
init_db()
auto_import_project_excels()
sync_auto_project_related_links()
normalize_all_collection_entry_storage()
logger.info("DB ready at %s", DB_PATH)
def normalize_text(value: Any) -> str:
if value is None:
return ""
if isinstance(value, str):
return value.strip()
return str(value).strip()
def normalize_amount(value: Any) -> float:
if value in (None, ""):
return 0.0
if isinstance(value, (int, float)):
return float(value)
cleaned = (
str(value)
.strip()
.replace(",", "")
.replace("원", "")
.replace("(", "-")
.replace(")", "")
)
if not cleaned:
return 0.0
try:
return float(cleaned)
except ValueError:
return 0.0
def normalize_date_text(value: Any) -> str:
if value in (None, ""):
return ""
if isinstance(value, datetime):
return value.date().isoformat()
if isinstance(value, date):
return value.isoformat()
text_value = normalize_text(value)
for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y%m%d"):
try:
return datetime.strptime(text_value, fmt).date().isoformat()
except ValueError:
continue
return text_value
def normalize_collection_progress_type(value: Any) -> str:
text_value = normalize_text(value)
if not text_value:
return ""
if "준공" in text_value:
return "준공금"
if any(keyword in text_value for keyword in ("선수", "선급")):
return "선급금"
if "기성" in text_value:
return "기성금"
return ""
def normalize_collection_billing_type(value: Any) -> str:
text_value = normalize_text(value)
if not text_value:
return ""
if "계약" in text_value:
return "계약분"
if "기타" in text_value:
return "기타"
return ""
def normalize_collection_entry_fields(row: dict[str, Any]) -> dict[str, Any]:
normalized = dict(row)
progress_type = normalize_collection_progress_type(normalized.get("progress_type"))
raw_billing_type = normalize_text(normalized.get("billing_type"))
billing_type = normalize_collection_billing_type(raw_billing_type)
if not progress_type and raw_billing_type in {"선수금", "선급금", "기성금", "준공금"}:
progress_type = normalize_collection_progress_type(raw_billing_type)
normalized["progress_type"] = progress_type
normalized["billing_type"] = billing_type or ("계약분" if progress_type else "")
return normalized
def normalize_project_code(value: Any, default_prefix: str = "Y") -> str:
text_value = normalize_text(value).replace("\u3164", "").replace("\xa0", "")
prefix = ""
for character in text_value:
if character.isalpha():
prefix = character.upper()
break
digits = "".join(character for character in text_value if character.isdigit())
if not digits:
return ""
return f"{prefix or default_prefix}{int(digits)}"
def normalize_round_value(value: Any) -> str:
text_value = normalize_text(value)
if not text_value:
return ""
digits = "".join(character for character in text_value if character.isdigit())
if digits:
return str(int(digits))
return text_value
PROJECT_TITLE_LINK_STRIP_PATTERNS = (
r"\((?:\d+\s*차|[가-힣A-Za-z0-9\s]*변경[가-힣A-Za-z0-9\s]*|[가-힣A-Za-z0-9\s]*보완[가-힣A-Za-z0-9\s]*|[가-힣A-Za-z0-9\s]*입찰[가-힣A-Za-z0-9\s]*|가칭)\)",
r"\d+\s*차",
r"변경",
r"보완",
r"입찰",
r"기술제안",
r"실시설계",
r"조사[·ㆍ]설계",
r"조사설계",
r"기본\s*및\s*실시설계",
r"기본및실시설계",
r"기본설계",
r"설계",
r"용역",
r"시공단계",
r"건설사업관리",
r"가칭",
)
def normalize_project_title_for_linking(value: Any) -> str:
text_value = normalize_text(value)
if not text_value:
return ""
normalized = text_value
for pattern in PROJECT_TITLE_LINK_STRIP_PATTERNS:
normalized = re.sub(pattern, "", normalized, flags=re.IGNORECASE)
normalized = re.sub(r"[^가-힣A-Za-z0-9]", "", normalized)
return normalized
def has_project_variant_marker(value: Any) -> bool:
text_value = normalize_text(value)
if not text_value:
return False
return bool(re.search(r"(\d+\s*차|[nN]\s*차|변경|보완|지연보상금|가칭|연차|년분)", text_value))
def classify_special_x_project(name: Any) -> str:
normalized_name = normalize_text(name)
if not normalized_name:
return "precontract"
rules = get_special_x_classification_rules()
for category_key, keywords in rules.items():
if any(keyword in normalized_name for keyword in keywords):
return category_key
return "precontract"
def decode_json_rows(value: Any) -> list[dict[str, Any]]:
text_value = normalize_text(value)
if not text_value:
return []
try:
rows = json.loads(text_value)
except json.JSONDecodeError:
return []
return [row for row in rows if isinstance(row, dict)]
def encode_json_rows(rows: list[dict[str, Any]]) -> str:
return json.dumps(rows, ensure_ascii=False)
def normalize_collection_entry_row(row: dict[str, Any]) -> dict[str, Any]:
normalized = {
"vendor": clean_row_text(row.get("vendor")),
"progress_type": clean_row_text(row.get("progress_type")),
"billing_round": normalize_round_value(row.get("billing_round")),
"billing_type": clean_row_text(row.get("billing_type")),
"billing_date": normalize_date_text(row.get("billing_date")),
"billed_amount": normalize_amount(row.get("billed_amount")),
"round": normalize_round_value(row.get("round")),
"date": normalize_date_text(row.get("date")),
"due_date": normalize_date_text(row.get("due_date")),
"amount": normalize_amount(row.get("amount")),
"balance_amount": normalize_amount(row.get("balance_amount")),
"collection_rate": normalize_amount(row.get("collection_rate")),
"note": clean_row_text(row.get("note")),
}
return normalize_collection_entry_fields(normalized)
def normalize_task_plan_entry_row(row: dict[str, Any]) -> dict[str, Any]:
return {
"group": clean_row_text(row.get("group")),
"dept_name": clean_row_text(row.get("dept_name")),
"work_name": clean_row_text(row.get("work_name")),
"amount": normalize_amount(row.get("amount")),
}
def normalize_exec_budget_entry_row(row: dict[str, Any]) -> dict[str, Any]:
return {
"group": clean_row_text(row.get("group")),
"grade": clean_row_text(row.get("grade")),
"hours": clean_row_text(row.get("hours")),
"dept_name": clean_row_text(row.get("dept_name")),
"work_name": clean_row_text(row.get("work_name")),
"account_code": clean_row_text(row.get("account_code")),
"account_name": clean_row_text(row.get("account_name")),
"amount": normalize_amount(row.get("amount")),
}
def normalize_actual_input_entry_row(row: dict[str, Any]) -> dict[str, Any]:
return {
"group": clean_row_text(row.get("group")),
"grade": clean_row_text(row.get("grade")),
"minutes": clean_row_text(row.get("minutes")),
"label": clean_row_text(row.get("label")),
"reference": clean_row_text(row.get("reference")),
"note": clean_row_text(row.get("note")),
"amount": normalize_amount(row.get("amount")),
}
def extract_project_status_entry_sets(row: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
collection_entries = [normalize_collection_entry_row(item) for item in decode_json_rows(row.get("collection_entries_json"))]
task_plan_entries = [normalize_task_plan_entry_row(item) for item in decode_json_rows(row.get("task_plan_entries_json"))]
exec_budget_entries = [normalize_exec_budget_entry_row(item) for item in decode_json_rows(row.get("exec_budget_entries_json"))]
actual_input_entries = [normalize_actual_input_entry_row(item) for item in decode_json_rows(row.get("actual_input_entries_json"))]
if not collection_entries and normalize_amount(row.get("collection_amount")):
collection_entries = [
normalize_collection_entry_row(
{
"vendor": "",
"round": "",
"amount": row.get("collection_amount", ""),
"date": "",
"due_date": "",
"note": "기존 수기 입력값",
}
)
]
if not task_plan_entries:
fallback_task_rows = []
if normalize_amount(row.get("task_plan_department_budget")):
fallback_task_rows.append(
{
"group": "department",
"dept_name": "기존 부서별 배분",
"work_name": "",
"amount": row.get("task_plan_department_budget", ""),
}
)
if normalize_amount(row.get("task_plan_outsource_budget")):
fallback_task_rows.append(
{
"group": "outsource",
"dept_name": "기존 외주비",
"work_name": row.get("task_plan_outsource_detail", ""),
"amount": row.get("task_plan_outsource_budget", ""),
}
)
if normalize_amount(row.get("task_plan_joint_operating_cost")):
fallback_task_rows.append(
{
"group": "joint",
"dept_name": "기존 합사운영비",
"work_name": "",
"amount": row.get("task_plan_joint_operating_cost", ""),
}
)
task_plan_entries = [normalize_task_plan_entry_row(item) for item in fallback_task_rows]
if not exec_budget_entries:
fallback_exec_rows = []
if normalize_amount(row.get("exec_budget_labor_by_grade")):
fallback_exec_rows.append(
{
"group": "labor",
"grade": "기존 인건비",
"hours": "",
"amount": row.get("exec_budget_labor_by_grade", ""),
}
)
if normalize_amount(row.get("exec_budget_outsource")):
fallback_exec_rows.append(
{
"group": "outsource",
"dept_name": "기존 외주비",
"work_name": "",
"amount": row.get("exec_budget_outsource", ""),
}
)
if normalize_amount(row.get("exec_budget_cost_plan")):
fallback_exec_rows.append(
{
"group": "cost_plan",
"account_code": "기존",
"account_name": "비용계획",
"amount": row.get("exec_budget_cost_plan", ""),
}
)
exec_budget_entries = [normalize_exec_budget_entry_row(item) for item in fallback_exec_rows]
if not actual_input_entries and normalize_amount(row.get("item_investment")):
actual_input_entries = [
normalize_actual_input_entry_row(
{
"reference": "",
"amount": row.get("item_investment", ""),
"note": "기존 항목별투입액",
}
)
]
return {
"collection_entries": collection_entries,
"task_plan_entries": task_plan_entries,
"exec_budget_entries": exec_budget_entries,
"actual_input_entries": actual_input_entries,
}
def replace_project_status_child_entries(
conn: Any,
support_dept_code: str,
collection_entries: list[dict[str, Any]],
task_plan_entries: list[dict[str, Any]],
exec_budget_entries: list[dict[str, Any]],
actual_input_entries: list[dict[str, Any]],
) -> None:
conn.execute(
text("DELETE FROM project_collection_entries WHERE support_dept_code = :support_dept_code"),
{"support_dept_code": support_dept_code},
)
conn.execute(
text("DELETE FROM project_task_plan_entries WHERE support_dept_code = :support_dept_code"),
{"support_dept_code": support_dept_code},
)
conn.execute(
text("DELETE FROM project_exec_budget_entries WHERE support_dept_code = :support_dept_code"),
{"support_dept_code": support_dept_code},
)
conn.execute(
text("DELETE FROM project_actual_input_entries WHERE support_dept_code = :support_dept_code"),
{"support_dept_code": support_dept_code},
)
for position, row in enumerate(collection_entries):
normalized = normalize_collection_entry_row(row)
conn.execute(
text(
"""
INSERT INTO project_collection_entries (
support_dept_code, position, vendor, progress_type, billing_round,
billing_type, billing_date, billed_amount, round, date, due_date,
amount, balance_amount, collection_rate, note, updated_at
) VALUES (
:support_dept_code, :position, :vendor, :progress_type, :billing_round,
:billing_type, :billing_date, :billed_amount, :round, :date, :due_date,
:amount, :balance_amount, :collection_rate, :note, CURRENT_TIMESTAMP
)
"""
),
{"support_dept_code": support_dept_code, "position": position, **normalized},
)
for position, row in enumerate(task_plan_entries):
normalized = normalize_task_plan_entry_row(row)
conn.execute(
text(
"""
INSERT INTO project_task_plan_entries (
support_dept_code, position, group_name, dept_name, work_name, amount, updated_at
) VALUES (
:support_dept_code, :position, :group_name, :dept_name, :work_name, :amount, CURRENT_TIMESTAMP
)
"""
),
{
"support_dept_code": support_dept_code,
"position": position,
"group_name": normalized["group"],
"dept_name": normalized["dept_name"],
"work_name": normalized["work_name"],
"amount": normalized["amount"],
},
)
for position, row in enumerate(exec_budget_entries):
normalized = normalize_exec_budget_entry_row(row)
conn.execute(
text(
"""
INSERT INTO project_exec_budget_entries (
support_dept_code, position, group_name, grade, hours, dept_name,
work_name, account_code, account_name, amount, updated_at
) VALUES (
:support_dept_code, :position, :group_name, :grade, :hours, :dept_name,
:work_name, :account_code, :account_name, :amount, CURRENT_TIMESTAMP
)
"""
),
{
"support_dept_code": support_dept_code,
"position": position,
"group_name": normalized["group"],
"grade": normalized["grade"],
"hours": normalized["hours"],
"dept_name": normalized["dept_name"],
"work_name": normalized["work_name"],
"account_code": normalized["account_code"],
"account_name": normalized["account_name"],
"amount": normalized["amount"],
},
)
for position, row in enumerate(actual_input_entries):
normalized = normalize_actual_input_entry_row(row)
conn.execute(
text(
"""
INSERT INTO project_actual_input_entries (
support_dept_code, position, group_name, grade, minutes, label,
reference, note, amount, updated_at
) VALUES (
:support_dept_code, :position, :group_name, :grade, :minutes, :label,
:reference, :note, :amount, CURRENT_TIMESTAMP
)
"""
),
{
"support_dept_code": support_dept_code,
"position": position,
"group_name": normalized["group"],
"grade": normalized["grade"],
"minutes": normalized["minutes"],
"label": normalized["label"],
"reference": normalized["reference"],
"note": normalized["note"],
"amount": normalized["amount"],
},
)
def load_project_status_entry_maps(conn: Any) -> dict[str, dict[str, list[dict[str, Any]]]]:
result: dict[str, dict[str, list[dict[str, Any]]]] = {}
collection_rows = conn.execute(
text(
"""
SELECT support_dept_code, position, vendor, progress_type, billing_round,
billing_type, billing_date, billed_amount, round, date, due_date,
amount, balance_amount, collection_rate, note
FROM project_collection_entries
ORDER BY support_dept_code, position, id
"""
)
).mappings().all()
for row in collection_rows:
code = normalize_text(row["support_dept_code"])
result.setdefault(code, {})["collection_entries"] = result.setdefault(code, {}).get("collection_entries", [])
result[code]["collection_entries"].append(
normalize_collection_entry_row({key: row[key] for key in row.keys() if key not in {"support_dept_code", "position"}})
)
task_rows = conn.execute(
text(
"""
SELECT support_dept_code, position, group_name, dept_name, work_name, amount
FROM project_task_plan_entries
ORDER BY support_dept_code, position, id
"""
)
).mappings().all()
for row in task_rows:
code = normalize_text(row["support_dept_code"])
result.setdefault(code, {})["task_plan_entries"] = result.setdefault(code, {}).get("task_plan_entries", [])
result[code]["task_plan_entries"].append(
normalize_task_plan_entry_row(
{
"group": row["group_name"],
"dept_name": row["dept_name"],
"work_name": row["work_name"],
"amount": row["amount"],
}
)
)
exec_rows = conn.execute(
text(
"""
SELECT support_dept_code, position, group_name, grade, hours, dept_name,
work_name, account_code, account_name, amount
FROM project_exec_budget_entries
ORDER BY support_dept_code, position, id
"""
)
).mappings().all()
for row in exec_rows:
code = normalize_text(row["support_dept_code"])
result.setdefault(code, {})["exec_budget_entries"] = result.setdefault(code, {}).get("exec_budget_entries", [])
result[code]["exec_budget_entries"].append(
normalize_exec_budget_entry_row(
{
"group": row["group_name"],
"grade": row["grade"],
"hours": row["hours"],
"dept_name": row["dept_name"],
"work_name": row["work_name"],
"account_code": row["account_code"],
"account_name": row["account_name"],
"amount": row["amount"],
}
)
)
actual_rows = conn.execute(
text(
"""
SELECT support_dept_code, position, group_name, grade, minutes, label,
reference, note, amount
FROM project_actual_input_entries
ORDER BY support_dept_code, position, id
"""
)
).mappings().all()
for row in actual_rows:
code = normalize_text(row["support_dept_code"])
result.setdefault(code, {})["actual_input_entries"] = result.setdefault(code, {}).get("actual_input_entries", [])
result[code]["actual_input_entries"].append(
normalize_actual_input_entry_row(
{
"group": row["group_name"],
"grade": row["grade"],
"minutes": row["minutes"],
"label": row["label"],
"reference": row["reference"],
"note": row["note"],
"amount": row["amount"],
}
)
)
return result
def ensure_project_entry_set(entry_set: dict[str, list[dict[str, Any]]] | None) -> dict[str, list[dict[str, Any]]]:
source = entry_set or {}
return {
"collection_entries": list(source.get("collection_entries", [])),
"task_plan_entries": list(source.get("task_plan_entries", [])),
"exec_budget_entries": list(source.get("exec_budget_entries", [])),
"actual_input_entries": list(source.get("actual_input_entries", [])),
}
def migrate_project_status_entries(conn: Any) -> None:
migrated_codes = {
normalize_text(row[0])
for row in conn.execute(
text(
"""
SELECT DISTINCT support_dept_code
FROM (
SELECT support_dept_code FROM project_collection_entries
UNION ALL
SELECT support_dept_code FROM project_task_plan_entries
UNION ALL
SELECT support_dept_code FROM project_exec_budget_entries
UNION ALL
SELECT support_dept_code FROM project_actual_input_entries
)
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).fetchall()
if normalize_text(row[0])
}
source_rows = conn.execute(
text(
"""
SELECT support_dept_code,
collection_entries_json,
task_plan_entries_json,
exec_budget_entries_json,
actual_input_entries_json,
collection_amount,
task_plan_department_budget,
task_plan_outsource_budget,
task_plan_outsource_detail,
task_plan_joint_operating_cost,
exec_budget_labor_by_grade,
exec_budget_outsource,
exec_budget_cost_plan,
item_investment
FROM project_status
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).mappings().all()
for row in source_rows:
support_dept_code = normalize_text(row["support_dept_code"])
if not support_dept_code or support_dept_code in migrated_codes:
continue
entry_sets = extract_project_status_entry_sets(dict(row))
replace_project_status_child_entries(
conn,
support_dept_code,
entry_sets["collection_entries"],
entry_sets["task_plan_entries"],
entry_sets["exec_budget_entries"],
entry_sets["actual_input_entries"],
)
def migrate_project_basic_info(conn: Any) -> None:
existing_codes = {
normalize_text(row[0])
for row in conn.execute(
text("SELECT support_dept_code FROM project_basic_info WHERE COALESCE(support_dept_code, '') <> ''")
).fetchall()
if normalize_text(row[0])
}
source_rows = conn.execute(
text(
"""
SELECT support_dept_code, support_dept_name, contract_amount, project_type,
expected_as_rate, expected_sga_rate, expected_as_cost, expected_sga_budget,
exec_labor_rates_json, change_round, project_start_date, project_end_date,
completion_status, notes, last_editor_session_id, last_client_submitted_at
FROM project_status
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).mappings().all()
for row in source_rows:
support_dept_code = normalize_text(row["support_dept_code"])
if not support_dept_code or support_dept_code in existing_codes:
continue
conn.execute(
text(
"""
INSERT INTO project_basic_info (
support_dept_code, support_dept_name, contract_amount, project_type,
expected_as_rate, expected_sga_rate, expected_as_cost, expected_sga_budget,
exec_labor_rates_json, change_round, project_start_date, project_end_date,
completion_status, notes, last_editor_session_id, last_client_submitted_at,
updated_at
) VALUES (
:support_dept_code, :support_dept_name, :contract_amount, :project_type,
:expected_as_rate, :expected_sga_rate, :expected_as_cost, :expected_sga_budget,
:exec_labor_rates_json, :change_round, :project_start_date, :project_end_date,
:completion_status, :notes, :last_editor_session_id, :last_client_submitted_at,
CURRENT_TIMESTAMP
)
"""
),
dict(row),
)
def save_project_basic_info_section(conn: Any, payload: dict[str, Any]) -> None:
conn.execute(
text(
"""
INSERT INTO project_basic_info (
support_dept_code, support_dept_name, contract_amount, project_type,
expected_as_rate, expected_sga_rate, expected_as_cost, expected_sga_budget,
exec_labor_rates_json, change_round, project_start_date, project_end_date,
completion_status, notes, last_editor_session_id, last_client_submitted_at,
updated_at
) VALUES (
:support_dept_code, :support_dept_name, :contract_amount, :project_type,
:expected_as_rate, :expected_sga_rate, :expected_as_cost, :expected_sga_budget,
:exec_labor_rates_json, :change_round, :project_start_date, :project_end_date,
:completion_status, :notes, :last_editor_session_id, :last_client_submitted_at,
CURRENT_TIMESTAMP
)
ON CONFLICT(support_dept_code) DO UPDATE SET
support_dept_name = excluded.support_dept_name,
contract_amount = excluded.contract_amount,
project_type = excluded.project_type,
expected_as_rate = excluded.expected_as_rate,
expected_sga_rate = excluded.expected_sga_rate,
expected_as_cost = excluded.expected_as_cost,
expected_sga_budget = excluded.expected_sga_budget,
exec_labor_rates_json = excluded.exec_labor_rates_json,
change_round = excluded.change_round,
project_start_date = excluded.project_start_date,
project_end_date = excluded.project_end_date,
completion_status = excluded.completion_status,
notes = excluded.notes,
last_editor_session_id = excluded.last_editor_session_id,
last_client_submitted_at = excluded.last_client_submitted_at,
updated_at = CURRENT_TIMESTAMP
"""
),
payload,
)
def load_project_status_entries_for_code(conn: Any, support_dept_code: str) -> dict[str, list[dict[str, Any]]]:
code = normalize_text(support_dept_code)
entry_map = ensure_project_entry_set(None)
if not code:
return entry_map
collection_rows = conn.execute(
text(
"""
SELECT vendor, progress_type, billing_round, billing_type, billing_date,
billed_amount, round, date, due_date, amount, balance_amount,
collection_rate, note
FROM project_collection_entries
WHERE support_dept_code = :support_dept_code
ORDER BY position, id
"""
),
{"support_dept_code": code},
).mappings().all()
entry_map["collection_entries"] = [
normalize_collection_entry_row(dict(row))
for row in collection_rows
]
task_rows = conn.execute(
text(
"""
SELECT group_name, dept_name, work_name, amount
FROM project_task_plan_entries
WHERE support_dept_code = :support_dept_code
ORDER BY position, id
"""
),
{"support_dept_code": code},
).mappings().all()
entry_map["task_plan_entries"] = [
normalize_task_plan_entry_row(
{
"group": row["group_name"],
"dept_name": row["dept_name"],
"work_name": row["work_name"],
"amount": row["amount"],
}
)
for row in task_rows
]
exec_rows = conn.execute(
text(
"""
SELECT group_name, grade, hours, dept_name, work_name, account_code, account_name, amount
FROM project_exec_budget_entries
WHERE support_dept_code = :support_dept_code
ORDER BY position, id
"""
),
{"support_dept_code": code},
).mappings().all()
entry_map["exec_budget_entries"] = [
normalize_exec_budget_entry_row(
{
"group": row["group_name"],
"grade": row["grade"],
"hours": row["hours"],
"dept_name": row["dept_name"],
"work_name": row["work_name"],
"account_code": row["account_code"],
"account_name": row["account_name"],
"amount": row["amount"],
}
)
for row in exec_rows
]
actual_rows = conn.execute(
text(
"""
SELECT group_name, grade, minutes, label, reference, note, amount
FROM project_actual_input_entries
WHERE support_dept_code = :support_dept_code
ORDER BY position, id
"""
),
{"support_dept_code": code},
).mappings().all()
entry_map["actual_input_entries"] = [
normalize_actual_input_entry_row(
{
"group": row["group_name"],
"grade": row["grade"],
"minutes": row["minutes"],
"label": row["label"],
"reference": row["reference"],
"note": row["note"],
"amount": row["amount"],
}
)
for row in actual_rows
]
return entry_map
def sync_project_status_cache_row(conn: Any, support_dept_code: str) -> None:
code = normalize_text(support_dept_code)
if not code:
return
basic_info = conn.execute(
text("SELECT * FROM project_basic_info WHERE support_dept_code = :support_dept_code"),
{"support_dept_code": code},
).mappings().first()
entry_set = load_project_status_entries_for_code(conn, code)
collection_entries = entry_set["collection_entries"]
task_plan_entries = entry_set["task_plan_entries"]
exec_budget_entries = entry_set["exec_budget_entries"]
actual_input_entries = entry_set["actual_input_entries"]
contract_amount = normalize_amount((basic_info or {}).get("contract_amount"))
collection_amount = sum_row_amounts(collection_entries)
progress_rate = (collection_amount / contract_amount * 100) if contract_amount else 0.0
task_plan_department_rows = [row for row in task_plan_entries if normalize_text(row.get("group")) == "department"]
task_plan_outsource_rows = [row for row in task_plan_entries if normalize_text(row.get("group")) == "outsource"]
task_plan_joint_rows = [row for row in task_plan_entries if normalize_text(row.get("group")) == "joint"]
exec_labor_rows = [row for row in exec_budget_entries if normalize_text(row.get("group")) == "labor"]
exec_outsource_rows = [row for row in exec_budget_entries if normalize_text(row.get("group")) == "outsource"]
exec_cost_plan_rows = [row for row in exec_budget_entries if normalize_text(row.get("group")) == "cost_plan"]
item_investment = sum_row_amounts(actual_input_entries)
support_name = normalize_text((basic_info or {}).get("support_dept_name"))
if not support_name:
support_name = normalize_text(
conn.execute(
text(
"""
SELECT support_dept_name
FROM (
SELECT support_dept_name, 1 AS priority FROM project_contract_info WHERE support_dept_code = :support_dept_code
UNION ALL
SELECT support_dept_name, 2 AS priority FROM project_billing_entries WHERE support_dept_code = :support_dept_code
UNION ALL
SELECT support_dept_name, 3 AS priority FROM transactions WHERE support_dept_code = :support_dept_code
)
WHERE COALESCE(support_dept_name, '') <> ''
ORDER BY priority
LIMIT 1
"""
),
{"support_dept_code": code},
).scalar()
)
payload = {
"support_dept_code": code,
"support_dept_name": support_name,
"progress_rate": progress_rate,
"contract_amount": contract_amount,
"collection_amount": collection_amount,
"collection_entries_json": encode_json_rows(collection_entries),
"change_round": normalize_text((basic_info or {}).get("change_round")),
"item_investment": item_investment,
"task_plan_department_budget": sum_row_amounts(task_plan_department_rows),
"task_plan_outsource_budget": sum_row_amounts(task_plan_outsource_rows),
"task_plan_outsource_detail": "\n".join(
f"{normalize_text(row.get('dept_name'))} / {normalize_text(row.get('work_name'))}: {format_amount_for_text(row.get('amount'))}".strip(" /:")
for row in task_plan_outsource_rows
),
"task_plan_joint_operating_cost": sum_row_amounts(task_plan_joint_rows),
"task_plan_entries_json": encode_json_rows(task_plan_entries),
"exec_budget_labor_by_grade": sum_row_amounts(exec_labor_rows),
"exec_labor_rates_json": normalize_text((basic_info or {}).get("exec_labor_rates_json")) or "{}",
"exec_budget_outsource": sum_row_amounts(exec_outsource_rows),
"exec_budget_cost_plan": sum_row_amounts(exec_cost_plan_rows),
"exec_budget_entries_json": encode_json_rows(exec_budget_entries),
"actual_input_entries_json": encode_json_rows(actual_input_entries),
"project_type": normalize_text((basic_info or {}).get("project_type")),
"expected_as_rate": normalize_amount((basic_info or {}).get("expected_as_rate")),
"expected_sga_rate": normalize_amount((basic_info or {}).get("expected_sga_rate")),
"expected_as_cost": normalize_amount((basic_info or {}).get("expected_as_cost")),
"expected_sga_budget": normalize_amount((basic_info or {}).get("expected_sga_budget")),
"last_editor_session_id": normalize_text((basic_info or {}).get("last_editor_session_id")),
"last_client_submitted_at": normalize_text((basic_info or {}).get("last_client_submitted_at")),
"project_start_date": normalize_text((basic_info or {}).get("project_start_date")),
"project_end_date": normalize_text((basic_info or {}).get("project_end_date")),
"completion_status": normalize_text((basic_info or {}).get("completion_status")),
"notes": normalize_text((basic_info or {}).get("notes")),
}
conn.execute(
text(
"""
INSERT INTO project_status (
support_dept_code, support_dept_name, progress_rate, contract_amount,
collection_amount, collection_entries_json, change_round, item_investment,
task_plan_department_budget, task_plan_outsource_budget, task_plan_outsource_detail,
task_plan_joint_operating_cost, task_plan_entries_json, exec_budget_labor_by_grade,
exec_labor_rates_json, exec_budget_outsource, exec_budget_cost_plan,
exec_budget_entries_json, actual_input_entries_json, project_type,
expected_as_rate, expected_sga_rate, expected_as_cost, expected_sga_budget,
last_editor_session_id, last_client_submitted_at, project_start_date,
project_end_date, completion_status, notes, updated_at
) VALUES (
:support_dept_code, :support_dept_name, :progress_rate, :contract_amount,
:collection_amount, :collection_entries_json, :change_round, :item_investment,
:task_plan_department_budget, :task_plan_outsource_budget, :task_plan_outsource_detail,
:task_plan_joint_operating_cost, :task_plan_entries_json, :exec_budget_labor_by_grade,
:exec_labor_rates_json, :exec_budget_outsource, :exec_budget_cost_plan,
:exec_budget_entries_json, :actual_input_entries_json, :project_type,
:expected_as_rate, :expected_sga_rate, :expected_as_cost, :expected_sga_budget,
:last_editor_session_id, :last_client_submitted_at, :project_start_date,
:project_end_date, :completion_status, :notes, CURRENT_TIMESTAMP
)
ON CONFLICT(support_dept_code) DO UPDATE SET
support_dept_name = excluded.support_dept_name,
progress_rate = excluded.progress_rate,
contract_amount = excluded.contract_amount,
collection_amount = excluded.collection_amount,
collection_entries_json = excluded.collection_entries_json,
change_round = excluded.change_round,
item_investment = excluded.item_investment,
task_plan_department_budget = excluded.task_plan_department_budget,
task_plan_outsource_budget = excluded.task_plan_outsource_budget,
task_plan_outsource_detail = excluded.task_plan_outsource_detail,
task_plan_joint_operating_cost = excluded.task_plan_joint_operating_cost,
task_plan_entries_json = excluded.task_plan_entries_json,
exec_budget_labor_by_grade = excluded.exec_budget_labor_by_grade,
exec_labor_rates_json = excluded.exec_labor_rates_json,
exec_budget_outsource = excluded.exec_budget_outsource,
exec_budget_cost_plan = excluded.exec_budget_cost_plan,
exec_budget_entries_json = excluded.exec_budget_entries_json,
actual_input_entries_json = excluded.actual_input_entries_json,
project_type = excluded.project_type,
expected_as_rate = excluded.expected_as_rate,
expected_sga_rate = excluded.expected_sga_rate,
expected_as_cost = excluded.expected_as_cost,
expected_sga_budget = excluded.expected_sga_budget,
last_editor_session_id = excluded.last_editor_session_id,
last_client_submitted_at = excluded.last_client_submitted_at,
project_start_date = excluded.project_start_date,
project_end_date = excluded.project_end_date,
completion_status = excluded.completion_status,
notes = excluded.notes,
updated_at = CURRENT_TIMESTAMP
"""
),
payload,
)
def clean_row_text(value: Any) -> str:
return normalize_text(value)
def filter_amount_rows(rows: list[dict[str, Any]], amount_key: str = "amount") -> list[dict[str, Any]]:
cleaned_rows: list[dict[str, Any]] = []
for row in rows:
normalized_row = {key: clean_row_text(value) for key, value in row.items()}
amount = normalize_amount(normalized_row.get(amount_key))
if amount or any(value for key, value in normalized_row.items() if key != amount_key):
normalized_row[amount_key] = amount
cleaned_rows.append(normalized_row)
return cleaned_rows
def sum_row_amounts(rows: list[dict[str, Any]], amount_key: str = "amount") -> float:
return sum(normalize_amount(row.get(amount_key)) for row in rows)
def get_support_department_options() -> list[dict[str, str]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT support_dept_code, support_dept_name
FROM (
SELECT support_dept_code, support_dept_name
FROM transactions
UNION ALL
SELECT support_dept_code, support_dept_name
FROM project_contract_info
UNION ALL
SELECT support_dept_code, support_dept_name
FROM project_billing_entries
) AS merged
WHERE COALESCE(support_dept_code, '') <> ''
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
ORDER BY support_dept_code, support_dept_name
"""
)
).mappings().all()
return [
{
"support_dept_code": normalize_text(row["support_dept_code"]),
"support_dept_name": normalize_text(row["support_dept_name"]),
}
for row in rows
]
def get_cost_department_options() -> list[dict[str, str]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT cost_dept_name
FROM transactions
WHERE COALESCE(cost_dept_name, '') <> ''
ORDER BY cost_dept_name
"""
)
).mappings().all()
return [
{"cost_dept_name": normalize_text(row["cost_dept_name"])}
for row in rows
if normalize_text(row["cost_dept_name"])
]
def get_cost_account_options() -> list[dict[str, str]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT account_code, account_name
FROM transactions
WHERE accounting_category = '원가'
AND COALESCE(account_code, '') <> ''
AND COALESCE(account_name, '') <> ''
ORDER BY account_code, account_name
"""
)
).mappings().all()
deduped: dict[tuple[str, str], dict[str, str]] = {}
for row in rows:
account_code = normalize_text(row["account_code"])
account_name = normalize_text(row["account_name"])
if not account_code or not account_name:
continue
normalized_code, normalized_name, _ = normalize_account_display(account_code, account_name)
key = (normalized_code, normalized_name)
deduped[key] = {
"account_code": normalized_code,
"account_name": normalized_name,
}
return sorted(deduped.values(), key=lambda item: (item["account_code"], item["account_name"]))
def get_import_sync_summary() -> dict[str, Any]:
with engine.begin() as conn:
row = conn.execute(
text(
"""
SELECT
(SELECT COUNT(*) FROM project_contract_info) AS contract_project_count,
(SELECT COUNT(*) FROM project_billing_entries) AS billing_entry_count,
(SELECT COUNT(DISTINCT support_dept_code) FROM project_billing_entries) AS billing_project_count,
(SELECT COUNT(*) FROM project_contract_info WHERE COALESCE(review_tag, '') <> '') AS review_needed_count,
(SELECT SUM(hanmac_contract_amount) FROM project_contract_info) AS total_hanmac_contract_amount,
(SELECT SUM(collected_amount) FROM project_billing_entries) AS total_collected_amount,
(SELECT MAX(updated_at) FROM project_contract_info) AS latest_contract_sync,
(SELECT MAX(updated_at) FROM project_billing_entries) AS latest_billing_sync
"""
)
).mappings().first()
return dict(row) if row else {}
def get_project_contract_info_map() -> dict[str, dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT * FROM project_contract_info ORDER BY support_dept_code")
).mappings().all()
return {normalize_text(row["support_dept_code"]): dict(row) for row in rows}
def get_project_billing_summary_map() -> dict[str, dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT support_dept_code,
MAX(support_dept_name) AS support_dept_name,
MAX(contract_amount) AS contract_amount,
MAX(client_name) AS client_name,
MAX(support_department) AS support_department,
MAX(business_division) AS business_division,
SUM(billed_amount) AS billed_amount,
SUM(collected_amount) AS collected_amount,
SUM(balance_amount) AS balance_amount,
MAX(billing_date) AS latest_billing_date
FROM project_billing_entries
GROUP BY support_dept_code
ORDER BY support_dept_code
"""
)
).mappings().all()
entry_rows = conn.execute(
text(
"""
SELECT support_dept_code,
billing_type,
progress_round,
billing_date,
tax_invoice_date,
expected_collection_date,
billed_amount,
collected_amount,
balance_amount,
collection_rate,
note
FROM project_billing_entries
ORDER BY support_dept_code, billing_date, progress_round, id
"""
)
).mappings().all()
result = {normalize_text(row["support_dept_code"]): dict(row) for row in rows}
for item in result.values():
item["entries"] = []
for row in entry_rows:
support_dept_code = normalize_text(row["support_dept_code"])
if support_dept_code not in result:
continue
result[support_dept_code]["entries"].append(
normalize_collection_entry_row(
{
"progress_type": "",
"billing_round": normalize_round_value(row["progress_round"]),
"billing_type": normalize_text(row["billing_type"]),
"billing_date": normalize_date_text(row["billing_date"]),
"billed_amount": normalize_amount(row["billed_amount"]),
"round": normalize_round_value(row["progress_round"]),
"date": normalize_date_text(row["tax_invoice_date"]) or normalize_date_text(row["expected_collection_date"]),
"amount": normalize_amount(row["collected_amount"]),
"balance_amount": normalize_amount(row["balance_amount"]),
"collection_rate": normalize_amount(row["collection_rate"]),
"note": normalize_text(row["note"]),
}
)
)
return result
def merge_project_external_fields(
item: dict[str, Any],
contract_info: dict[str, Any] | None,
billing_summary: dict[str, Any] | None,
latest_summary_change: dict[str, Any] | None = None,
latest_round_change: dict[str, Any] | None = None,
change_representative_code: str = "",
change_title_key: str = "",
) -> dict[str, Any]:
contract_info = contract_info or {}
billing_summary = billing_summary or {}
support_dept_name = normalize_text(item.get("support_dept_name")) or normalize_text(contract_info.get("support_dept_name")) or normalize_text(billing_summary.get("support_dept_name"))
contract_amount = normalize_amount(item.get("contract_amount"))
latest_summary_change = latest_summary_change or {}
latest_round_change = latest_round_change or {}
latest_changed_contract_amount = (
normalize_amount(latest_summary_change.get("changed_contract_amount"))
or normalize_amount(latest_round_change.get("changed_contract_amount"))
)
current_code = normalize_text(item.get("support_dept_code"))
if change_representative_code and current_code == change_representative_code and latest_changed_contract_amount:
contract_amount = latest_changed_contract_amount
elif not contract_amount:
contract_amount = (
normalize_amount(contract_info.get("hanmac_contract_amount"))
or normalize_amount(billing_summary.get("contract_amount"))
)
if not contract_amount and latest_changed_contract_amount and (not change_representative_code or current_code == change_representative_code):
contract_amount = latest_changed_contract_amount
collection_amount = normalize_amount(item.get("collection_amount"))
if not collection_amount:
collection_amount = normalize_amount(billing_summary.get("collected_amount"))
collection_entries = item.get("collection_entries")
if not collection_entries:
collection_entries = billing_summary.get("entries", [])
project_start_date = normalize_text(item.get("project_start_date")) or normalize_text(contract_info.get("project_start_date"))
project_end_date = (
normalize_text(item.get("project_end_date"))
or normalize_text(contract_info.get("project_end_date"))
or normalize_text(latest_summary_change.get("changed_project_end_date"))
or normalize_text(latest_round_change.get("changed_project_end_date"))
)
completion_status = normalize_text(item.get("completion_status")) or normalize_text(contract_info.get("progress_status"))
project_type = (
normalize_text(item.get("project_type"))
or normalize_text(contract_info.get("business_division"))
or normalize_text(billing_summary.get("business_division"))
or normalize_text(latest_summary_change.get("business_division"))
or normalize_text(latest_round_change.get("business_division"))
)
progress_rate = normalize_amount(item.get("progress_rate"))
if not progress_rate and contract_amount:
progress_rate = (collection_amount / contract_amount * 100) if contract_amount else 0.0
item["support_dept_name"] = support_dept_name
item["contract_amount"] = contract_amount
item["collection_amount"] = collection_amount
item["collection_entries"] = collection_entries or []
item["project_start_date"] = project_start_date
item["project_end_date"] = project_end_date
item["completion_status"] = completion_status
item["project_type"] = project_type
item["progress_rate"] = progress_rate
item["client_name"] = (
normalize_text(contract_info.get("client_name"))
or normalize_text(billing_summary.get("client_name"))
or normalize_text(latest_summary_change.get("client_name"))
or normalize_text(latest_round_change.get("client_name"))
)
item["order_method"] = normalize_text(contract_info.get("order_method"))
item["joint_contract"] = normalize_text(contract_info.get("joint_contract"))
item["pm_name"] = normalize_text(contract_info.get("pm_name"))
item["contract_status"] = normalize_text(contract_info.get("contract_status"))
item["progress_status"] = normalize_text(contract_info.get("progress_status"))
item["work_category"] = normalize_text(contract_info.get("work_category"))
item["review_tag"] = normalize_text(contract_info.get("review_tag"))
item["review_note"] = normalize_text(contract_info.get("review_note"))
item["total_contract_amount"] = normalize_amount(contract_info.get("total_contract_amount"))
item["hanmac_contract_amount"] = normalize_amount(contract_info.get("hanmac_contract_amount"))
item["billing_contract_amount"] = normalize_amount(billing_summary.get("contract_amount"))
item["billed_amount"] = normalize_amount(billing_summary.get("billed_amount"))
item["collection_balance_amount"] = normalize_amount(billing_summary.get("balance_amount"))
item["latest_billing_date"] = normalize_text(billing_summary.get("latest_billing_date"))
item["changed_contract_amount"] = latest_changed_contract_amount
item["changed_contract_date"] = normalize_text(latest_summary_change.get("change_date")) or normalize_text(latest_round_change.get("change_date"))
item["changed_project_end_date"] = normalize_text(latest_summary_change.get("changed_project_end_date")) or normalize_text(latest_round_change.get("changed_project_end_date"))
item["change_contract_representative_code"] = change_representative_code
item["change_contract_title_key"] = change_title_key
return item
def normalize_account_display(account_code: Any, account_name: Any) -> tuple[str, str, str]:
normalized_code = normalize_text(account_code)[:6]
normalized_name = re.sub(r"\s*\(.*$", "", normalize_text(account_name)).strip()
if normalized_code and normalized_name:
label = f"{normalized_code} · {normalized_name}"
else:
label = normalized_name or normalized_code or "미분류"
return normalized_code, normalized_name, label
def get_data_version() -> str:
with engine.begin() as conn:
transaction_updated = conn.execute(text("SELECT MAX(updated_at) FROM transactions")).scalar()
project_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_status")).scalar()
contract_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_contract_info")).scalar()
billing_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_billing_entries")).scalar()
change_summary_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_contract_change_summary")).scalar()
change_round_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_contract_change_round")).scalar()
versions = [
normalize_text(transaction_updated),
normalize_text(project_updated),
normalize_text(contract_updated),
normalize_text(billing_updated),
normalize_text(change_summary_updated),
normalize_text(change_round_updated),
]
return max((version for version in versions if version), default="")
def build_health_payload() -> dict[str, str]:
return {
"status": "ok",
"server_time": datetime.now().isoformat(timespec="seconds"),
"data_version": get_data_version(),
}
def check_record_revision(conn: Any, table_name: str, key_column: str, key_value: Any, edit_revision: str) -> None:
if not key_value or not edit_revision:
return
current_revision = conn.execute(
text(f"SELECT updated_at FROM {table_name} WHERE {key_column} = :key_value"),
{"key_value": key_value},
).scalar()
current_revision_text = normalize_text(current_revision)
if current_revision_text and current_revision_text != normalize_text(edit_revision):
raise ValueError("다른 사용자가 먼저 수정했습니다. 최신 화면으로 다시 확인한 뒤 저장해주세요.")
def date_diff_days(start_date: str, end_date: str) -> int | None:
if not start_date or not end_date:
return None
try:
start = datetime.strptime(start_date, "%Y-%m-%d").date()
end = datetime.strptime(end_date, "%Y-%m-%d").date()
return (end - start).days
except ValueError:
return None
def detect_category(account_code: str) -> str:
if account_code.startswith("5"):
return "원가"
if account_code.startswith("4"):
return "수입/매출액"
if account_code.startswith("6"):
return "판관비"
return "기타"
def extract_period(voucher_number: str, posting_date: str) -> tuple[int | None, int | None, int | None]:
voucher_match = VOUCHER_PATTERN.match(voucher_number)
if voucher_match:
year_text, month_text, day_text = voucher_match.groups()
return int(year_text), int(month_text), int(day_text)
parsed_date = normalize_date_text(posting_date)
if re.match(r"^\d{4}-\d{2}-\d{2}$", parsed_date):
parsed = datetime.strptime(parsed_date, "%Y-%m-%d")
return parsed.year, parsed.month, parsed.day
return None, None, None
def choose_amount(debit_supply: float, credit_supply: float) -> float:
if debit_supply:
return abs(debit_supply)
if credit_supply:
return abs(credit_supply)
return 0.0
def canonical_header_name(value: Any) -> str | None:
normalized = normalize_text(value).replace(" ", "")
if not normalized:
return None
if normalized in DIRECT_HEADER_MAP:
return DIRECT_HEADER_MAP[normalized]
if "확정전표" in normalized:
return "confirmed_voucher_number"
return None
def empty_record() -> dict[str, str]:
record = {field: "" for field in FORM_FIELDS}
record["id"] = ""
return record
def build_transaction_payload(raw: dict[str, Any], source_file: str = "") -> dict[str, Any]:
payload: dict[str, Any] = {}
for field in FORM_FIELDS:
if field in {"debit_supply", "debit_vat", "credit_supply", "credit_vat"}:
payload[field] = normalize_amount(raw.get(field))
elif field == "posting_date":
payload[field] = normalize_date_text(raw.get(field))
else:
payload[field] = normalize_text(raw.get(field))
payload["accounting_category"] = detect_category(payload["account_code"])
payload["amount"] = choose_amount(payload["debit_supply"], payload["credit_supply"])
year, month, day = extract_period(payload["voucher_number"], payload["posting_date"])
payload["year"] = year
payload["month"] = month
payload["day"] = day
payload["source_file"] = source_file
return payload
def save_transaction(payload: dict[str, Any], record_id: int | None = None) -> None:
init_db()
params = {
**payload,
"record_id": record_id,
"last_editor_session_id": normalize_text(payload.get("client_session_id")),
"last_client_submitted_at": normalize_text(payload.get("client_submitted_at")),
}
with engine.begin() as conn:
if record_id:
check_record_revision(conn, "transactions", "id", record_id, normalize_text(payload.get("edit_revision")))
conn.execute(
text(
"""
UPDATE transactions
SET approval_status = :approval_status,
voucher_number = :voucher_number,
account_code = :account_code,
account_name = :account_name,
debit_supply = :debit_supply,
debit_vat = :debit_vat,
credit_supply = :credit_supply,
credit_vat = :credit_vat,
issuing_dept_code = :issuing_dept_code,
issuing_dept_name = :issuing_dept_name,
confirmed_voucher_number = :confirmed_voucher_number,
support_dept_code = :support_dept_code,
support_dept_name = :support_dept_name,
cost_dept_code = :cost_dept_code,
cost_dept_name = :cost_dept_name,
memo1 = :memo1,
memo2 = :memo2,
partner_code = :partner_code,
partner_name = :partner_name,
tax_code = :tax_code,
posting_date = :posting_date,
voucher_type = :voucher_type,
management_item = :management_item,
accounting_category = :accounting_category,
amount = :amount,
year = :year,
month = :month,
day = :day,
source_file = COALESCE(NULLIF(:source_file, ''), source_file),
last_editor_session_id = :last_editor_session_id,
last_client_submitted_at = :last_client_submitted_at,
updated_at = CURRENT_TIMESTAMP
WHERE id = :record_id
"""
),
params,
)
return
conn.execute(
text(
"""
INSERT INTO transactions (
approval_status,
voucher_number,
account_code,
account_name,
debit_supply,
debit_vat,
credit_supply,
credit_vat,
issuing_dept_code,
issuing_dept_name,
confirmed_voucher_number,
support_dept_code,
support_dept_name,
cost_dept_code,
cost_dept_name,
memo1,
memo2,
partner_code,
partner_name,
tax_code,
posting_date,
voucher_type,
management_item,
accounting_category,
amount,
year,
month,
day,
source_file,
last_editor_session_id,
last_client_submitted_at
) VALUES (
:approval_status,
:voucher_number,
:account_code,
:account_name,
:debit_supply,
:debit_vat,
:credit_supply,
:credit_vat,
:issuing_dept_code,
:issuing_dept_name,
:confirmed_voucher_number,
:support_dept_code,
:support_dept_name,
:cost_dept_code,
:cost_dept_name,
:memo1,
:memo2,
:partner_code,
:partner_name,
:tax_code,
:posting_date,
:voucher_type,
:management_item,
:accounting_category,
:amount,
:year,
:month,
:day,
:source_file,
:last_editor_session_id,
:last_client_submitted_at
)
"""
),
params,
)
def get_record_for_edit(record_id: int | None) -> dict[str, Any]:
if not record_id:
return empty_record()
with engine.begin() as conn:
row = conn.execute(
text("SELECT * FROM transactions WHERE id = :record_id"),
{"record_id": record_id},
).mappings().first()
if not row:
return empty_record()
data = dict(row)
for key, value in list(data.items()):
if value is None:
data[key] = ""
return data
def get_support_businesses() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT support_dept_code, support_dept_name, COUNT(*) AS row_count
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN (
'공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
)
GROUP BY support_dept_code, support_dept_name
ORDER BY support_dept_code, support_dept_name
"""
)
).mappings().all()
result = [dict(row) for row in rows]
for item in result:
item["project_duration_days"] = date_diff_days(item.get("project_start_date", ""), item.get("project_end_date", ""))
item["planned_total"] = (
(item.get("task_plan_department_budget") or 0)
+ (item.get("task_plan_outsource_budget") or 0)
+ (item.get("task_plan_joint_operating_cost") or 0)
+ (item.get("exec_budget_labor_by_grade") or 0)
+ (item.get("exec_budget_outsource") or 0)
+ (item.get("exec_budget_cost_plan") or 0)
+ (item.get("expected_as_cost") or 0)
+ (item.get("expected_sga_budget") or 0)
)
item["actual_total_expense"] = (
(item.get("total_cost") or 0)
+ (item.get("total_sga") or 0)
)
return result
def get_monthly_summary() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT year,
month,
SUM(
CASE
WHEN accounting_category = '원가'
AND account_code NOT LIKE '5012%'
AND account_code NOT LIKE '5017%'
THEN amount
ELSE 0
END
) AS cost_sum,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_sum,
SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_sum,
SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_sum
FROM transactions
WHERE year IS NOT NULL
AND month IS NOT NULL
GROUP BY year, month
ORDER BY year, month
"""
)
).mappings().all()
return [dict(row) for row in rows]
def get_yearly_summary() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT year,
SUM(
CASE
WHEN accounting_category = '원가'
AND account_code NOT LIKE '5012%'
AND account_code NOT LIKE '5017%'
THEN amount
ELSE 0
END
) AS cost_sum,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_sum,
SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_sum,
SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_sum
FROM transactions
WHERE year IS NOT NULL
GROUP BY year
ORDER BY year
"""
)
).mappings().all()
return [dict(row) for row in rows]
def get_business_monthly_summary() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT year,
month,
support_dept_code,
support_dept_name,
SUM(CASE WHEN accounting_category = '원가' THEN amount ELSE 0 END) AS cost_sum,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_sum
FROM transactions
WHERE year IS NOT NULL
AND month IS NOT NULL
AND COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN (
'공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
)
GROUP BY year, month, support_dept_code, support_dept_name
ORDER BY year, month, support_dept_code, support_dept_name
"""
)
).mappings().all()
return [dict(row) for row in rows]
def get_project_status_rows() -> list[dict[str, Any]]:
contract_info_map = get_project_contract_info_map()
billing_summary_map = get_project_billing_summary_map()
latest_summary_by_title, latest_round_by_code, representative_by_title, title_by_code = get_project_contract_change_maps()
with engine.begin() as conn:
entry_maps = load_project_status_entry_maps(conn)
rows = conn.execute(
text(
f"""
SELECT b.support_dept_code,
b.support_dept_name,
b.row_count,
COALESCE(ps.progress_rate, 0) AS progress_rate,
COALESCE(ps.contract_amount, 0) AS contract_amount,
COALESCE(ps.collection_amount, 0) AS collection_amount,
COALESCE(ps.collection_entries_json, '[]') AS collection_entries_json,
COALESCE(ps.change_round, '') AS change_round,
COALESCE(ps.item_investment, 0) AS item_investment,
COALESCE(ps.task_plan_department_budget, 0) AS task_plan_department_budget,
COALESCE(ps.task_plan_outsource_budget, 0) AS task_plan_outsource_budget,
COALESCE(ps.task_plan_outsource_detail, '') AS task_plan_outsource_detail,
COALESCE(ps.task_plan_joint_operating_cost, 0) AS task_plan_joint_operating_cost,
COALESCE(ps.task_plan_entries_json, '[]') AS task_plan_entries_json,
COALESCE(ps.exec_budget_labor_by_grade, 0) AS exec_budget_labor_by_grade,
COALESCE(ps.exec_budget_outsource, 0) AS exec_budget_outsource,
COALESCE(ps.exec_budget_cost_plan, 0) AS exec_budget_cost_plan,
COALESCE(ps.exec_budget_entries_json, '[]') AS exec_budget_entries_json,
COALESCE(ps.actual_input_entries_json, '[]') AS actual_input_entries_json,
COALESCE(ps.expected_as_cost, 0) AS expected_as_cost,
COALESCE(ps.expected_sga_budget, 0) AS expected_sga_budget,
COALESCE(ps.project_start_date, '') AS project_start_date,
COALESCE(ps.project_end_date, '') AS project_end_date,
COALESCE(ps.completion_status, '') AS completion_status,
COALESCE(ps.notes, '') AS notes,
COALESCE(agg.total_cost, 0) AS total_cost,
COALESCE(agg.total_sga, 0) AS total_sga,
COALESCE(agg.total_revenue, 0) AS total_revenue,
COALESCE(agg.actual_labor, 0) AS actual_labor,
COALESCE(agg.actual_outsource, 0) AS actual_outsource,
COALESCE(agg.latest_year, 0) AS latest_year,
COALESCE(agg.latest_month, 0) AS latest_month
FROM (
SELECT support_dept_code, support_dept_name, COUNT(*) AS row_count
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN (
'공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
)
GROUP BY support_dept_code, support_dept_name
) AS b
LEFT JOIN project_status AS ps
ON ps.support_dept_code = b.support_dept_code
LEFT JOIN (
SELECT support_dept_code,
SUM(CASE WHEN accounting_category = '원가' THEN amount ELSE 0 END) AS total_cost,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS total_sga,
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS total_revenue,
SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS actual_labor,
SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS actual_outsource,
MAX(year) AS latest_year,
MAX(month) AS latest_month
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
GROUP BY support_dept_code
) AS agg
ON agg.support_dept_code = b.support_dept_code
ORDER BY b.support_dept_code, b.support_dept_name
"""
)
).mappings().all()
result = []
seen_codes: set[str] = set()
for row in rows:
item = dict(row)
support_dept_code = normalize_text(item.get("support_dept_code"))
entry_set = ensure_project_entry_set(entry_maps.get(support_dept_code)) if support_dept_code in entry_maps else extract_project_status_entry_sets(item)
item["collection_entries"] = entry_set["collection_entries"]
item["task_plan_entries"] = entry_set["task_plan_entries"]
item["exec_budget_entries"] = entry_set["exec_budget_entries"]
item["actual_input_entries"] = entry_set["actual_input_entries"]
item.pop("collection_entries_json", None)
item.pop("task_plan_entries_json", None)
item.pop("exec_budget_entries_json", None)
item.pop("actual_input_entries_json", None)
item = merge_project_external_fields(
item,
contract_info_map.get(normalize_text(item.get("support_dept_code"))),
billing_summary_map.get(normalize_text(item.get("support_dept_code"))),
latest_summary_by_title.get(title_by_code.get(normalize_text(item.get("support_dept_code"))) or normalize_project_title_for_linking(item.get("support_dept_name"))),
latest_round_by_code.get(normalize_text(item.get("support_dept_code"))),
representative_by_title.get(title_by_code.get(normalize_text(item.get("support_dept_code"))) or normalize_project_title_for_linking(item.get("support_dept_name")), ""),
title_by_code.get(normalize_text(item.get("support_dept_code"))) or normalize_project_title_for_linking(item.get("support_dept_name")),
)
seen_codes.add(normalize_text(item.get("support_dept_code")))
result.append(item)
for support_dept_code in sorted((set(contract_info_map) | set(billing_summary_map)) - seen_codes):
fallback_title_key = (
title_by_code.get(support_dept_code)
or normalize_project_title_for_linking(
contract_info_map.get(support_dept_code, {}).get("support_dept_name")
or billing_summary_map.get(support_dept_code, {}).get("support_dept_name")
)
)
result.append(
merge_project_external_fields(
{
"support_dept_code": support_dept_code,
"support_dept_name": "",
"row_count": 0,
"progress_rate": 0,
"contract_amount": 0,
"collection_amount": 0,
"collection_entries": [],
"change_round": "",
"item_investment": 0,
"task_plan_department_budget": 0,
"task_plan_outsource_budget": 0,
"task_plan_outsource_detail": "",
"task_plan_joint_operating_cost": 0,
"task_plan_entries": [],
"exec_budget_labor_by_grade": 0,
"exec_budget_outsource": 0,
"exec_budget_cost_plan": 0,
"exec_budget_entries": [],
"actual_input_entries": [],
"expected_as_cost": 0,
"expected_sga_budget": 0,
"project_start_date": "",
"project_end_date": "",
"completion_status": "",
"notes": "",
"total_cost": 0,
"total_sga": 0,
"total_revenue": 0,
"actual_labor": 0,
"actual_outsource": 0,
"latest_year": 0,
"latest_month": 0,
"project_type": "",
},
contract_info_map.get(support_dept_code),
billing_summary_map.get(support_dept_code),
latest_summary_by_title.get(fallback_title_key),
latest_round_by_code.get(support_dept_code),
representative_by_title.get(fallback_title_key, ""),
fallback_title_key,
)
)
return result
def get_project_status_row_for_code(support_dept_code: str | None) -> dict[str, Any] | None:
normalized_code = normalize_text(support_dept_code)
if not normalized_code:
return None
for item in get_project_status_rows():
if normalize_text(item.get("support_dept_code")) == normalized_code:
return item
return None
def get_project_comparison_notes_map() -> dict[str, dict[str, str]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT support_dept_code, item_key, COALESCE(note, '') AS note
FROM project_comparison_notes
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).mappings().all()
result: dict[str, dict[str, str]] = {}
for row in rows:
code = normalize_text(row.get("support_dept_code"))
item_key = normalize_text(row.get("item_key"))
if not code or not item_key:
continue
result.setdefault(code, {})[item_key] = normalize_text(row.get("note"))
return result
def save_project_comparison_note(support_dept_code: str | None, item_key: str | None, note: str | None) -> None:
code = normalize_text(support_dept_code)
normalized_item_key = normalize_text(item_key)
if not code or not normalized_item_key:
return
normalized_note = normalize_text(note)
with engine.begin() as conn:
if normalized_note:
conn.execute(
text(
"""
INSERT INTO project_comparison_notes (
support_dept_code, item_key, note, updated_at
) VALUES (
:support_dept_code, :item_key, :note, CURRENT_TIMESTAMP
)
ON CONFLICT(support_dept_code, item_key) DO UPDATE SET
note = excluded.note,
updated_at = CURRENT_TIMESTAMP
"""
),
{
"support_dept_code": code,
"item_key": normalized_item_key,
"note": normalized_note,
},
)
else:
conn.execute(
text(
"""
DELETE FROM project_comparison_notes
WHERE support_dept_code = :support_dept_code
AND item_key = :item_key
"""
),
{
"support_dept_code": code,
"item_key": normalized_item_key,
},
)
def get_project_status_for_edit(support_dept_code: str | None) -> dict[str, Any]:
contract_info_map = get_project_contract_info_map()
billing_summary_map = get_project_billing_summary_map()
latest_summary_by_title, latest_round_by_code, representative_by_title, title_by_code = get_project_contract_change_maps()
if not support_dept_code:
return {
"support_dept_code": "",
"support_dept_name": "",
"progress_rate": "",
"contract_amount": "",
"collection_amount": "",
"collection_entries": [],
"change_round": "",
"item_investment": "",
"task_plan_department_budget": "",
"task_plan_outsource_budget": "",
"task_plan_outsource_detail": "",
"task_plan_joint_operating_cost": "",
"task_plan_entries": [],
"exec_budget_labor_by_grade": "",
"exec_labor_rates": {},
"exec_budget_outsource": "",
"exec_budget_cost_plan": "",
"exec_budget_entries": [],
"actual_input_entries": [],
"project_type": "",
"expected_as_rate": "",
"expected_sga_rate": "",
"expected_as_cost": "",
"expected_sga_budget": "",
"project_start_date": "",
"project_end_date": "",
"completion_status": "",
"notes": "",
"client_name": "",
"order_method": "",
"joint_contract": "",
"pm_name": "",
"contract_status": "",
"progress_status": "",
"work_category": "",
"review_tag": "",
"review_note": "",
"total_contract_amount": 0,
"hanmac_contract_amount": 0,
"billing_contract_amount": 0,
"billed_amount": 0,
"collection_balance_amount": 0,
"latest_billing_date": "",
"updated_at": "",
}
with engine.begin() as conn:
entry_maps = load_project_status_entry_maps(conn)
row = conn.execute(
text(
"""
SELECT b.support_dept_code,
b.support_dept_name,
COALESCE(ps.progress_rate, '') AS progress_rate,
COALESCE(ps.contract_amount, '') AS contract_amount,
COALESCE(ps.collection_amount, '') AS collection_amount,
COALESCE(ps.collection_entries_json, '[]') AS collection_entries_json,
COALESCE(ps.change_round, '') AS change_round,
COALESCE(ps.item_investment, '') AS item_investment,
COALESCE(ps.task_plan_department_budget, '') AS task_plan_department_budget,
COALESCE(ps.task_plan_outsource_budget, '') AS task_plan_outsource_budget,
COALESCE(ps.task_plan_outsource_detail, '') AS task_plan_outsource_detail,
COALESCE(ps.task_plan_joint_operating_cost, '') AS task_plan_joint_operating_cost,
COALESCE(ps.task_plan_entries_json, '[]') AS task_plan_entries_json,
COALESCE(ps.exec_budget_labor_by_grade, '') AS exec_budget_labor_by_grade,
COALESCE(ps.exec_labor_rates_json, '{}') AS exec_labor_rates_json,
COALESCE(ps.exec_budget_outsource, '') AS exec_budget_outsource,
COALESCE(ps.exec_budget_cost_plan, '') AS exec_budget_cost_plan,
COALESCE(ps.exec_budget_entries_json, '[]') AS exec_budget_entries_json,
COALESCE(ps.actual_input_entries_json, '[]') AS actual_input_entries_json,
COALESCE(ps.project_type, '') AS project_type,
COALESCE(ps.expected_as_rate, '') AS expected_as_rate,
COALESCE(ps.expected_sga_rate, '') AS expected_sga_rate,
COALESCE(ps.expected_as_cost, '') AS expected_as_cost,
COALESCE(ps.expected_sga_budget, '') AS expected_sga_budget,
COALESCE(ps.last_editor_session_id, '') AS last_editor_session_id,
COALESCE(ps.last_client_submitted_at, '') AS last_client_submitted_at,
COALESCE(ps.project_start_date, '') AS project_start_date,
COALESCE(ps.project_end_date, '') AS project_end_date,
COALESCE(ps.completion_status, '') AS completion_status,
COALESCE(ps.notes, '') AS notes,
COALESCE(ps.updated_at, '') AS updated_at
FROM (
SELECT DISTINCT support_dept_code, support_dept_name
FROM transactions
WHERE support_dept_code = :support_dept_code
) AS b
LEFT JOIN project_status AS ps
ON ps.support_dept_code = b.support_dept_code
"""
),
{"support_dept_code": support_dept_code},
).mappings().first()
if not row:
return {
"support_dept_code": support_dept_code,
"support_dept_name": "",
"progress_rate": "",
"contract_amount": "",
"collection_amount": "",
"collection_entries": [],
"change_round": "",
"item_investment": "",
"task_plan_department_budget": "",
"task_plan_outsource_budget": "",
"task_plan_outsource_detail": "",
"task_plan_joint_operating_cost": "",
"task_plan_entries": [],
"exec_budget_labor_by_grade": "",
"exec_labor_rates": {},
"exec_budget_outsource": "",
"exec_budget_cost_plan": "",
"exec_budget_entries": [],
"actual_input_entries": [],
"project_type": "",
"expected_as_rate": "",
"expected_sga_rate": "",
"expected_as_cost": "",
"expected_sga_budget": "",
"project_start_date": "",
"project_end_date": "",
"completion_status": "",
"notes": "",
"client_name": "",
"order_method": "",
"joint_contract": "",
"pm_name": "",
"contract_status": "",
"progress_status": "",
"work_category": "",
"review_tag": "",
"review_note": "",
"total_contract_amount": 0,
"hanmac_contract_amount": 0,
"billing_contract_amount": 0,
"billed_amount": 0,
"collection_balance_amount": 0,
"latest_billing_date": "",
"updated_at": "",
}
result = dict(row)
support_dept_code = normalize_text(result.get("support_dept_code"))
entry_set = ensure_project_entry_set(entry_maps.get(support_dept_code)) if support_dept_code in entry_maps else extract_project_status_entry_sets(result)
result["collection_entries"] = entry_set["collection_entries"]
result["task_plan_entries"] = entry_set["task_plan_entries"]
result["exec_budget_entries"] = entry_set["exec_budget_entries"]
result["actual_input_entries"] = entry_set["actual_input_entries"]
result.pop("collection_entries_json", None)
result.pop("task_plan_entries_json", None)
result.pop("exec_budget_entries_json", None)
result.pop("actual_input_entries_json", None)
try:
result["exec_labor_rates"] = json.loads(normalize_text(result.pop("exec_labor_rates_json", "{}")) or "{}")
except json.JSONDecodeError:
result["exec_labor_rates"] = {}
return merge_project_external_fields(
result,
contract_info_map.get(normalize_text(result.get("support_dept_code"))),
billing_summary_map.get(normalize_text(result.get("support_dept_code"))),
latest_summary_by_title.get(title_by_code.get(normalize_text(result.get("support_dept_code"))) or normalize_project_title_for_linking(result.get("support_dept_name"))),
latest_round_by_code.get(normalize_text(result.get("support_dept_code"))),
representative_by_title.get(title_by_code.get(normalize_text(result.get("support_dept_code"))) or normalize_project_title_for_linking(result.get("support_dept_name")), ""),
title_by_code.get(normalize_text(result.get("support_dept_code"))) or normalize_project_title_for_linking(result.get("support_dept_name")),
)
def get_project_page_state(session_id: str | None = None) -> dict[str, Any]:
normalized_session_id = normalize_text(session_id)
with engine.begin() as conn:
row = conn.execute(
text(
"""
SELECT COALESCE(selected_code, '') AS selected_code,
COALESCE(selected_year, '') AS selected_year,
COALESCE(analysis_open, 0) AS analysis_open,
COALESCE(uncontracted_year_start, '') AS uncontracted_year_start,
COALESCE(uncontracted_year_end, '') AS uncontracted_year_end,
COALESCE(related_project_selections_json, '{}') AS related_project_selections_json
FROM project_page_state
WHERE page_key = 'projects'
AND session_id = :session_id
"""
),
{"session_id": normalized_session_id},
).mappings().first()
if not row and normalized_session_id:
with engine.begin() as conn:
row = conn.execute(
text(
"""
SELECT COALESCE(selected_code, '') AS selected_code,
COALESCE(selected_year, '') AS selected_year,
COALESCE(analysis_open, 0) AS analysis_open,
COALESCE(uncontracted_year_start, '') AS uncontracted_year_start,
COALESCE(uncontracted_year_end, '') AS uncontracted_year_end,
COALESCE(related_project_selections_json, '{}') AS related_project_selections_json
FROM project_page_state
WHERE page_key = 'projects'
AND session_id = ''
"""
)
).mappings().first()
if not row:
return {
"selected_code": "",
"selected_year": "",
"analysis_open": False,
"uncontracted_year_start": "",
"uncontracted_year_end": "",
"related_project_selections": {},
}
try:
related_project_selections_raw = json.loads(normalize_text(row["related_project_selections_json"]) or "{}")
except json.JSONDecodeError:
related_project_selections_raw = {}
related_project_selections = {}
if isinstance(related_project_selections_raw, dict):
related_project_selections = {
normalize_text(key): [
normalize_text(value)
for value in values
if normalize_text(value)
]
for key, values in related_project_selections_raw.items()
if normalize_text(key) and isinstance(values, list)
}
return {
"selected_code": normalize_text(row["selected_code"]),
"selected_year": normalize_text(row["selected_year"]),
"analysis_open": bool(row["analysis_open"]),
"uncontracted_year_start": normalize_text(row["uncontracted_year_start"]),
"uncontracted_year_end": normalize_text(row["uncontracted_year_end"]),
"related_project_selections": related_project_selections,
}
def save_project_page_state(payload: dict[str, Any]) -> None:
session_id = normalize_text(payload.get("session_id"))
selected_code = normalize_text(payload.get("selected_code"))
selected_year = normalize_text(payload.get("selected_year"))
analysis_open = 1 if payload.get("analysis_open") else 0
uncontracted_year_start = normalize_text(payload.get("uncontracted_year_start"))
uncontracted_year_end = normalize_text(payload.get("uncontracted_year_end"))
raw_related = payload.get("related_project_selections") or {}
related_project_selections = {}
if isinstance(raw_related, dict):
related_project_selections = {
normalize_text(key): [
normalize_text(value)
for value in values
if normalize_text(value)
]
for key, values in raw_related.items()
if normalize_text(key) and isinstance(values, list)
}
with engine.begin() as conn:
conn.execute(
text(
"""
INSERT INTO project_page_state (
page_key,
session_id,
selected_code,
selected_year,
analysis_open,
uncontracted_year_start,
uncontracted_year_end,
related_project_selections_json,
updated_at
) VALUES (
'projects',
:session_id,
:selected_code,
:selected_year,
:analysis_open,
:uncontracted_year_start,
:uncontracted_year_end,
:related_project_selections_json,
CURRENT_TIMESTAMP
)
ON CONFLICT(page_key, session_id) DO UPDATE SET
selected_code = excluded.selected_code,
selected_year = excluded.selected_year,
analysis_open = excluded.analysis_open,
uncontracted_year_start = excluded.uncontracted_year_start,
uncontracted_year_end = excluded.uncontracted_year_end,
related_project_selections_json = excluded.related_project_selections_json,
updated_at = CURRENT_TIMESTAMP
"""
),
{
"session_id": session_id,
"selected_code": selected_code,
"selected_year": selected_year,
"analysis_open": analysis_open,
"uncontracted_year_start": uncontracted_year_start,
"uncontracted_year_end": uncontracted_year_end,
"related_project_selections_json": json.dumps(related_project_selections, ensure_ascii=False),
},
)
for base_code, related_codes in related_project_selections.items():
save_project_related_links(base_code, related_codes)
def get_project_related_links_map() -> dict[str, list[str]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT base_support_dept_code, related_support_dept_code
FROM project_related_links
ORDER BY base_support_dept_code, related_support_dept_code
"""
)
).mappings().all()
related_map: dict[str, list[str]] = {}
for row in rows:
base_code = normalize_text(row["base_support_dept_code"])
related_code = normalize_text(row["related_support_dept_code"])
if not base_code or not related_code:
continue
related_map.setdefault(base_code, []).append(related_code)
return related_map
def get_project_quick_links(session_id: str | None = None) -> list[str]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT support_dept_code
FROM project_quick_links
WHERE page_key = 'projects'
ORDER BY sort_order, updated_at DESC, support_dept_code
"""
)
).mappings().all()
return [normalize_text(row["support_dept_code"]) for row in rows if normalize_text(row["support_dept_code"])]
def save_project_quick_links(session_id: str | None, codes: list[str]) -> None:
normalized_codes: list[str] = []
for code in codes:
normalized_code = normalize_text(code)
if normalized_code and normalized_code not in normalized_codes:
normalized_codes.append(normalized_code)
with engine.begin() as conn:
conn.execute(
text(
"""
DELETE FROM project_quick_links
WHERE page_key = 'projects'
"""
)
)
for sort_order, support_dept_code in enumerate(normalized_codes):
conn.execute(
text(
"""
INSERT INTO project_quick_links (
page_key, support_dept_code, sort_order, updated_at
) VALUES (
'projects', :support_dept_code, :sort_order, CURRENT_TIMESTAMP
)
"""
),
{
"support_dept_code": support_dept_code,
"sort_order": sort_order,
},
)
def get_project_uncontracted_classification_map() -> dict[str, str]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT support_dept_code, category
FROM project_uncontracted_classification
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).fetchall()
return {
normalize_text(row[0]): normalize_text(row[1])
for row in rows
if normalize_text(row[0])
}
def save_project_uncontracted_classification(support_dept_code: Any, category: Any) -> None:
normalized_code = normalize_text(support_dept_code)
normalized_category = normalize_text(category)
allowed_categories = {"general", "precontract", "corporate_rnd", "external_research"}
if not normalized_code:
raise ValueError("프로젝트 코드가 필요합니다.")
if normalized_category not in allowed_categories:
raise ValueError("허용되지 않는 미계약 분류입니다.")
with engine.begin() as conn:
conn.execute(
text(
"""
INSERT INTO project_uncontracted_classification (
support_dept_code,
category,
updated_at
) VALUES (
:support_dept_code,
:category,
CURRENT_TIMESTAMP
)
ON CONFLICT(support_dept_code) DO UPDATE SET
category = excluded.category,
updated_at = CURRENT_TIMESTAMP
"""
),
{
"support_dept_code": normalized_code,
"category": normalized_category,
},
)
def save_project_related_links(base_support_dept_code: str, related_codes: list[Any]) -> None:
base_code = normalize_text(base_support_dept_code)
if not base_code:
return
normalized_codes = sorted(
{
normalize_text(code)
for code in related_codes
if normalize_text(code) and normalize_text(code) != base_code
}
)
with engine.begin() as conn:
conn.execute(
text(
"""
DELETE FROM project_related_links
WHERE base_support_dept_code = :base_support_dept_code
AND COALESCE(link_source, 'manual') = 'manual'
"""
),
{"base_support_dept_code": base_code},
)
for related_code in normalized_codes:
conn.execute(
text(
"""
INSERT INTO project_related_links (
base_support_dept_code,
related_support_dept_code,
link_source,
updated_at
) VALUES (
:base_support_dept_code,
:related_support_dept_code,
'manual',
CURRENT_TIMESTAMP
)
ON CONFLICT(base_support_dept_code, related_support_dept_code) DO UPDATE SET
link_source = excluded.link_source,
updated_at = CURRENT_TIMESTAMP
"""
),
{
"base_support_dept_code": base_code,
"related_support_dept_code": related_code,
},
)
def get_project_year_options() -> list[int]:
return get_available_years()
def resolve_selected_year(selected_year: int | None) -> int | None:
return selected_year
def parse_optional_year(value: Any) -> int | None:
text = normalize_text(value)
if not text:
return None
return int(text) if text.isdigit() else None
def get_recent_10_start_year() -> int | None:
available_years = get_available_years()
if not available_years:
return None
return max(available_years) - 9
def get_project_dashboard_summary(selected_year: int | None) -> dict[str, Any]:
selected_year = resolve_selected_year(selected_year)
project_year_clause = ""
collection_year_clause = ""
params: dict[str, Any] = {}
if selected_year:
project_year_clause = "AND p.year = :selected_year"
collection_year_clause = "AND year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
project_year_clause = "AND p.year >= :recent_10_start_year"
collection_year_clause = "AND year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
project_row = conn.execute(
text(
f"""
SELECT COUNT(*) AS related_projects,
SUM(COALESCE(p.expense_amount, 0)) AS expense_amount,
COUNT(CASE WHEN COALESCE(ps.completion_status, '') IN ('종료', '완료', 'Y', 'YES') THEN 1 END) AS completed_projects
FROM (
SELECT year,
support_dept_code,
support_dept_name,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
GROUP BY year, support_dept_code, support_dept_name
) AS p
LEFT JOIN project_status AS ps
ON ps.support_dept_code = p.support_dept_code
WHERE 1=1
{project_year_clause}
"""
),
params,
).mappings().first()
collection_row = conn.execute(
text(
f"""
SELECT COUNT(*) AS collection_transaction_count,
COUNT(DISTINCT support_dept_code) AS collection_project_count,
SUM(COALESCE(credit_supply, 0)) AS collection_amount
FROM transactions
WHERE COALESCE(credit_supply, 0) <> 0
AND ({REVENUE_SQL})
AND COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
{collection_year_clause}
"""
),
params,
).mappings().first()
input_row = conn.execute(
text(
"""
SELECT COUNT(CASE WHEN COALESCE(collection_amount, 0) <> 0 THEN 1 END) AS collection_input_projects,
COUNT(CASE WHEN COALESCE(completion_status, '') <> '' THEN 1 END) AS completion_input_projects
FROM project_status
"""
)
).mappings().first()
return {
**(dict(project_row) if project_row else {}),
**(dict(collection_row) if collection_row else {}),
**(dict(input_row) if input_row else {}),
}
def get_uncontracted_project_dashboard(selected_year: int | None) -> dict[str, Any]:
selected_year = resolve_selected_year(selected_year)
transaction_year_clause = ""
params: dict[str, Any] = {}
if selected_year:
transaction_year_clause = "AND t.year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
transaction_year_clause = "AND t.year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
summary = conn.execute(
text(
f"""
WITH project_universe AS (
SELECT DISTINCT support_dept_code, support_dept_name
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
UNION
SELECT support_dept_code, support_dept_name
FROM project_contract_info
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT support_dept_code, support_dept_name
FROM project_billing_entries
WHERE COALESCE(support_dept_code, '') <> ''
),
contract_flags AS (
SELECT support_dept_code,
COALESCE(hanmac_contract_amount, 0) AS hanmac_contract_amount,
COALESCE(review_tag, '') AS review_tag
FROM project_contract_info
),
project_amounts AS (
SELECT t.support_dept_code,
SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount
FROM transactions AS t
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
{transaction_year_clause}
GROUP BY t.support_dept_code
)
SELECT
SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 THEN 1 ELSE 0 END) AS uncontracted_projects,
SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 AND COALESCE(project_amounts.expense_amount, 0) > 0 THEN 1 ELSE 0 END) AS cost_incurred_projects,
SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 THEN COALESCE(project_amounts.expense_amount, 0) ELSE 0 END) AS expense_amount,
SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 THEN COALESCE(project_amounts.revenue_amount, 0) ELSE 0 END) AS revenue_amount,
SUM(CASE WHEN COALESCE(contract_flags.review_tag, '') <> '' THEN 1 ELSE 0 END) AS review_needed_projects
FROM project_universe
LEFT JOIN contract_flags
ON contract_flags.support_dept_code = project_universe.support_dept_code
LEFT JOIN project_amounts
ON project_amounts.support_dept_code = project_universe.support_dept_code
"""
),
params,
).mappings().first()
yearly_rows = conn.execute(
text(
f"""
WITH yearly_costs AS (
SELECT t.year,
t.support_dept_code,
SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount
FROM transactions AS t
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
AND t.year IS NOT NULL
{transaction_year_clause}
GROUP BY t.year, t.support_dept_code
)
SELECT yearly_costs.year,
COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN yearly_costs.support_dept_code END) AS uncontracted_projects,
COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 AND COALESCE(yearly_costs.expense_amount, 0) > 0 THEN yearly_costs.support_dept_code END) AS cost_incurred_projects,
SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(yearly_costs.expense_amount, 0) ELSE 0 END) AS expense_amount,
SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(yearly_costs.revenue_amount, 0) ELSE 0 END) AS revenue_amount
FROM yearly_costs
LEFT JOIN project_contract_info AS c
ON c.support_dept_code = yearly_costs.support_dept_code
GROUP BY yearly_costs.year
ORDER BY yearly_costs.year
"""
),
params,
).mappings().all()
monthly_focus_year = selected_year
if monthly_focus_year is None:
monthly_focus_year = conn.execute(
text(
"""
WITH monthly_candidates AS (
SELECT MAX(t.year) AS latest_year
FROM transactions AS t
LEFT JOIN project_contract_info AS c
ON c.support_dept_code = t.support_dept_code
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(c.hanmac_contract_amount, 0) <= 0
AND (t.account_code LIKE '5%' OR t.account_code LIKE '6%')
)
SELECT latest_year FROM monthly_candidates
"""
)
).scalar()
monthly_rows: list[dict[str, Any]] = []
if monthly_focus_year:
monthly_rows = conn.execute(
text(
f"""
WITH monthly_costs AS (
SELECT t.month,
t.support_dept_code,
SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount
FROM transactions AS t
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
AND t.year = :monthly_focus_year
AND t.month IS NOT NULL
GROUP BY t.month, t.support_dept_code
)
SELECT monthly_costs.month,
COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN monthly_costs.support_dept_code END) AS uncontracted_projects,
COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 AND COALESCE(monthly_costs.expense_amount, 0) > 0 THEN monthly_costs.support_dept_code END) AS cost_incurred_projects,
SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(monthly_costs.expense_amount, 0) ELSE 0 END) AS expense_amount,
SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(monthly_costs.revenue_amount, 0) ELSE 0 END) AS revenue_amount
FROM monthly_costs
LEFT JOIN project_contract_info AS c
ON c.support_dept_code = monthly_costs.support_dept_code
GROUP BY monthly_costs.month
ORDER BY monthly_costs.month
"""
),
{"monthly_focus_year": monthly_focus_year},
).mappings().all()
top_rows = conn.execute(
text(
f"""
WITH project_costs AS (
SELECT t.support_dept_code,
MAX(t.support_dept_name) AS support_dept_name,
SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount,
MAX(t.year) AS latest_year,
MAX(t.month) AS latest_month
FROM transactions AS t
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
{transaction_year_clause}
GROUP BY t.support_dept_code
)
SELECT project_costs.support_dept_code,
project_costs.support_dept_name,
project_costs.expense_amount,
project_costs.revenue_amount,
project_costs.latest_year,
project_costs.latest_month,
COALESCE(c.review_tag, '') AS review_tag
FROM project_costs
LEFT JOIN project_contract_info AS c
ON c.support_dept_code = project_costs.support_dept_code
WHERE COALESCE(c.hanmac_contract_amount, 0) <= 0
AND COALESCE(project_costs.expense_amount, 0) > 0
ORDER BY project_costs.expense_amount DESC, project_costs.support_dept_code
LIMIT 12
"""
),
params,
).mappings().all()
return {
"summary": dict(summary) if summary else {},
"yearly_rows": [dict(row) for row in yearly_rows],
"monthly_rows": [dict(row) for row in monthly_rows],
"monthly_focus_year": int(monthly_focus_year) if monthly_focus_year else None,
"top_rows": [dict(row) for row in top_rows],
}
def get_project_revenue_mix(selected_year: int | None = None) -> list[dict[str, Any]]:
params: dict[str, Any] = {}
if selected_year:
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT
year,
month,
SUM(CASE WHEN account_code LIKE '40110101%' AND memo1 LIKE '%설계%' THEN amount ELSE 0 END) AS design_revenue,
SUM(CASE WHEN account_code LIKE '40110101%' AND (memo1 NOT LIKE '%설계%' OR COALESCE(memo1, '') = '') THEN amount ELSE 0 END) AS design_other_revenue,
SUM(CASE WHEN account_code LIKE '40110102%' THEN amount ELSE 0 END) AS supervision_revenue,
SUM(CASE WHEN account_code LIKE '40110103%' THEN amount ELSE 0 END) AS inspection_revenue
FROM transactions
WHERE year = :selected_year
AND month IS NOT NULL
GROUP BY year, month
ORDER BY year, month
"""
),
{"selected_year": selected_year},
).mappings().all()
result = [dict(row) for row in rows]
for item in result:
item["label"] = f"{int(item['month'])}월" if item.get("month") is not None else str(item.get("year", ""))
return result
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT
year,
SUM(CASE WHEN account_code LIKE '40110101%' AND memo1 LIKE '%설계%' THEN amount ELSE 0 END) AS design_revenue,
SUM(CASE WHEN account_code LIKE '40110101%' AND (memo1 NOT LIKE '%설계%' OR COALESCE(memo1, '') = '') THEN amount ELSE 0 END) AS design_other_revenue,
SUM(CASE WHEN account_code LIKE '40110102%' THEN amount ELSE 0 END) AS supervision_revenue,
SUM(CASE WHEN account_code LIKE '40110103%' THEN amount ELSE 0 END) AS inspection_revenue
FROM transactions
WHERE year IS NOT NULL
GROUP BY year
ORDER BY year
"""
)
).mappings().all()
result = [dict(row) for row in rows]
for item in result:
item["label"] = str(item.get("year", ""))
return result[-10:]
def get_project_revenue_mix_monthly() -> list[dict[str, Any]]:
recent_10_start_year = get_recent_10_start_year()
params: dict[str, Any] = {}
year_clause = ""
if recent_10_start_year is not None:
year_clause = "AND year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT
year,
month,
SUM(CASE WHEN account_code LIKE '40110101%' AND memo1 LIKE '%설계%' THEN amount ELSE 0 END) AS design_revenue,
SUM(CASE WHEN account_code LIKE '40110101%' AND (memo1 NOT LIKE '%설계%' OR COALESCE(memo1, '') = '') THEN amount ELSE 0 END) AS design_other_revenue,
SUM(CASE WHEN account_code LIKE '40110102%' THEN amount ELSE 0 END) AS supervision_revenue,
SUM(CASE WHEN account_code LIKE '40110103%' THEN amount ELSE 0 END) AS inspection_revenue
FROM transactions
WHERE month IS NOT NULL
{year_clause}
GROUP BY year, month
ORDER BY year, month
"""
),
params,
).mappings().all()
result = [dict(row) for row in rows]
for item in result:
item["label"] = f"{int(item['month'])}월" if item.get("month") is not None else str(item.get("year", ""))
return result
def get_project_cost_by_year(selected_year: int | None) -> list[dict[str, Any]]:
year_clause = ""
params: dict[str, Any] = {}
if selected_year:
year_clause = "AND year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
year_clause = "AND year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
contract_info_map = get_project_contract_info_map()
billing_summary_map = get_project_billing_summary_map()
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT year,
support_dept_code,
support_dept_name,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS revenue_amount
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
{year_clause}
GROUP BY year, support_dept_code, support_dept_name
ORDER BY year DESC, expense_amount DESC, support_dept_code
"""
),
params,
).mappings().all()
result = [dict(row) for row in rows]
existing_codes = {normalize_text(row["support_dept_code"]) for row in result}
candidate_codes = sorted((set(contract_info_map) | set(billing_summary_map)) - existing_codes)
recent_10_start_year = get_recent_10_start_year()
for support_dept_code in candidate_codes:
contract_info = contract_info_map.get(support_dept_code, {})
billing_summary = billing_summary_map.get(support_dept_code, {})
fallback_date = (
normalize_text(contract_info.get("project_start_date"))
or normalize_text(contract_info.get("contract_date"))
or normalize_text(billing_summary.get("latest_billing_date"))
)
fallback_year = 0
if re.match(r"^\d{4}-\d{2}-\d{2}$", fallback_date):
fallback_year = int(fallback_date[:4])
if selected_year and fallback_year and fallback_year != selected_year:
continue
if not selected_year and recent_10_start_year is not None and fallback_year and fallback_year < recent_10_start_year:
continue
result.append(
{
"year": fallback_year,
"support_dept_code": support_dept_code,
"support_dept_name": normalize_text(contract_info.get("support_dept_name")) or normalize_text(billing_summary.get("support_dept_name")),
"expense_amount": 0,
"revenue_amount": normalize_amount(billing_summary.get("collected_amount")),
}
)
result.sort(key=lambda item: (-(int(item.get("year") or 0)), -normalize_amount(item.get("expense_amount")), normalize_text(item.get("support_dept_code"))))
return result
def get_project_account_breakdowns(selected_year: int | None) -> dict[str, dict[str, list[dict[str, Any]]]]:
year_clause = ""
params: dict[str, Any] = {}
if selected_year:
year_clause = "AND year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
year_clause = "AND year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT support_dept_code,
CASE
WHEN {REVENUE_SQL} THEN 'revenue'
WHEN accounting_category = '원가' THEN 'cost'
WHEN accounting_category = '판관비' THEN 'sga'
ELSE 'other'
END AS breakdown_kind,
COALESCE(account_code, '') AS account_code,
COALESCE(account_name, '') AS account_name,
SUM(COALESCE(amount, 0)) AS total_amount
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
AND ({REVENUE_SQL} OR accounting_category IN ('원가', '판관비'))
{year_clause}
GROUP BY support_dept_code, breakdown_kind, account_code, account_name
ORDER BY support_dept_code, breakdown_kind, total_amount DESC, account_code, account_name
"""
),
params,
).mappings().all()
result: dict[str, dict[str, dict[str, float]]] = {}
for row in rows:
code = row["support_dept_code"]
kind = row["breakdown_kind"]
if kind == "other":
continue
_, _, label = normalize_account_display(row["account_code"], row["account_name"])
result.setdefault(code, {"revenue": {}, "cost": {}, "sga": {}})
result[code][kind][label] = result[code][kind].get(label, 0.0) + float(row["total_amount"] or 0)
normalized_result: dict[str, dict[str, list[dict[str, Any]]]] = {}
for code, buckets in result.items():
normalized_result[code] = {}
for kind, entries in buckets.items():
normalized_result[code][kind] = [
{"label": label, "amount": amount}
for label, amount in sorted(entries.items(), key=lambda item: item[1], reverse=True)
]
return normalized_result
def get_recent_transactions(limit: int = 50) -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT id,
year,
month,
voucher_number,
account_code,
account_name,
support_dept_code,
support_dept_name,
cost_dept_code,
cost_dept_name,
accounting_category,
amount,
memo1,
management_item,
source_file,
updated_at
FROM transactions
ORDER BY COALESCE(year, 0) DESC,
COALESCE(month, 0) DESC,
id DESC
LIMIT :limit_count
"""
),
{"limit_count": limit},
).mappings().all()
return [dict(row) for row in rows]
def get_overview_stats(selected_year: int | None = None) -> dict[str, Any]:
year_clause = ""
params: dict[str, Any] = {}
if selected_year:
year_clause = "WHERE year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
year_clause = "WHERE year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
row = conn.execute(
text(
f"""
SELECT COUNT(*) AS total_rows,
COUNT(DISTINCT source_file) AS source_files,
COUNT(DISTINCT CASE
WHEN COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN (
'공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
)
THEN support_dept_code || '|' || support_dept_name
END) AS business_count,
SUM(CASE WHEN accounting_category = '원가' THEN amount ELSE 0 END) AS total_cost,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS total_sga,
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS total_revenue
FROM transactions
{year_clause}
"""
),
params,
).mappings().first()
return dict(row) if row else {}
def get_available_years() -> list[int]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT year
FROM transactions
WHERE year IS NOT NULL
ORDER BY year
"""
)
).fetchall()
return [int(row[0]) for row in rows if row[0] is not None]
def get_financial_series(granularity: str) -> list[dict[str, Any]]:
group_fields = "year" if granularity == "yearly" else "year, month"
order_fields = "year" if granularity == "yearly" else "year, month"
month_where = "" if granularity == "yearly" else "AND month IS NOT NULL"
project_cost_sql = (
"accounting_category = '원가' "
"AND account_code NOT LIKE '5012%' "
"AND account_code NOT LIKE '5017%' "
"AND support_dept_code <> 'ZZZZZZ' "
f"AND {FIELD_COST_DEPT_SQL}"
)
support_cost_sql = (
"accounting_category = '원가' "
"AND account_code NOT LIKE '5012%' "
"AND account_code NOT LIKE '5017%' "
"AND support_dept_code = 'ZZZZZZ' "
f"AND {FIELD_COST_DEPT_SQL}"
)
support_sga_sql = f"accounting_category = '판관비' AND {SUPPORT_COST_DEPT_SQL}"
field_sga_sql = f"accounting_category = '판관비' AND {FIELD_COST_DEPT_SQL}"
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT {group_fields},
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS revenue_sum,
SUM(CASE WHEN {project_cost_sql} THEN amount ELSE 0 END) AS project_cost_sum,
SUM(CASE WHEN {support_cost_sql} THEN amount ELSE 0 END) AS support_cost_sum,
SUM(CASE WHEN {support_sga_sql} THEN amount ELSE 0 END) AS support_sga_sum,
SUM(CASE WHEN {field_sga_sql} THEN amount ELSE 0 END) AS field_sga_sum,
SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_sum,
SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_sum
FROM transactions
WHERE year IS NOT NULL
{month_where}
GROUP BY {group_fields}
ORDER BY {order_fields}
"""
)
).mappings().all()
result: list[dict[str, Any]] = []
for row in rows:
item = dict(row)
item["total_expense"] = (
(item.get("project_cost_sum") or 0)
+ (item.get("support_cost_sum") or 0)
+ (item.get("support_sga_sum") or 0)
+ (item.get("field_sga_sum") or 0)
+ (item.get("labor_sum") or 0)
+ (item.get("outsourcing_sum") or 0)
)
item["operating_balance"] = (item.get("revenue_sum") or 0) - item["total_expense"]
item["label"] = str(item["year"]) if granularity == "yearly" else f"{item['year']}-{int(item['month']):02d}"
result.append(item)
return result
def get_source_files_summary() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT source_file, COUNT(*) AS row_count
FROM transactions
WHERE COALESCE(source_file, '') <> ''
GROUP BY source_file
ORDER BY row_count DESC, source_file
"""
)
).mappings().all()
return [dict(row) for row in rows]
def parse_excel_upload(upload_file: UploadFile) -> int:
init_db()
workbook = load_workbook(upload_file.file, data_only=True)
import_kind = detect_excel_import_kind(workbook, upload_file.filename or "")
if import_kind == "contract_status":
return import_contract_status_workbook(workbook, upload_file.filename or "")
if import_kind == "change_contract_summary":
return import_change_contract_summary_workbook(workbook, upload_file.filename or "")
if import_kind == "change_contract_round":
return import_change_contract_round_workbook(workbook, upload_file.filename or "")
if import_kind == "billing_status":
return import_billing_status_workbook(workbook, upload_file.filename or "")
sheet = workbook.active
headers = [canonical_header_name(cell.value) for cell in next(sheet.iter_rows(min_row=1, max_row=1))]
inserted = 0
for row in sheet.iter_rows(min_row=2, values_only=True):
raw: dict[str, Any] = {}
has_value = False
for index, value in enumerate(row):
field_name = headers[index] if index < len(headers) else None
if field_name:
raw[field_name] = value
if normalize_text(value):
has_value = True
if not has_value:
continue
payload = build_transaction_payload(raw, source_file=upload_file.filename or "")
if not payload["voucher_number"] and not payload["account_code"] and not payload["account_name"]:
continue
save_transaction(payload)
inserted += 1
return inserted
def import_excel_path(path: Path) -> int:
with path.open("rb") as excel_file:
upload = UploadFile(filename=path.name, file=excel_file)
return parse_excel_upload(upload)
def auto_import_project_excels() -> None:
init_db()
excel_files = [
path
for path in sorted(BASE_DIR.glob("*.xlsx"))
if not path.name.startswith("~$")
]
if not excel_files:
return
known_files = existing_source_files()
known_contract_files = existing_contract_source_files()
known_billing_files = existing_billing_source_files()
known_change_summary_files = existing_change_contract_summary_source_files()
known_change_round_files = existing_change_contract_round_source_files()
if count_transactions() > 0 and all(file.name in known_files for file in excel_files):
if all(
file.name in known_contract_files
or file.name in known_billing_files
or file.name in known_change_summary_files
or file.name in known_change_round_files
for file in excel_files
):
return
for excel_path in excel_files:
if not zipfile.is_zipfile(excel_path):
logger.warning("Skipping non-Excel or temporary workbook during auto-import: %s", excel_path.name)
continue
try:
workbook = load_workbook(excel_path, data_only=True)
except zipfile.BadZipFile:
logger.warning("Skipping invalid workbook during auto-import: %s", excel_path.name)
continue
except Exception:
logger.exception("Failed to inspect workbook during auto-import: %s", excel_path.name)
continue
import_kind = detect_excel_import_kind(workbook, excel_path.name)
if import_kind == "contract_status" and excel_path.name in known_contract_files:
continue
if import_kind == "change_contract_summary" and excel_path.name in known_change_summary_files:
continue
if import_kind == "change_contract_round" and excel_path.name in known_change_round_files:
continue
if import_kind == "billing_status" and excel_path.name in known_billing_files:
continue
if import_kind == "transactions" and excel_path.name in known_files:
continue
try:
with excel_path.open("rb") as excel_file:
upload = UploadFile(filename=excel_path.name, file=excel_file)
inserted = parse_excel_upload(upload)
logger.info("Auto-imported %s rows from %s", inserted, excel_path.name)
except Exception:
logger.exception("Failed to auto-import workbook: %s", excel_path.name)
def normalize_all_collection_entry_storage() -> None:
with engine.begin() as conn:
entry_rows = conn.execute(
text(
"""
SELECT id, support_dept_code, vendor, progress_type, billing_round, billing_type,
billing_date, billed_amount, round, date, due_date, amount,
balance_amount, collection_rate, note
FROM project_collection_entries
ORDER BY support_dept_code, position, id
"""
)
).mappings().all()
grouped_entries: dict[str, list[dict[str, Any]]] = {}
for row in entry_rows:
row_dict = dict(row)
entry_id = row_dict.pop("id", None)
support_dept_code = normalize_text(row_dict.pop("support_dept_code", ""))
normalized = normalize_collection_entry_row(row_dict)
grouped_entries.setdefault(support_dept_code, []).append(normalized)
if (
normalize_text(row.get("progress_type")) != normalized["progress_type"]
or normalize_text(row.get("billing_type")) != normalized["billing_type"]
):
conn.execute(
text(
"""
UPDATE project_collection_entries
SET progress_type = :progress_type,
billing_type = :billing_type,
updated_at = CURRENT_TIMESTAMP
WHERE id = :id
"""
),
{
"id": entry_id,
"progress_type": normalized["progress_type"],
"billing_type": normalized["billing_type"],
},
)
cached_rows = conn.execute(
text(
"""
SELECT support_dept_code, collection_entries_json
FROM project_status
WHERE collection_entries_json IS NOT NULL AND collection_entries_json <> ''
"""
)
).mappings().all()
for row in cached_rows:
support_dept_code = normalize_text(row["support_dept_code"])
normalized_entries = grouped_entries.get(support_dept_code)
if normalized_entries is None:
raw_entries = decode_json_rows(row["collection_entries_json"])
normalized_entries = [normalize_collection_entry_row(item) for item in raw_entries]
conn.execute(
text(
"""
UPDATE project_status
SET collection_entries_json = :collection_entries_json,
updated_at = CURRENT_TIMESTAMP
WHERE support_dept_code = :support_dept_code
"""
),
{
"support_dept_code": support_dept_code,
"collection_entries_json": encode_json_rows(normalized_entries),
},
)
def parse_manual_form(raw_body: bytes) -> dict[str, Any]:
parsed = parse_qs(raw_body.decode("utf-8"))
payload = {key: values[0] if values else "" for key, values in parsed.items()}
return payload
def parse_project_form(raw_body: bytes) -> dict[str, Any]:
parsed = parse_qs(raw_body.decode("utf-8"))
payload: dict[str, Any] = {}
for key, values in parsed.items():
payload[key] = values if key.endswith("[]") else (values[0] if values else "")
return payload
def build_named_amount_rows(
labels: list[Any],
amounts: list[Any],
*,
label_key: str = "label",
amount_key: str = "amount",
) -> list[dict[str, Any]]:
rows = []
for index, label in enumerate(labels):
rows.append(
{
label_key: label,
amount_key: amounts[index] if index < len(amounts) else "",
}
)
return filter_amount_rows(rows, amount_key=amount_key)
def build_triplet_amount_rows(
first_values: list[Any],
second_values: list[Any],
amounts: list[Any],
*,
first_key: str,
second_key: str,
amount_key: str = "amount",
) -> list[dict[str, Any]]:
max_length = max(len(first_values), len(second_values), len(amounts))
rows = []
for index in range(max_length):
rows.append(
{
first_key: first_values[index] if index < len(first_values) else "",
second_key: second_values[index] if index < len(second_values) else "",
amount_key: amounts[index] if index < len(amounts) else "",
}
)
return filter_amount_rows(rows, amount_key=amount_key)
def build_collection_rows(payload: dict[str, Any]) -> list[dict[str, Any]]:
progress_types = payload.get("collection_progress_type[]", [])
billing_rounds = payload.get("collection_billing_round[]", [])
billing_types = payload.get("collection_billing_type[]", [])
billing_dates = payload.get("collection_billing_date[]", [])
billed_amounts = payload.get("collection_billed_amount[]", [])
collection_rounds = payload.get("collection_round[]", [])
collection_dates = payload.get("collection_date[]", [])
collection_amounts = payload.get("collection_amount_row[]", [])
rows = []
total_rows = max(
len(progress_types),
len(billing_rounds),
len(billing_types),
len(billing_dates),
len(billed_amounts),
len(collection_rounds),
len(collection_dates),
len(collection_amounts),
)
for index in range(total_rows):
rows.append(
{
"progress_type": progress_types[index] if index < len(progress_types) else "",
"billing_round": billing_rounds[index] if index < len(billing_rounds) else "",
"billing_type": billing_types[index] if index < len(billing_types) else "",
"billing_date": billing_dates[index] if index < len(billing_dates) else "",
"billed_amount": billed_amounts[index] if index < len(billed_amounts) else "",
"round": collection_rounds[index] if index < len(collection_rounds) else "",
"date": collection_dates[index] if index < len(collection_dates) else "",
"amount": collection_amounts[index] if index < len(collection_amounts) else "",
}
)
filtered_rows: list[dict[str, Any]] = []
for row in rows:
normalized_row = {key: clean_row_text(value) for key, value in row.items()}
amount = normalize_amount(normalized_row.get("amount"))
billed_amount = normalize_amount(normalized_row.get("billed_amount"))
has_other_value = any(
value for key, value in normalized_row.items()
if key not in {"amount", "billed_amount"}
)
if amount or billed_amount or has_other_value:
normalized_row["amount"] = amount
normalized_row["billed_amount"] = billed_amount
filtered_rows.append(normalized_row)
for row in filtered_rows:
normalized_fields = normalize_collection_entry_fields(row)
row["progress_type"] = normalized_fields["progress_type"]
row["billing_type"] = normalized_fields["billing_type"]
row["billing_round"] = normalize_round_value(row.get("billing_round"))
row["round"] = normalize_round_value(row.get("round"))
row["billing_date"] = normalize_date_text(row.get("billing_date"))
row["date"] = normalize_date_text(row.get("date"))
row["billed_amount"] = normalize_amount(row.get("billed_amount"))
return filtered_rows
def build_project_status_payload(payload: dict[str, Any]) -> dict[str, Any]:
contract_amount = normalize_amount(payload.get("contract_amount"))
collection_rows = build_collection_rows(payload)
collection_amount = sum_row_amounts(collection_rows)
progress_rate = (collection_amount / contract_amount * 100) if contract_amount else 0.0
task_plan_department_rows = build_triplet_amount_rows(
payload.get("task_plan_department_dept[]", []),
payload.get("task_plan_department_work[]", []),
payload.get("task_plan_department_amount[]", []),
first_key="dept_name",
second_key="work_name",
)
for row in task_plan_department_rows:
row["group"] = "department"
task_plan_outsource_rows = build_triplet_amount_rows(
payload.get("task_plan_outsource_dept[]", []),
payload.get("task_plan_outsource_work[]", []),
payload.get("task_plan_outsource_amount[]", []),
first_key="dept_name",
second_key="work_name",
)
for row in task_plan_outsource_rows:
row["group"] = "outsource"
task_plan_joint_rows = build_triplet_amount_rows(
payload.get("task_plan_joint_dept[]", []),
payload.get("task_plan_joint_work[]", []),
payload.get("task_plan_joint_amount[]", []),
first_key="dept_name",
second_key="work_name",
)
for row in task_plan_joint_rows:
row["group"] = "joint"
task_plan_rows = task_plan_department_rows + task_plan_outsource_rows + task_plan_joint_rows
exec_labor_rows = build_triplet_amount_rows(
payload.get("exec_labor_grade[]", []),
payload.get("exec_labor_hours[]", []),
payload.get("exec_labor_amount[]", []),
first_key="grade",
second_key="hours",
)
for row in exec_labor_rows:
row["group"] = "labor"
exec_outsource_rows = build_triplet_amount_rows(
payload.get("exec_outsource_dept[]", []),
payload.get("exec_outsource_work[]", []),
payload.get("exec_outsource_amount[]", []),
first_key="dept_name",
second_key="work_name",
)
for row in exec_outsource_rows:
row["group"] = "outsource"
exec_cost_plan_rows = build_triplet_amount_rows(
payload.get("exec_cost_plan_code[]", []),
payload.get("exec_cost_plan_name[]", []),
payload.get("exec_cost_plan_amount[]", []),
first_key="account_code",
second_key="account_name",
)
for row in exec_cost_plan_rows:
row["group"] = "cost_plan"
exec_budget_rows = exec_labor_rows + exec_outsource_rows + exec_cost_plan_rows
actual_labor_grades = payload.get("actual_labor_grade[]", [])
actual_labor_minutes = payload.get("actual_labor_minutes[]", [])
actual_labor_amounts = payload.get("actual_labor_amount[]", [])
actual_labor_rows: list[dict[str, Any]] = []
actual_labor_max_length = max(
len(actual_labor_grades),
len(actual_labor_minutes),
len(actual_labor_amounts),
)
for index in range(actual_labor_max_length):
row = {
"grade": actual_labor_grades[index] if index < len(actual_labor_grades) else "",
"minutes": actual_labor_minutes[index] if index < len(actual_labor_minutes) else "",
"amount": actual_labor_amounts[index] if index < len(actual_labor_amounts) else "",
}
normalized_row = {key: clean_row_text(value) for key, value in row.items()}
amount = normalize_amount(normalized_row.get("amount"))
has_other_value = any(
value for key, value in normalized_row.items()
if key != "amount"
)
if amount or has_other_value:
normalized_row["amount"] = amount
actual_labor_rows.append(normalized_row)
for row in actual_labor_rows:
row["group"] = "labor"
actual_labor_adjustment_total = normalize_amount(payload.get("actual_labor_adjustment_total"))
actual_labor_adjustment_rows = []
if actual_labor_adjustment_total:
actual_labor_adjustment_rows.append(
{
"group": "labor_adjustment",
"label": "인건비 조정",
"amount": actual_labor_adjustment_total,
}
)
actual_as_rows = build_named_amount_rows(
payload.get("actual_as_label[]", []),
payload.get("actual_as_amount[]", []),
label_key="label",
amount_key="amount",
)
for row in actual_as_rows:
row["group"] = "as"
actual_labor_joint_rows = build_named_amount_rows(
payload.get("actual_labor_joint_label[]", []),
payload.get("actual_labor_joint_amount[]", []),
label_key="label",
amount_key="amount",
)
for row in actual_labor_joint_rows:
row["group"] = "labor_joint"
actual_sga_rows = build_named_amount_rows(
payload.get("actual_sga_label[]", []),
payload.get("actual_sga_amount[]", []),
label_key="label",
amount_key="amount",
)
for row in actual_sga_rows:
row["group"] = "sga"
actual_input_rows = actual_labor_rows + actual_labor_adjustment_rows + actual_labor_joint_rows + actual_as_rows + actual_sga_rows
if not actual_input_rows:
legacy_refs = payload.get("actual_input_ref[]", [])
legacy_amounts = payload.get("actual_input_amount[]", [])
legacy_notes = payload.get("actual_input_note[]", [])
for index, ref in enumerate(legacy_refs):
actual_input_rows.append(
{
"reference": ref,
"amount": legacy_amounts[index] if index < len(legacy_amounts) else "",
"note": legacy_notes[index] if index < len(legacy_notes) else "",
}
)
actual_input_rows = filter_amount_rows(actual_input_rows, amount_key="amount")
expected_as_rate = normalize_amount(payload.get("expected_as_rate"))
expected_sga_rate = normalize_amount(payload.get("expected_sga_rate"))
expected_as_cost = normalize_amount(payload.get("expected_as_cost"))
expected_sga_budget = normalize_amount(payload.get("expected_sga_budget"))
if not expected_as_cost and contract_amount and expected_as_rate:
expected_as_cost = contract_amount * expected_as_rate / 100
if not expected_sga_budget and contract_amount and expected_sga_rate:
expected_sga_budget = contract_amount * expected_sga_rate / 100
exec_labor_rates = normalize_text(payload.get("exec_labor_rates_json")) or "{}"
return {
"support_dept_code": normalize_text(payload.get("support_dept_code")),
"support_dept_name": normalize_text(payload.get("support_dept_name")),
"progress_rate": progress_rate,
"contract_amount": contract_amount,
"collection_amount": collection_amount,
"collection_entries_json": encode_json_rows(collection_rows),
"change_round": normalize_text(payload.get("change_round")),
"item_investment": sum_row_amounts(actual_input_rows),
"task_plan_department_budget": sum_row_amounts(task_plan_department_rows),
"task_plan_outsource_budget": sum_row_amounts(task_plan_outsource_rows),
"task_plan_outsource_detail": "\n".join(
f"{normalize_text(row.get('dept_name'))} / {normalize_text(row.get('work_name'))}: {format_amount_for_text(row.get('amount'))}".strip(" /:")
for row in task_plan_outsource_rows
),
"task_plan_joint_operating_cost": sum_row_amounts(task_plan_joint_rows),
"task_plan_entries_json": encode_json_rows(task_plan_rows),
"exec_budget_labor_by_grade": sum_row_amounts(exec_labor_rows),
"exec_labor_rates_json": exec_labor_rates,
"exec_budget_outsource": sum_row_amounts(exec_outsource_rows),
"exec_budget_cost_plan": sum_row_amounts(exec_cost_plan_rows),
"exec_budget_entries_json": encode_json_rows(exec_budget_rows),
"actual_input_entries_json": encode_json_rows(actual_input_rows),
"project_type": normalize_text(payload.get("project_type")),
"expected_as_rate": expected_as_rate,
"expected_sga_rate": expected_sga_rate,
"expected_as_cost": expected_as_cost,
"expected_sga_budget": expected_sga_budget,
"last_editor_session_id": normalize_text(payload.get("client_session_id")),
"last_client_submitted_at": normalize_text(payload.get("client_submitted_at")),
"project_start_date": normalize_date_text(payload.get("project_start_date")),
"project_end_date": normalize_date_text(payload.get("project_end_date")),
"completion_status": normalize_text(payload.get("completion_status")),
"notes": normalize_text(payload.get("notes")),
"_collection_rows": collection_rows,
"_task_plan_rows": task_plan_rows,
"_exec_budget_rows": exec_budget_rows,
"_actual_input_rows": actual_input_rows,
}
def project_status_payload_has_meaningful_data(payload: dict[str, Any]) -> bool:
if normalize_amount(payload.get("contract_amount")):
return True
if normalize_amount(payload.get("collection_amount")):
return True
if normalize_amount(payload.get("task_plan_department_budget")):
return True
if normalize_amount(payload.get("task_plan_outsource_budget")):
return True
if normalize_amount(payload.get("task_plan_joint_operating_cost")):
return True
if normalize_amount(payload.get("exec_budget_labor_by_grade")):
return True
if normalize_amount(payload.get("exec_budget_outsource")):
return True
if normalize_amount(payload.get("exec_budget_cost_plan")):
return True
if normalize_amount(payload.get("item_investment")):
return True
for key in (
"project_type",
"project_start_date",
"project_end_date",
"completion_status",
"notes",
"change_round",
"support_dept_name",
"task_plan_outsource_detail",
):
if normalize_text(payload.get(key)):
return True
for key in (
"collection_entries_json",
"task_plan_entries_json",
"exec_budget_entries_json",
"actual_input_entries_json",
):
if decode_json_rows(payload.get(key)):
return True
return False
def format_amount_for_text(value: Any) -> str:
amount = normalize_amount(value)
return f"{amount:,.0f}"
def save_project_status(payload: dict[str, Any]) -> None:
normalized_payload = build_project_status_payload(payload)
support_dept_code = normalize_text(normalized_payload.get("support_dept_code"))
if not support_dept_code:
return
collection_rows = normalized_payload.pop("_collection_rows", [])
task_plan_rows = normalized_payload.pop("_task_plan_rows", [])
exec_budget_rows = normalized_payload.pop("_exec_budget_rows", [])
actual_input_rows = normalized_payload.pop("_actual_input_rows", [])
basic_info_field_keys = (
"support_dept_name",
"contract_amount",
"project_type",
"expected_as_rate",
"expected_sga_rate",
"expected_as_cost",
"expected_sga_budget",
"exec_labor_rates_json",
"change_round",
"project_start_date",
"project_end_date",
"completion_status",
"notes",
"last_editor_session_id",
"last_client_submitted_at",
)
collection_field_keys = (
"collection_progress_type[]",
"collection_billing_round[]",
"collection_billing_type[]",
"collection_billing_date[]",
"collection_billed_amount[]",
"collection_round[]",
"collection_date[]",
"collection_amount_row[]",
)
task_plan_field_keys = (
"task_plan_department_dept[]",
"task_plan_department_work[]",
"task_plan_department_amount[]",
"task_plan_outsource_dept[]",
"task_plan_outsource_work[]",
"task_plan_outsource_amount[]",
"task_plan_joint_dept[]",
"task_plan_joint_work[]",
"task_plan_joint_amount[]",
)
exec_budget_field_keys = (
"exec_labor_grade[]",
"exec_labor_hours[]",
"exec_labor_amount[]",
"exec_outsource_dept[]",
"exec_outsource_work[]",
"exec_outsource_amount[]",
"exec_cost_plan_code[]",
"exec_cost_plan_name[]",
"exec_cost_plan_amount[]",
)
actual_input_field_keys = (
"actual_labor_grade[]",
"actual_labor_minutes[]",
"actual_labor_amount[]",
"actual_labor_adjustment_total",
"actual_labor_joint_label[]",
"actual_labor_joint_amount[]",
"actual_as_label[]",
"actual_as_amount[]",
"actual_sga_label[]",
"actual_sga_amount[]",
)
def payload_has_any(keys: tuple[str, ...]) -> bool:
return any(key in payload for key in keys)
with engine.begin() as conn:
existing_row = conn.execute(
text("SELECT * FROM project_status WHERE support_dept_code = :support_dept_code"),
{"support_dept_code": support_dept_code},
).mappings().first()
if existing_row:
existing_payload = dict(existing_row)
if (
project_status_payload_has_meaningful_data(existing_payload)
and not project_status_payload_has_meaningful_data(normalized_payload)
):
raise ValueError("기존 입력값을 불러오지 않은 빈 상태로는 저장할 수 없습니다.")
existing_basic_info = conn.execute(
text("SELECT * FROM project_basic_info WHERE support_dept_code = :support_dept_code"),
{"support_dept_code": support_dept_code},
).mappings().first()
existing_entry_set = load_project_status_entries_for_code(conn, support_dept_code)
if existing_basic_info:
for key in basic_info_field_keys:
if key not in payload:
normalized_payload[key] = existing_basic_info.get(key)
if not payload_has_any(collection_field_keys):
collection_rows = existing_entry_set["collection_entries"]
if not payload_has_any(task_plan_field_keys):
task_plan_rows = existing_entry_set["task_plan_entries"]
else:
existing_task_rows = existing_entry_set["task_plan_entries"]
if not payload_has_any(("task_plan_department_dept[]", "task_plan_department_work[]", "task_plan_department_amount[]")):
task_plan_rows.extend(row for row in existing_task_rows if normalize_text(row.get("group")) == "department")
if not payload_has_any(("task_plan_outsource_dept[]", "task_plan_outsource_work[]", "task_plan_outsource_amount[]")):
task_plan_rows.extend(row for row in existing_task_rows if normalize_text(row.get("group")) == "outsource")
if not payload_has_any(("task_plan_joint_dept[]", "task_plan_joint_work[]", "task_plan_joint_amount[]")):
task_plan_rows.extend(row for row in existing_task_rows if normalize_text(row.get("group")) == "joint")
if not payload_has_any(exec_budget_field_keys):
exec_budget_rows = existing_entry_set["exec_budget_entries"]
else:
existing_exec_rows = existing_entry_set["exec_budget_entries"]
if not payload_has_any(("exec_labor_grade[]", "exec_labor_hours[]", "exec_labor_amount[]")):
exec_budget_rows.extend(row for row in existing_exec_rows if normalize_text(row.get("group")) == "labor")
if not payload_has_any(("exec_outsource_dept[]", "exec_outsource_work[]", "exec_outsource_amount[]")):
exec_budget_rows.extend(row for row in existing_exec_rows if normalize_text(row.get("group")) == "outsource")
if not payload_has_any(("exec_cost_plan_code[]", "exec_cost_plan_name[]", "exec_cost_plan_amount[]")):
exec_budget_rows.extend(row for row in existing_exec_rows if normalize_text(row.get("group")) == "cost_plan")
if not payload_has_any(actual_input_field_keys):
actual_input_rows = existing_entry_set["actual_input_entries"]
else:
existing_actual_rows = existing_entry_set["actual_input_entries"]
if not payload_has_any(("actual_labor_grade[]", "actual_labor_minutes[]", "actual_labor_amount[]")):
actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "labor")
if "actual_labor_adjustment_total" not in payload:
actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "labor_adjustment")
if not payload_has_any(("actual_labor_joint_label[]", "actual_labor_joint_amount[]")):
actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "labor_joint")
if not payload_has_any(("actual_as_label[]", "actual_as_amount[]")):
actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "as")
if not payload_has_any(("actual_sga_label[]", "actual_sga_amount[]")):
actual_input_rows.extend(row for row in existing_actual_rows if normalize_text(row.get("group")) == "sga")
normalized_payload["collection_amount"] = sum_row_amounts(collection_rows)
normalized_payload["progress_rate"] = (
normalized_payload["collection_amount"] / normalize_amount(normalized_payload.get("contract_amount")) * 100
if normalize_amount(normalized_payload.get("contract_amount"))
else 0.0
)
task_plan_department_rows = [row for row in task_plan_rows if normalize_text(row.get("group")) == "department"]
task_plan_outsource_rows = [row for row in task_plan_rows if normalize_text(row.get("group")) == "outsource"]
task_plan_joint_rows = [row for row in task_plan_rows if normalize_text(row.get("group")) == "joint"]
exec_labor_rows = [row for row in exec_budget_rows if normalize_text(row.get("group")) == "labor"]
exec_outsource_rows = [row for row in exec_budget_rows if normalize_text(row.get("group")) == "outsource"]
exec_cost_plan_rows = [row for row in exec_budget_rows if normalize_text(row.get("group")) == "cost_plan"]
normalized_payload["collection_entries_json"] = encode_json_rows(collection_rows)
normalized_payload["task_plan_department_budget"] = sum_row_amounts(task_plan_department_rows)
normalized_payload["task_plan_outsource_budget"] = sum_row_amounts(task_plan_outsource_rows)
normalized_payload["task_plan_outsource_detail"] = "\n".join(
f"{normalize_text(row.get('dept_name'))} / {normalize_text(row.get('work_name'))}: {format_amount_for_text(row.get('amount'))}".strip(" /:")
for row in task_plan_outsource_rows
)
normalized_payload["task_plan_joint_operating_cost"] = sum_row_amounts(task_plan_joint_rows)
normalized_payload["task_plan_entries_json"] = encode_json_rows(task_plan_rows)
normalized_payload["exec_budget_labor_by_grade"] = sum_row_amounts(exec_labor_rows)
normalized_payload["exec_budget_outsource"] = sum_row_amounts(exec_outsource_rows)
normalized_payload["exec_budget_cost_plan"] = sum_row_amounts(exec_cost_plan_rows)
normalized_payload["exec_budget_entries_json"] = encode_json_rows(exec_budget_rows)
normalized_payload["item_investment"] = sum_row_amounts(actual_input_rows)
normalized_payload["actual_input_entries_json"] = encode_json_rows(actual_input_rows)
check_record_revision(
conn,
"project_status",
"support_dept_code",
support_dept_code,
normalize_text(payload.get("edit_revision")),
)
save_project_basic_info_section(conn, normalized_payload)
replace_project_status_child_entries(
conn,
support_dept_code,
collection_rows,
task_plan_rows,
exec_budget_rows,
actual_input_rows,
)
sync_project_status_cache_row(conn, support_dept_code)
def base_context(request: Request, message: str = "") -> dict[str, Any]:
health_payload = build_health_payload()
return {
"request": request,
"message": message,
"data_version": health_payload["data_version"],
"server_time": health_payload["server_time"],
"import_sync_summary": get_import_sync_summary(),
}
def render_home(
request: Request,
edit_id: int | None = None,
message: str = "",
overview_year: int | None = None,
) -> HTMLResponse:
init_db()
available_years = get_available_years()
context = {
**base_context(request, message),
"overview": get_overview_stats(overview_year),
"overview_selected_year": overview_year,
"project_dashboard": get_project_dashboard_summary(overview_year),
"project_revenue_mix_yearly": get_project_revenue_mix(),
"project_revenue_mix_monthly": get_project_revenue_mix_monthly(),
"yearly_summary": get_yearly_summary(),
"monthly_summary": get_monthly_summary(),
"available_years": available_years,
"dashboard_revenue_metric_options": get_option_items("dashboard_revenue_metrics"),
"dashboard_expense_metric_options": get_option_items("dashboard_expense_metrics"),
}
return templates.TemplateResponse(request, "dashboard.html", context)
def render_projects_page(
request: Request,
edit_code: str | None = None,
focus_code: str | None = None,
selected_year: int | None = None,
message: str = "",
) -> HTMLResponse:
init_db()
selected_year = resolve_selected_year(selected_year)
context = {
**base_context(request, message),
"project_year_options": get_project_year_options(),
"selected_year": selected_year,
"project_dashboard": get_project_dashboard_summary(selected_year),
"project_revenue_mix": get_project_revenue_mix(selected_year),
"project_cost_by_year": get_project_cost_by_year(selected_year),
"project_monthly_cost_rows": get_business_monthly_summary(),
"project_account_breakdowns": get_project_account_breakdowns(selected_year),
"project_status_rows": get_project_status_rows(),
"project_comparison_notes": get_project_comparison_notes_map(),
"project_edit": get_project_status_for_edit(edit_code),
"project_focus_code": normalize_text(focus_code),
"project_page_state": get_project_page_state(),
"project_related_links": get_project_related_links_map(),
"project_uncontracted_classifications": get_project_uncontracted_classification_map(),
"support_department_options": get_support_department_options(),
"cost_department_options": get_cost_department_options(),
"cost_account_options": get_cost_account_options(),
"labor_grade_options": get_labor_grade_options(),
"expected_as_rate_options": get_expected_as_rate_options(),
"expected_sga_rate_options": get_expected_sga_rate_options(),
"collection_progress_type_options": get_collection_progress_type_options(),
"collection_billing_type_options": get_collection_billing_type_options(),
"uncontracted_category_options": get_uncontracted_category_options(),
"special_x_classification_rules": get_special_x_classification_rules(),
"project_runtime_settings": get_project_runtime_settings(),
}
return templates.TemplateResponse(request, "projects.html", context)
def render_annual_summary_page(request: Request, message: str = "") -> HTMLResponse:
init_db()
context = {
**base_context(request, message),
"available_years": get_available_years(),
"yearly_financial_series": get_financial_series("yearly"),
"monthly_financial_series": get_financial_series("monthly"),
"annual_metric_cards": get_option_items("annual_metric_cards"),
"annual_expense_chart_metrics": get_option_items("annual_expense_chart_metrics"),
"annual_balance_chart_metrics": get_option_items("annual_balance_chart_metrics"),
}
return templates.TemplateResponse(request, "annual_summary.html", context)
@app.get("/health")
async def health() -> dict[str, str]:
return build_health_payload()
@app.get("/")
async def home(request: Request, edit_id: int | None = None, overview_year: int | None = None):
try:
return render_home(request, edit_id=edit_id, overview_year=overview_year)
except Exception as exc:
logger.exception("홈페이지 에러: %s", exc)
return HTMLResponse("<h1>서버 오류</h1><p>로그를 확인해주세요.</p>", status_code=500)
@app.get("/projects")
async def projects(
request: Request,
edit_code: str | None = None,
focus_code: str | None = None,
year: str | None = None,
):
try:
return render_projects_page(
request,
edit_code=edit_code,
focus_code=focus_code,
selected_year=parse_optional_year(year),
)
except Exception as exc:
logger.exception("사업현황 페이지 에러: %s", exc)
return HTMLResponse("<h1>서버 오류</h1><p>로그를 확인해주세요.</p>", status_code=500)
@app.get("/projects/edit-data")
async def project_edit_data(code: str | None = None):
try:
return JSONResponse(content=jsonable_encoder(get_project_status_for_edit(code)))
except Exception as exc:
logger.exception("사업현황 편집 데이터 조회 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.post("/projects/page-state")
async def project_page_state_save(request: Request):
try:
payload = await request.json()
if not isinstance(payload, dict):
raise ValueError("잘못된 페이지 상태 형식입니다.")
save_project_page_state(payload)
return JSONResponse(content={"status": "ok"})
except Exception as exc:
logger.exception("사업현황 페이지 상태 저장 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.get("/projects/page-state")
async def project_page_state_load(session_id: str = ""):
try:
return JSONResponse(content=jsonable_encoder(get_project_page_state(session_id)))
except Exception as exc:
logger.exception("사업현황 페이지 상태 조회 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.post("/projects/related-links")
async def project_related_links_save(request: Request):
try:
payload = await request.json()
if not isinstance(payload, dict):
raise ValueError("잘못된 연관 프로젝트 형식입니다.")
base_code = normalize_text(payload.get("base_code"))
related_codes = payload.get("related_codes") or []
if not isinstance(related_codes, list):
raise ValueError("연관 프로젝트 목록 형식이 올바르지 않습니다.")
save_project_related_links(base_code, related_codes)
return JSONResponse(content={"status": "ok"})
except Exception as exc:
logger.exception("연관 프로젝트 저장 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.get("/projects/quick-links")
async def project_quick_links_load(session_id: str = ""):
try:
return JSONResponse(content={"codes": get_project_quick_links(session_id)})
except Exception as exc:
logger.exception("프로젝트 바로가기 조회 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.post("/projects/quick-links")
async def project_quick_links_save(request: Request):
try:
payload = await request.json()
if not isinstance(payload, dict):
raise ValueError("잘못된 바로가기 형식입니다.")
codes = payload.get("codes") or []
if not isinstance(codes, list):
raise ValueError("바로가기 목록 형식이 잘못되었습니다.")
session_id = normalize_text(payload.get("session_id"))
save_project_quick_links(session_id, [normalize_text(code) for code in codes if isinstance(code, str)])
return JSONResponse(content={"status": "ok", "codes": get_project_quick_links(session_id)})
except Exception as exc:
logger.exception("프로젝트 바로가기 저장 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.post("/projects/uncontracted-category")
async def project_uncontracted_category_save(request: Request):
try:
payload = await request.json()
if not isinstance(payload, dict):
raise ValueError("잘못된 미계약 분류 형식입니다.")
save_project_uncontracted_classification(
payload.get("support_dept_code"),
payload.get("category"),
)
return JSONResponse(content={"status": "ok"})
except Exception as exc:
logger.exception("미계약 분류 저장 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.post("/projects/runtime-setting")
async def project_runtime_setting_save(request: Request):
try:
payload = await request.json()
if not isinstance(payload, dict):
raise ValueError("잘못된 설정 형식입니다.")
save_project_runtime_setting(payload.get("item_key"), payload.get("value"))
return JSONResponse(content={"status": "ok", "settings": get_project_runtime_settings()})
except Exception as exc:
logger.exception("프로젝트 런타임 설정 저장 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.post("/projects/comparison-note")
async def project_comparison_note_save(request: Request):
try:
payload = await request.json()
if not isinstance(payload, dict):
raise ValueError("잘못된 비고 형식입니다.")
save_project_comparison_note(
payload.get("support_dept_code"),
payload.get("item_key"),
payload.get("note"),
)
return JSONResponse(content={"status": "ok"})
except Exception as exc:
logger.exception("계획 대비 실제 비교 비고 저장 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.get("/annual-summary")
async def annual_summary(request: Request):
try:
return render_annual_summary_page(request)
except Exception as exc:
logger.exception("연도별 수익 비용 정리 페이지 에러: %s", exc)
return HTMLResponse("<h1>서버 오류</h1><p>로그를 확인해주세요.</p>", status_code=500)
@app.post("/upload")
async def upload_excel(request: Request, excel_file: UploadFile = File(...)):
try:
inserted = parse_excel_upload(excel_file)
return render_home(request, message=f"{inserted}건의 엑셀 데이터를 DB에 저장했습니다.")
except Exception as exc:
logger.exception("엑셀 업로드 에러: %s", exc)
return render_home(request, message=f"엑셀 업로드 중 오류가 발생했습니다: {exc}")
@app.post("/records/save")
async def save_record(request: Request):
form_data: dict[str, Any] = {}
try:
form_data = parse_manual_form(await request.body())
record_id = normalize_text(form_data.get("id"))
payload = build_transaction_payload(form_data)
save_transaction(payload, int(record_id) if record_id else None)
return RedirectResponse("/", status_code=303)
except Exception as exc:
logger.exception("데이터 저장 에러: %s", exc)
record_id = normalize_text(form_data.get("id"))
return render_home(
request,
edit_id=int(record_id) if record_id else None,
message=f"데이터 저장 중 오류가 발생했습니다: {exc}",
)
@app.post("/projects/save")
async def save_project(request: Request):
form_data: dict[str, Any] = {}
try:
form_data = parse_project_form(await request.body())
save_project_status(form_data)
code = normalize_text(form_data.get("support_dept_code"))
selected_year = normalize_text(form_data.get("selected_year"))
redirect_url = "/projects"
query_parts = []
if code:
query_parts.append(f"focus_code={quote_plus(code)}")
if selected_year:
query_parts.append(f"year={quote_plus(selected_year)}")
if query_parts:
redirect_url += "?" + "&".join(query_parts)
return RedirectResponse(redirect_url, status_code=303)
except Exception as exc:
logger.exception("사업현황 저장 에러: %s", exc)
code = normalize_text(form_data.get("support_dept_code"))
selected_year_text = normalize_text(form_data.get("selected_year"))
selected_year = int(selected_year_text) if selected_year_text.isdigit() else None
return render_projects_page(
request,
edit_code=code or None,
focus_code=code or None,
selected_year=selected_year,
message=f"사업현황 저장 중 오류가 발생했습니다: {exc}",
)
@app.post("/projects/save-json")
async def save_project_json(request: Request):
form_data: dict[str, Any] = {}
try:
form_data = parse_project_form(await request.body())
save_project_status(form_data)
code = normalize_text(form_data.get("support_dept_code"))
health_payload = build_health_payload()
return JSONResponse(
content=jsonable_encoder(
{
"ok": True,
"support_dept_code": code,
"project_edit": get_project_status_for_edit(code),
"project_row": get_project_status_row_for_code(code),
"data_version": health_payload.get("data_version", ""),
"server_time": health_payload.get("server_time", ""),
}
)
)
except Exception as exc:
logger.exception("사업현황 JSON 저장 에러: %s", exc)
return JSONResponse(
status_code=500,
content={
"ok": False,
"error": str(exc) or "사업현황 저장 중 오류가 발생했습니다.",
},
)
if __name__ == "__main__":
auto_reload = os.getenv("INTRANET_AUTO_RELOAD", "1").lower() not in {"0", "false", "no"}
uvicorn.run("main:app", host="0.0.0.0", port=8010, reload=auto_reload, reload_dirs=[str(BASE_DIR)])
Reference in New Issue View Git Blame Copy Permalink