PTC Data API
PTC 원장 데이터 서버
이 서버는 선택된 PTC 원본 엑셀 파일을 읽어 요약, 프로젝트 집계, 계정 집계, 거래 미리보기를 JSON API로 제공합니다.
메인 화면은 http://localhost:4000/PTC/ 에서 바로 확인할 수 있습니다.
현재 적재 건수
{count:,}
원본 파일 기준 전체 거래 행 수
원본 파일: {xlsx_path}
메인 화면
localhost:4000/PTC/
사용자가 보는 메인 대시보드와 API를 같은 서버에서 제공합니다.
헬스체크
/api/health
서버 상태와 row count 확인
요약
/api/summary
건수, 기간, 공급가액, 누락값
주요 API
GET/api/health
API 서버가 정상 동작하는지와 적재 건수를 반환합니다.
GET/api/summary
건수, 입금/출금, 공급가액, 부가세, 기간, 누락값 요약을 반환합니다.
GET/api/top-accounts
계정코드별 상위 집계를 반환합니다.
GET/api/top-projects
프로젝트별 상위 집계를 반환합니다.
GET/api/project-mismatches
프로젝트코드 대비 프로젝트명/구분 불일치를 반환합니다.
GET/api/transactions?limit=30
원본 거래 미리보기를 반환합니다.
필터 예시
/api/summary?project_type=시공
/api/top-accounts?in_out=출금
/api/transactions?keyword=여비교통비&limit=20
"""
self._send_html(200, html)
return
if parsed.path == "/api/health":
count = conn.execute("select count(*) as count from ptc_transactions").fetchone()["count"]
self._send(200, {"ok": True, "row_count": count})
return
if parsed.path == "/api/project-master-options":
self._send(
200,
{
"project_type_options": PROJECT_TYPE_OPTIONS,
"method_family_options": METHOD_FAMILY_OPTIONS,
"method_options": METHOD_OPTIONS,
"method_family_map": METHOD_FAMILY_MAP,
"account_master": ACCOUNT_MASTER,
"allowed_account_codes_by_project_type": {
key: sorted(value) for key, value in ALLOWED_ACCOUNT_CODES_BY_PROJECT_TYPE.items()
},
},
)
return
if parsed.path == "/api/summary":
summary = query_summary(conn, params)
self._send(200, summary)
return
if parsed.path == "/api/dashboard-prototype":
project_type = params.get("project_type", ["전체"])[0]
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
exclude_asset_accounts = params.get("exclude_asset_accounts", ["0"])[0].strip() in {"1", "true", "yes"}
base_clauses = ["t.project_code is not null", "t.project_code <> ''"]
base_values: list[str] = []
if project_type and project_type != "전체":
base_clauses.append("coalesce(pm.project_type, t.project_type) = ?")
base_values.append(project_type)
if exclude_asset_accounts and MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
base_clauses.append(f"coalesce(t.account_code_final, '') not in ({excluded_placeholders})")
base_values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
clauses = list(base_clauses)
values = list(base_values)
if date_from:
clauses.append("coalesce(t.transaction_date, '') >= ?")
values.append(date_from)
if date_to:
clauses.append("coalesce(t.transaction_date, '') <= ?")
values.append(date_to)
where = " where " + " and ".join(clauses)
rows = conn.execute(
f"""
select
t.project_code,
coalesce(pm.project_name, max(t.project_name)) as project_name,
coalesce(pm.project_type, max(t.project_type)) as project_type,
coalesce(pm.construction_family, '') as construction_family,
coalesce(pm.construction_method, '') as construction_method,
coalesce(pb.revenue_budget_total, 0) as revenue_budget_total,
coalesce(pp.progress_rate, 0) as progress_rate,
coalesce(pp.contract_pile_count, 0) as contract_pile_count,
coalesce(pp.constructed_pile_count, 0) as constructed_pile_count,
coalesce(sum(case when t.in_out = '입금' then t.supply_amount else 0 end), 0) as income_supply,
coalesce(sum(case when t.in_out = '출금' then t.supply_amount else 0 end), 0) as expense_supply,
count(*) as txn_count
from ptc_transactions t
left join project_master pm on pm.project_code = t.project_code
left join (
select project_code, coalesce(sum(budget_amount), 0) as revenue_budget_total
from project_budget_lines
where section = '수입'
group by project_code
) pb on pb.project_code = t.project_code
left join project_progress pp on pp.project_code = t.project_code
{where}
group by t.project_code, pm.project_name, pm.project_type, pm.construction_family, pm.construction_method, pb.revenue_budget_total, pp.progress_rate, pp.contract_pile_count, pp.constructed_pile_count
""",
values,
).fetchall()
amount_buckets = [
{"key": "under_5", "label": "5억 미만", "min": 0, "max": 500_000_000},
{"key": "5_to_20", "label": "5억~20억", "min": 500_000_000, "max": 2_000_000_000},
{"key": "20_to_50", "label": "20억~50억", "min": 2_000_000_000, "max": 5_000_000_000},
{"key": "over_50", "label": "50억 이상", "min": 5_000_000_000, "max": None},
]
status_bands = [
{"key": "normal", "label": "정상"},
{"key": "upfront", "label": "선투입"},
{"key": "delay", "label": "회수지연"},
{"key": "risk", "label": "원가위험"},
]
def bucket_amount(value: float) -> str:
for bucket in amount_buckets:
if bucket["max"] is None and value >= bucket["min"]:
return bucket["key"]
if bucket["min"] <= value < bucket["max"]:
return bucket["key"]
return amount_buckets[0]["key"]
def classify_status(income_supply: float, expense_supply: float, progress_rate: float, contract_pile_count: float, constructed_pile_count: float) -> str:
profit_supply = income_supply - expense_supply
expense_ratio = (expense_supply / income_supply) if income_supply > 0 else None
has_progress_signal = progress_rate > 0 or contract_pile_count > 0 or constructed_pile_count > 0
if profit_supply >= 0:
return "normal"
if income_supply <= 0 and expense_supply > 0:
return "upfront"
if expense_ratio is not None and expense_ratio >= 1.15:
return "risk"
if expense_supply >= 300_000_000 and profit_supply < -100_000_000:
return "risk"
if income_supply > 0 and income_supply < expense_supply:
return "delay" if has_progress_signal else "risk"
return "normal"
by_method: dict[str, dict] = {}
overview = {
"project_count": 0,
"income_supply": 0.0,
"expense_supply": 0.0,
"profit_supply": 0.0,
"status_counts": {band["key"]: 0 for band in status_bands},
}
project_items = []
for row in rows:
method = normalize_dashboard_method(row["construction_method"] or "")
family = normalize_dashboard_family(row["construction_family"] or "")
income_supply = float(row["income_supply"] or 0)
expense_supply = float(row["expense_supply"] or 0)
profit_supply = income_supply - expense_supply
margin_rate = (profit_supply / income_supply * 100) if income_supply > 0 else 0.0
progress_rate = float(row["progress_rate"] or 0)
contract_pile_count = float(row["contract_pile_count"] or 0)
constructed_pile_count = float(row["constructed_pile_count"] or 0)
revenue_budget_total = float(row["revenue_budget_total"] or 0)
amount_bucket_key = bucket_amount(income_supply)
status_key = classify_status(income_supply, expense_supply, progress_rate, contract_pile_count, constructed_pile_count)
if method not in by_method:
by_method[method] = {
"method": method,
"family": family,
"project_count": 0,
"income_supply": 0.0,
"expense_supply": 0.0,
"profit_supply": 0.0,
"status_counts": {band["key"]: 0 for band in status_bands},
"cells": {
bucket["key"]: {
"bucket_key": bucket["key"],
"bucket_label": bucket["label"],
"project_count": 0,
"income_supply": 0.0,
"expense_supply": 0.0,
"profit_supply": 0.0,
"status_counts": {band["key"]: 0 for band in status_bands},
"projects": [],
}
for bucket in amount_buckets
},
}
cell = by_method[method]["cells"][amount_bucket_key]
cell["project_count"] += 1
cell["income_supply"] += income_supply
cell["expense_supply"] += expense_supply
cell["profit_supply"] += profit_supply
cell["status_counts"][status_key] += 1
if len(cell["projects"]) < 5:
cell["projects"].append({
"project_code": row["project_code"],
"project_name": row["project_name"],
"margin_rate": margin_rate,
"income_supply": income_supply,
"status_key": status_key,
})
by_method[method]["project_count"] += 1
by_method[method]["income_supply"] += income_supply
by_method[method]["expense_supply"] += expense_supply
by_method[method]["profit_supply"] += profit_supply
by_method[method]["status_counts"][status_key] += 1
overview["project_count"] += 1
overview["income_supply"] += income_supply
overview["expense_supply"] += expense_supply
overview["profit_supply"] += profit_supply
overview["status_counts"][status_key] += 1
project_items.append({
"project_code": row["project_code"],
"project_name": row["project_name"],
"project_type": row["project_type"],
"construction_method": method,
"construction_family": family,
"progress_rate": progress_rate,
"revenue_budget_total": revenue_budget_total,
"income_supply": income_supply,
"expense_supply": expense_supply,
"profit_supply": profit_supply,
"margin_rate": margin_rate,
"amount_bucket_key": amount_bucket_key,
"status_key": status_key,
})
overview["margin_rate"] = (overview["profit_supply"] / overview["income_supply"] * 100) if overview["income_supply"] > 0 else 0.0
method_items = []
for method, item in sorted(by_method.items(), key=lambda pair: (-pair[1]["income_supply"], pair[0])):
item["margin_rate"] = (item["profit_supply"] / item["income_supply"] * 100) if item["income_supply"] > 0 else 0.0
item["cells"] = [item["cells"][bucket["key"]] for bucket in amount_buckets]
method_items.append(item)
loan_clauses = ["account_code_final = '196'"]
loan_values: list[str] = []
if date_from:
loan_clauses.append("coalesce(transaction_date, '') >= ?")
loan_values.append(date_from)
if date_to:
loan_clauses.append("coalesce(transaction_date, '') <= ?")
loan_values.append(date_to)
loan_where = " where " + " and ".join(loan_clauses)
loan_summary = conn.execute(
f"""
select
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as recovered_supply,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as loaned_supply,
sum(case when in_out = '입금' then 1 else 0 end) as recovered_count,
sum(case when in_out = '출금' then 1 else 0 end) as loaned_count,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{loan_where}
""",
loan_values,
).fetchone()
loan_project_rows = conn.execute(
f"""
select
project_code,
max(project_name) as project_name,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as recovered_supply,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as loaned_supply,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) - coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as outstanding_supply,
count(*) as txn_count
from ptc_transactions
{loan_where}
group by project_code
having coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) > 0
order by outstanding_supply desc, loaned_supply desc, project_code desc
limit 8
""",
loan_values,
).fetchall()
loan_vendor_rows = conn.execute(
f"""
select
coalesce(vendor_name, '거래처없음') as vendor_name,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as recovered_supply,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as loaned_supply,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) - coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as outstanding_supply,
count(*) as txn_count
from ptc_transactions
{loan_where}
group by coalesce(vendor_name, '거래처없음')
having coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) > 0
order by outstanding_supply desc, loaned_supply desc, vendor_name
limit 8
""",
loan_values,
).fetchall()
loan_summary_dict = dict(loan_summary) if loan_summary else {}
recovered_supply = float(loan_summary_dict.get("recovered_supply") or 0)
loaned_supply = float(loan_summary_dict.get("loaned_supply") or 0)
loan_summary_dict["outstanding_supply"] = max(loaned_supply - recovered_supply, 0)
yearly_rows = conn.execute(
f"""
select
substr(coalesce(t.transaction_date, ''), 1, 4) as year,
coalesce(sum(case when t.in_out = '입금' then t.supply_amount else 0 end), 0) as income_supply,
coalesce(sum(case when t.in_out = '출금' then t.supply_amount else 0 end), 0) as expense_supply
from ptc_transactions t
left join project_master pm on pm.project_code = t.project_code
{where}
group by substr(coalesce(t.transaction_date, ''), 1, 4)
having coalesce(substr(coalesce(t.transaction_date, ''), 1, 4), '') <> ''
order by year asc
""",
values,
).fetchall()
yearly_items = []
for row in yearly_rows:
income_supply = float(row["income_supply"] or 0)
expense_supply = float(row["expense_supply"] or 0)
profit_supply = income_supply - expense_supply
yearly_items.append(
{
"year": row["year"],
"income_supply": income_supply,
"expense_supply": expense_supply,
"profit_supply": profit_supply,
"margin_rate": (profit_supply / income_supply * 100) if income_supply > 0 else 0.0,
}
)
month_anchor = datetime.today().replace(day=1)
ongoing_month_index = (month_anchor.year * 12 + month_anchor.month - 1) - 5
ongoing_start_year = ongoing_month_index // 12
ongoing_start_month = (ongoing_month_index % 12) + 1
ongoing_start = f"{ongoing_start_year:04d}-{ongoing_start_month:02d}-01"
selected_project_codes = {str(row["project_code"] or "").strip() for row in rows if str(row["project_code"] or "").strip()}
ongoing_rows = conn.execute(
f"""
select
t.project_code,
coalesce(pm.project_type, max(t.project_type)) as project_type,
coalesce(pm.project_name, max(t.project_name)) as project_name,
max(coalesce(t.transaction_date, '')) as latest_transaction_date
from ptc_transactions t
left join project_master pm on pm.project_code = t.project_code
{" where " + " and ".join(base_clauses)}
group by t.project_code, pm.project_name, pm.project_type
""",
base_values,
).fetchall()
ongoing_project_count = 0
ongoing_project_codes: list[str] = []
for row in ongoing_rows:
project_code = str(row["project_code"] or "").strip()
project_type_value = str(row["project_type"] or "").strip()
project_name = str(row["project_name"] or "").strip()
latest_transaction_date = str(row["latest_transaction_date"] or "").strip()
if not (
"-시공-" in project_code
and "-관리-" not in project_code
and "-설계-" not in project_code
and "시공관리" not in project_name
and project_type_value not in {"관리", "설계"}
):
continue
if selected_project_codes and project_code not in selected_project_codes:
continue
if latest_transaction_date >= ongoing_start:
ongoing_project_count += 1
ongoing_project_codes.append(project_code)
self._send(
200,
{
"overview": overview,
"yearly_overview": yearly_items,
"ongoing_project_count": ongoing_project_count,
"ongoing_project_codes": ongoing_project_codes,
"ongoing_window_start": ongoing_start,
"amount_buckets": amount_buckets,
"status_bands": status_bands,
"date_from": date_from,
"date_to": date_to,
"loan_risk": {
"summary": loan_summary_dict,
"projects": rows_to_dicts(loan_project_rows),
"vendors": rows_to_dicts(loan_vendor_rows),
},
"methods": method_items,
"projects": sorted(project_items, key=lambda item: (-item["income_supply"], item["project_code"])),
},
)
return
if parsed.path == "/api/project-types":
rows = conn.execute(
"select distinct project_type from ptc_transactions where coalesce(project_type,'') <> '' order by project_type"
).fetchall()
self._send(200, {"items": [row["project_type"] for row in rows]})
return
if parsed.path == "/api/projects":
keyword = params.get("keyword", [""])[0].strip().lower()
project_type = params.get("project_type", ["전체"])[0]
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
clauses = ["project_code is not null", "project_code <> ''"]
values = []
if PROJECT_VIEW_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in PROJECT_VIEW_EXCLUDED_ACCOUNT_CODES)
clauses.append(f"coalesce(account_code_final, '') not in ({excluded_placeholders})")
values.extend(sorted(PROJECT_VIEW_EXCLUDED_ACCOUNT_CODES))
if project_type and project_type != "전체":
clauses.append("project_type = ?")
values.append(project_type)
if date_from:
clauses.append("coalesce(transaction_date, '') >= ?")
values.append(date_from)
if date_to:
clauses.append("coalesce(transaction_date, '') <= ?")
values.append(date_to)
where = " where " + " and ".join(clauses)
rows = conn.execute(
f"""
select
project_code,
max(project_name) as project_name,
max(project_type) as project_type,
count(*) as txn_count,
coalesce(sum(supply_amount), 0) as supply_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{where}
group by project_code
order by supply_sum desc, project_code
""",
values,
).fetchall()
items = rows_to_dicts(rows)
project_codes = [item["project_code"] for item in items if item.get("project_code")]
master_rows = {}
if project_codes:
placeholders = ",".join("?" for _ in project_codes)
master_rows = {
row["project_code"]: dict(row)
for row in conn.execute(
f"""
select project_code, project_name, project_type, construction_family, construction_method, note
from project_master
where project_code in ({placeholders})
""",
project_codes,
).fetchall()
}
for item in items:
master = master_rows.get(item["project_code"])
item["project_type"] = resolve_project_type(
item["project_code"],
item["project_type"],
master.get("project_type") if master else "",
)
if master:
item["project_name"] = master.get("project_name") or item["project_name"]
item["construction_family"] = resolve_construction_family(
master.get("construction_method"),
master.get("construction_family"),
)
item["construction_method"] = master.get("construction_method") or ""
item["note"] = master.get("note") or ""
else:
item["construction_family"] = ""
item["construction_method"] = ""
item["note"] = ""
related_projects = build_related_projects(
conn,
item["project_code"],
item.get("project_name") or "",
)
related_search_terms = []
for rel in related_projects:
related_search_terms.extend(
[
rel.get("project_code") or "",
rel.get("project_name") or "",
rel.get("project_type") or "",
]
)
item["related_projects"] = related_projects
item["related_search_text"] = " ".join(
filter(
None,
[
item.get("project_code") or "",
item.get("project_name") or "",
item.get("project_type") or "",
*related_search_terms,
],
)
).lower()
if keyword:
items = [
item
for item in items
if keyword in (item.get("related_search_text") or "")
]
self._send(200, {"items": items})
return
if parsed.path == "/api/vendors":
keyword = params.get("keyword", [""])[0].strip().lower()
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
clauses = ["coalesce(vendor_name, '') <> ''"]
values: list[str] = []
if keyword:
clauses.append(
"""
(
lower(coalesce(vendor_name, '')) like ?
or lower(coalesce(account_code_final, '')) like ?
or lower(coalesce(account_name_final, '')) like ?
)
"""
)
like = f"%{keyword}%"
values.extend([like, like, like])
if date_from:
clauses.append("coalesce(transaction_date, '') >= ?")
values.append(date_from)
if date_to:
clauses.append("coalesce(transaction_date, '') <= ?")
values.append(date_to)
where = f"where {' and '.join(clauses)}"
rows = conn.execute(
f"""
select
vendor_name,
count(*) as txn_count,
coalesce(sum(supply_amount), 0) as supply_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{where}
group by vendor_name
order by supply_sum desc, vendor_name
""",
values,
).fetchall()
self._send(200, {"items": rows_to_dicts(rows)})
return
if parsed.path == "/api/accounts":
keyword = params.get("keyword", [""])[0].strip().lower()
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
clauses = ["coalesce(account_code_final, '') <> ''"]
values: list[str] = []
if keyword:
clauses.append(
"""
(
lower(coalesce(account_code_final, '')) like ?
or lower(coalesce(account_name_final, '')) like ?
or lower(coalesce(vendor_name, '')) like ?
)
"""
)
like = f"%{keyword}%"
values.extend([like, like, like])
if date_from:
clauses.append("coalesce(transaction_date, '') >= ?")
values.append(date_from)
if date_to:
clauses.append("coalesce(transaction_date, '') <= ?")
values.append(date_to)
where = f"where {' and '.join(clauses)}"
rows = conn.execute(
f"""
select
account_code_final as account_code,
account_name_final as account_name,
count(*) as txn_count,
coalesce(sum(supply_amount), 0) as supply_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{where}
group by account_code_final, account_name_final
order by cast(account_code_final as integer) asc, account_code_final asc
""",
values,
).fetchall()
self._send(200, {"items": rows_to_dicts(rows)})
return
if parsed.path == "/api/management-accounts":
keyword = params.get("keyword", [""])[0].strip().lower()
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
clauses = [
"coalesce(account_code_final, '') <> ''",
"(coalesce(project_type, '') = '관리' or coalesce(project_code, '') like '%-관리-%')",
]
values: list[str] = []
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
clauses.append(f"coalesce(account_code_final, '') not in ({excluded_placeholders})")
values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
if keyword:
clauses.append(
"""
(
lower(coalesce(account_code_final, '')) like ?
or lower(coalesce(account_name_final, '')) like ?
or lower(coalesce(vendor_name, '')) like ?
)
"""
)
like = f"%{keyword}%"
values.extend([like, like, like])
if date_from:
clauses.append("coalesce(transaction_date, '') >= ?")
values.append(date_from)
if date_to:
clauses.append("coalesce(transaction_date, '') <= ?")
values.append(date_to)
where = f"where {' and '.join(clauses)}"
rows = conn.execute(
f"""
select
account_code_final as account_code,
account_name_final as account_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{where}
group by account_code_final, account_name_final
order by cast(account_code_final as integer) asc, account_code_final asc
""",
values,
).fetchall()
self._send(200, {"items": rows_to_dicts(rows)})
return
if parsed.path == "/api/management-overview":
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
clauses = [
"(coalesce(project_type, '') = '관리' or coalesce(project_code, '') like '%-관리-%')",
"coalesce(account_code_final, '') <> ''",
]
values: list[str] = []
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
clauses.append(f"coalesce(account_code_final, '') not in ({excluded_placeholders})")
values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
if date_from:
clauses.append("coalesce(transaction_date, '') >= ?")
values.append(date_from)
if date_to:
clauses.append("coalesce(transaction_date, '') <= ?")
values.append(date_to)
where = f"where {' and '.join(clauses)}"
rows = conn.execute(
f"""
select
substr(coalesce(transaction_date, ''), 1, 4) as year,
account_code_final as account_code,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply
from ptc_transactions
{where}
group by substr(coalesce(transaction_date, ''), 1, 4), account_code_final
order by year asc, account_code_final asc
""",
values,
).fetchall()
excluded_rows = []
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
excluded_clauses = [
"(coalesce(project_type, '') = '관리' or coalesce(project_code, '') like '%-관리-%')",
"coalesce(account_code_final, '') in (" + excluded_placeholders + ")",
]
excluded_values: list[str] = list(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
if date_from:
excluded_clauses.append("coalesce(transaction_date, '') >= ?")
excluded_values.append(date_from)
if date_to:
excluded_clauses.append("coalesce(transaction_date, '') <= ?")
excluded_values.append(date_to)
excluded_where = f"where {' and '.join(excluded_clauses)}"
excluded_rows = conn.execute(
f"""
select
substr(coalesce(transaction_date, ''), 1, 4) as year,
account_code_final as account_code,
coalesce(max(account_name_final), '') as account_name,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply,
count(*) as txn_count
from ptc_transactions
{excluded_where}
group by substr(coalesce(transaction_date, ''), 1, 4), account_code_final
order by year asc, cast(account_code_final as integer) asc, account_code_final asc
""",
excluded_values,
).fetchall()
by_year: dict[str, dict] = {}
for row in rows:
year = (row["year"] or "").strip() or "미상"
account_code = (row["account_code"] or "").strip()
master = ACCOUNT_MASTER.get(account_code) or {}
if master.get("project_type") != "관리":
continue
category = master.get("category") or "기타"
if category not in MANAGEMENT_ACCOUNT_CATEGORY_ORDER:
continue
if year not in by_year:
by_year[year] = {
"year": year,
"income_supply": 0.0,
"total_expense": 0.0,
"project_applied_admin_expense": 0.0,
"common_admin_expense": 0.0,
"categories": {key: 0.0 for key in MANAGEMENT_ACCOUNT_CATEGORY_ORDER},
"excluded_total": 0.0,
"excluded_income_total": 0.0,
"excluded_expense_total": 0.0,
"excluded_accounts": [],
}
income_amount = float(row["income_supply"] or 0)
amount = float(row["expense_supply"] or 0)
by_year[year]["income_supply"] += income_amount
by_year[year]["total_expense"] += amount
by_year[year]["categories"][category] += amount
for row in excluded_rows:
year = (row["year"] or "").strip() or "미상"
if year not in by_year:
by_year[year] = {
"year": year,
"income_supply": 0.0,
"total_expense": 0.0,
"project_applied_admin_expense": 0.0,
"common_admin_expense": 0.0,
"categories": {key: 0.0 for key in MANAGEMENT_ACCOUNT_CATEGORY_ORDER},
"excluded_total": 0.0,
"excluded_income_total": 0.0,
"excluded_expense_total": 0.0,
"excluded_accounts": [],
}
income_supply = float(row["income_supply"] or 0)
expense_supply = float(row["expense_supply"] or 0)
total_supply = income_supply + expense_supply
by_year[year]["excluded_total"] += total_supply
by_year[year]["excluded_income_total"] += income_supply
by_year[year]["excluded_expense_total"] += expense_supply
by_year[year]["excluded_accounts"].append(
{
"account_code": row["account_code"],
"account_name": row["account_name"] or (ACCOUNT_MASTER.get((row["account_code"] or "").strip()) or {}).get("name") or "",
"income_supply": income_supply,
"expense_supply": expense_supply,
"total_supply": total_supply,
"txn_count": int(row["txn_count"] or 0),
}
)
split_clauses = ["coalesce(account_code_final, '') <> ''"]
split_values: list[str] = []
if date_from:
split_clauses.append("coalesce(transaction_date, '') >= ?")
split_values.append(date_from)
if date_to:
split_clauses.append("coalesce(transaction_date, '') <= ?")
split_values.append(date_to)
split_where = f"where {' and '.join(split_clauses)}"
management_split_rows = conn.execute(
f"""
select
substr(coalesce(transaction_date, ''), 1, 4) as year,
coalesce(project_code, '') as project_code,
account_code_final as account_code,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply
from ptc_transactions
{split_where}
group by substr(coalesce(transaction_date, ''), 1, 4), coalesce(project_code, ''), account_code_final
having year <> ''
""",
split_values,
).fetchall()
relation_rows = conn.execute(
"""
select
pr.project_code as project_code,
pr.related_project_code as related_project_code,
coalesce(pm.project_type, '') as project_type,
coalesce(rpm.project_type, '') as related_project_type
from project_relations pr
left join project_master pm on pm.project_code = pr.project_code
left join project_master rpm on rpm.project_code = pr.related_project_code
"""
).fetchall()
connected_codes: set[str] = set()
for rel in relation_rows:
left_code = (rel["project_code"] or "").strip()
right_code = (rel["related_project_code"] or "").strip()
left_type = (rel["project_type"] or "").strip()
right_type = (rel["related_project_type"] or "").strip()
if left_type == "시공" and right_code:
connected_codes.add(right_code)
connected_codes.add(left_code)
if right_type == "시공" and left_code:
connected_codes.add(left_code)
connected_codes.add(right_code)
for row in management_split_rows:
year = (row["year"] or "").strip() or "미상"
project_code = (row["project_code"] or "").strip()
account_code = (row["account_code"] or "").strip()
expense_supply = float(row["expense_supply"] or 0)
if not account_code or expense_supply == 0:
continue
master = ACCOUNT_MASTER.get(account_code) or {}
if master.get("project_type") != "관리":
continue
if (master.get("category") or "") == "인건비":
continue
if year not in by_year:
by_year[year] = {
"year": year,
"income_supply": 0.0,
"total_expense": 0.0,
"project_applied_admin_expense": 0.0,
"common_admin_expense": 0.0,
"categories": {key: 0.0 for key in MANAGEMENT_ACCOUNT_CATEGORY_ORDER},
"excluded_total": 0.0,
"excluded_income_total": 0.0,
"excluded_expense_total": 0.0,
"excluded_accounts": [],
}
is_project_applied = (
(bool(project_code) and "-시공-" in project_code)
or (project_code in connected_codes)
)
if is_project_applied:
by_year[year]["project_applied_admin_expense"] += expense_supply
else:
by_year[year]["common_admin_expense"] += expense_supply
items = []
for year in sorted(by_year.keys()):
item = by_year[year]
items.append({
"year": item["year"],
"income_supply": item["income_supply"],
"total_expense": item["total_expense"],
"project_applied_admin_expense": item["project_applied_admin_expense"],
"common_admin_expense": item["common_admin_expense"],
"categories": [
{"name": category, "amount": item["categories"][category]}
for category in MANAGEMENT_ACCOUNT_CATEGORY_ORDER
],
"excluded_total": item["excluded_total"],
"excluded_income_total": item["excluded_income_total"],
"excluded_expense_total": item["excluded_expense_total"],
"excluded_accounts": item["excluded_accounts"],
})
company_clauses = ["coalesce(account_code_final, '') <> ''"]
company_values: list[str] = []
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
company_excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
company_clauses.append(f"coalesce(account_code_final, '') not in ({company_excluded_placeholders})")
company_values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
if date_from:
company_clauses.append("coalesce(transaction_date, '') >= ?")
company_values.append(date_from)
if date_to:
company_clauses.append("coalesce(transaction_date, '') <= ?")
company_values.append(date_to)
company_where = f"where {' and '.join(company_clauses)}"
yearly_profit_rows = conn.execute(
f"""
select
substr(coalesce(transaction_date, ''), 1, 4) as year,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply
from ptc_transactions
{company_where}
group by substr(coalesce(transaction_date, ''), 1, 4)
having coalesce(substr(coalesce(transaction_date, ''), 1, 4), '') <> ''
order by year asc
""",
company_values,
).fetchall()
yearly_profit_items = []
for row in yearly_profit_rows:
income_supply = float(row["income_supply"] or 0)
expense_supply = float(row["expense_supply"] or 0)
yearly_profit_items.append(
{
"year": row["year"],
"income_supply": income_supply,
"expense_supply": expense_supply,
"profit_supply": income_supply - expense_supply,
}
)
construction_clauses = [
"coalesce(account_code_final, '') <> ''",
"coalesce(project_code, '') like '%-시공-%'",
"coalesce(project_name, '') not like '%시공관리%'",
"coalesce(project_type, '') not in ('관리', '설계')",
]
construction_values: list[str] = []
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
construction_excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
construction_clauses.append(f"coalesce(account_code_final, '') not in ({construction_excluded_placeholders})")
construction_values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
if date_from:
construction_clauses.append("coalesce(transaction_date, '') >= ?")
construction_values.append(date_from)
if date_to:
construction_clauses.append("coalesce(transaction_date, '') <= ?")
construction_values.append(date_to)
construction_where = f"where {' and '.join(construction_clauses)}"
yearly_construction_rows = conn.execute(
f"""
select
substr(coalesce(transaction_date, ''), 1, 4) as year,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply
from ptc_transactions
{construction_where}
group by substr(coalesce(transaction_date, ''), 1, 4)
having coalesce(substr(coalesce(transaction_date, ''), 1, 4), '') <> ''
order by year asc
""",
construction_values,
).fetchall()
yearly_construction_margin_items = []
for row in yearly_construction_rows:
income_supply = float(row["income_supply"] or 0)
expense_supply = float(row["expense_supply"] or 0)
profit_supply = income_supply - expense_supply
yearly_construction_margin_items.append(
{
"year": row["year"],
"income_supply": income_supply,
"expense_supply": expense_supply,
"profit_supply": profit_supply,
"margin_rate": (profit_supply / income_supply * 100) if income_supply > 0 else 0.0,
}
)
self._send(
200,
{
"items": items,
"category_order": MANAGEMENT_ACCOUNT_CATEGORY_ORDER,
"yearly_profit_items": yearly_profit_items,
"yearly_construction_margin_items": yearly_construction_margin_items,
},
)
return
if parsed.path == "/api/management-company-overview":
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
project_type_case = """
case
when coalesce(t.project_code, '') like '%-시공-%' then '시공'
when coalesce(t.project_code, '') like '%-영업-%' then '영업'
when coalesce(t.project_code, '') like '%-설계-%' then '설계'
when coalesce(t.project_code, '') like '%-관리-%' then '관리'
else coalesce(pm.project_type, t.project_type, '미지정')
end
"""
clauses = ["coalesce(t.account_code_final, '') <> ''"]
values: list[str] = []
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
clauses.append(f"coalesce(t.account_code_final, '') not in ({excluded_placeholders})")
values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
if date_from:
clauses.append("coalesce(t.transaction_date, '') >= ?")
values.append(date_from)
if date_to:
clauses.append("coalesce(t.transaction_date, '') <= ?")
values.append(date_to)
where = f"where {' and '.join(clauses)}"
rows = conn.execute(
f"""
select
substr(coalesce(t.transaction_date, ''), 1, 4) as year,
{project_type_case} as project_type,
coalesce(sum(case when t.in_out = '입금' then t.supply_amount else 0 end), 0) as income_supply,
coalesce(sum(case when t.in_out = '출금' then t.supply_amount else 0 end), 0) as expense_supply
from ptc_transactions t
left join project_master pm on pm.project_code = t.project_code
{where}
group by substr(coalesce(t.transaction_date, ''), 1, 4), {project_type_case}
having coalesce(substr(coalesce(t.transaction_date, ''), 1, 4), '') <> ''
order by year asc, project_type asc
""",
values,
).fetchall()
preferred_types = ["시공", "영업", "설계", "관리", "미지정"]
by_year: dict[str, dict] = {}
discovered_types: set[str] = set()
for row in rows:
year = (row["year"] or "").strip() or "미상"
project_type = (row["project_type"] or "").strip() or "미지정"
discovered_types.add(project_type)
if year not in by_year:
by_year[year] = {
"year": year,
"income_supply": 0.0,
"expense_supply": 0.0,
"types": {},
}
income_supply = float(row["income_supply"] or 0)
expense_supply = float(row["expense_supply"] or 0)
by_year[year]["income_supply"] += income_supply
by_year[year]["expense_supply"] += expense_supply
by_year[year]["types"][project_type] = {
"project_type": project_type,
"income_supply": income_supply,
"expense_supply": expense_supply,
}
ordered_types = [item for item in preferred_types if item in discovered_types]
ordered_types.extend(sorted(discovered_types - set(ordered_types)))
items = []
for year in sorted(by_year.keys()):
item = by_year[year]
total_income = float(item["income_supply"] or 0)
total_expense = float(item["expense_supply"] or 0)
type_items = []
for project_type in ordered_types:
type_item = item["types"].get(project_type) or {
"project_type": project_type,
"income_supply": 0.0,
"expense_supply": 0.0,
}
type_items.append(
{
**type_item,
"income_ratio": (float(type_item["income_supply"] or 0) / total_income * 100) if total_income > 0 else 0.0,
"expense_ratio": (float(type_item["expense_supply"] or 0) / total_expense * 100) if total_expense > 0 else 0.0,
}
)
items.append(
{
"year": year,
"income_supply": total_income,
"expense_supply": total_expense,
"profit_supply": total_income - total_expense,
"margin_rate": ((total_income - total_expense) / total_income * 100) if total_income > 0 else 0.0,
"types": type_items,
}
)
self._send(200, {"project_type_order": ordered_types, "items": items})
return
if parsed.path == "/api/management-overview-accounts":
year = params.get("year", [""])[0].strip()
category = params.get("category", [""])[0].strip()
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
if not year or not category:
self._send(400, {"ok": False, "message": "year and category are required"})
return
if category not in MANAGEMENT_ACCOUNT_CATEGORY_ORDER:
self._send(400, {"ok": False, "message": "invalid category"})
return
category_codes = [
code
for code, meta in ACCOUNT_MASTER.items()
if meta.get("project_type") == "관리" and meta.get("category") == category
]
if not category_codes:
self._send(200, {"items": []})
return
placeholders = ",".join("?" for _ in category_codes)
clauses = [
"(coalesce(project_type, '') = '관리' or coalesce(project_code, '') like '%-관리-%')",
"in_out = '출금'",
"substr(coalesce(transaction_date, ''), 1, 4) = ?",
f"account_code_final in ({placeholders})",
]
values: list[str] = [year, *category_codes]
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
clauses.append(f"coalesce(account_code_final, '') not in ({excluded_placeholders})")
values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
if date_from:
clauses.append("coalesce(transaction_date, '') >= ?")
values.append(date_from)
if date_to:
clauses.append("coalesce(transaction_date, '') <= ?")
values.append(date_to)
where = " where " + " and ".join(clauses)
rows = conn.execute(
f"""
select
account_code_final as account_code,
max(account_name_final) as account_name,
count(*) as transaction_count,
coalesce(sum(supply_amount), 0) as expense_amount
from ptc_transactions
{where}
group by account_code_final
order by expense_amount desc, account_code_final asc
""",
values,
).fetchall()
self._send(200, {"items": rows_to_dicts(rows)})
return
if parsed.path == "/api/management-company-accounts":
year = params.get("year", [""])[0].strip()
project_type = params.get("project_type", [""])[0].strip()
if not year or not project_type:
self._send(400, {"ok": False, "message": "year and project_type are required"})
return
project_type_case = """
case
when coalesce(t.project_code, '') like '%-시공-%' then '시공'
when coalesce(t.project_code, '') like '%-영업-%' then '영업'
when coalesce(t.project_code, '') like '%-설계-%' then '설계'
when coalesce(t.project_code, '') like '%-관리-%' then '관리'
else coalesce(pm.project_type, t.project_type, '미지정')
end
"""
clauses = [
"substr(coalesce(t.transaction_date, ''), 1, 4) = ?",
"coalesce(t.account_code_final, '') <> ''",
f"{project_type_case} = ?",
]
values: list[str] = [year, project_type]
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
clauses.append(f"coalesce(t.account_code_final, '') not in ({excluded_placeholders})")
values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
where = f"where {' and '.join(clauses)}"
rows = conn.execute(
f"""
select
t.account_code_final as account_code,
max(t.account_name_final) as account_name,
sum(case when t.in_out = '입금' then 1 else 0 end) as income_count,
sum(case when t.in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when t.in_out = '입금' then t.supply_amount else 0 end), 0) as income_supply,
coalesce(sum(case when t.in_out = '출금' then t.supply_amount else 0 end), 0) as expense_supply,
count(*) as txn_count
from ptc_transactions t
left join project_master pm on pm.project_code = t.project_code
{where}
group by t.account_code_final
order by (coalesce(sum(case when t.in_out = '입금' then t.supply_amount else 0 end), 0) +
coalesce(sum(case when t.in_out = '출금' then t.supply_amount else 0 end), 0)) desc,
t.account_code_final asc
""",
values,
).fetchall()
self._send(200, {"items": rows_to_dicts(rows)})
return
if parsed.path == "/api/company-account-detail":
year = params.get("year", [""])[0].strip()
project_type = params.get("project_type", [""])[0].strip()
account_code = params.get("account_code", [""])[0].strip()
if not year or not project_type or not account_code:
self._send(400, {"ok": False, "message": "year, project_type and account_code are required"})
return
project_type_case = """
case
when coalesce(t.project_code, '') like '%-시공-%' then '시공'
when coalesce(t.project_code, '') like '%-영업-%' then '영업'
when coalesce(t.project_code, '') like '%-설계-%' then '설계'
when coalesce(t.project_code, '') like '%-관리-%' then '관리'
else coalesce(pm.project_type, t.project_type, '미지정')
end
"""
detail_clauses = [
"substr(coalesce(t.transaction_date, ''), 1, 4) = ?",
f"{project_type_case} = ?",
"coalesce(t.account_code_final, '') = ?",
]
detail_values: list[str] = [year, project_type, account_code]
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
detail_clauses.append(f"coalesce(t.account_code_final, '') not in ({excluded_placeholders})")
detail_values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
detail_where = " where " + " and ".join(detail_clauses)
summary = conn.execute(
f"""
select
t.account_code_final as account_code,
max(t.account_name_final) as account_name,
count(*) as txn_count,
sum(case when t.in_out = '입금' then 1 else 0 end) as income_count,
sum(case when t.in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when t.in_out = '입금' then t.supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when t.in_out = '출금' then t.supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(t.supply_amount), 0) as supply_sum,
min(t.transaction_date) as min_date,
max(t.transaction_date) as max_date
from ptc_transactions t
left join project_master pm on pm.project_code = t.project_code
{detail_where}
group by t.account_code_final
""",
detail_values,
).fetchone()
project_rows = conn.execute(
f"""
select
t.project_code,
max(t.project_name) as project_name,
count(*) as txn_count,
sum(case when t.in_out = '입금' then 1 else 0 end) as income_count,
sum(case when t.in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when t.in_out = '입금' then t.supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when t.in_out = '출금' then t.supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(t.supply_amount), 0) as supply_sum
from ptc_transactions t
left join project_master pm on pm.project_code = t.project_code
{detail_where}
group by t.project_code
order by supply_sum desc, t.project_code desc
limit 50
""",
detail_values,
).fetchall()
transaction_rows = conn.execute(
f"""
select
t.source_row_no,
t.transaction_date,
t.in_out,
t.project_code,
t.project_name,
t.vendor_name,
t.description,
t.supply_amount
from ptc_transactions t
left join project_master pm on pm.project_code = t.project_code
{detail_where}
order by t.transaction_date desc, t.source_row_no desc
limit 100
""",
detail_values,
).fetchall()
allocated_projects, allocation_meta = build_company_allocated_project_rows(
conn, project_rows, project_type
)
self._send(
200,
{
"summary": dict(summary) if summary else None,
"projects": allocated_projects,
"project_allocation": allocation_meta,
"transactions": rows_to_dicts(transaction_rows),
},
)
return
if parsed.path == "/api/account-detail":
account_code = params.get("account_code", [""])[0].strip()
project_code = params.get("project_code", [""])[0].strip()
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
if not account_code:
self._send(400, {"ok": False, "message": "account_code is required"})
return
detail_clauses = ["account_code_final = ?"]
detail_values: list[str] = [account_code]
if project_code:
detail_clauses.append("project_code = ?")
detail_values.append(project_code)
if date_from:
detail_clauses.append("coalesce(transaction_date, '') >= ?")
detail_values.append(date_from)
if date_to:
detail_clauses.append("coalesce(transaction_date, '') <= ?")
detail_values.append(date_to)
detail_where = " where " + " and ".join(detail_clauses)
summary = conn.execute(
f"""
select
account_code_final as account_code,
account_name_final as account_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{detail_where}
group by account_code_final, account_name_final
""",
detail_values,
).fetchone()
project_rows = conn.execute(
f"""
select
project_code,
max(project_name) as project_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum
from ptc_transactions
{detail_where}
group by project_code
order by supply_sum desc, project_code
limit 30
""",
detail_values,
).fetchall()
vendor_rows = conn.execute(
f"""
select
vendor_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum
from ptc_transactions
{detail_where}
group by vendor_name
order by supply_sum desc, vendor_name
limit 30
""",
detail_values,
).fetchall()
transaction_rows = conn.execute(
f"""
select
source_row_no,
transaction_date,
in_out,
project_code,
project_name,
vendor_name,
department_name,
description,
supply_amount
from ptc_transactions
{detail_where}
order by transaction_date desc, source_row_no desc
limit 100
""",
detail_values,
).fetchall()
self._send(
200,
{
"summary": dict(summary) if summary else None,
"projects": rows_to_dicts(project_rows),
"vendors": rows_to_dicts(vendor_rows),
"transactions": rows_to_dicts(transaction_rows),
},
)
return
if parsed.path == "/api/management-account-detail":
account_code = params.get("account_code", [""])[0].strip()
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
if not account_code:
self._send(400, {"ok": False, "message": "account_code is required"})
return
detail_clauses = [
"account_code_final = ?",
"(coalesce(project_type, '') = '관리' or coalesce(project_code, '') like '%-관리-%')",
]
detail_values: list[str] = [account_code]
if MANAGEMENT_EXCLUDED_ACCOUNT_CODES:
excluded_placeholders = ",".join("?" for _ in MANAGEMENT_EXCLUDED_ACCOUNT_CODES)
detail_clauses.append(f"coalesce(account_code_final, '') not in ({excluded_placeholders})")
detail_values.extend(sorted(MANAGEMENT_EXCLUDED_ACCOUNT_CODES))
if date_from:
detail_clauses.append("coalesce(transaction_date, '') >= ?")
detail_values.append(date_from)
if date_to:
detail_clauses.append("coalesce(transaction_date, '') <= ?")
detail_values.append(date_to)
detail_where = " where " + " and ".join(detail_clauses)
summary = conn.execute(
f"""
select
account_code_final as account_code,
account_name_final as account_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{detail_where}
group by account_code_final, account_name_final
""",
detail_values,
).fetchone()
project_rows = conn.execute(
f"""
select
project_code,
max(project_name) as project_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum
from ptc_transactions
{detail_where}
group by project_code
order by supply_sum desc, project_code desc
limit 50
""",
detail_values,
).fetchall()
transaction_rows = conn.execute(
f"""
select
source_row_no,
transaction_date,
in_out,
project_code,
project_name,
vendor_name,
department_name,
description,
supply_amount
from ptc_transactions
{detail_where}
order by transaction_date desc, source_row_no desc
limit 100
""",
detail_values,
).fetchall()
self._send(
200,
{
"summary": dict(summary) if summary else None,
"projects": rows_to_dicts(project_rows),
"transactions": rows_to_dicts(transaction_rows),
},
)
return
if parsed.path == "/api/lifecycle-account-detail":
project_code = params.get("project_code", [""])[0].strip()
bucket_label = params.get("bucket_label", [""])[0].strip()
account_code = params.get("account_code", [""])[0].strip()
if not project_code or not bucket_label or not account_code:
self._send(400, {"ok": False, "message": "project_code, bucket_label and account_code are required"})
return
master = fetch_project_master(conn, project_code) or fetch_project_defaults(conn, project_code)
project_name = (master or {}).get("project_name") or ""
resolved_project_type = resolve_project_type(project_code, (master or {}).get("project_type") or "")
related_projects = build_related_projects(conn, project_code, project_name)
detail = build_lifecycle_account_detail(
conn,
related_projects,
project_code,
resolved_project_type,
bucket_label,
account_code,
)
if not detail:
self._send(404, {"ok": False, "message": "lifecycle account detail not found"})
return
self._send(200, detail)
return
if parsed.path == "/api/management-excluded-account-detail":
account_code = params.get("account_code", [""])[0].strip()
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
if not account_code:
self._send(400, {"ok": False, "message": "account_code is required"})
return
detail_clauses = [
"account_code_final = ?",
"(coalesce(project_type, '') = '관리' or coalesce(project_code, '') like '%-관리-%')",
]
detail_values: list[str] = [account_code]
if date_from:
detail_clauses.append("coalesce(transaction_date, '') >= ?")
detail_values.append(date_from)
if date_to:
detail_clauses.append("coalesce(transaction_date, '') <= ?")
detail_values.append(date_to)
detail_where = " where " + " and ".join(detail_clauses)
summary = conn.execute(
f"""
select
account_code_final as account_code,
account_name_final as account_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{detail_where}
group by account_code_final, account_name_final
""",
detail_values,
).fetchone()
project_rows = conn.execute(
f"""
select
project_code,
max(project_name) as project_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum
from ptc_transactions
{detail_where}
group by project_code
order by supply_sum desc, project_code desc
limit 50
""",
detail_values,
).fetchall()
transaction_rows = conn.execute(
f"""
select
source_row_no,
transaction_date,
in_out,
project_code,
project_name,
vendor_name,
department_name,
description,
supply_amount
from ptc_transactions
{detail_where}
order by transaction_date desc, source_row_no desc
limit 100
""",
detail_values,
).fetchall()
self._send(
200,
{
"summary": dict(summary) if summary else None,
"projects": rows_to_dicts(project_rows),
"transactions": rows_to_dicts(transaction_rows),
},
)
return
if parsed.path == "/api/vendor-detail":
vendor_name = params.get("vendor_name", [""])[0].strip()
project_code = params.get("project_code", [""])[0].strip()
account_code = params.get("account_code", [""])[0].strip()
date_from = params.get("date_from", [""])[0].strip()
date_to = params.get("date_to", [""])[0].strip()
if not vendor_name:
self._send(400, {"ok": False, "message": "vendor_name is required"})
return
detail_clauses = ["vendor_name = ?"]
detail_values: list[str] = [vendor_name]
if project_code:
detail_clauses.append("project_code = ?")
detail_values.append(project_code)
if account_code:
detail_clauses.append("account_code_final = ?")
detail_values.append(account_code)
if date_from:
detail_clauses.append("coalesce(transaction_date, '') >= ?")
detail_values.append(date_from)
if date_to:
detail_clauses.append("coalesce(transaction_date, '') <= ?")
detail_values.append(date_to)
detail_where = " where " + " and ".join(detail_clauses)
summary = conn.execute(
f"""
select
vendor_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{detail_where}
group by vendor_name
""",
detail_values,
).fetchone()
project_rows = conn.execute(
f"""
select
project_code,
max(project_name) as project_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum
from ptc_transactions
{detail_where}
group by project_code
order by supply_sum desc, project_code
limit 20
""",
detail_values,
).fetchall()
account_rows = conn.execute(
f"""
select
account_code_final as account_code,
account_name_final as account_name,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_supply_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_supply_sum,
coalesce(sum(supply_amount), 0) as supply_sum
from ptc_transactions
{detail_where}
group by account_code_final, account_name_final
order by supply_sum desc, account_code_final
limit 30
""",
detail_values,
).fetchall()
transaction_rows = conn.execute(
f"""
select
source_row_no,
transaction_date,
in_out,
project_code,
project_name,
account_code_final as account_code,
account_name_final as account_name,
department_name,
description,
supply_amount
from ptc_transactions
{detail_where}
order by transaction_date desc, source_row_no desc
limit 100
""",
detail_values,
).fetchall()
self._send(
200,
{
"summary": dict(summary) if summary else None,
"projects": rows_to_dicts(project_rows),
"accounts": rows_to_dicts(account_rows),
"transactions": rows_to_dicts(transaction_rows),
},
)
return
if parsed.path == "/api/project-detail":
project_code = params.get("project_code", [""])[0].strip()
if not project_code:
self._send(400, {"ok": False, "message": "project_code is required"})
return
keyword = params.get("keyword", [""])[0]
in_out = params.get("in_out", ["전체"])[0]
where, values = build_project_where(project_code, keyword, in_out)
summary = conn.execute(
f"""
select
project_code,
max(project_name) as project_name,
max(project_type) as project_type,
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(supply_amount), 0) as supply_sum,
coalesce(sum(vat_amount), 0) as vat_sum,
coalesce(sum(total_amount), 0) as total_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
{where}
""",
values,
).fetchone()
account_rows = conn.execute(
f"""
select
account_code_final as code,
account_name_final as name,
count(*) as count,
coalesce(sum(supply_amount), 0) as total
from ptc_transactions
{where}
group by account_code_final, account_name_final
order by total desc
limit 12
""",
values,
).fetchall()
transaction_rows = conn.execute(
f"""
select
source_row_no,
transaction_date,
in_out,
account_code_final as account_code,
account_name_final as account_name,
department_name,
vendor_name,
description,
supply_amount,
vat_amount,
total_amount
from ptc_transactions
{where}
order by transaction_date desc, source_row_no desc
limit 20
""",
values,
).fetchall()
summary_dict = dict(summary) if summary else None
master = fetch_project_master(conn, project_code)
if summary_dict and master:
summary_dict["project_name"] = master.get("project_name") or summary_dict["project_name"]
summary_dict["project_type"] = resolve_project_type(
project_code,
summary_dict["project_type"],
master.get("project_type"),
)
summary_dict["construction_family"] = resolve_construction_family(
master.get("construction_method"),
master.get("construction_family"),
)
summary_dict["construction_method"] = master.get("construction_method") or ""
summary_dict["start_date"] = master.get("start_date") or ""
summary_dict["end_date"] = master.get("end_date") or ""
summary_dict["note"] = master.get("note") or ""
elif summary_dict:
summary_dict["project_type"] = resolve_project_type(project_code, summary_dict["project_type"])
summary_dict["construction_family"] = resolve_construction_family("")
summary_dict["construction_method"] = ""
summary_dict["start_date"] = ""
summary_dict["end_date"] = ""
summary_dict["note"] = ""
related_projects = build_related_projects(
conn,
project_code,
summary_dict.get("project_name") if summary_dict else "",
)
lifecycle_cost = build_project_lifecycle_cost(
conn,
related_projects,
summary_dict["project_type"] if summary_dict else "",
project_code,
)
account_issues = get_project_account_issues(
conn,
project_code,
summary_dict["project_type"] if summary_dict else "",
)
budget_analysis = build_budget_analysis(conn, project_code, build_account_structure_rows(account_rows))
self._send(
200,
{
"summary": summary_dict,
"project_master": master,
"account_structure": build_account_structure_rows(account_rows),
"budget_analysis": budget_analysis,
"accounts": rows_to_dicts(account_rows),
"account_issues": account_issues,
"transactions": rows_to_dicts(transaction_rows),
"related_projects": related_projects,
"lifecycle_cost": lifecycle_cost,
},
)
return
if parsed.path == "/api/project-account-issue-detail":
project_code = params.get("project_code", [""])[0].strip()
account_code = params.get("account_code", [""])[0].strip()
if not project_code or not account_code:
self._send(400, {"ok": False, "message": "project_code and account_code are required"})
return
rows = conn.execute(
"""
select
source_row_no,
transaction_date,
in_out,
account_code_final as account_code,
account_name_final as account_name,
department_name,
vendor_name,
description,
supply_amount,
vat_amount,
total_amount
from ptc_transactions
where project_code = ?
and account_code_final = ?
order by transaction_date desc, source_row_no desc
limit 100
""",
(project_code, account_code),
).fetchall()
summary = conn.execute(
"""
select
count(*) as txn_count,
coalesce(sum(supply_amount), 0) as supply_sum,
min(transaction_date) as min_date,
max(transaction_date) as max_date
from ptc_transactions
where project_code = ?
and account_code_final = ?
""",
(project_code, account_code),
).fetchone()
self._send(
200,
{
"project_code": project_code,
"account_code": account_code,
"account_name": resolve_account_name(account_code, rows[0]["account_name"] if rows else ""),
"summary": dict(summary) if summary else None,
"items": rows_to_dicts(rows),
},
)
return
if parsed.path == "/api/project-budget-actual-detail":
project_code = params.get("project_code", [""])[0].strip()
section = params.get("section", [""])[0].strip()
group_name = params.get("group_name", [""])[0].strip()
category = params.get("category", [""])[0].strip()
if not project_code or not section or not group_name or not category:
self._send(400, {"ok": False, "message": "project_code, section, group_name, category are required"})
return
category_accounts = get_category_account_items(section, group_name, category)
account_codes = [item["account_code"] for item in category_accounts if item.get("account_code")]
if not account_codes:
self._send(
200,
{
"project_code": project_code,
"section": section,
"group_name": group_name,
"category": category,
"summary": {"txn_count": 0, "income_count": 0, "expense_count": 0, "income_sum": 0, "expense_sum": 0, "supply_sum": 0},
"accounts": [],
"transactions": [],
},
)
return
placeholders = ",".join("?" for _ in account_codes)
values = [project_code, *account_codes]
summary = conn.execute(
f"""
select
count(*) as txn_count,
sum(case when in_out = '입금' then 1 else 0 end) as income_count,
sum(case when in_out = '출금' then 1 else 0 end) as expense_count,
coalesce(sum(case when in_out = '입금' then supply_amount else 0 end), 0) as income_sum,
coalesce(sum(case when in_out = '출금' then supply_amount else 0 end), 0) as expense_sum,
coalesce(sum(supply_amount), 0) as supply_sum
from ptc_transactions
where project_code = ?
and account_code_final in ({placeholders})
""",
values,
).fetchone()
account_rows = conn.execute(
f"""
select
account_code_final as account_code,
max(account_name_final) as account_name,
count(*) as txn_count,
coalesce(sum(supply_amount), 0) as supply_sum
from ptc_transactions
where project_code = ?
and account_code_final in ({placeholders})
group by account_code_final
order by account_code_final
""",
values,
).fetchall()
transaction_rows = conn.execute(
f"""
select
source_row_no,
transaction_date,
in_out,
account_code_final as account_code,
account_name_final as account_name,
department_name,
vendor_name,
description,
supply_amount,
vat_amount,
total_amount
from ptc_transactions
where project_code = ?
and account_code_final in ({placeholders})
order by transaction_date desc, source_row_no desc
limit 100
""",
values,
).fetchall()
self._send(
200,
{
"project_code": project_code,
"section": section,
"group_name": group_name,
"category": category,
"summary": dict(summary) if summary else None,
"accounts": rows_to_dicts(account_rows),
"transactions": rows_to_dicts(transaction_rows),
},
)
return
if parsed.path == "/api/top-accounts":
where, values = build_where(params)
rows = conn.execute(
f"""
select
account_code_final as code,
account_name_final as name,
count(*) as count,
coalesce(sum(supply_amount), 0) as total
from ptc_transactions
{where}
group by account_code_final, account_name_final
order by total desc
limit 10
""",
values,
).fetchall()
self._send(200, {"items": rows_to_dicts(rows)})
return
if parsed.path == "/api/top-projects":
where, values = build_where(params)
rows = conn.execute(
f"""
select
coalesce(project_code, '(없음)') as project_code,
coalesce(project_name, '(없음)') as project_name,
coalesce(project_type, '(없음)') as project_type,
count(*) as count,
coalesce(sum(supply_amount), 0) as total
from ptc_transactions
{where}
group by project_code, project_name, project_type
order by total desc
limit 10
""",
values,
).fetchall()
self._send(200, {"items": rows_to_dicts(rows)})
return
if parsed.path == "/api/project-mismatches":
rows = conn.execute(
"""
select project_code, count(distinct project_name) as name_count, count(distinct project_type) as type_count
from ptc_transactions
where coalesce(project_code, '') <> ''
group by project_code
having count(distinct project_name) > 1 or count(distinct project_type) > 1
order by project_code
limit 20
"""
).fetchall()
self._send(200, {"items": rows_to_dicts(rows)})
return
if parsed.path == "/api/transactions":
where, values = build_where(params)
limit = int(params.get("limit", ["30"])[0])
rows = conn.execute(
f"""
select
source_row_no,
transaction_date,
in_out,
account_code_final as account_code,
account_name_final as account_name,
department_name,
vendor_name,
project_code,
project_name,
project_type,
description,
supply_amount,
vat_amount,
total_amount
from ptc_transactions
{where}
order by source_row_no
limit ?
""",
values + [limit],
).fetchall()
self._send(200, {"items": rows_to_dicts(rows)})
return
self._send(404, {"ok": False, "message": "Not found"})
finally:
conn.close()
def main() -> None:
init_db()
server = ThreadingHTTPServer(("0.0.0.0", 4000), Handler)
print("PTC API server listening on http://0.0.0.0:4000")
server.serve_forever()
if __name__ == "__main__":
main()