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]
clauses = ["t.project_code is not null", "t.project_code <> ''"]
values: list[str] = []
if project_type and project_type != "전체":
clauses.append("coalesce(pm.project_type, t.project_type) = ?")
values.append(project_type)
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(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 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, 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)
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,
"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)
self._send(
200,
{
"overview": overview,
"amount_buckets": amount_buckets,
"status_bands": status_bands,
"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]
clauses = ["project_code is not null", "project_code <> ''"]
values = []
if keyword:
like = f"%{keyword}%"
clauses.append(
"""
(
lower(coalesce(project_code, '')) like ?
or lower(coalesce(project_name, '')) like ?
or lower(coalesce(project_type, '')) like ?
)
"""
)
values.extend([like, like, like])
if project_type and project_type != "전체":
clauses.append("project_type = ?")
values.append(project_type)
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"] = ""
self._send(200, {"items": items})
return
if parsed.path == "/api/vendors":
keyword = params.get("keyword", [""])[0].strip().lower()
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])
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()
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])
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/account-detail":
account_code = params.get("account_code", [""])[0].strip()
project_code = params.get("project_code", [""])[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)
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(
"""
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
where account_code_final = ?
group by project_code
order by supply_sum desc, project_code
limit 30
""",
(account_code,),
).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/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()
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)
detail_where = " where " + " and ".join(detail_clauses)
summary = conn.execute(
"""
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
where vendor_name = ?
group by vendor_name
""",
(vendor_name,),
).fetchone()
project_rows = conn.execute(
"""
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
where vendor_name = ?
group by project_code
order by supply_sum desc, project_code
limit 20
""",
(vendor_name,),
).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"] = ""
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),
},
)
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()