YJKIM/HM_project_Viewer_Board
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refine project status workflows and dashboards

Browse Source
This commit is contained in:
b17301
2026-04-07 20:31:34 +09:00
parent 27a29e8e4c
commit d72a8377fb
6 changed files with 1621 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+231 -60
View File
@@ -1,3 +1,4 @@
import os
import logging
import json
import re
@@ -8,7 +9,8 @@ from urllib.parse import parse_qs, quote_plus
import uvicorn
from fastapi import FastAPI, File, Request, UploadFile
from fastapi.responses import HTMLResponse, RedirectResponse
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
@@ -192,6 +194,7 @@ def init_db() -> None:
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 '[]',
@@ -227,6 +230,7 @@ def init_db() -> None:
"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 '[]'",
@@ -368,6 +372,64 @@ def get_support_department_options() -> list[dict[str, str]]:
]
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()
@@ -753,16 +815,21 @@ def get_project_status_rows() -> list[dict[str, Any]]:
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,
@@ -807,7 +874,15 @@ def get_project_status_rows() -> list[dict[str, Any]]:
"""
)
).mappings().all()
return [dict(row) for row in rows]
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]:
@@ -827,6 +902,7 @@ def get_project_status_for_edit(support_dept_code: str | None) -> dict[str, Any]
"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": [],
@@ -861,6 +937,7 @@ def get_project_status_for_edit(support_dept_code: str | None) -> dict[str, Any]
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,
@@ -904,6 +981,7 @@ def get_project_status_for_edit(support_dept_code: str | None) -> dict[str, Any]
"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": [],
@@ -924,6 +1002,10 @@ def get_project_status_for_edit(support_dept_code: str | None) -> dict[str, Any]
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"] = [
{
@@ -938,20 +1020,62 @@ def get_project_status_for_edit(support_dept_code: str | None) -> dict[str, Any]
if not result["task_plan_entries"]:
fallback_task_rows = []
if normalize_amount(result.get("task_plan_department_budget")):
fallback_task_rows.append({"group": "department", "label": "기존 부서별 배분액", "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", "label": "기존 외주비", "amount": result.get("task_plan_outsource_budget", ""), "note": result.get("task_plan_outsource_detail", "")})
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", "label": "기존 합사운영비", "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", "label": "기존 직급별 인건비", "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", "label": "기존 외주비", "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", "label": "기존 비용계획", "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"] = [
@@ -1175,7 +1299,8 @@ def get_project_account_breakdowns(selected_year: int | None) -> dict[str, dict[
WHEN accounting_category = '판관비' THEN 'sga'
ELSE 'other'
END AS breakdown_kind,
COALESCE(account_name, account_code, '미분류') AS account_label,
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, '') <> ''
@@ -1184,27 +1309,32 @@ def get_project_account_breakdowns(selected_year: int | None) -> dict[str, dict[
AND support_dept_name NOT IN ('공통', '경영지원부', '기술개발센터', '임원실', '기술개발부', '총괄기획실')
AND ({REVENUE_SQL} OR accounting_category IN ('원가', '판관비'))
{year_clause}
GROUP BY support_dept_code, breakdown_kind, account_label
ORDER BY support_dept_code, breakdown_kind, total_amount DESC, account_label
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, list[dict[str, Any]]]] = {}
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
result.setdefault(code, {"revenue": [], "cost": [], "sga": []})
result[code][kind].append(
{
"label": row["account_label"],
"amount": row["total_amount"] or 0,
}
)
return result
_, _, 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]]:
@@ -1454,6 +1584,28 @@ def build_named_amount_rows(
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[]", [])
@@ -1513,57 +1665,59 @@ def build_project_status_payload(payload: dict[str, Any]) -> dict[str, Any]:
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_named_amount_rows(
payload.get("task_plan_department_label[]", []),
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[]", []),
)
task_plan_outsource_rows = []
outsource_names = payload.get("task_plan_outsource_vendor[]", [])
outsource_amounts = payload.get("task_plan_outsource_amount[]", [])
outsource_notes = payload.get("task_plan_outsource_note[]", [])
for index, vendor in enumerate(outsource_names):
task_plan_outsource_rows.append(
{
"label": vendor,
"amount": outsource_amounts[index] if index < len(outsource_amounts) else "",
"note": outsource_notes[index] if index < len(outsource_notes) else "",
"group": "outsource",
}
)
task_plan_outsource_rows = filter_amount_rows(task_plan_outsource_rows, amount_key="amount")
task_plan_joint_rows = build_named_amount_rows(
payload.get("task_plan_joint_label[]", []),
payload.get("task_plan_joint_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_named_amount_rows(
payload.get("exec_labor_role[]", []),
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 = []
exec_outsource_names = payload.get("exec_outsource_vendor[]", [])
exec_outsource_amounts = payload.get("exec_outsource_amount[]", [])
exec_outsource_notes = payload.get("exec_outsource_note[]", [])
for index, vendor in enumerate(exec_outsource_names):
exec_outsource_rows.append(
{
"label": vendor,
"amount": exec_outsource_amounts[index] if index < len(exec_outsource_amounts) else "",
"note": exec_outsource_notes[index] if index < len(exec_outsource_notes) else "",
"group": "outsource",
}
)
exec_outsource_rows = filter_amount_rows(exec_outsource_rows, amount_key="amount")
exec_cost_plan_rows = build_named_amount_rows(
payload.get("exec_cost_plan_label[]", []),
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"
@@ -1587,6 +1741,7 @@ def build_project_status_payload(payload: dict[str, Any]) -> dict[str, Any]:
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")),
@@ -1600,12 +1755,13 @@ def build_project_status_payload(payload: dict[str, Any]) -> dict[str, Any]:
"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"{row.get('label', '')}: {format_amount_for_text(row.get('amount'))}{' / ' + row.get('note', '') if row.get('note') else ''}"
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),
@@ -1714,6 +1870,7 @@ def save_project_status(payload: dict[str, Any]) -> None:
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,
@@ -1745,6 +1902,7 @@ def save_project_status(payload: dict[str, Any]) -> None:
: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,
@@ -1776,6 +1934,7 @@ def save_project_status(payload: dict[str, Any]) -> None:
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,
@@ -1846,6 +2005,8 @@ def render_projects_page(
"project_status_rows": get_project_status_rows(),
"project_edit": get_project_status_for_edit(edit_code),
"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)
@@ -1884,6 +2045,15 @@ async def projects(request: Request, edit_code: str | None = None, year: str | N
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.get("/annual-summary")
async def annual_summary(request: Request):
try:
@@ -1953,4 +2123,5 @@ async def save_project(request: Request):
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8010, reload=False)
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)])