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

3404 lines
147 KiB
Python
Raw Blame History

import os
import logging
import json
import re
from datetime import date, datetime
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, quote_plus
import uvicorn
from fastapi import FastAPI, File, Request, UploadFile
from fastapi.encoders import jsonable_encoder
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from openpyxl import load_workbook
from sqlalchemy import create_engine, text
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
BASE_DIR = Path(__file__).resolve().parent
STATIC_DIR = BASE_DIR / "static"
TEMPLATES_DIR = BASE_DIR / "templates"
DB_PATH = BASE_DIR / "data.db"
STATIC_DIR.mkdir(exist_ok=True)
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
engine = create_engine(
f"sqlite:///{DB_PATH}",
connect_args={"check_same_thread": False},
)
EXCLUDED_SUPPORT_CODES = {"ZZZZZZ"}
EXCLUDED_SUPPORT_NAMES = {"공통", "경영지원부", "기술개발센터", "임원실", "기술개발부", "총괄기획실"}
SUPPORT_DEPARTMENT_NAMES = ("경영지원부", "임원실", "총괄기획실", "기술개발센터", "기술개발부", "공통")
VOUCHER_PATTERN = re.compile(r"^11-(\d{4})(\d{2})(\d{2})-[^-]+-[^-]+-[^-]+$")
REVENUE_SQL = "(account_code LIKE '401101%' OR account_code LIKE '401102%')"
SUPPORT_COST_DEPT_SQL = (
"cost_dept_name IN ('경영지원부', '임원실', '총괄기획실', '기술개발센터', '기술개발부', '공통')"
)
FIELD_COST_DEPT_SQL = (
"COALESCE(cost_dept_name, '') NOT IN ('경영지원부', '임원실', '총괄기획실', '기술개발센터', '기술개발부', '공통')"
)
FIELD_LABELS = {
"approval_status": "결재상태",
"voucher_number": "가전표번호",
"account_code": "계정코드",
"account_name": "계정명칭",
"debit_supply": "차변공급가",
"debit_vat": "차변부가세",
"credit_supply": "대변공급가",
"credit_vat": "대변부가세",
"issuing_dept_code": "발의부서코드",
"issuing_dept_name": "발의부서명",
"confirmed_voucher_number": "확정전표번호",
"support_dept_code": "지원부서코드",
"support_dept_name": "지원부서명",
"cost_dept_code": "원가부서코드",
"cost_dept_name": "원가부서명",
"memo1": "적요1",
"memo2": "적요2",
"partner_code": "거래처코드",
"partner_name": "거래처명칭",
"tax_code": "세무코드",
"posting_date": "증빙일자",
"voucher_type": "전표종류",
"management_item": "관리항목",
}
FORM_FIELDS = list(FIELD_LABELS.keys())
DIRECT_HEADER_MAP = {
"결재상태": "approval_status",
"가전표번호": "voucher_number",
"계정코드": "account_code",
"계정명칭": "account_name",
"차변공급가": "debit_supply",
"차변부가세": "debit_vat",
"대변공급가": "credit_supply",
"대변부가세": "credit_vat",
"발의부서코드": "issuing_dept_code",
"발의부서명": "issuing_dept_name",
"발의부서명칭": "issuing_dept_name",
"지원부서코드": "support_dept_code",
"지원부서명": "support_dept_name",
"지원부서명칭": "support_dept_name",
"원가부서코드": "cost_dept_code",
"원가부서명": "cost_dept_name",
"원가부서명칭": "cost_dept_name",
"적요1": "memo1",
"적요2": "memo2",
"거래처코드": "partner_code",
"거래처명칭": "partner_name",
"세무코드": "tax_code",
"증빙일자": "posting_date",
"전표종류": "voucher_type",
"관리항목": "management_item",
}
def init_db() -> None:
with engine.begin() as conn:
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
approval_status TEXT,
voucher_number TEXT,
account_code TEXT,
account_name TEXT,
debit_supply REAL DEFAULT 0,
debit_vat REAL DEFAULT 0,
credit_supply REAL DEFAULT 0,
credit_vat REAL DEFAULT 0,
issuing_dept_code TEXT,
issuing_dept_name TEXT,
confirmed_voucher_number TEXT,
support_dept_code TEXT,
support_dept_name TEXT,
cost_dept_code TEXT,
cost_dept_name TEXT,
memo1 TEXT,
memo2 TEXT,
partner_code TEXT,
partner_name TEXT,
tax_code TEXT,
posting_date TEXT,
voucher_type TEXT,
management_item TEXT,
accounting_category TEXT,
amount REAL DEFAULT 0,
year INTEGER,
month INTEGER,
day INTEGER,
source_file TEXT,
last_editor_session_id TEXT DEFAULT '',
last_client_submitted_at TEXT DEFAULT '',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_transactions_year_month
ON transactions (year, month)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_transactions_support
ON transactions (support_dept_code, support_dept_name)
"""
)
)
transaction_columns = {
row[1]
for row in conn.execute(text("PRAGMA table_info(transactions)")).fetchall()
}
required_transaction_columns = {
"last_editor_session_id": "TEXT DEFAULT ''",
"last_client_submitted_at": "TEXT DEFAULT ''",
}
for column_name, column_type in required_transaction_columns.items():
if column_name not in transaction_columns:
conn.execute(text(f"ALTER TABLE transactions ADD COLUMN {column_name} {column_type}"))
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_status (
support_dept_code TEXT PRIMARY KEY,
support_dept_name TEXT NOT NULL,
progress_rate REAL DEFAULT 0,
contract_amount REAL DEFAULT 0,
collection_amount REAL DEFAULT 0,
collection_entries_json TEXT DEFAULT '[]',
change_round TEXT DEFAULT '',
item_investment REAL DEFAULT 0,
task_plan_department_budget REAL DEFAULT 0,
task_plan_outsource_budget REAL DEFAULT 0,
task_plan_outsource_detail TEXT DEFAULT '',
task_plan_joint_operating_cost REAL DEFAULT 0,
task_plan_entries_json TEXT DEFAULT '[]',
exec_budget_labor_by_grade REAL DEFAULT 0,
exec_labor_rates_json TEXT DEFAULT '{}',
exec_budget_outsource REAL DEFAULT 0,
exec_budget_cost_plan REAL DEFAULT 0,
exec_budget_entries_json TEXT DEFAULT '[]',
actual_input_entries_json TEXT DEFAULT '[]',
project_type TEXT DEFAULT '',
expected_as_rate REAL DEFAULT 0,
expected_sga_rate REAL DEFAULT 0,
expected_as_cost REAL DEFAULT 0,
expected_sga_budget REAL DEFAULT 0,
last_editor_session_id TEXT DEFAULT '',
last_client_submitted_at TEXT DEFAULT '',
project_start_date TEXT DEFAULT '',
project_end_date TEXT DEFAULT '',
completion_status TEXT DEFAULT '',
notes TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_page_state (
page_key TEXT PRIMARY KEY,
selected_code TEXT DEFAULT '',
selected_year TEXT DEFAULT '',
analysis_open INTEGER DEFAULT 0,
related_project_selections_json TEXT DEFAULT '{}',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_related_links (
base_support_dept_code TEXT NOT NULL,
related_support_dept_code TEXT NOT NULL,
link_source TEXT DEFAULT 'manual',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (base_support_dept_code, related_support_dept_code)
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_related_links_base
ON project_related_links (base_support_dept_code)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_contract_info (
support_dept_code TEXT PRIMARY KEY,
raw_contract_code TEXT DEFAULT '',
business_division TEXT DEFAULT '',
order_method TEXT DEFAULT '',
owner_department TEXT DEFAULT '',
client_name TEXT DEFAULT '',
support_dept_name TEXT DEFAULT '',
work_category TEXT DEFAULT '',
order_date TEXT DEFAULT '',
contract_date TEXT DEFAULT '',
project_start_date TEXT DEFAULT '',
project_end_date TEXT DEFAULT '',
contract_status TEXT DEFAULT '',
joint_contract TEXT DEFAULT '',
pm_name TEXT DEFAULT '',
progress_status TEXT DEFAULT '',
total_contract_amount REAL DEFAULT 0,
hanmac_contract_amount REAL DEFAULT 0,
review_tag TEXT DEFAULT '',
review_note TEXT DEFAULT '',
source_file TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE TABLE IF NOT EXISTS project_billing_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
support_dept_code TEXT,
raw_project_code TEXT DEFAULT '',
round_code TEXT DEFAULT '',
support_department TEXT DEFAULT '',
business_division TEXT DEFAULT '',
support_dept_name TEXT DEFAULT '',
contract_amount REAL DEFAULT 0,
client_name TEXT DEFAULT '',
billing_type TEXT DEFAULT '',
progress_round TEXT DEFAULT '',
billing_date TEXT DEFAULT '',
tax_invoice_date TEXT DEFAULT '',
expected_collection_date TEXT DEFAULT '',
billed_amount REAL DEFAULT 0,
collected_amount REAL DEFAULT 0,
balance_amount REAL DEFAULT 0,
collection_rate REAL DEFAULT 0,
note TEXT DEFAULT '',
source_file TEXT DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
)
conn.execute(
text(
"""
CREATE INDEX IF NOT EXISTS idx_project_billing_entries_code
ON project_billing_entries (support_dept_code, billing_date)
"""
)
)
existing_columns = {
row[1]
for row in conn.execute(text("PRAGMA table_info(project_status)")).fetchall()
}
required_columns = {
"last_editor_session_id": "TEXT DEFAULT ''",
"last_client_submitted_at": "TEXT DEFAULT ''",
"contract_amount": "REAL DEFAULT 0",
"collection_entries_json": "TEXT DEFAULT '[]'",
"task_plan_department_budget": "REAL DEFAULT 0",
"task_plan_outsource_budget": "REAL DEFAULT 0",
"task_plan_outsource_detail": "TEXT DEFAULT ''",
"task_plan_joint_operating_cost": "REAL DEFAULT 0",
"task_plan_entries_json": "TEXT DEFAULT '[]'",
"exec_budget_labor_by_grade": "REAL DEFAULT 0",
"exec_labor_rates_json": "TEXT DEFAULT '{}'",
"exec_budget_outsource": "REAL DEFAULT 0",
"exec_budget_cost_plan": "REAL DEFAULT 0",
"exec_budget_entries_json": "TEXT DEFAULT '[]'",
"actual_input_entries_json": "TEXT DEFAULT '[]'",
"project_type": "TEXT DEFAULT ''",
"expected_as_rate": "REAL DEFAULT 0",
"expected_sga_rate": "REAL DEFAULT 0",
"expected_as_cost": "REAL DEFAULT 0",
"expected_sga_budget": "REAL DEFAULT 0",
"project_start_date": "TEXT DEFAULT ''",
"project_end_date": "TEXT DEFAULT ''",
"completion_status": "TEXT DEFAULT ''",
}
for column_name, column_type in required_columns.items():
if column_name not in existing_columns:
conn.execute(text(f"ALTER TABLE project_status ADD COLUMN {column_name} {column_type}"))
page_state_columns = {
row[1]
for row in conn.execute(text("PRAGMA table_info(project_page_state)")).fetchall()
}
required_page_state_columns = {
"selected_code": "TEXT DEFAULT ''",
"selected_year": "TEXT DEFAULT ''",
"analysis_open": "INTEGER DEFAULT 0",
"related_project_selections_json": "TEXT DEFAULT '{}'",
}
for column_name, column_type in required_page_state_columns.items():
if column_name not in page_state_columns:
conn.execute(text(f"ALTER TABLE project_page_state ADD COLUMN {column_name} {column_type}"))
related_link_columns = {
row[1]
for row in conn.execute(text("PRAGMA table_info(project_related_links)")).fetchall()
}
if "link_source" not in related_link_columns:
conn.execute(text("ALTER TABLE project_related_links ADD COLUMN link_source TEXT DEFAULT 'manual'"))
def count_transactions() -> int:
with engine.begin() as conn:
return conn.execute(text("SELECT COUNT(*) FROM transactions")).scalar_one()
def existing_source_files() -> set[str]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT DISTINCT source_file FROM transactions WHERE COALESCE(source_file, '') <> ''")
).fetchall()
return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}
def existing_contract_source_files() -> set[str]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT DISTINCT source_file FROM project_contract_info WHERE COALESCE(source_file, '') <> ''")
).fetchall()
return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}
def existing_billing_source_files() -> set[str]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT DISTINCT source_file FROM project_billing_entries WHERE COALESCE(source_file, '') <> ''")
).fetchall()
return {normalize_text(row[0]) for row in rows if normalize_text(row[0])}
def workbook_row_values(sheet: Any, row_number: int) -> list[str]:
return [normalize_text(sheet.cell(row_number, column).value) for column in range(1, sheet.max_column + 1)]
def detect_excel_import_kind(workbook: Any, filename: str = "") -> str:
sheet = workbook.active
row1 = workbook_row_values(sheet, 1)
row5 = workbook_row_values(sheet, 5) if sheet.max_row >= 5 else []
filename = normalize_text(filename)
if {"총괄코드", "총 계약금액", "한맥계약금액"}.issubset(set(row1)):
return "contract_status"
if {"차수코드", "차수사업명", "청구금액", "수금금액"}.issubset(set(row5)):
return "billing_status"
if "계약현황" in filename:
return "contract_status"
if "기성청구현황" in filename:
return "billing_status"
return "transactions"
def import_contract_status_workbook(workbook: Any, source_file: str) -> int:
sheet = workbook.active
with engine.begin() as conn:
conn.execute(
text("DELETE FROM project_contract_info WHERE source_file = :source_file"),
{"source_file": source_file},
)
inserted = 0
for row in sheet.iter_rows(min_row=2, values_only=True):
support_dept_code = normalize_project_code(row[1] if len(row) > 1 else "")
if not support_dept_code:
continue
payload = {
"support_dept_code": support_dept_code,
"raw_contract_code": normalize_text(row[1] if len(row) > 1 else ""),
"business_division": normalize_text(row[0] if len(row) > 0 else ""),
"order_method": normalize_text(row[2] if len(row) > 2 else ""),
"owner_department": normalize_text(row[3] if len(row) > 3 else ""),
"client_name": normalize_text(row[4] if len(row) > 4 else ""),
"support_dept_name": normalize_text(row[5] if len(row) > 5 else ""),
"work_category": normalize_text(row[6] if len(row) > 6 else ""),
"order_date": normalize_date_text(row[7] if len(row) > 7 else ""),
"contract_date": normalize_date_text(row[8] if len(row) > 8 else ""),
"project_start_date": normalize_date_text(row[9] if len(row) > 9 else ""),
"project_end_date": normalize_date_text(row[10] if len(row) > 10 else ""),
"contract_status": normalize_text(row[11] if len(row) > 11 else ""),
"joint_contract": normalize_text(row[12] if len(row) > 12 else ""),
"pm_name": normalize_text(row[13] if len(row) > 13 else ""),
"progress_status": normalize_text(row[14] if len(row) > 14 else ""),
"total_contract_amount": normalize_amount(row[15] if len(row) > 15 else 0),
"hanmac_contract_amount": normalize_amount(row[16] if len(row) > 16 else 0),
"review_tag": "",
"review_note": "",
"source_file": source_file,
}
conn.execute(
text(
"""
INSERT INTO project_contract_info (
support_dept_code, raw_contract_code, business_division, order_method,
owner_department, client_name, support_dept_name, work_category,
order_date, contract_date, project_start_date, project_end_date,
contract_status, joint_contract, pm_name, progress_status,
total_contract_amount, hanmac_contract_amount, review_tag, review_note,
source_file, updated_at
) VALUES (
:support_dept_code, :raw_contract_code, :business_division, :order_method,
:owner_department, :client_name, :support_dept_name, :work_category,
:order_date, :contract_date, :project_start_date, :project_end_date,
:contract_status, :joint_contract, :pm_name, :progress_status,
:total_contract_amount, :hanmac_contract_amount, :review_tag, :review_note,
:source_file, CURRENT_TIMESTAMP
)
ON CONFLICT(support_dept_code) DO UPDATE SET
raw_contract_code = excluded.raw_contract_code,
business_division = excluded.business_division,
order_method = excluded.order_method,
owner_department = excluded.owner_department,
client_name = excluded.client_name,
support_dept_name = excluded.support_dept_name,
work_category = excluded.work_category,
order_date = excluded.order_date,
contract_date = excluded.contract_date,
project_start_date = excluded.project_start_date,
project_end_date = excluded.project_end_date,
contract_status = excluded.contract_status,
joint_contract = excluded.joint_contract,
pm_name = excluded.pm_name,
progress_status = excluded.progress_status,
total_contract_amount = excluded.total_contract_amount,
hanmac_contract_amount = excluded.hanmac_contract_amount,
source_file = excluded.source_file,
updated_at = CURRENT_TIMESTAMP
"""
),
payload,
)
inserted += 1
refresh_contract_review_tags()
sync_auto_project_related_links()
return inserted
def import_billing_status_workbook(workbook: Any, source_file: str) -> int:
sheet = workbook.active
with engine.begin() as conn:
conn.execute(
text("DELETE FROM project_billing_entries WHERE source_file = :source_file"),
{"source_file": source_file},
)
inserted = 0
current: dict[str, Any] = {}
for row in sheet.iter_rows(min_row=6, values_only=True):
values = list(row)
if values and all(value in (None, "") for value in values):
continue
if values[0] is not None:
current["support_department"] = normalize_text(values[0])
if len(values) > 1 and values[1] is not None:
current["business_division"] = normalize_text(values[1])
if len(values) > 2 and values[2] is not None:
current["raw_project_code"] = normalize_text(values[2])
if len(values) > 3 and values[3] is not None:
current["round_code"] = normalize_text(values[3])
if len(values) > 4 and values[4] is not None:
current["support_dept_name"] = normalize_text(values[4])
if len(values) > 5 and values[5] is not None:
current["contract_amount"] = normalize_amount(values[5])
if len(values) > 6 and values[6] is not None:
current["client_name"] = normalize_text(values[6])
round_code_text = normalize_text(current.get("round_code"))
round_prefix = next((character.upper() for character in round_code_text if character.isalpha()), "Y")
support_dept_code = normalize_project_code(
current.get("raw_project_code") or current.get("round_code"),
default_prefix=round_prefix,
)
if not support_dept_code:
continue
summary_row = normalize_text(values[11] if len(values) > 11 else "") == "합계" or normalize_text(values[10] if len(values) > 10 else "").startswith("수금 :")
department_summary = "합계" in normalize_text(values[4] if len(values) > 4 else "")
if summary_row or department_summary:
continue
payload = {
"support_dept_code": support_dept_code,
"raw_project_code": normalize_text(current.get("raw_project_code")),
"round_code": normalize_text(current.get("round_code")),
"support_department": normalize_text(current.get("support_department")),
"business_division": normalize_text(current.get("business_division")),
"support_dept_name": normalize_text(current.get("support_dept_name")),
"contract_amount": normalize_amount(current.get("contract_amount")),
"client_name": normalize_text(current.get("client_name")),
"billing_type": normalize_text(values[7] if len(values) > 7 else ""),
"progress_round": normalize_round_value(values[8] if len(values) > 8 else ""),
"billing_date": normalize_date_text(values[9] if len(values) > 9 else ""),
"tax_invoice_date": normalize_date_text(values[10] if len(values) > 10 else ""),
"expected_collection_date": normalize_date_text(values[11] if len(values) > 11 else ""),
"billed_amount": normalize_amount(values[12] if len(values) > 12 else 0),
"collected_amount": normalize_amount(values[13] if len(values) > 13 else 0),
"balance_amount": normalize_amount(values[14] if len(values) > 14 else 0),
"collection_rate": normalize_amount(values[15] if len(values) > 15 else 0),
"note": normalize_text(values[16] if len(values) > 16 else ""),
"source_file": source_file,
}
conn.execute(
text(
"""
INSERT INTO project_billing_entries (
support_dept_code, raw_project_code, round_code, support_department,
business_division, support_dept_name, contract_amount, client_name,
billing_type, progress_round, billing_date, tax_invoice_date,
expected_collection_date, billed_amount, collected_amount,
balance_amount, collection_rate, note, source_file, updated_at
) VALUES (
:support_dept_code, :raw_project_code, :round_code, :support_department,
:business_division, :support_dept_name, :contract_amount, :client_name,
:billing_type, :progress_round, :billing_date, :tax_invoice_date,
:expected_collection_date, :billed_amount, :collected_amount,
:balance_amount, :collection_rate, :note, :source_file, CURRENT_TIMESTAMP
)
"""
),
payload,
)
inserted += 1
refresh_contract_review_tags()
return inserted
def refresh_contract_review_tags() -> None:
with engine.begin() as conn:
billing_rows = conn.execute(
text(
"""
SELECT support_dept_code, MAX(contract_amount) AS billing_contract_amount
FROM project_billing_entries
GROUP BY support_dept_code
"""
)
).mappings().all()
billing_map = {
normalize_text(row["support_dept_code"]): normalize_amount(row["billing_contract_amount"])
for row in billing_rows
if normalize_text(row["support_dept_code"])
}
contract_rows = conn.execute(
text("SELECT support_dept_code, hanmac_contract_amount FROM project_contract_info")
).mappings().all()
for row in contract_rows:
support_dept_code = normalize_text(row["support_dept_code"])
hanmac_contract_amount = normalize_amount(row["hanmac_contract_amount"])
billing_contract_amount = normalize_amount(billing_map.get(support_dept_code))
review_tag = ""
review_note = ""
if billing_contract_amount and abs(hanmac_contract_amount - billing_contract_amount) > 0.5:
review_tag = "변경계약 검토 필요"
review_note = (
f"계약현황 한맥계약금액 {hanmac_contract_amount:,.0f}원 / "
f"기성청구현황 계약금액 {billing_contract_amount:,.0f}원"
)
conn.execute(
text(
"""
UPDATE project_contract_info
SET review_tag = :review_tag,
review_note = :review_note,
updated_at = CURRENT_TIMESTAMP
WHERE support_dept_code = :support_dept_code
"""
),
{
"support_dept_code": support_dept_code,
"review_tag": review_tag,
"review_note": review_note,
},
)
def sync_auto_project_related_links() -> None:
with engine.begin() as conn:
billing_rows = conn.execute(
text(
"""
SELECT support_dept_code, raw_project_code, round_code
FROM project_billing_entries
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).mappings().all()
existing_codes = {
normalize_text(row[0])
for row in conn.execute(
text(
"""
SELECT DISTINCT support_dept_code FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_status
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_contract_info
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT DISTINCT support_dept_code FROM project_billing_entries
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).fetchall()
if normalize_text(row[0])
}
cluster_map: dict[str, set[str]] = {}
for row in billing_rows:
base_code = normalize_text(row["support_dept_code"])
raw_project_code = normalize_text(row["raw_project_code"])
round_code = normalize_project_code(row["round_code"], default_prefix=base_code[:1] or "Y")
if not base_code:
continue
cluster_key = raw_project_code or base_code
cluster = cluster_map.setdefault(cluster_key, set())
if base_code in existing_codes:
cluster.add(base_code)
if round_code and round_code in existing_codes:
cluster.add(round_code)
conn.execute(text("DELETE FROM project_related_links WHERE COALESCE(link_source, 'manual') = 'auto'"))
for cluster_codes in cluster_map.values():
normalized_cluster = sorted(cluster_codes)
if len(normalized_cluster) < 2:
continue
for base_code in normalized_cluster:
for related_code in normalized_cluster:
if base_code == related_code:
continue
conn.execute(
text(
"""
INSERT INTO project_related_links (
base_support_dept_code,
related_support_dept_code,
link_source,
updated_at
) VALUES (
:base_support_dept_code,
:related_support_dept_code,
'auto',
CURRENT_TIMESTAMP
)
ON CONFLICT(base_support_dept_code, related_support_dept_code) DO UPDATE SET
link_source = excluded.link_source,
updated_at = CURRENT_TIMESTAMP
"""
),
{
"base_support_dept_code": base_code,
"related_support_dept_code": related_code,
},
)
@app.on_event("startup")
def on_startup() -> None:
init_db()
auto_import_project_excels()
sync_auto_project_related_links()
logger.info("DB ready at %s", DB_PATH)
def normalize_text(value: Any) -> str:
if value is None:
return ""
if isinstance(value, str):
return value.strip()
return str(value).strip()
def normalize_amount(value: Any) -> float:
if value in (None, ""):
return 0.0
if isinstance(value, (int, float)):
return float(value)
cleaned = (
str(value)
.strip()
.replace(",", "")
.replace("원", "")
.replace("(", "-")
.replace(")", "")
)
if not cleaned:
return 0.0
try:
return float(cleaned)
except ValueError:
return 0.0
def normalize_date_text(value: Any) -> str:
if value in (None, ""):
return ""
if isinstance(value, datetime):
return value.date().isoformat()
if isinstance(value, date):
return value.isoformat()
text_value = normalize_text(value)
for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y%m%d"):
try:
return datetime.strptime(text_value, fmt).date().isoformat()
except ValueError:
continue
return text_value
def normalize_project_code(value: Any, default_prefix: str = "Y") -> str:
text_value = normalize_text(value).replace("\u3164", "").replace("\xa0", "")
prefix = ""
for character in text_value:
if character.isalpha():
prefix = character.upper()
break
digits = "".join(character for character in text_value if character.isdigit())
if not digits:
return ""
return f"{prefix or default_prefix}{int(digits)}"
def normalize_round_value(value: Any) -> str:
text_value = normalize_text(value)
if not text_value:
return ""
digits = "".join(character for character in text_value if character.isdigit())
if digits:
return str(int(digits))
return text_value
def decode_json_rows(value: Any) -> list[dict[str, Any]]:
text_value = normalize_text(value)
if not text_value:
return []
try:
rows = json.loads(text_value)
except json.JSONDecodeError:
return []
return [row for row in rows if isinstance(row, dict)]
def encode_json_rows(rows: list[dict[str, Any]]) -> str:
return json.dumps(rows, ensure_ascii=False)
def clean_row_text(value: Any) -> str:
return normalize_text(value)
def filter_amount_rows(rows: list[dict[str, Any]], amount_key: str = "amount") -> list[dict[str, Any]]:
cleaned_rows: list[dict[str, Any]] = []
for row in rows:
normalized_row = {key: clean_row_text(value) for key, value in row.items()}
amount = normalize_amount(normalized_row.get(amount_key))
if amount or any(value for key, value in normalized_row.items() if key != amount_key):
normalized_row[amount_key] = amount
cleaned_rows.append(normalized_row)
return cleaned_rows
def sum_row_amounts(rows: list[dict[str, Any]], amount_key: str = "amount") -> float:
return sum(normalize_amount(row.get(amount_key)) for row in rows)
def get_support_department_options() -> list[dict[str, str]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT support_dept_code, support_dept_name
FROM (
SELECT support_dept_code, support_dept_name
FROM transactions
UNION ALL
SELECT support_dept_code, support_dept_name
FROM project_contract_info
UNION ALL
SELECT support_dept_code, support_dept_name
FROM project_billing_entries
) AS merged
WHERE COALESCE(support_dept_code, '') <> ''
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
ORDER BY support_dept_code, support_dept_name
"""
)
).mappings().all()
return [
{
"support_dept_code": normalize_text(row["support_dept_code"]),
"support_dept_name": normalize_text(row["support_dept_name"]),
}
for row in rows
]
def get_cost_department_options() -> list[dict[str, str]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT cost_dept_name
FROM transactions
WHERE COALESCE(cost_dept_name, '') <> ''
ORDER BY cost_dept_name
"""
)
).mappings().all()
return [
{"cost_dept_name": normalize_text(row["cost_dept_name"])}
for row in rows
if normalize_text(row["cost_dept_name"])
]
def get_cost_account_options() -> list[dict[str, str]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT account_code, account_name
FROM transactions
WHERE accounting_category = '원가'
AND COALESCE(account_code, '') <> ''
AND COALESCE(account_name, '') <> ''
ORDER BY account_code, account_name
"""
)
).mappings().all()
deduped: dict[tuple[str, str], dict[str, str]] = {}
for row in rows:
account_code = normalize_text(row["account_code"])
account_name = normalize_text(row["account_name"])
if not account_code or not account_name:
continue
normalized_code, normalized_name, _ = normalize_account_display(account_code, account_name)
key = (normalized_code, normalized_name)
deduped[key] = {
"account_code": normalized_code,
"account_name": normalized_name,
}
return sorted(deduped.values(), key=lambda item: (item["account_code"], item["account_name"]))
def get_import_sync_summary() -> dict[str, Any]:
with engine.begin() as conn:
row = conn.execute(
text(
"""
SELECT
(SELECT COUNT(*) FROM project_contract_info) AS contract_project_count,
(SELECT COUNT(*) FROM project_billing_entries) AS billing_entry_count,
(SELECT COUNT(DISTINCT support_dept_code) FROM project_billing_entries) AS billing_project_count,
(SELECT COUNT(*) FROM project_contract_info WHERE COALESCE(review_tag, '') <> '') AS review_needed_count,
(SELECT SUM(hanmac_contract_amount) FROM project_contract_info) AS total_hanmac_contract_amount,
(SELECT SUM(collected_amount) FROM project_billing_entries) AS total_collected_amount,
(SELECT MAX(updated_at) FROM project_contract_info) AS latest_contract_sync,
(SELECT MAX(updated_at) FROM project_billing_entries) AS latest_billing_sync
"""
)
).mappings().first()
return dict(row) if row else {}
def get_project_contract_info_map() -> dict[str, dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text("SELECT * FROM project_contract_info ORDER BY support_dept_code")
).mappings().all()
return {normalize_text(row["support_dept_code"]): dict(row) for row in rows}
def get_project_billing_summary_map() -> dict[str, dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT support_dept_code,
MAX(support_dept_name) AS support_dept_name,
MAX(contract_amount) AS contract_amount,
MAX(client_name) AS client_name,
MAX(support_department) AS support_department,
MAX(business_division) AS business_division,
SUM(billed_amount) AS billed_amount,
SUM(collected_amount) AS collected_amount,
SUM(balance_amount) AS balance_amount,
MAX(billing_date) AS latest_billing_date
FROM project_billing_entries
GROUP BY support_dept_code
ORDER BY support_dept_code
"""
)
).mappings().all()
entry_rows = conn.execute(
text(
"""
SELECT support_dept_code,
billing_type,
progress_round,
billing_date,
tax_invoice_date,
expected_collection_date,
billed_amount,
collected_amount,
balance_amount,
collection_rate,
note
FROM project_billing_entries
ORDER BY support_dept_code, billing_date, progress_round, id
"""
)
).mappings().all()
result = {normalize_text(row["support_dept_code"]): dict(row) for row in rows}
for item in result.values():
item["entries"] = []
for row in entry_rows:
support_dept_code = normalize_text(row["support_dept_code"])
if support_dept_code not in result:
continue
result[support_dept_code]["entries"].append(
{
"progress_type": "",
"billing_round": normalize_round_value(row["progress_round"]),
"billing_type": normalize_text(row["billing_type"]),
"billing_date": normalize_date_text(row["billing_date"]),
"billed_amount": normalize_amount(row["billed_amount"]),
"round": normalize_round_value(row["progress_round"]),
"date": normalize_date_text(row["tax_invoice_date"]) or normalize_date_text(row["expected_collection_date"]),
"amount": normalize_amount(row["collected_amount"]),
"balance_amount": normalize_amount(row["balance_amount"]),
"collection_rate": normalize_amount(row["collection_rate"]),
"note": normalize_text(row["note"]),
}
)
return result
def merge_project_external_fields(
item: dict[str, Any],
contract_info: dict[str, Any] | None,
billing_summary: dict[str, Any] | None,
) -> dict[str, Any]:
contract_info = contract_info or {}
billing_summary = billing_summary or {}
support_dept_name = normalize_text(item.get("support_dept_name")) or normalize_text(contract_info.get("support_dept_name")) or normalize_text(billing_summary.get("support_dept_name"))
contract_amount = normalize_amount(item.get("contract_amount"))
if not contract_amount:
contract_amount = normalize_amount(contract_info.get("hanmac_contract_amount")) or normalize_amount(billing_summary.get("contract_amount"))
collection_amount = normalize_amount(item.get("collection_amount"))
if not collection_amount:
collection_amount = normalize_amount(billing_summary.get("collected_amount"))
collection_entries = item.get("collection_entries")
if not collection_entries:
collection_entries = billing_summary.get("entries", [])
project_start_date = normalize_text(item.get("project_start_date")) or normalize_text(contract_info.get("project_start_date"))
project_end_date = normalize_text(item.get("project_end_date")) or normalize_text(contract_info.get("project_end_date"))
completion_status = normalize_text(item.get("completion_status")) or normalize_text(contract_info.get("progress_status"))
project_type = normalize_text(item.get("project_type")) or normalize_text(contract_info.get("business_division")) or normalize_text(billing_summary.get("business_division"))
progress_rate = normalize_amount(item.get("progress_rate"))
if not progress_rate and contract_amount:
progress_rate = (collection_amount / contract_amount * 100) if contract_amount else 0.0
item["support_dept_name"] = support_dept_name
item["contract_amount"] = contract_amount
item["collection_amount"] = collection_amount
item["collection_entries"] = collection_entries or []
item["project_start_date"] = project_start_date
item["project_end_date"] = project_end_date
item["completion_status"] = completion_status
item["project_type"] = project_type
item["progress_rate"] = progress_rate
item["client_name"] = normalize_text(contract_info.get("client_name")) or normalize_text(billing_summary.get("client_name"))
item["order_method"] = normalize_text(contract_info.get("order_method"))
item["joint_contract"] = normalize_text(contract_info.get("joint_contract"))
item["pm_name"] = normalize_text(contract_info.get("pm_name"))
item["contract_status"] = normalize_text(contract_info.get("contract_status"))
item["progress_status"] = normalize_text(contract_info.get("progress_status"))
item["work_category"] = normalize_text(contract_info.get("work_category"))
item["review_tag"] = normalize_text(contract_info.get("review_tag"))
item["review_note"] = normalize_text(contract_info.get("review_note"))
item["total_contract_amount"] = normalize_amount(contract_info.get("total_contract_amount"))
item["hanmac_contract_amount"] = normalize_amount(contract_info.get("hanmac_contract_amount"))
item["billing_contract_amount"] = normalize_amount(billing_summary.get("contract_amount"))
item["billed_amount"] = normalize_amount(billing_summary.get("billed_amount"))
item["collection_balance_amount"] = normalize_amount(billing_summary.get("balance_amount"))
item["latest_billing_date"] = normalize_text(billing_summary.get("latest_billing_date"))
return item
def normalize_account_display(account_code: Any, account_name: Any) -> tuple[str, str, str]:
normalized_code = normalize_text(account_code)[:6]
normalized_name = re.sub(r"\s*\(.*$", "", normalize_text(account_name)).strip()
if normalized_code and normalized_name:
label = f"{normalized_code} · {normalized_name}"
else:
label = normalized_name or normalized_code or "미분류"
return normalized_code, normalized_name, label
def get_data_version() -> str:
with engine.begin() as conn:
transaction_updated = conn.execute(text("SELECT MAX(updated_at) FROM transactions")).scalar()
project_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_status")).scalar()
contract_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_contract_info")).scalar()
billing_updated = conn.execute(text("SELECT MAX(updated_at) FROM project_billing_entries")).scalar()
versions = [
normalize_text(transaction_updated),
normalize_text(project_updated),
normalize_text(contract_updated),
normalize_text(billing_updated),
]
return max((version for version in versions if version), default="")
def build_health_payload() -> dict[str, str]:
return {
"status": "ok",
"server_time": datetime.now().isoformat(timespec="seconds"),
"data_version": get_data_version(),
}
def check_record_revision(conn: Any, table_name: str, key_column: str, key_value: Any, edit_revision: str) -> None:
if not key_value or not edit_revision:
return
current_revision = conn.execute(
text(f"SELECT updated_at FROM {table_name} WHERE {key_column} = :key_value"),
{"key_value": key_value},
).scalar()
current_revision_text = normalize_text(current_revision)
if current_revision_text and current_revision_text != normalize_text(edit_revision):
raise ValueError("다른 사용자가 먼저 수정했습니다. 최신 화면으로 다시 확인한 뒤 저장해주세요.")
def date_diff_days(start_date: str, end_date: str) -> int | None:
if not start_date or not end_date:
return None
try:
start = datetime.strptime(start_date, "%Y-%m-%d").date()
end = datetime.strptime(end_date, "%Y-%m-%d").date()
return (end - start).days
except ValueError:
return None
def detect_category(account_code: str) -> str:
if account_code.startswith("5"):
return "원가"
if account_code.startswith("4"):
return "수입/매출액"
if account_code.startswith("6"):
return "판관비"
return "기타"
def extract_period(voucher_number: str, posting_date: str) -> tuple[int | None, int | None, int | None]:
voucher_match = VOUCHER_PATTERN.match(voucher_number)
if voucher_match:
year_text, month_text, day_text = voucher_match.groups()
return int(year_text), int(month_text), int(day_text)
parsed_date = normalize_date_text(posting_date)
if re.match(r"^\d{4}-\d{2}-\d{2}$", parsed_date):
parsed = datetime.strptime(parsed_date, "%Y-%m-%d")
return parsed.year, parsed.month, parsed.day
return None, None, None
def choose_amount(debit_supply: float, credit_supply: float) -> float:
if debit_supply:
return abs(debit_supply)
if credit_supply:
return abs(credit_supply)
return 0.0
def canonical_header_name(value: Any) -> str | None:
normalized = normalize_text(value).replace(" ", "")
if not normalized:
return None
if normalized in DIRECT_HEADER_MAP:
return DIRECT_HEADER_MAP[normalized]
if "확정전표" in normalized:
return "confirmed_voucher_number"
return None
def empty_record() -> dict[str, str]:
record = {field: "" for field in FORM_FIELDS}
record["id"] = ""
return record
def build_transaction_payload(raw: dict[str, Any], source_file: str = "") -> dict[str, Any]:
payload: dict[str, Any] = {}
for field in FORM_FIELDS:
if field in {"debit_supply", "debit_vat", "credit_supply", "credit_vat"}:
payload[field] = normalize_amount(raw.get(field))
elif field == "posting_date":
payload[field] = normalize_date_text(raw.get(field))
else:
payload[field] = normalize_text(raw.get(field))
payload["accounting_category"] = detect_category(payload["account_code"])
payload["amount"] = choose_amount(payload["debit_supply"], payload["credit_supply"])
year, month, day = extract_period(payload["voucher_number"], payload["posting_date"])
payload["year"] = year
payload["month"] = month
payload["day"] = day
payload["source_file"] = source_file
return payload
def save_transaction(payload: dict[str, Any], record_id: int | None = None) -> None:
init_db()
params = {
**payload,
"record_id": record_id,
"last_editor_session_id": normalize_text(payload.get("client_session_id")),
"last_client_submitted_at": normalize_text(payload.get("client_submitted_at")),
}
with engine.begin() as conn:
if record_id:
check_record_revision(conn, "transactions", "id", record_id, normalize_text(payload.get("edit_revision")))
conn.execute(
text(
"""
UPDATE transactions
SET approval_status = :approval_status,
voucher_number = :voucher_number,
account_code = :account_code,
account_name = :account_name,
debit_supply = :debit_supply,
debit_vat = :debit_vat,
credit_supply = :credit_supply,
credit_vat = :credit_vat,
issuing_dept_code = :issuing_dept_code,
issuing_dept_name = :issuing_dept_name,
confirmed_voucher_number = :confirmed_voucher_number,
support_dept_code = :support_dept_code,
support_dept_name = :support_dept_name,
cost_dept_code = :cost_dept_code,
cost_dept_name = :cost_dept_name,
memo1 = :memo1,
memo2 = :memo2,
partner_code = :partner_code,
partner_name = :partner_name,
tax_code = :tax_code,
posting_date = :posting_date,
voucher_type = :voucher_type,
management_item = :management_item,
accounting_category = :accounting_category,
amount = :amount,
year = :year,
month = :month,
day = :day,
source_file = COALESCE(NULLIF(:source_file, ''), source_file),
last_editor_session_id = :last_editor_session_id,
last_client_submitted_at = :last_client_submitted_at,
updated_at = CURRENT_TIMESTAMP
WHERE id = :record_id
"""
),
params,
)
return
conn.execute(
text(
"""
INSERT INTO transactions (
approval_status,
voucher_number,
account_code,
account_name,
debit_supply,
debit_vat,
credit_supply,
credit_vat,
issuing_dept_code,
issuing_dept_name,
confirmed_voucher_number,
support_dept_code,
support_dept_name,
cost_dept_code,
cost_dept_name,
memo1,
memo2,
partner_code,
partner_name,
tax_code,
posting_date,
voucher_type,
management_item,
accounting_category,
amount,
year,
month,
day,
source_file,
last_editor_session_id,
last_client_submitted_at
) VALUES (
:approval_status,
:voucher_number,
:account_code,
:account_name,
:debit_supply,
:debit_vat,
:credit_supply,
:credit_vat,
:issuing_dept_code,
:issuing_dept_name,
:confirmed_voucher_number,
:support_dept_code,
:support_dept_name,
:cost_dept_code,
:cost_dept_name,
:memo1,
:memo2,
:partner_code,
:partner_name,
:tax_code,
:posting_date,
:voucher_type,
:management_item,
:accounting_category,
:amount,
:year,
:month,
:day,
:source_file,
:last_editor_session_id,
:last_client_submitted_at
)
"""
),
params,
)
def get_record_for_edit(record_id: int | None) -> dict[str, Any]:
if not record_id:
return empty_record()
with engine.begin() as conn:
row = conn.execute(
text("SELECT * FROM transactions WHERE id = :record_id"),
{"record_id": record_id},
).mappings().first()
if not row:
return empty_record()
data = dict(row)
for key, value in list(data.items()):
if value is None:
data[key] = ""
return data
def get_support_businesses() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT support_dept_code, support_dept_name, COUNT(*) AS row_count
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN (
'공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
)
GROUP BY support_dept_code, support_dept_name
ORDER BY support_dept_code, support_dept_name
"""
)
).mappings().all()
result = [dict(row) for row in rows]
for item in result:
item["project_duration_days"] = date_diff_days(item.get("project_start_date", ""), item.get("project_end_date", ""))
item["planned_total"] = (
(item.get("task_plan_department_budget") or 0)
+ (item.get("task_plan_outsource_budget") or 0)
+ (item.get("task_plan_joint_operating_cost") or 0)
+ (item.get("exec_budget_labor_by_grade") or 0)
+ (item.get("exec_budget_outsource") or 0)
+ (item.get("exec_budget_cost_plan") or 0)
+ (item.get("expected_as_cost") or 0)
+ (item.get("expected_sga_budget") or 0)
)
item["actual_total_expense"] = (
(item.get("total_cost") or 0)
+ (item.get("total_sga") or 0)
)
return result
def get_monthly_summary() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT year,
month,
SUM(
CASE
WHEN accounting_category = '원가'
AND account_code NOT LIKE '5012%'
AND account_code NOT LIKE '5017%'
THEN amount
ELSE 0
END
) AS cost_sum,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_sum,
SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_sum,
SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_sum
FROM transactions
WHERE year IS NOT NULL
AND month IS NOT NULL
GROUP BY year, month
ORDER BY year, month
"""
)
).mappings().all()
return [dict(row) for row in rows]
def get_yearly_summary() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT year,
SUM(
CASE
WHEN accounting_category = '원가'
AND account_code NOT LIKE '5012%'
AND account_code NOT LIKE '5017%'
THEN amount
ELSE 0
END
) AS cost_sum,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_sum,
SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_sum,
SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_sum
FROM transactions
WHERE year IS NOT NULL
GROUP BY year
ORDER BY year
"""
)
).mappings().all()
return [dict(row) for row in rows]
def get_business_monthly_summary() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT year,
month,
support_dept_code,
support_dept_name,
SUM(CASE WHEN accounting_category = '원가' THEN amount ELSE 0 END) AS cost_sum,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_sum
FROM transactions
WHERE year IS NOT NULL
AND month IS NOT NULL
AND COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN (
'공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
)
GROUP BY year, month, support_dept_code, support_dept_name
ORDER BY year, month, support_dept_code, support_dept_name
"""
)
).mappings().all()
return [dict(row) for row in rows]
def get_project_status_rows() -> list[dict[str, Any]]:
contract_info_map = get_project_contract_info_map()
billing_summary_map = get_project_billing_summary_map()
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT b.support_dept_code,
b.support_dept_name,
b.row_count,
COALESCE(ps.progress_rate, 0) AS progress_rate,
COALESCE(ps.contract_amount, 0) AS contract_amount,
COALESCE(ps.collection_amount, 0) AS collection_amount,
COALESCE(ps.collection_entries_json, '[]') AS collection_entries_json,
COALESCE(ps.change_round, '') AS change_round,
COALESCE(ps.item_investment, 0) AS item_investment,
COALESCE(ps.task_plan_department_budget, 0) AS task_plan_department_budget,
COALESCE(ps.task_plan_outsource_budget, 0) AS task_plan_outsource_budget,
COALESCE(ps.task_plan_outsource_detail, '') AS task_plan_outsource_detail,
COALESCE(ps.task_plan_joint_operating_cost, 0) AS task_plan_joint_operating_cost,
COALESCE(ps.task_plan_entries_json, '[]') AS task_plan_entries_json,
COALESCE(ps.exec_budget_labor_by_grade, 0) AS exec_budget_labor_by_grade,
COALESCE(ps.exec_budget_outsource, 0) AS exec_budget_outsource,
COALESCE(ps.exec_budget_cost_plan, 0) AS exec_budget_cost_plan,
COALESCE(ps.exec_budget_entries_json, '[]') AS exec_budget_entries_json,
COALESCE(ps.actual_input_entries_json, '[]') AS actual_input_entries_json,
COALESCE(ps.expected_as_cost, 0) AS expected_as_cost,
COALESCE(ps.expected_sga_budget, 0) AS expected_sga_budget,
COALESCE(ps.project_start_date, '') AS project_start_date,
COALESCE(ps.project_end_date, '') AS project_end_date,
COALESCE(ps.completion_status, '') AS completion_status,
COALESCE(ps.notes, '') AS notes,
COALESCE(agg.total_cost, 0) AS total_cost,
COALESCE(agg.total_sga, 0) AS total_sga,
COALESCE(agg.total_revenue, 0) AS total_revenue,
COALESCE(agg.actual_labor, 0) AS actual_labor,
COALESCE(agg.actual_outsource, 0) AS actual_outsource,
COALESCE(agg.latest_year, 0) AS latest_year,
COALESCE(agg.latest_month, 0) AS latest_month
FROM (
SELECT support_dept_code, support_dept_name, COUNT(*) AS row_count
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN (
'공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
)
GROUP BY support_dept_code, support_dept_name
) AS b
LEFT JOIN project_status AS ps
ON ps.support_dept_code = b.support_dept_code
LEFT JOIN (
SELECT support_dept_code,
SUM(CASE WHEN accounting_category = '원가' THEN amount ELSE 0 END) AS total_cost,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS total_sga,
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS total_revenue,
SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS actual_labor,
SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS actual_outsource,
MAX(year) AS latest_year,
MAX(month) AS latest_month
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
GROUP BY support_dept_code
) AS agg
ON agg.support_dept_code = b.support_dept_code
ORDER BY b.support_dept_code, b.support_dept_name
"""
)
).mappings().all()
result = []
seen_codes: set[str] = set()
for row in rows:
item = dict(row)
item["collection_entries"] = decode_json_rows(item.pop("collection_entries_json", "[]"))
item["task_plan_entries"] = decode_json_rows(item.pop("task_plan_entries_json", "[]"))
item["exec_budget_entries"] = decode_json_rows(item.pop("exec_budget_entries_json", "[]"))
item["actual_input_entries"] = decode_json_rows(item.pop("actual_input_entries_json", "[]"))
item = merge_project_external_fields(
item,
contract_info_map.get(normalize_text(item.get("support_dept_code"))),
billing_summary_map.get(normalize_text(item.get("support_dept_code"))),
)
seen_codes.add(normalize_text(item.get("support_dept_code")))
result.append(item)
for support_dept_code in sorted((set(contract_info_map) | set(billing_summary_map)) - seen_codes):
result.append(
merge_project_external_fields(
{
"support_dept_code": support_dept_code,
"support_dept_name": "",
"row_count": 0,
"progress_rate": 0,
"contract_amount": 0,
"collection_amount": 0,
"collection_entries": [],
"change_round": "",
"item_investment": 0,
"task_plan_department_budget": 0,
"task_plan_outsource_budget": 0,
"task_plan_outsource_detail": "",
"task_plan_joint_operating_cost": 0,
"task_plan_entries": [],
"exec_budget_labor_by_grade": 0,
"exec_budget_outsource": 0,
"exec_budget_cost_plan": 0,
"exec_budget_entries": [],
"actual_input_entries": [],
"expected_as_cost": 0,
"expected_sga_budget": 0,
"project_start_date": "",
"project_end_date": "",
"completion_status": "",
"notes": "",
"total_cost": 0,
"total_sga": 0,
"total_revenue": 0,
"actual_labor": 0,
"actual_outsource": 0,
"latest_year": 0,
"latest_month": 0,
"project_type": "",
},
contract_info_map.get(support_dept_code),
billing_summary_map.get(support_dept_code),
)
)
return result
def get_project_status_for_edit(support_dept_code: str | None) -> dict[str, Any]:
contract_info_map = get_project_contract_info_map()
billing_summary_map = get_project_billing_summary_map()
if not support_dept_code:
return {
"support_dept_code": "",
"support_dept_name": "",
"progress_rate": "",
"contract_amount": "",
"collection_amount": "",
"collection_entries": [],
"change_round": "",
"item_investment": "",
"task_plan_department_budget": "",
"task_plan_outsource_budget": "",
"task_plan_outsource_detail": "",
"task_plan_joint_operating_cost": "",
"task_plan_entries": [],
"exec_budget_labor_by_grade": "",
"exec_labor_rates": {},
"exec_budget_outsource": "",
"exec_budget_cost_plan": "",
"exec_budget_entries": [],
"actual_input_entries": [],
"project_type": "",
"expected_as_rate": "",
"expected_sga_rate": "",
"expected_as_cost": "",
"expected_sga_budget": "",
"project_start_date": "",
"project_end_date": "",
"completion_status": "",
"notes": "",
"client_name": "",
"order_method": "",
"joint_contract": "",
"pm_name": "",
"contract_status": "",
"progress_status": "",
"work_category": "",
"review_tag": "",
"review_note": "",
"total_contract_amount": 0,
"hanmac_contract_amount": 0,
"billing_contract_amount": 0,
"billed_amount": 0,
"collection_balance_amount": 0,
"latest_billing_date": "",
"updated_at": "",
}
with engine.begin() as conn:
row = conn.execute(
text(
"""
SELECT b.support_dept_code,
b.support_dept_name,
COALESCE(ps.progress_rate, '') AS progress_rate,
COALESCE(ps.contract_amount, '') AS contract_amount,
COALESCE(ps.collection_amount, '') AS collection_amount,
COALESCE(ps.collection_entries_json, '[]') AS collection_entries_json,
COALESCE(ps.change_round, '') AS change_round,
COALESCE(ps.item_investment, '') AS item_investment,
COALESCE(ps.task_plan_department_budget, '') AS task_plan_department_budget,
COALESCE(ps.task_plan_outsource_budget, '') AS task_plan_outsource_budget,
COALESCE(ps.task_plan_outsource_detail, '') AS task_plan_outsource_detail,
COALESCE(ps.task_plan_joint_operating_cost, '') AS task_plan_joint_operating_cost,
COALESCE(ps.task_plan_entries_json, '[]') AS task_plan_entries_json,
COALESCE(ps.exec_budget_labor_by_grade, '') AS exec_budget_labor_by_grade,
COALESCE(ps.exec_labor_rates_json, '{}') AS exec_labor_rates_json,
COALESCE(ps.exec_budget_outsource, '') AS exec_budget_outsource,
COALESCE(ps.exec_budget_cost_plan, '') AS exec_budget_cost_plan,
COALESCE(ps.exec_budget_entries_json, '[]') AS exec_budget_entries_json,
COALESCE(ps.actual_input_entries_json, '[]') AS actual_input_entries_json,
COALESCE(ps.project_type, '') AS project_type,
COALESCE(ps.expected_as_rate, '') AS expected_as_rate,
COALESCE(ps.expected_sga_rate, '') AS expected_sga_rate,
COALESCE(ps.expected_as_cost, '') AS expected_as_cost,
COALESCE(ps.expected_sga_budget, '') AS expected_sga_budget,
COALESCE(ps.last_editor_session_id, '') AS last_editor_session_id,
COALESCE(ps.last_client_submitted_at, '') AS last_client_submitted_at,
COALESCE(ps.project_start_date, '') AS project_start_date,
COALESCE(ps.project_end_date, '') AS project_end_date,
COALESCE(ps.completion_status, '') AS completion_status,
COALESCE(ps.notes, '') AS notes,
COALESCE(ps.updated_at, '') AS updated_at
FROM (
SELECT DISTINCT support_dept_code, support_dept_name
FROM transactions
WHERE support_dept_code = :support_dept_code
) AS b
LEFT JOIN project_status AS ps
ON ps.support_dept_code = b.support_dept_code
"""
),
{"support_dept_code": support_dept_code},
).mappings().first()
if not row:
return {
"support_dept_code": support_dept_code,
"support_dept_name": "",
"progress_rate": "",
"contract_amount": "",
"collection_amount": "",
"collection_entries": [],
"change_round": "",
"item_investment": "",
"task_plan_department_budget": "",
"task_plan_outsource_budget": "",
"task_plan_outsource_detail": "",
"task_plan_joint_operating_cost": "",
"task_plan_entries": [],
"exec_budget_labor_by_grade": "",
"exec_labor_rates": {},
"exec_budget_outsource": "",
"exec_budget_cost_plan": "",
"exec_budget_entries": [],
"actual_input_entries": [],
"project_type": "",
"expected_as_rate": "",
"expected_sga_rate": "",
"expected_as_cost": "",
"expected_sga_budget": "",
"project_start_date": "",
"project_end_date": "",
"completion_status": "",
"notes": "",
"client_name": "",
"order_method": "",
"joint_contract": "",
"pm_name": "",
"contract_status": "",
"progress_status": "",
"work_category": "",
"review_tag": "",
"review_note": "",
"total_contract_amount": 0,
"hanmac_contract_amount": 0,
"billing_contract_amount": 0,
"billed_amount": 0,
"collection_balance_amount": 0,
"latest_billing_date": "",
"updated_at": "",
}
result = dict(row)
result["collection_entries"] = decode_json_rows(result.pop("collection_entries_json", "[]"))
result["task_plan_entries"] = decode_json_rows(result.pop("task_plan_entries_json", "[]"))
result["exec_budget_entries"] = decode_json_rows(result.pop("exec_budget_entries_json", "[]"))
result["actual_input_entries"] = decode_json_rows(result.pop("actual_input_entries_json", "[]"))
try:
result["exec_labor_rates"] = json.loads(normalize_text(result.pop("exec_labor_rates_json", "{}")) or "{}")
except json.JSONDecodeError:
result["exec_labor_rates"] = {}
if not result["collection_entries"] and normalize_amount(result.get("collection_amount")):
result["collection_entries"] = [
{
"vendor": "",
"round": "",
"amount": result.get("collection_amount", ""),
"date": "",
"due_date": "",
"note": "기존 수기 입력값",
}
]
if not result["task_plan_entries"]:
fallback_task_rows = []
if normalize_amount(result.get("task_plan_department_budget")):
fallback_task_rows.append(
{
"group": "department",
"dept_name": "기존 부서별 배분",
"work_name": "",
"amount": result.get("task_plan_department_budget", ""),
}
)
if normalize_amount(result.get("task_plan_outsource_budget")):
fallback_task_rows.append(
{
"group": "outsource",
"dept_name": "기존 외주비",
"work_name": result.get("task_plan_outsource_detail", ""),
"amount": result.get("task_plan_outsource_budget", ""),
}
)
if normalize_amount(result.get("task_plan_joint_operating_cost")):
fallback_task_rows.append(
{
"group": "joint",
"dept_name": "기존 합사운영비",
"work_name": "",
"amount": result.get("task_plan_joint_operating_cost", ""),
}
)
result["task_plan_entries"] = fallback_task_rows
if not result["exec_budget_entries"]:
fallback_exec_rows = []
if normalize_amount(result.get("exec_budget_labor_by_grade")):
fallback_exec_rows.append(
{
"group": "labor",
"grade": "기존 인건비",
"hours": "",
"amount": result.get("exec_budget_labor_by_grade", ""),
}
)
if normalize_amount(result.get("exec_budget_outsource")):
fallback_exec_rows.append(
{
"group": "outsource",
"dept_name": "기존 외주비",
"work_name": "",
"amount": result.get("exec_budget_outsource", ""),
}
)
if normalize_amount(result.get("exec_budget_cost_plan")):
fallback_exec_rows.append(
{
"group": "cost_plan",
"account_code": "기존",
"account_name": "비용계획",
"amount": result.get("exec_budget_cost_plan", ""),
}
)
result["exec_budget_entries"] = fallback_exec_rows
if not result["actual_input_entries"] and normalize_amount(result.get("item_investment")):
result["actual_input_entries"] = [
{
"reference": "",
"amount": result.get("item_investment", ""),
"note": "기존 항목별투입액",
}
]
return merge_project_external_fields(
result,
contract_info_map.get(normalize_text(result.get("support_dept_code"))),
billing_summary_map.get(normalize_text(result.get("support_dept_code"))),
)
def get_project_page_state() -> dict[str, Any]:
with engine.begin() as conn:
row = conn.execute(
text(
"""
SELECT COALESCE(selected_code, '') AS selected_code,
COALESCE(selected_year, '') AS selected_year,
COALESCE(analysis_open, 0) AS analysis_open,
COALESCE(related_project_selections_json, '{}') AS related_project_selections_json
FROM project_page_state
WHERE page_key = 'projects'
"""
)
).mappings().first()
if not row:
return {
"selected_code": "",
"selected_year": "",
"analysis_open": False,
"related_project_selections": {},
}
try:
related_project_selections_raw = json.loads(normalize_text(row["related_project_selections_json"]) or "{}")
except json.JSONDecodeError:
related_project_selections_raw = {}
related_project_selections = {}
if isinstance(related_project_selections_raw, dict):
related_project_selections = {
normalize_text(key): [
normalize_text(value)
for value in values
if normalize_text(value)
]
for key, values in related_project_selections_raw.items()
if normalize_text(key) and isinstance(values, list)
}
return {
"selected_code": normalize_text(row["selected_code"]),
"selected_year": normalize_text(row["selected_year"]),
"analysis_open": bool(row["analysis_open"]),
"related_project_selections": related_project_selections,
}
def save_project_page_state(payload: dict[str, Any]) -> None:
selected_code = normalize_text(payload.get("selected_code"))
selected_year = normalize_text(payload.get("selected_year"))
analysis_open = 1 if payload.get("analysis_open") else 0
raw_related = payload.get("related_project_selections") or {}
related_project_selections = {}
if isinstance(raw_related, dict):
related_project_selections = {
normalize_text(key): [
normalize_text(value)
for value in values
if normalize_text(value)
]
for key, values in raw_related.items()
if normalize_text(key) and isinstance(values, list)
}
with engine.begin() as conn:
conn.execute(
text(
"""
INSERT INTO project_page_state (
page_key,
selected_code,
selected_year,
analysis_open,
related_project_selections_json,
updated_at
) VALUES (
'projects',
:selected_code,
:selected_year,
:analysis_open,
:related_project_selections_json,
CURRENT_TIMESTAMP
)
ON CONFLICT(page_key) DO UPDATE SET
selected_code = excluded.selected_code,
selected_year = excluded.selected_year,
analysis_open = excluded.analysis_open,
related_project_selections_json = excluded.related_project_selections_json,
updated_at = CURRENT_TIMESTAMP
"""
),
{
"selected_code": selected_code,
"selected_year": selected_year,
"analysis_open": analysis_open,
"related_project_selections_json": json.dumps(related_project_selections, ensure_ascii=False),
},
)
for base_code, related_codes in related_project_selections.items():
save_project_related_links(base_code, related_codes)
def get_project_related_links_map() -> dict[str, list[str]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT base_support_dept_code, related_support_dept_code
FROM project_related_links
ORDER BY base_support_dept_code, related_support_dept_code
"""
)
).mappings().all()
related_map: dict[str, list[str]] = {}
for row in rows:
base_code = normalize_text(row["base_support_dept_code"])
related_code = normalize_text(row["related_support_dept_code"])
if not base_code or not related_code:
continue
related_map.setdefault(base_code, []).append(related_code)
return related_map
def save_project_related_links(base_support_dept_code: str, related_codes: list[Any]) -> None:
base_code = normalize_text(base_support_dept_code)
if not base_code:
return
normalized_codes = sorted(
{
normalize_text(code)
for code in related_codes
if normalize_text(code) and normalize_text(code) != base_code
}
)
with engine.begin() as conn:
conn.execute(
text(
"""
DELETE FROM project_related_links
WHERE base_support_dept_code = :base_support_dept_code
AND COALESCE(link_source, 'manual') = 'manual'
"""
),
{"base_support_dept_code": base_code},
)
for related_code in normalized_codes:
conn.execute(
text(
"""
INSERT INTO project_related_links (
base_support_dept_code,
related_support_dept_code,
link_source,
updated_at
) VALUES (
:base_support_dept_code,
:related_support_dept_code,
'manual',
CURRENT_TIMESTAMP
)
ON CONFLICT(base_support_dept_code, related_support_dept_code) DO UPDATE SET
link_source = excluded.link_source,
updated_at = CURRENT_TIMESTAMP
"""
),
{
"base_support_dept_code": base_code,
"related_support_dept_code": related_code,
},
)
def get_project_year_options() -> list[int]:
return get_available_years()
def resolve_selected_year(selected_year: int | None) -> int | None:
return selected_year
def parse_optional_year(value: Any) -> int | None:
text = normalize_text(value)
if not text:
return None
return int(text) if text.isdigit() else None
def get_recent_10_start_year() -> int | None:
available_years = get_available_years()
if not available_years:
return None
return max(available_years) - 9
def get_project_dashboard_summary(selected_year: int | None) -> dict[str, Any]:
selected_year = resolve_selected_year(selected_year)
project_year_clause = ""
collection_year_clause = ""
params: dict[str, Any] = {}
if selected_year:
project_year_clause = "AND p.year = :selected_year"
collection_year_clause = "AND year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
project_year_clause = "AND p.year >= :recent_10_start_year"
collection_year_clause = "AND year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
project_row = conn.execute(
text(
f"""
SELECT COUNT(*) AS related_projects,
SUM(COALESCE(p.expense_amount, 0)) AS expense_amount,
COUNT(CASE WHEN COALESCE(ps.completion_status, '') IN ('종료', '완료', 'Y', 'YES') THEN 1 END) AS completed_projects
FROM (
SELECT year,
support_dept_code,
support_dept_name,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
GROUP BY year, support_dept_code, support_dept_name
) AS p
LEFT JOIN project_status AS ps
ON ps.support_dept_code = p.support_dept_code
WHERE 1=1
{project_year_clause}
"""
),
params,
).mappings().first()
collection_row = conn.execute(
text(
f"""
SELECT COUNT(*) AS collection_transaction_count,
COUNT(DISTINCT support_dept_code) AS collection_project_count,
SUM(COALESCE(credit_supply, 0)) AS collection_amount
FROM transactions
WHERE COALESCE(credit_supply, 0) <> 0
AND ({REVENUE_SQL})
AND COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
{collection_year_clause}
"""
),
params,
).mappings().first()
input_row = conn.execute(
text(
"""
SELECT COUNT(CASE WHEN COALESCE(collection_amount, 0) <> 0 THEN 1 END) AS collection_input_projects,
COUNT(CASE WHEN COALESCE(completion_status, '') <> '' THEN 1 END) AS completion_input_projects
FROM project_status
"""
)
).mappings().first()
return {
**(dict(project_row) if project_row else {}),
**(dict(collection_row) if collection_row else {}),
**(dict(input_row) if input_row else {}),
}
def get_uncontracted_project_dashboard(selected_year: int | None) -> dict[str, Any]:
selected_year = resolve_selected_year(selected_year)
transaction_year_clause = ""
params: dict[str, Any] = {}
if selected_year:
transaction_year_clause = "AND t.year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
transaction_year_clause = "AND t.year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
summary = conn.execute(
text(
f"""
WITH project_universe AS (
SELECT DISTINCT support_dept_code, support_dept_name
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
UNION
SELECT support_dept_code, support_dept_name
FROM project_contract_info
WHERE COALESCE(support_dept_code, '') <> ''
UNION
SELECT support_dept_code, support_dept_name
FROM project_billing_entries
WHERE COALESCE(support_dept_code, '') <> ''
),
contract_flags AS (
SELECT support_dept_code,
COALESCE(hanmac_contract_amount, 0) AS hanmac_contract_amount,
COALESCE(review_tag, '') AS review_tag
FROM project_contract_info
),
project_amounts AS (
SELECT t.support_dept_code,
SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount
FROM transactions AS t
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
{transaction_year_clause}
GROUP BY t.support_dept_code
)
SELECT
SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 THEN 1 ELSE 0 END) AS uncontracted_projects,
SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 AND COALESCE(project_amounts.expense_amount, 0) > 0 THEN 1 ELSE 0 END) AS cost_incurred_projects,
SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 THEN COALESCE(project_amounts.expense_amount, 0) ELSE 0 END) AS expense_amount,
SUM(CASE WHEN COALESCE(contract_flags.hanmac_contract_amount, 0) <= 0 THEN COALESCE(project_amounts.revenue_amount, 0) ELSE 0 END) AS revenue_amount,
SUM(CASE WHEN COALESCE(contract_flags.review_tag, '') <> '' THEN 1 ELSE 0 END) AS review_needed_projects
FROM project_universe
LEFT JOIN contract_flags
ON contract_flags.support_dept_code = project_universe.support_dept_code
LEFT JOIN project_amounts
ON project_amounts.support_dept_code = project_universe.support_dept_code
"""
),
params,
).mappings().first()
yearly_rows = conn.execute(
text(
f"""
WITH yearly_costs AS (
SELECT t.year,
t.support_dept_code,
SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount
FROM transactions AS t
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
AND t.year IS NOT NULL
{transaction_year_clause}
GROUP BY t.year, t.support_dept_code
)
SELECT yearly_costs.year,
COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN yearly_costs.support_dept_code END) AS uncontracted_projects,
COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 AND COALESCE(yearly_costs.expense_amount, 0) > 0 THEN yearly_costs.support_dept_code END) AS cost_incurred_projects,
SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(yearly_costs.expense_amount, 0) ELSE 0 END) AS expense_amount,
SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(yearly_costs.revenue_amount, 0) ELSE 0 END) AS revenue_amount
FROM yearly_costs
LEFT JOIN project_contract_info AS c
ON c.support_dept_code = yearly_costs.support_dept_code
GROUP BY yearly_costs.year
ORDER BY yearly_costs.year
"""
),
params,
).mappings().all()
monthly_focus_year = selected_year
if monthly_focus_year is None:
monthly_focus_year = conn.execute(
text(
"""
WITH monthly_candidates AS (
SELECT MAX(t.year) AS latest_year
FROM transactions AS t
LEFT JOIN project_contract_info AS c
ON c.support_dept_code = t.support_dept_code
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(c.hanmac_contract_amount, 0) <= 0
AND (t.account_code LIKE '5%' OR t.account_code LIKE '6%')
)
SELECT latest_year FROM monthly_candidates
"""
)
).scalar()
monthly_rows: list[dict[str, Any]] = []
if monthly_focus_year:
monthly_rows = conn.execute(
text(
f"""
WITH monthly_costs AS (
SELECT t.month,
t.support_dept_code,
SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount
FROM transactions AS t
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
AND t.year = :monthly_focus_year
AND t.month IS NOT NULL
GROUP BY t.month, t.support_dept_code
)
SELECT monthly_costs.month,
COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN monthly_costs.support_dept_code END) AS uncontracted_projects,
COUNT(DISTINCT CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 AND COALESCE(monthly_costs.expense_amount, 0) > 0 THEN monthly_costs.support_dept_code END) AS cost_incurred_projects,
SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(monthly_costs.expense_amount, 0) ELSE 0 END) AS expense_amount,
SUM(CASE WHEN COALESCE(c.hanmac_contract_amount, 0) <= 0 THEN COALESCE(monthly_costs.revenue_amount, 0) ELSE 0 END) AS revenue_amount
FROM monthly_costs
LEFT JOIN project_contract_info AS c
ON c.support_dept_code = monthly_costs.support_dept_code
GROUP BY monthly_costs.month
ORDER BY monthly_costs.month
"""
),
{"monthly_focus_year": monthly_focus_year},
).mappings().all()
top_rows = conn.execute(
text(
f"""
WITH project_costs AS (
SELECT t.support_dept_code,
MAX(t.support_dept_name) AS support_dept_name,
SUM(CASE WHEN t.account_code LIKE '5%' OR t.account_code LIKE '6%' THEN t.amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL.replace('account_code', 't.account_code')} THEN t.amount ELSE 0 END) AS revenue_amount,
MAX(t.year) AS latest_year,
MAX(t.month) AS latest_month
FROM transactions AS t
WHERE COALESCE(t.support_dept_code, '') <> ''
AND t.support_dept_code NOT IN ('ZZZZZZ')
{transaction_year_clause}
GROUP BY t.support_dept_code
)
SELECT project_costs.support_dept_code,
project_costs.support_dept_name,
project_costs.expense_amount,
project_costs.revenue_amount,
project_costs.latest_year,
project_costs.latest_month,
COALESCE(c.review_tag, '') AS review_tag
FROM project_costs
LEFT JOIN project_contract_info AS c
ON c.support_dept_code = project_costs.support_dept_code
WHERE COALESCE(c.hanmac_contract_amount, 0) <= 0
AND COALESCE(project_costs.expense_amount, 0) > 0
ORDER BY project_costs.expense_amount DESC, project_costs.support_dept_code
LIMIT 12
"""
),
params,
).mappings().all()
return {
"summary": dict(summary) if summary else {},
"yearly_rows": [dict(row) for row in yearly_rows],
"monthly_rows": [dict(row) for row in monthly_rows],
"monthly_focus_year": int(monthly_focus_year) if monthly_focus_year else None,
"top_rows": [dict(row) for row in top_rows],
}
def get_project_revenue_mix(selected_year: int | None = None) -> list[dict[str, Any]]:
params: dict[str, Any] = {}
if selected_year:
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT
year,
month,
SUM(CASE WHEN account_code LIKE '40110101%' AND memo1 LIKE '%설계%' THEN amount ELSE 0 END) AS design_revenue,
SUM(CASE WHEN account_code LIKE '40110101%' AND (memo1 NOT LIKE '%설계%' OR COALESCE(memo1, '') = '') THEN amount ELSE 0 END) AS design_other_revenue,
SUM(CASE WHEN account_code LIKE '40110102%' THEN amount ELSE 0 END) AS supervision_revenue,
SUM(CASE WHEN account_code LIKE '40110103%' THEN amount ELSE 0 END) AS inspection_revenue
FROM transactions
WHERE year = :selected_year
AND month IS NOT NULL
GROUP BY year, month
ORDER BY year, month
"""
),
{"selected_year": selected_year},
).mappings().all()
result = [dict(row) for row in rows]
for item in result:
item["label"] = f"{int(item['month'])}월" if item.get("month") is not None else str(item.get("year", ""))
return result
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT
year,
SUM(CASE WHEN account_code LIKE '40110101%' AND memo1 LIKE '%설계%' THEN amount ELSE 0 END) AS design_revenue,
SUM(CASE WHEN account_code LIKE '40110101%' AND (memo1 NOT LIKE '%설계%' OR COALESCE(memo1, '') = '') THEN amount ELSE 0 END) AS design_other_revenue,
SUM(CASE WHEN account_code LIKE '40110102%' THEN amount ELSE 0 END) AS supervision_revenue,
SUM(CASE WHEN account_code LIKE '40110103%' THEN amount ELSE 0 END) AS inspection_revenue
FROM transactions
WHERE year IS NOT NULL
GROUP BY year
ORDER BY year
"""
)
).mappings().all()
result = [dict(row) for row in rows]
for item in result:
item["label"] = str(item.get("year", ""))
return result[-10:]
def get_project_revenue_mix_monthly() -> list[dict[str, Any]]:
recent_10_start_year = get_recent_10_start_year()
params: dict[str, Any] = {}
year_clause = ""
if recent_10_start_year is not None:
year_clause = "AND year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT
year,
month,
SUM(CASE WHEN account_code LIKE '40110101%' AND memo1 LIKE '%설계%' THEN amount ELSE 0 END) AS design_revenue,
SUM(CASE WHEN account_code LIKE '40110101%' AND (memo1 NOT LIKE '%설계%' OR COALESCE(memo1, '') = '') THEN amount ELSE 0 END) AS design_other_revenue,
SUM(CASE WHEN account_code LIKE '40110102%' THEN amount ELSE 0 END) AS supervision_revenue,
SUM(CASE WHEN account_code LIKE '40110103%' THEN amount ELSE 0 END) AS inspection_revenue
FROM transactions
WHERE month IS NOT NULL
{year_clause}
GROUP BY year, month
ORDER BY year, month
"""
),
params,
).mappings().all()
result = [dict(row) for row in rows]
for item in result:
item["label"] = f"{int(item['month'])}월" if item.get("month") is not None else str(item.get("year", ""))
return result
def get_project_cost_by_year(selected_year: int | None) -> list[dict[str, Any]]:
year_clause = ""
params: dict[str, Any] = {}
if selected_year:
year_clause = "AND year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
year_clause = "AND year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
contract_info_map = get_project_contract_info_map()
billing_summary_map = get_project_billing_summary_map()
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT year,
support_dept_code,
support_dept_name,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS revenue_amount
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
{year_clause}
GROUP BY year, support_dept_code, support_dept_name
ORDER BY year DESC, expense_amount DESC, support_dept_code
"""
),
params,
).mappings().all()
result = [dict(row) for row in rows]
existing_codes = {normalize_text(row["support_dept_code"]) for row in result}
candidate_codes = sorted((set(contract_info_map) | set(billing_summary_map)) - existing_codes)
recent_10_start_year = get_recent_10_start_year()
for support_dept_code in candidate_codes:
contract_info = contract_info_map.get(support_dept_code, {})
billing_summary = billing_summary_map.get(support_dept_code, {})
fallback_date = (
normalize_text(contract_info.get("project_start_date"))
or normalize_text(contract_info.get("contract_date"))
or normalize_text(billing_summary.get("latest_billing_date"))
)
fallback_year = 0
if re.match(r"^\d{4}-\d{2}-\d{2}$", fallback_date):
fallback_year = int(fallback_date[:4])
if selected_year and fallback_year and fallback_year != selected_year:
continue
if not selected_year and recent_10_start_year is not None and fallback_year and fallback_year < recent_10_start_year:
continue
result.append(
{
"year": fallback_year,
"support_dept_code": support_dept_code,
"support_dept_name": normalize_text(contract_info.get("support_dept_name")) or normalize_text(billing_summary.get("support_dept_name")),
"expense_amount": 0,
"revenue_amount": normalize_amount(billing_summary.get("collected_amount")),
}
)
result.sort(key=lambda item: (-(int(item.get("year") or 0)), -normalize_amount(item.get("expense_amount")), normalize_text(item.get("support_dept_code"))))
return result
def get_project_account_breakdowns(selected_year: int | None) -> dict[str, dict[str, list[dict[str, Any]]]]:
year_clause = ""
params: dict[str, Any] = {}
if selected_year:
year_clause = "AND year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
year_clause = "AND year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT support_dept_code,
CASE
WHEN {REVENUE_SQL} THEN 'revenue'
WHEN accounting_category = '원가' THEN 'cost'
WHEN accounting_category = '판관비' THEN 'sga'
ELSE 'other'
END AS breakdown_kind,
COALESCE(account_code, '') AS account_code,
COALESCE(account_name, '') AS account_name,
SUM(COALESCE(amount, 0)) AS total_amount
FROM transactions
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
AND ({REVENUE_SQL} OR accounting_category IN ('원가', '판관비'))
{year_clause}
GROUP BY support_dept_code, breakdown_kind, account_code, account_name
ORDER BY support_dept_code, breakdown_kind, total_amount DESC, account_code, account_name
"""
),
params,
).mappings().all()
result: dict[str, dict[str, dict[str, float]]] = {}
for row in rows:
code = row["support_dept_code"]
kind = row["breakdown_kind"]
if kind == "other":
continue
_, _, label = normalize_account_display(row["account_code"], row["account_name"])
result.setdefault(code, {"revenue": {}, "cost": {}, "sga": {}})
result[code][kind][label] = result[code][kind].get(label, 0.0) + float(row["total_amount"] or 0)
normalized_result: dict[str, dict[str, list[dict[str, Any]]]] = {}
for code, buckets in result.items():
normalized_result[code] = {}
for kind, entries in buckets.items():
normalized_result[code][kind] = [
{"label": label, "amount": amount}
for label, amount in sorted(entries.items(), key=lambda item: item[1], reverse=True)
]
return normalized_result
def get_recent_transactions(limit: int = 50) -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT id,
year,
month,
voucher_number,
account_code,
account_name,
support_dept_code,
support_dept_name,
cost_dept_code,
cost_dept_name,
accounting_category,
amount,
memo1,
management_item,
source_file,
updated_at
FROM transactions
ORDER BY COALESCE(year, 0) DESC,
COALESCE(month, 0) DESC,
id DESC
LIMIT :limit_count
"""
),
{"limit_count": limit},
).mappings().all()
return [dict(row) for row in rows]
def get_overview_stats(selected_year: int | None = None) -> dict[str, Any]:
year_clause = ""
params: dict[str, Any] = {}
if selected_year:
year_clause = "WHERE year = :selected_year"
params["selected_year"] = selected_year
else:
recent_10_start_year = get_recent_10_start_year()
if recent_10_start_year is not None:
year_clause = "WHERE year >= :recent_10_start_year"
params["recent_10_start_year"] = recent_10_start_year
with engine.begin() as conn:
row = conn.execute(
text(
f"""
SELECT COUNT(*) AS total_rows,
COUNT(DISTINCT source_file) AS source_files,
COUNT(DISTINCT CASE
WHEN COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
AND COALESCE(support_dept_name, '') <> ''
AND support_dept_name NOT IN (
'공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실'
)
THEN support_dept_code || '|' || support_dept_name
END) AS business_count,
SUM(CASE WHEN accounting_category = '원가' THEN amount ELSE 0 END) AS total_cost,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS total_sga,
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS total_revenue
FROM transactions
{year_clause}
"""
),
params,
).mappings().first()
return dict(row) if row else {}
def get_available_years() -> list[int]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT year
FROM transactions
WHERE year IS NOT NULL
ORDER BY year
"""
)
).fetchall()
return [int(row[0]) for row in rows if row[0] is not None]
def get_financial_series(granularity: str) -> list[dict[str, Any]]:
group_fields = "year" if granularity == "yearly" else "year, month"
order_fields = "year" if granularity == "yearly" else "year, month"
month_where = "" if granularity == "yearly" else "AND month IS NOT NULL"
project_cost_sql = (
"accounting_category = '원가' "
"AND account_code NOT LIKE '5012%' "
"AND account_code NOT LIKE '5017%' "
"AND support_dept_code <> 'ZZZZZZ' "
f"AND {FIELD_COST_DEPT_SQL}"
)
support_cost_sql = (
"accounting_category = '원가' "
"AND account_code NOT LIKE '5012%' "
"AND account_code NOT LIKE '5017%' "
"AND support_dept_code = 'ZZZZZZ' "
f"AND {FIELD_COST_DEPT_SQL}"
)
support_sga_sql = f"accounting_category = '판관비' AND {SUPPORT_COST_DEPT_SQL}"
field_sga_sql = f"accounting_category = '판관비' AND {FIELD_COST_DEPT_SQL}"
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT {group_fields},
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS revenue_sum,
SUM(CASE WHEN {project_cost_sql} THEN amount ELSE 0 END) AS project_cost_sum,
SUM(CASE WHEN {support_cost_sql} THEN amount ELSE 0 END) AS support_cost_sum,
SUM(CASE WHEN {support_sga_sql} THEN amount ELSE 0 END) AS support_sga_sum,
SUM(CASE WHEN {field_sga_sql} THEN amount ELSE 0 END) AS field_sga_sum,
SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_sum,
SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_sum
FROM transactions
WHERE year IS NOT NULL
{month_where}
GROUP BY {group_fields}
ORDER BY {order_fields}
"""
)
).mappings().all()
result: list[dict[str, Any]] = []
for row in rows:
item = dict(row)
item["total_expense"] = (
(item.get("project_cost_sum") or 0)
+ (item.get("support_cost_sum") or 0)
+ (item.get("support_sga_sum") or 0)
+ (item.get("field_sga_sum") or 0)
+ (item.get("labor_sum") or 0)
+ (item.get("outsourcing_sum") or 0)
)
item["operating_balance"] = (item.get("revenue_sum") or 0) - item["total_expense"]
item["label"] = str(item["year"]) if granularity == "yearly" else f"{item['year']}-{int(item['month']):02d}"
result.append(item)
return result
def get_source_files_summary() -> list[dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT source_file, COUNT(*) AS row_count
FROM transactions
WHERE COALESCE(source_file, '') <> ''
GROUP BY source_file
ORDER BY row_count DESC, source_file
"""
)
).mappings().all()
return [dict(row) for row in rows]
def parse_excel_upload(upload_file: UploadFile) -> int:
init_db()
workbook = load_workbook(upload_file.file, data_only=True)
import_kind = detect_excel_import_kind(workbook, upload_file.filename or "")
if import_kind == "contract_status":
return import_contract_status_workbook(workbook, upload_file.filename or "")
if import_kind == "billing_status":
return import_billing_status_workbook(workbook, upload_file.filename or "")
sheet = workbook.active
headers = [canonical_header_name(cell.value) for cell in next(sheet.iter_rows(min_row=1, max_row=1))]
inserted = 0
for row in sheet.iter_rows(min_row=2, values_only=True):
raw: dict[str, Any] = {}
has_value = False
for index, value in enumerate(row):
field_name = headers[index] if index < len(headers) else None
if field_name:
raw[field_name] = value
if normalize_text(value):
has_value = True
if not has_value:
continue
payload = build_transaction_payload(raw, source_file=upload_file.filename or "")
if not payload["voucher_number"] and not payload["account_code"] and not payload["account_name"]:
continue
save_transaction(payload)
inserted += 1
return inserted
def import_excel_path(path: Path) -> int:
with path.open("rb") as excel_file:
upload = UploadFile(filename=path.name, file=excel_file)
return parse_excel_upload(upload)
def auto_import_project_excels() -> None:
init_db()
excel_files = sorted(BASE_DIR.glob("*.xlsx"))
if not excel_files:
return
known_files = existing_source_files()
known_contract_files = existing_contract_source_files()
known_billing_files = existing_billing_source_files()
if count_transactions() > 0 and all(file.name in known_files for file in excel_files):
if all(file.name in known_contract_files or file.name in known_billing_files for file in excel_files):
return
for excel_path in excel_files:
workbook = load_workbook(excel_path, data_only=True)
import_kind = detect_excel_import_kind(workbook, excel_path.name)
if import_kind == "contract_status" and excel_path.name in known_contract_files:
continue
if import_kind == "billing_status" and excel_path.name in known_billing_files:
continue
if import_kind == "transactions" and excel_path.name in known_files:
continue
with excel_path.open("rb") as excel_file:
upload = UploadFile(filename=excel_path.name, file=excel_file)
inserted = parse_excel_upload(upload)
logger.info("Auto-imported %s rows from %s", inserted, excel_path.name)
def parse_manual_form(raw_body: bytes) -> dict[str, Any]:
parsed = parse_qs(raw_body.decode("utf-8"))
payload = {key: values[0] if values else "" for key, values in parsed.items()}
return payload
def parse_project_form(raw_body: bytes) -> dict[str, Any]:
parsed = parse_qs(raw_body.decode("utf-8"))
payload: dict[str, Any] = {}
for key, values in parsed.items():
payload[key] = values if key.endswith("[]") else (values[0] if values else "")
return payload
def build_named_amount_rows(
labels: list[Any],
amounts: list[Any],
*,
label_key: str = "label",
amount_key: str = "amount",
) -> list[dict[str, Any]]:
rows = []
for index, label in enumerate(labels):
rows.append(
{
label_key: label,
amount_key: amounts[index] if index < len(amounts) else "",
}
)
return filter_amount_rows(rows, amount_key=amount_key)
def build_triplet_amount_rows(
first_values: list[Any],
second_values: list[Any],
amounts: list[Any],
*,
first_key: str,
second_key: str,
amount_key: str = "amount",
) -> list[dict[str, Any]]:
max_length = max(len(first_values), len(second_values), len(amounts))
rows = []
for index in range(max_length):
rows.append(
{
first_key: first_values[index] if index < len(first_values) else "",
second_key: second_values[index] if index < len(second_values) else "",
amount_key: amounts[index] if index < len(amounts) else "",
}
)
return filter_amount_rows(rows, amount_key=amount_key)
def build_collection_rows(payload: dict[str, Any]) -> list[dict[str, Any]]:
progress_types = payload.get("collection_progress_type[]", [])
billing_rounds = payload.get("collection_billing_round[]", [])
billing_types = payload.get("collection_billing_type[]", [])
billing_dates = payload.get("collection_billing_date[]", [])
billed_amounts = payload.get("collection_billed_amount[]", [])
collection_rounds = payload.get("collection_round[]", [])
collection_dates = payload.get("collection_date[]", [])
collection_amounts = payload.get("collection_amount_row[]", [])
rows = []
total_rows = max(
len(progress_types),
len(billing_rounds),
len(billing_types),
len(billing_dates),
len(billed_amounts),
len(collection_rounds),
len(collection_dates),
len(collection_amounts),
)
for index in range(total_rows):
rows.append(
{
"progress_type": progress_types[index] if index < len(progress_types) else "",
"billing_round": billing_rounds[index] if index < len(billing_rounds) else "",
"billing_type": billing_types[index] if index < len(billing_types) else "",
"billing_date": billing_dates[index] if index < len(billing_dates) else "",
"billed_amount": billed_amounts[index] if index < len(billed_amounts) else "",
"round": collection_rounds[index] if index < len(collection_rounds) else "",
"date": collection_dates[index] if index < len(collection_dates) else "",
"amount": collection_amounts[index] if index < len(collection_amounts) else "",
}
)
filtered_rows: list[dict[str, Any]] = []
for row in rows:
normalized_row = {key: clean_row_text(value) for key, value in row.items()}
amount = normalize_amount(normalized_row.get("amount"))
billed_amount = normalize_amount(normalized_row.get("billed_amount"))
has_other_value = any(
value for key, value in normalized_row.items()
if key not in {"amount", "billed_amount"}
)
if amount or billed_amount or has_other_value:
normalized_row["amount"] = amount
normalized_row["billed_amount"] = billed_amount
filtered_rows.append(normalized_row)
for row in filtered_rows:
row["billing_round"] = normalize_round_value(row.get("billing_round"))
row["round"] = normalize_round_value(row.get("round"))
row["billing_date"] = normalize_date_text(row.get("billing_date"))
row["date"] = normalize_date_text(row.get("date"))
row["billed_amount"] = normalize_amount(row.get("billed_amount"))
return filtered_rows
def build_project_status_payload(payload: dict[str, Any]) -> dict[str, Any]:
contract_amount = normalize_amount(payload.get("contract_amount"))
collection_rows = build_collection_rows(payload)
collection_amount = sum_row_amounts(collection_rows)
progress_rate = (collection_amount / contract_amount * 100) if contract_amount else 0.0
task_plan_department_rows = build_triplet_amount_rows(
payload.get("task_plan_department_dept[]", []),
payload.get("task_plan_department_work[]", []),
payload.get("task_plan_department_amount[]", []),
first_key="dept_name",
second_key="work_name",
)
for row in task_plan_department_rows:
row["group"] = "department"
task_plan_outsource_rows = build_triplet_amount_rows(
payload.get("task_plan_outsource_dept[]", []),
payload.get("task_plan_outsource_work[]", []),
payload.get("task_plan_outsource_amount[]", []),
first_key="dept_name",
second_key="work_name",
)
for row in task_plan_outsource_rows:
row["group"] = "outsource"
task_plan_joint_rows = build_triplet_amount_rows(
payload.get("task_plan_joint_dept[]", []),
payload.get("task_plan_joint_work[]", []),
payload.get("task_plan_joint_amount[]", []),
first_key="dept_name",
second_key="work_name",
)
for row in task_plan_joint_rows:
row["group"] = "joint"
task_plan_rows = task_plan_department_rows + task_plan_outsource_rows + task_plan_joint_rows
exec_labor_rows = build_triplet_amount_rows(
payload.get("exec_labor_grade[]", []),
payload.get("exec_labor_hours[]", []),
payload.get("exec_labor_amount[]", []),
first_key="grade",
second_key="hours",
)
for row in exec_labor_rows:
row["group"] = "labor"
exec_outsource_rows = build_triplet_amount_rows(
payload.get("exec_outsource_dept[]", []),
payload.get("exec_outsource_work[]", []),
payload.get("exec_outsource_amount[]", []),
first_key="dept_name",
second_key="work_name",
)
for row in exec_outsource_rows:
row["group"] = "outsource"
exec_cost_plan_rows = build_triplet_amount_rows(
payload.get("exec_cost_plan_code[]", []),
payload.get("exec_cost_plan_name[]", []),
payload.get("exec_cost_plan_amount[]", []),
first_key="account_code",
second_key="account_name",
)
for row in exec_cost_plan_rows:
row["group"] = "cost_plan"
exec_budget_rows = exec_labor_rows + exec_outsource_rows + exec_cost_plan_rows
actual_labor_grades = payload.get("actual_labor_grade[]", [])
actual_labor_minutes = payload.get("actual_labor_minutes[]", [])
actual_labor_amounts = payload.get("actual_labor_amount[]", [])
actual_labor_rows: list[dict[str, Any]] = []
actual_labor_max_length = max(
len(actual_labor_grades),
len(actual_labor_minutes),
len(actual_labor_amounts),
)
for index in range(actual_labor_max_length):
row = {
"grade": actual_labor_grades[index] if index < len(actual_labor_grades) else "",
"minutes": actual_labor_minutes[index] if index < len(actual_labor_minutes) else "",
"amount": actual_labor_amounts[index] if index < len(actual_labor_amounts) else "",
}
normalized_row = {key: clean_row_text(value) for key, value in row.items()}
amount = normalize_amount(normalized_row.get("amount"))
has_other_value = any(
value for key, value in normalized_row.items()
if key != "amount"
)
if amount or has_other_value:
normalized_row["amount"] = amount
actual_labor_rows.append(normalized_row)
for row in actual_labor_rows:
row["group"] = "labor"
actual_labor_adjustment_total = normalize_amount(payload.get("actual_labor_adjustment_total"))
actual_labor_adjustment_rows = []
if actual_labor_adjustment_total:
actual_labor_adjustment_rows.append(
{
"group": "labor_adjustment",
"label": "인건비 조정",
"amount": actual_labor_adjustment_total,
}
)
actual_as_rows = build_named_amount_rows(
payload.get("actual_as_label[]", []),
payload.get("actual_as_amount[]", []),
label_key="label",
amount_key="amount",
)
for row in actual_as_rows:
row["group"] = "as"
actual_labor_joint_rows = build_named_amount_rows(
payload.get("actual_labor_joint_label[]", []),
payload.get("actual_labor_joint_amount[]", []),
label_key="label",
amount_key="amount",
)
for row in actual_labor_joint_rows:
row["group"] = "labor_joint"
actual_sga_rows = build_named_amount_rows(
payload.get("actual_sga_label[]", []),
payload.get("actual_sga_amount[]", []),
label_key="label",
amount_key="amount",
)
for row in actual_sga_rows:
row["group"] = "sga"
actual_input_rows = actual_labor_rows + actual_labor_adjustment_rows + actual_labor_joint_rows + actual_as_rows + actual_sga_rows
if not actual_input_rows:
legacy_refs = payload.get("actual_input_ref[]", [])
legacy_amounts = payload.get("actual_input_amount[]", [])
legacy_notes = payload.get("actual_input_note[]", [])
for index, ref in enumerate(legacy_refs):
actual_input_rows.append(
{
"reference": ref,
"amount": legacy_amounts[index] if index < len(legacy_amounts) else "",
"note": legacy_notes[index] if index < len(legacy_notes) else "",
}
)
actual_input_rows = filter_amount_rows(actual_input_rows, amount_key="amount")
expected_as_rate = normalize_amount(payload.get("expected_as_rate"))
expected_sga_rate = normalize_amount(payload.get("expected_sga_rate"))
expected_as_cost = contract_amount * expected_as_rate / 100 if contract_amount else 0.0
expected_sga_budget = contract_amount * expected_sga_rate / 100 if contract_amount else 0.0
exec_labor_rates = normalize_text(payload.get("exec_labor_rates_json")) or "{}"
return {
"support_dept_code": normalize_text(payload.get("support_dept_code")),
"support_dept_name": normalize_text(payload.get("support_dept_name")),
"progress_rate": progress_rate,
"contract_amount": contract_amount,
"collection_amount": collection_amount,
"collection_entries_json": encode_json_rows(collection_rows),
"change_round": normalize_text(payload.get("change_round")),
"item_investment": sum_row_amounts(actual_input_rows),
"task_plan_department_budget": sum_row_amounts(task_plan_department_rows),
"task_plan_outsource_budget": sum_row_amounts(task_plan_outsource_rows),
"task_plan_outsource_detail": "\n".join(
f"{normalize_text(row.get('dept_name'))} / {normalize_text(row.get('work_name'))}: {format_amount_for_text(row.get('amount'))}".strip(" /:")
for row in task_plan_outsource_rows
),
"task_plan_joint_operating_cost": sum_row_amounts(task_plan_joint_rows),
"task_plan_entries_json": encode_json_rows(task_plan_rows),
"exec_budget_labor_by_grade": sum_row_amounts(exec_labor_rows),
"exec_labor_rates_json": exec_labor_rates,
"exec_budget_outsource": sum_row_amounts(exec_outsource_rows),
"exec_budget_cost_plan": sum_row_amounts(exec_cost_plan_rows),
"exec_budget_entries_json": encode_json_rows(exec_budget_rows),
"actual_input_entries_json": encode_json_rows(actual_input_rows),
"project_type": normalize_text(payload.get("project_type")),
"expected_as_rate": expected_as_rate,
"expected_sga_rate": expected_sga_rate,
"expected_as_cost": expected_as_cost,
"expected_sga_budget": expected_sga_budget,
"last_editor_session_id": normalize_text(payload.get("client_session_id")),
"last_client_submitted_at": normalize_text(payload.get("client_submitted_at")),
"project_start_date": normalize_date_text(payload.get("project_start_date")),
"project_end_date": normalize_date_text(payload.get("project_end_date")),
"completion_status": normalize_text(payload.get("completion_status")),
"notes": normalize_text(payload.get("notes")),
}
def project_status_payload_has_meaningful_data(payload: dict[str, Any]) -> bool:
if normalize_amount(payload.get("contract_amount")):
return True
if normalize_amount(payload.get("collection_amount")):
return True
if normalize_amount(payload.get("task_plan_department_budget")):
return True
if normalize_amount(payload.get("task_plan_outsource_budget")):
return True
if normalize_amount(payload.get("task_plan_joint_operating_cost")):
return True
if normalize_amount(payload.get("exec_budget_labor_by_grade")):
return True
if normalize_amount(payload.get("exec_budget_outsource")):
return True
if normalize_amount(payload.get("exec_budget_cost_plan")):
return True
if normalize_amount(payload.get("item_investment")):
return True
for key in (
"project_type",
"project_start_date",
"project_end_date",
"completion_status",
"notes",
"change_round",
"support_dept_name",
"task_plan_outsource_detail",
):
if normalize_text(payload.get(key)):
return True
for key in (
"collection_entries_json",
"task_plan_entries_json",
"exec_budget_entries_json",
"actual_input_entries_json",
):
if decode_json_rows(payload.get(key)):
return True
return False
def format_amount_for_text(value: Any) -> str:
amount = normalize_amount(value)
return f"{amount:,.0f}"
def save_project_status(payload: dict[str, Any]) -> None:
normalized_payload = build_project_status_payload(payload)
support_dept_code = normalize_text(normalized_payload.get("support_dept_code"))
if not support_dept_code:
return
with engine.begin() as conn:
existing_row = conn.execute(
text("SELECT * FROM project_status WHERE support_dept_code = :support_dept_code"),
{"support_dept_code": support_dept_code},
).mappings().first()
if existing_row:
existing_payload = dict(existing_row)
if (
project_status_payload_has_meaningful_data(existing_payload)
and not project_status_payload_has_meaningful_data(normalized_payload)
):
raise ValueError("기존 입력값을 불러오지 않은 빈 상태로는 저장할 수 없습니다.")
check_record_revision(
conn,
"project_status",
"support_dept_code",
support_dept_code,
normalize_text(payload.get("edit_revision")),
)
conn.execute(
text(
"""
INSERT INTO project_status (
support_dept_code,
support_dept_name,
progress_rate,
contract_amount,
collection_amount,
collection_entries_json,
change_round,
item_investment,
task_plan_department_budget,
task_plan_outsource_budget,
task_plan_outsource_detail,
task_plan_joint_operating_cost,
task_plan_entries_json,
exec_budget_labor_by_grade,
exec_labor_rates_json,
exec_budget_outsource,
exec_budget_cost_plan,
exec_budget_entries_json,
actual_input_entries_json,
project_type,
expected_as_rate,
expected_sga_rate,
expected_as_cost,
expected_sga_budget,
last_editor_session_id,
last_client_submitted_at,
project_start_date,
project_end_date,
completion_status,
notes,
updated_at
) VALUES (
:support_dept_code,
:support_dept_name,
:progress_rate,
:contract_amount,
:collection_amount,
:collection_entries_json,
:change_round,
:item_investment,
:task_plan_department_budget,
:task_plan_outsource_budget,
:task_plan_outsource_detail,
:task_plan_joint_operating_cost,
:task_plan_entries_json,
:exec_budget_labor_by_grade,
:exec_labor_rates_json,
:exec_budget_outsource,
:exec_budget_cost_plan,
:exec_budget_entries_json,
:actual_input_entries_json,
:project_type,
:expected_as_rate,
:expected_sga_rate,
:expected_as_cost,
:expected_sga_budget,
:last_editor_session_id,
:last_client_submitted_at,
:project_start_date,
:project_end_date,
:completion_status,
:notes,
CURRENT_TIMESTAMP
)
ON CONFLICT(support_dept_code) DO UPDATE SET
support_dept_name = excluded.support_dept_name,
progress_rate = excluded.progress_rate,
contract_amount = excluded.contract_amount,
collection_amount = excluded.collection_amount,
collection_entries_json = excluded.collection_entries_json,
change_round = excluded.change_round,
item_investment = excluded.item_investment,
task_plan_department_budget = excluded.task_plan_department_budget,
task_plan_outsource_budget = excluded.task_plan_outsource_budget,
task_plan_outsource_detail = excluded.task_plan_outsource_detail,
task_plan_joint_operating_cost = excluded.task_plan_joint_operating_cost,
task_plan_entries_json = excluded.task_plan_entries_json,
exec_budget_labor_by_grade = excluded.exec_budget_labor_by_grade,
exec_labor_rates_json = excluded.exec_labor_rates_json,
exec_budget_outsource = excluded.exec_budget_outsource,
exec_budget_cost_plan = excluded.exec_budget_cost_plan,
exec_budget_entries_json = excluded.exec_budget_entries_json,
actual_input_entries_json = excluded.actual_input_entries_json,
project_type = excluded.project_type,
expected_as_rate = excluded.expected_as_rate,
expected_sga_rate = excluded.expected_sga_rate,
expected_as_cost = excluded.expected_as_cost,
expected_sga_budget = excluded.expected_sga_budget,
last_editor_session_id = excluded.last_editor_session_id,
last_client_submitted_at = excluded.last_client_submitted_at,
project_start_date = excluded.project_start_date,
project_end_date = excluded.project_end_date,
completion_status = excluded.completion_status,
notes = excluded.notes,
updated_at = CURRENT_TIMESTAMP
"""
),
normalized_payload,
)
def base_context(request: Request, message: str = "") -> dict[str, Any]:
health_payload = build_health_payload()
return {
"request": request,
"message": message,
"data_version": health_payload["data_version"],
"server_time": health_payload["server_time"],
"import_sync_summary": get_import_sync_summary(),
}
def render_home(
request: Request,
edit_id: int | None = None,
message: str = "",
overview_year: int | None = None,
) -> HTMLResponse:
init_db()
available_years = get_available_years()
context = {
**base_context(request, message),
"overview": get_overview_stats(overview_year),
"overview_selected_year": overview_year,
"project_dashboard": get_project_dashboard_summary(overview_year),
"project_revenue_mix_yearly": get_project_revenue_mix(),
"project_revenue_mix_monthly": get_project_revenue_mix_monthly(),
"yearly_summary": get_yearly_summary(),
"monthly_summary": get_monthly_summary(),
"available_years": available_years,
}
return templates.TemplateResponse(request, "dashboard.html", context)
def render_projects_page(
request: Request,
edit_code: str | None = None,
selected_year: int | None = None,
message: str = "",
) -> HTMLResponse:
init_db()
selected_year = resolve_selected_year(selected_year)
context = {
**base_context(request, message),
"project_year_options": get_project_year_options(),
"selected_year": selected_year,
"project_dashboard": get_project_dashboard_summary(selected_year),
"project_revenue_mix": get_project_revenue_mix(selected_year),
"project_cost_by_year": get_project_cost_by_year(selected_year),
"project_monthly_cost_rows": get_business_monthly_summary(),
"project_account_breakdowns": get_project_account_breakdowns(selected_year),
"project_status_rows": get_project_status_rows(),
"project_edit": get_project_status_for_edit(edit_code),
"project_page_state": get_project_page_state(),
"project_related_links": get_project_related_links_map(),
"support_department_options": get_support_department_options(),
"cost_department_options": get_cost_department_options(),
"cost_account_options": get_cost_account_options(),
}
return templates.TemplateResponse(request, "projects.html", context)
def render_annual_summary_page(request: Request, message: str = "") -> HTMLResponse:
init_db()
context = {
**base_context(request, message),
"available_years": get_available_years(),
"yearly_financial_series": get_financial_series("yearly"),
"monthly_financial_series": get_financial_series("monthly"),
}
return templates.TemplateResponse(request, "annual_summary.html", context)
@app.get("/health")
async def health() -> dict[str, str]:
return build_health_payload()
@app.get("/")
async def home(request: Request, edit_id: int | None = None, overview_year: int | None = None):
try:
return render_home(request, edit_id=edit_id, overview_year=overview_year)
except Exception as exc:
logger.exception("홈페이지 에러: %s", exc)
return HTMLResponse("<h1>서버 오류</h1><p>로그를 확인해주세요.</p>", status_code=500)
@app.get("/projects")
async def projects(request: Request, edit_code: str | None = None, year: str | None = None):
try:
return render_projects_page(request, edit_code=edit_code, selected_year=parse_optional_year(year))
except Exception as exc:
logger.exception("사업현황 페이지 에러: %s", exc)
return HTMLResponse("<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.post("/projects/related-links")
async def project_related_links_save(request: Request):
try:
payload = await request.json()
if not isinstance(payload, dict):
raise ValueError("잘못된 연관 프로젝트 형식입니다.")
base_code = normalize_text(payload.get("base_code"))
related_codes = payload.get("related_codes") or []
if not isinstance(related_codes, list):
raise ValueError("연관 프로젝트 목록 형식이 올바르지 않습니다.")
save_project_related_links(base_code, related_codes)
return JSONResponse(content={"status": "ok"})
except Exception as exc:
logger.exception("연관 프로젝트 저장 에러: %s", exc)
return JSONResponse(content={"error": str(exc)}, status_code=500)
@app.get("/annual-summary")
async def annual_summary(request: Request):
try:
return render_annual_summary_page(request)
except Exception as exc:
logger.exception("연도별 수익 비용 정리 페이지 에러: %s", exc)
return HTMLResponse("<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"edit_code={quote_plus(code)}")
if selected_year:
query_parts.append(f"year={quote_plus(selected_year)}")
if query_parts:
redirect_url += "?" + "&".join(query_parts)
return RedirectResponse(redirect_url, status_code=303)
except Exception as exc:
logger.exception("사업현황 저장 에러: %s", exc)
code = normalize_text(form_data.get("support_dept_code"))
selected_year_text = normalize_text(form_data.get("selected_year"))
selected_year = int(selected_year_text) if selected_year_text.isdigit() else None
return render_projects_page(
request,
edit_code=code or None,
selected_year=selected_year,
message=f"사업현황 저장 중 오류가 발생했습니다: {exc}",
)
if __name__ == "__main__":
auto_reload = os.getenv("INTRANET_AUTO_RELOAD", "1").lower() not in {"0", "false", "no"}
uvicorn.run("main:app", host="0.0.0.0", port=8010, reload=auto_reload, reload_dirs=[str(BASE_DIR)])
Reference in New Issue View Git Blame Copy Permalink