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

2368 lines
97 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
)
"""
)
)
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}"))
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])}
@app.on_event("startup")
def on_startup() -> None:
init_db()
auto_import_project_excels()
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 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 transactions
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 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()
versions = [normalize_text(transaction_updated), normalize_text(project_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]]:
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 = []
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", "[]"))
result.append(item)
return result
def get_project_status_for_edit(support_dept_code: str | None) -> dict[str, Any]:
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": "",
"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": "",
"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 result
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),
},
)
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_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
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()
return [dict(row) for row in rows]
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)
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()
if count_transactions() > 0 and all(file.name in known_files for file in excel_files):
return
for excel_path in excel_files:
if excel_path.name in known_files:
continue
inserted = import_excel_path(excel_path)
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_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"],
}
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_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(),
"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.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