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

Add process cost page with Hanmac/WEHAGO views and tab

Browse Source
This commit is contained in:
b17301
2026-04-27 18:10:01 +09:00
parent b2a3802dff
commit 21901fe8fb
3 changed files with 1020 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+452
View File
@@ -5281,6 +5281,439 @@ def get_available_years() -> list[int]:
return [int(row[0]) for row in rows if row[0] is not None]
def _safe_ratio(numerator: Any, denominator: Any) -> float:
denom = normalize_amount(denominator)
if abs(denom) < 1e-9:
return 0.0
return (normalize_amount(numerator) / denom) * 100.0
def get_process_cost_available_years(source: str) -> list[int]:
normalized_source = normalize_text(source).lower()
if normalized_source == "wehago":
init_wehago_compare_db(engine)
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT DISTINCT fiscal_year
FROM wehago_voucher_rows
WHERE fiscal_year IS NOT NULL
ORDER BY fiscal_year
"""
)
).fetchall()
return [int(row[0]) for row in rows if row and row[0] is not None]
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 and row[0] is not None]
def _get_project_contract_meta() -> dict[str, dict[str, Any]]:
with engine.begin() as conn:
rows = conn.execute(
text(
"""
SELECT
COALESCE(support_dept_code, '') AS support_dept_code,
COALESCE(support_dept_name, '') AS support_dept_name,
COALESCE(hanmac_contract_amount, 0) AS hanmac_contract_amount,
COALESCE(client_name, '') AS client_name
FROM project_contract_info
WHERE COALESCE(support_dept_code, '') <> ''
"""
)
).mappings().all()
result: dict[str, dict[str, Any]] = {}
for row in rows:
code = normalize_text(row.get("support_dept_code"))
if not code:
continue
result[code] = dict(row)
return result
def get_process_cost_project_options(source: str, selected_year: int | None) -> list[dict[str, Any]]:
normalized_source = normalize_text(source).lower()
contract_meta = _get_project_contract_meta()
if normalized_source == "wehago":
init_wehago_compare_db(engine)
year_clause = ""
params: dict[str, Any] = {}
if selected_year:
year_clause = "AND fiscal_year = :selected_year"
params["selected_year"] = selected_year
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
WITH base AS (
SELECT
COALESCE(support_dept_code, '') AS support_dept_code,
COALESCE(support_dept_name, '') AS support_dept_name,
COALESCE(proof_date, '') AS proof_date,
COALESCE(account_code, '') AS account_code,
CASE
WHEN ABS(COALESCE(compare_amount, 0)) > 0 THEN ABS(COALESCE(compare_amount, 0))
WHEN ABS(COALESCE(debit_supply, 0)) >= ABS(COALESCE(credit_supply, 0)) THEN ABS(COALESCE(debit_supply, 0))
ELSE ABS(COALESCE(credit_supply, 0))
END AS amount,
COALESCE(confirmed_no, '') AS confirmed_no,
COALESCE(draft_no, '') AS draft_no
FROM wehago_voucher_rows
WHERE COALESCE(support_dept_code, '') <> ''
AND support_dept_code NOT IN ('ZZZZZZ')
{year_clause}
)
SELECT
support_dept_code,
MAX(support_dept_name) AS support_dept_name,
SUM(CASE WHEN account_code LIKE '401101%' OR account_code LIKE '401102%' THEN amount ELSE 0 END) AS revenue_amount,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount,
COUNT(DISTINCT CASE WHEN COALESCE(confirmed_no, '') <> '' THEN confirmed_no ELSE draft_no END) AS voucher_count,
MAX(proof_date) AS last_posting_date
FROM base
GROUP BY support_dept_code
ORDER BY expense_amount DESC, support_dept_code
"""
),
params,
).mappings().all()
else:
year_clause = ""
params = {}
if selected_year:
year_clause = "AND year = :selected_year"
params["selected_year"] = selected_year
with engine.begin() as conn:
rows = conn.execute(
text(
f"""
SELECT
COALESCE(support_dept_code, '') AS support_dept_code,
MAX(COALESCE(support_dept_name, '')) AS support_dept_name,
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS revenue_amount,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount,
COUNT(DISTINCT COALESCE(voucher_number, '')) AS voucher_count,
MAX(COALESCE(posting_date, '')) AS last_posting_date
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 support_dept_code
ORDER BY expense_amount DESC, support_dept_code
"""
),
params,
).mappings().all()
result: list[dict[str, Any]] = []
for row in rows:
code = normalize_text(row.get("support_dept_code"))
if not code:
continue
contract_row = contract_meta.get(code, {})
contract_amount = normalize_amount(contract_row.get("hanmac_contract_amount"))
revenue_amount = normalize_amount(row.get("revenue_amount"))
expense_amount = normalize_amount(row.get("expense_amount"))
profit_amount = revenue_amount - expense_amount
result.append(
{
"support_dept_code": code,
"support_dept_name": normalize_text(contract_row.get("support_dept_name")) or normalize_text(row.get("support_dept_name")) or code,
"client_name": normalize_text(contract_row.get("client_name")),
"contract_amount": contract_amount,
"revenue_amount": revenue_amount,
"expense_amount": expense_amount,
"profit_amount": profit_amount,
"profit_rate": _safe_ratio(profit_amount, revenue_amount),
"voucher_count": int(row.get("voucher_count") or 0),
"last_posting_date": normalize_text(row.get("last_posting_date")),
}
)
result.sort(key=lambda item: (-normalize_amount(item.get("expense_amount")), item.get("support_dept_code", "")))
return result
def get_process_cost_project_detail(source: str, selected_year: int | None, support_dept_code: str) -> dict[str, Any]:
code = normalize_text(support_dept_code)
if not code:
return {
"overview": {},
"phase_rows": [],
"account_rows": [],
"monthly_rows": [],
"diagnostics": {},
}
normalized_source = normalize_text(source).lower()
contract_meta = _get_project_contract_meta().get(code, {})
if normalized_source == "wehago":
init_wehago_compare_db(engine)
year_clause = ""
params: dict[str, Any] = {"support_dept_code": code}
if selected_year:
year_clause = "AND fiscal_year = :selected_year"
params["selected_year"] = selected_year
amount_expr = (
"CASE "
"WHEN ABS(COALESCE(compare_amount, 0)) > 0 THEN ABS(COALESCE(compare_amount, 0)) "
"WHEN ABS(COALESCE(debit_supply, 0)) >= ABS(COALESCE(credit_supply, 0)) THEN ABS(COALESCE(debit_supply, 0)) "
"ELSE ABS(COALESCE(credit_supply, 0)) "
"END"
)
with engine.begin() as conn:
summary = conn.execute(
text(
f"""
SELECT
MAX(COALESCE(support_dept_name, '')) AS support_dept_name,
SUM(CASE WHEN account_code LIKE '401101%' OR account_code LIKE '401102%' THEN {amount_expr} ELSE 0 END) AS revenue_amount,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN {amount_expr} ELSE 0 END) AS expense_amount,
SUM(CASE WHEN account_code LIKE '5012%' THEN {amount_expr} ELSE 0 END) AS labor_amount,
SUM(CASE WHEN account_code LIKE '5017%' THEN {amount_expr} ELSE 0 END) AS outsourcing_amount,
SUM(CASE WHEN account_code LIKE '6%' THEN {amount_expr} ELSE 0 END) AS sga_amount,
COUNT(*) AS row_count,
COUNT(DISTINCT CASE WHEN COALESCE(confirmed_no, '') <> '' THEN confirmed_no ELSE draft_no END) AS voucher_count,
MAX(COALESCE(proof_date, '')) AS last_posting_date
FROM wehago_voucher_rows
WHERE support_dept_code = :support_dept_code
{year_clause}
"""
),
params,
).mappings().first()
account_rows = conn.execute(
text(
f"""
SELECT
COALESCE(account_code, '') AS account_code,
COALESCE(account_name, '') AS account_name,
SUM({amount_expr}) AS amount,
COUNT(*) AS row_count,
MAX(COALESCE(proof_date, '')) AS last_posting_date
FROM wehago_voucher_rows
WHERE support_dept_code = :support_dept_code
AND (account_code LIKE '5%' OR account_code LIKE '6%')
{year_clause}
GROUP BY account_code, account_name
ORDER BY amount DESC, account_code
LIMIT 14
"""
),
params,
).mappings().all()
monthly_rows = conn.execute(
text(
f"""
SELECT
SUBSTR(COALESCE(proof_date, ''), 1, 7) AS month_label,
SUM(CASE WHEN account_code LIKE '401101%' OR account_code LIKE '401102%' THEN {amount_expr} ELSE 0 END) AS revenue_amount,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN {amount_expr} ELSE 0 END) AS expense_amount
FROM wehago_voucher_rows
WHERE support_dept_code = :support_dept_code
{year_clause}
AND LENGTH(COALESCE(proof_date, '')) >= 7
GROUP BY month_label
ORDER BY month_label DESC
LIMIT 8
"""
),
params,
).mappings().all()
else:
year_clause = ""
params = {"support_dept_code": code}
if selected_year:
year_clause = "AND year = :selected_year"
params["selected_year"] = selected_year
with engine.begin() as conn:
summary = conn.execute(
text(
f"""
SELECT
MAX(COALESCE(support_dept_name, '')) AS support_dept_name,
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS revenue_amount,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount,
SUM(CASE WHEN account_code LIKE '5012%' THEN amount ELSE 0 END) AS labor_amount,
SUM(CASE WHEN account_code LIKE '5017%' THEN amount ELSE 0 END) AS outsourcing_amount,
SUM(CASE WHEN accounting_category = '판관비' THEN amount ELSE 0 END) AS sga_amount,
COUNT(*) AS row_count,
COUNT(DISTINCT COALESCE(voucher_number, '')) AS voucher_count,
MAX(COALESCE(posting_date, '')) AS last_posting_date
FROM transactions
WHERE support_dept_code = :support_dept_code
{year_clause}
"""
),
params,
).mappings().first()
account_rows = conn.execute(
text(
f"""
SELECT
COALESCE(account_code, '') AS account_code,
COALESCE(account_name, '') AS account_name,
SUM(COALESCE(amount, 0)) AS amount,
COUNT(*) AS row_count,
MAX(COALESCE(posting_date, '')) AS last_posting_date
FROM transactions
WHERE support_dept_code = :support_dept_code
AND (account_code LIKE '5%' OR account_code LIKE '6%')
{year_clause}
GROUP BY account_code, account_name
ORDER BY amount DESC, account_code
LIMIT 14
"""
),
params,
).mappings().all()
monthly_rows = conn.execute(
text(
f"""
SELECT
printf('%04d-%02d', year, month) AS month_label,
SUM(CASE WHEN {REVENUE_SQL} THEN amount ELSE 0 END) AS revenue_amount,
SUM(CASE WHEN account_code LIKE '5%' OR account_code LIKE '6%' THEN amount ELSE 0 END) AS expense_amount
FROM transactions
WHERE support_dept_code = :support_dept_code
{year_clause}
AND year IS NOT NULL
AND month IS NOT NULL
GROUP BY year, month
ORDER BY year DESC, month DESC
LIMIT 8
"""
),
params,
).mappings().all()
summary_row = dict(summary) if summary else {}
contract_amount = normalize_amount(contract_meta.get("hanmac_contract_amount"))
revenue_amount = normalize_amount(summary_row.get("revenue_amount"))
expense_amount = normalize_amount(summary_row.get("expense_amount"))
labor_amount = normalize_amount(summary_row.get("labor_amount"))
outsourcing_amount = normalize_amount(summary_row.get("outsourcing_amount"))
sga_amount = normalize_amount(summary_row.get("sga_amount"))
design_cost_amount = max(expense_amount - labor_amount - outsourcing_amount - sga_amount, 0.0)
profit_amount = revenue_amount - expense_amount
target_base = contract_amount * 0.85 if contract_amount > 0 else max(revenue_amount * 0.85, expense_amount)
phase_rows = [
{"phase": "직접인건비", "target_amount": target_base * 0.30, "actual_amount": labor_amount},
{"phase": "외주비", "target_amount": target_base * 0.32, "actual_amount": outsourcing_amount},
{"phase": "설계경비", "target_amount": target_base * 0.23, "actual_amount": design_cost_amount},
{"phase": "판관비", "target_amount": target_base * 0.15, "actual_amount": sga_amount},
]
for row in phase_rows:
row["gap_amount"] = row["target_amount"] - row["actual_amount"]
row["progress_rate"] = _safe_ratio(row["actual_amount"], row["target_amount"])
normalized_accounts = []
for row in account_rows:
amount = normalize_amount(row.get("amount"))
normalized_accounts.append(
{
"account_code": normalize_text(row.get("account_code")),
"account_name": normalize_text(row.get("account_name")),
"amount": amount,
"row_count": int(row.get("row_count") or 0),
"share_rate": _safe_ratio(amount, expense_amount),
"last_posting_date": normalize_text(row.get("last_posting_date")),
}
)
normalized_monthly = []
for row in monthly_rows:
revenue = normalize_amount(row.get("revenue_amount"))
expense = normalize_amount(row.get("expense_amount"))
normalized_monthly.append(
{
"month_label": normalize_text(row.get("month_label")),
"revenue_amount": revenue,
"expense_amount": expense,
"profit_amount": revenue - expense,
}
)
normalized_monthly.reverse()
return {
"overview": {
"support_dept_code": code,
"support_dept_name": normalize_text(contract_meta.get("support_dept_name")) or normalize_text(summary_row.get("support_dept_name")) or code,
"client_name": normalize_text(contract_meta.get("client_name")),
"contract_amount": contract_amount,
"revenue_amount": revenue_amount,
"expense_amount": expense_amount,
"profit_amount": profit_amount,
"profit_rate": _safe_ratio(profit_amount, revenue_amount),
"target_cost_amount": target_base,
"execution_rate": _safe_ratio(expense_amount, target_base),
"last_posting_date": normalize_text(summary_row.get("last_posting_date")),
"voucher_count": int(summary_row.get("voucher_count") or 0),
"row_count": int(summary_row.get("row_count") or 0),
},
"phase_rows": phase_rows,
"account_rows": normalized_accounts,
"monthly_rows": normalized_monthly,
"diagnostics": {
"labor_ratio": _safe_ratio(labor_amount, expense_amount),
"outsourcing_ratio": _safe_ratio(outsourcing_amount, expense_amount),
"design_cost_ratio": _safe_ratio(design_cost_amount, expense_amount),
"sga_ratio": _safe_ratio(sga_amount, expense_amount),
"cost_to_revenue_ratio": _safe_ratio(expense_amount, revenue_amount),
},
}
def render_process_cost_page(
request: Request,
source: str | None = None,
year: int | None = None,
code: str | None = None,
message: str = "",
) -> HTMLResponse:
init_db()
init_wehago_compare_db(engine)
normalized_source = normalize_text(source).lower()
if normalized_source not in {"hanmac", "wehago"}:
normalized_source = "hanmac"
years = get_process_cost_available_years(normalized_source)
selected_year = year if year in years else (max(years) if years else None)
project_options = get_process_cost_project_options(normalized_source, selected_year)
selected_code = normalize_text(code)
if selected_code and not any(item["support_dept_code"] == selected_code for item in project_options):
selected_code = ""
if not selected_code and project_options:
selected_code = project_options[0]["support_dept_code"]
detail = get_process_cost_project_detail(normalized_source, selected_year, selected_code)
context = {
**base_context(request, message),
"process_cost_source": normalized_source,
"process_cost_years": years,
"process_cost_selected_year": selected_year,
"process_cost_selected_code": selected_code,
"process_cost_projects": project_options,
"process_cost_detail": detail,
}
return templates.TemplateResponse(request, "process_cost.html", context)
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"
@@ -6653,6 +7086,25 @@ async def annual_summary(request: Request):
return HTMLResponse("<h1>서버 오류</h1><p>로그를 확인해주세요.</p>", status_code=500)
@app.get("/process-cost")
async def process_cost(
request: Request,
source: str | None = None,
year: str | None = None,
code: str | None = None,
):
try:
return render_process_cost_page(
request,
source=source,
year=parse_optional_year(year),
code=code,
)
except Exception as exc:
logger.exception("프로세스 원가 페이지 에러: %s", exc)
return HTMLResponse("<h1>서버 오류</h1><p>로그를 확인해주세요.</p>", status_code=500)
@app.get("/wehago-compare")
async def wehago_compare(request: Request, start_year: str | None = None, end_year: str | None = None):
try: