458 lines
16 KiB
Python
458 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from runtime_config import DB_PATH
|
|
|
|
|
|
SHADOW_VERSION = "voucher-shadow-v1-wehago-anchor-diagnostic"
|
|
FINAL_STATUSES = ("voucher_matched", "voucher_recheck", "voucher_unmatched", "voucher_excepted")
|
|
|
|
|
|
def clean(value: Any) -> str:
|
|
return "" if value is None else str(value).strip()
|
|
|
|
|
|
def amount(value: Any) -> float:
|
|
try:
|
|
return float(value or 0)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def erp_base(value: Any) -> str:
|
|
text = clean(value)
|
|
return re.sub(r"-\d+$", "", text) if text else ""
|
|
|
|
|
|
def has_ledger(row: sqlite3.Row | dict[str, Any]) -> bool:
|
|
return bool(
|
|
clean(row["ledger_account_name"])
|
|
or clean(row["ledger_desc"])
|
|
or abs(amount(row["ledger_debit"])) >= 0.5
|
|
or abs(amount(row["ledger_credit"])) >= 0.5
|
|
)
|
|
|
|
|
|
def has_erp(row: sqlite3.Row | dict[str, Any]) -> bool:
|
|
return bool(
|
|
clean(row["voucher_account_name"])
|
|
or clean(row["voucher_desc"])
|
|
or abs(amount(row["voucher_debit"])) >= 0.5
|
|
or abs(amount(row["voucher_credit"])) >= 0.5
|
|
)
|
|
|
|
|
|
def latest_signature(conn: sqlite3.Connection, start_year: int, end_year: int, status: str) -> str:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT signature
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = ?
|
|
GROUP BY signature
|
|
ORDER BY MAX(updated_at) DESC
|
|
LIMIT 1
|
|
""",
|
|
(start_year, end_year, status),
|
|
).fetchone()
|
|
return clean(row[0]) if row else ""
|
|
|
|
|
|
def status_invariants(
|
|
conn: sqlite3.Connection,
|
|
start_year: int,
|
|
end_year: int,
|
|
signatures: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
result: dict[str, Any] = {}
|
|
for status in ("voucher_unmatched", "voucher_excepted"):
|
|
signature = signatures.get(status, "")
|
|
if not signature:
|
|
result[status] = {"missing_projection": True}
|
|
continue
|
|
mixed = conn.execute(
|
|
"""
|
|
WITH per_group AS (
|
|
SELECT group_index,
|
|
COUNT(DISTINCT CASE
|
|
WHEN COALESCE(ledger_account_name, '') <> ''
|
|
OR ABS(COALESCE(ledger_debit, 0)) >= 0.5
|
|
OR ABS(COALESCE(ledger_credit, 0)) >= 0.5
|
|
THEN COALESCE(ledger_date, '') || '|' || COALESCE(voucher_no, '')
|
|
END) AS wehago_keys
|
|
FROM wehago_compare_query_rows
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = ?
|
|
AND signature = ?
|
|
GROUP BY group_index
|
|
)
|
|
SELECT COUNT(*) FROM per_group WHERE wehago_keys > 1
|
|
""",
|
|
(start_year, end_year, status, signature),
|
|
).fetchone()[0]
|
|
zero_direct = conn.execute(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM wehago_compare_query_rows
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = ?
|
|
AND signature = ?
|
|
AND matched_case = 'DIRECT_MATCH_CANDIDATE'
|
|
AND (
|
|
MAX(ABS(ledger_debit), ABS(ledger_credit)) < 0.5
|
|
OR MAX(ABS(voucher_debit), ABS(voucher_credit)) < 0.5
|
|
)
|
|
""",
|
|
(start_year, end_year, status, signature),
|
|
).fetchone()[0]
|
|
result[status] = {
|
|
"signature": signature,
|
|
"mixed_wehago_groups": int(mixed or 0),
|
|
"zero_or_one_sided_direct_rows": int(zero_direct or 0),
|
|
"classification_changed": False,
|
|
}
|
|
|
|
source_anchors: dict[tuple[int, str, str], set[str]] = defaultdict(set)
|
|
for status in FINAL_STATUSES:
|
|
signature = signatures.get(status, "")
|
|
if not signature:
|
|
continue
|
|
for row in conn.execute(
|
|
"""
|
|
SELECT fiscal_year, ledger_date, voucher_no
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = ?
|
|
AND signature = ?
|
|
AND COALESCE(voucher_no, '') <> ''
|
|
""",
|
|
(start_year, end_year, status, signature),
|
|
):
|
|
source_anchors[(int(row[0] or 0), clean(row[1]), clean(row[2]))].add(status)
|
|
source_overlaps = {
|
|
"|".join(map(str, key)): sorted(statuses)
|
|
for key, statuses in source_anchors.items()
|
|
if len(statuses) > 1
|
|
}
|
|
final_signature_row = conn.execute(
|
|
"""
|
|
SELECT signature
|
|
FROM wehago_compare_final_status_projection
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
GROUP BY signature
|
|
ORDER BY MAX(updated_at) DESC
|
|
LIMIT 1
|
|
""",
|
|
(start_year, end_year),
|
|
).fetchone()
|
|
final_signature = clean(final_signature_row[0]) if final_signature_row else ""
|
|
final_duplicates = []
|
|
if final_signature:
|
|
final_duplicates = conn.execute(
|
|
"""
|
|
SELECT fiscal_year, ledger_date, voucher_no, COUNT(DISTINCT final_status) AS status_count
|
|
FROM wehago_compare_final_status_projection
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND signature = ?
|
|
GROUP BY fiscal_year, ledger_date, voucher_no
|
|
HAVING status_count > 1
|
|
LIMIT 20
|
|
""",
|
|
(start_year, end_year, final_signature),
|
|
).fetchall()
|
|
result["cross_status"] = {
|
|
"final_signature": final_signature,
|
|
"duplicate_final_status_vouchers": len(final_duplicates),
|
|
"final_duplicate_samples": [tuple(row) for row in final_duplicates],
|
|
"source_projection_overlap_count": len(source_overlaps),
|
|
"source_projection_overlap_note": "후보 source projection 간 중첩이며 final status 중복과는 다릅니다.",
|
|
"source_overlap_samples": dict(list(source_overlaps.items())[:20]),
|
|
}
|
|
return result
|
|
|
|
|
|
def build_shadow(
|
|
conn: sqlite3.Connection,
|
|
start_year: int,
|
|
end_year: int,
|
|
signature: str,
|
|
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
|
group_rows = conn.execute(
|
|
"""
|
|
SELECT *
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = 'voucher_matched'
|
|
AND signature = ?
|
|
ORDER BY group_index
|
|
""",
|
|
(start_year, end_year, signature),
|
|
).fetchall()
|
|
rows_by_group: dict[int, list[sqlite3.Row]] = defaultdict(list)
|
|
for row in conn.execute(
|
|
"""
|
|
SELECT *
|
|
FROM wehago_compare_query_rows
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = 'voucher_matched'
|
|
AND signature = ?
|
|
ORDER BY group_index, row_index
|
|
""",
|
|
(start_year, end_year, signature),
|
|
):
|
|
rows_by_group[int(row["group_index"])] .append(row)
|
|
|
|
shadow_groups: list[dict[str, Any]] = []
|
|
classifications: Counter[str] = Counter()
|
|
mixed_source_groups = 0
|
|
for group in group_rows:
|
|
source_index = int(group["group_index"])
|
|
rows = rows_by_group.get(source_index, [])
|
|
anchors = sorted(
|
|
{
|
|
(int(row["fiscal_year"] or 0), clean(row["ledger_date"]), clean(row["voucher_no"]))
|
|
for row in rows
|
|
if has_ledger(row)
|
|
}
|
|
)
|
|
if len(anchors) <= 1:
|
|
continue
|
|
mixed_source_groups += 1
|
|
erp_bases = {erp_base(row["draft_no"]) for row in rows if has_erp(row) and erp_base(row["draft_no"])}
|
|
direct_rows = [
|
|
row
|
|
for row in rows
|
|
if has_ledger(row)
|
|
and has_erp(row)
|
|
and max(abs(amount(row["ledger_debit"])), abs(amount(row["ledger_credit"]))) >= 0.5
|
|
and max(abs(amount(row["voucher_debit"])), abs(amount(row["voucher_credit"]))) >= 0.5
|
|
]
|
|
exact_direct = sum(
|
|
1
|
|
for row in direct_rows
|
|
if abs(
|
|
max(abs(amount(row["ledger_debit"])), abs(amount(row["ledger_credit"])))
|
|
- max(abs(amount(row["voucher_debit"])), abs(amount(row["voucher_credit"])))
|
|
)
|
|
< 0.5
|
|
)
|
|
ledger_debit = sum(amount(row["ledger_debit"]) for row in rows if has_ledger(row))
|
|
ledger_credit = sum(amount(row["ledger_credit"]) for row in rows if has_ledger(row))
|
|
voucher_debit = sum(amount(row["voucher_debit"]) for row in rows if has_erp(row))
|
|
voucher_credit = sum(amount(row["voucher_credit"]) for row in rows if has_erp(row))
|
|
balanced = abs(ledger_debit - voucher_debit) < 0.5 and abs(ledger_credit - voucher_credit) < 0.5
|
|
ledger_nonzero = sum(
|
|
1
|
|
for row in rows
|
|
if has_ledger(row)
|
|
and max(abs(amount(row["ledger_debit"])), abs(amount(row["ledger_credit"]))) >= 0.5
|
|
)
|
|
if len(erp_bases) == 1 and balanced:
|
|
classification = "normal_structural_n_to_one_or_one_to_n"
|
|
elif ledger_nonzero and exact_direct >= max(1, int(ledger_nonzero * 0.8)):
|
|
classification = "display_only_grouping_issue"
|
|
else:
|
|
classification = "likely_mismatch_requires_review"
|
|
classifications[classification] += 1
|
|
erp_context_rows = sum(1 for row in rows if has_erp(row) and not has_ledger(row))
|
|
for anchor in anchors:
|
|
anchor_rows = [
|
|
row
|
|
for row in rows
|
|
if has_ledger(row)
|
|
and (int(row["fiscal_year"] or 0), clean(row["ledger_date"]), clean(row["voucher_no"])) == anchor
|
|
]
|
|
shadow_groups.append(
|
|
{
|
|
"source_group_index": source_index,
|
|
"fiscal_year": anchor[0],
|
|
"ledger_date": anchor[1],
|
|
"voucher_no": anchor[2],
|
|
"draft_bases": sorted(erp_bases),
|
|
"classification": classification,
|
|
"source_anchor_count": len(anchors),
|
|
"ledger_row_count": len(anchor_rows),
|
|
"erp_context_row_count": erp_context_rows,
|
|
"source_review_reason": clean(group["review_reason"]),
|
|
}
|
|
)
|
|
summary = {
|
|
"source_groups": len(group_rows),
|
|
"mixed_source_groups": mixed_source_groups,
|
|
"shadow_wehago_groups": len(shadow_groups),
|
|
"classifications": dict(classifications),
|
|
"active_voucher_judgement_changed": False,
|
|
}
|
|
return shadow_groups, summary
|
|
|
|
|
|
def store_shadow(
|
|
conn: sqlite3.Connection,
|
|
start_year: int,
|
|
end_year: int,
|
|
source_signature: str,
|
|
groups: list[dict[str, Any]],
|
|
summary: dict[str, Any],
|
|
) -> int:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS wehago_voucher_shadow_runs (
|
|
shadow_version TEXT NOT NULL,
|
|
start_year INTEGER NOT NULL,
|
|
end_year INTEGER NOT NULL,
|
|
source_signature TEXT NOT NULL,
|
|
summary_json TEXT NOT NULL,
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (shadow_version, start_year, end_year, source_signature)
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS wehago_voucher_shadow_groups (
|
|
shadow_version TEXT NOT NULL,
|
|
start_year INTEGER NOT NULL,
|
|
end_year INTEGER NOT NULL,
|
|
source_signature TEXT NOT NULL,
|
|
shadow_group_index INTEGER NOT NULL,
|
|
source_group_index INTEGER NOT NULL,
|
|
fiscal_year INTEGER NOT NULL,
|
|
ledger_date TEXT NOT NULL,
|
|
voucher_no TEXT NOT NULL,
|
|
draft_no TEXT NOT NULL,
|
|
classification TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL,
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (
|
|
shadow_version, start_year, end_year, source_signature, shadow_group_index
|
|
)
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO wehago_voucher_shadow_runs (
|
|
shadow_version, start_year, end_year, source_signature, summary_json, created_at
|
|
) VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
|
|
ON CONFLICT(shadow_version, start_year, end_year, source_signature) DO UPDATE SET
|
|
summary_json = excluded.summary_json,
|
|
created_at = CURRENT_TIMESTAMP
|
|
""",
|
|
(SHADOW_VERSION, start_year, end_year, source_signature, json.dumps(summary, ensure_ascii=False)),
|
|
)
|
|
conn.execute(
|
|
"""
|
|
DELETE FROM wehago_voucher_shadow_groups
|
|
WHERE shadow_version = ?
|
|
AND start_year = ?
|
|
AND end_year = ?
|
|
AND source_signature = ?
|
|
""",
|
|
(SHADOW_VERSION, start_year, end_year, source_signature),
|
|
)
|
|
conn.executemany(
|
|
"""
|
|
INSERT INTO wehago_voucher_shadow_groups (
|
|
shadow_version, start_year, end_year, source_signature,
|
|
shadow_group_index, source_group_index, fiscal_year, ledger_date,
|
|
voucher_no, draft_no, classification, payload_json, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
|
|
""",
|
|
[
|
|
(
|
|
SHADOW_VERSION,
|
|
start_year,
|
|
end_year,
|
|
source_signature,
|
|
index,
|
|
int(group["source_group_index"]),
|
|
int(group["fiscal_year"]),
|
|
clean(group["ledger_date"]),
|
|
clean(group["voucher_no"]),
|
|
", ".join(group["draft_bases"]),
|
|
clean(group["classification"]),
|
|
json.dumps(group, ensure_ascii=False),
|
|
)
|
|
for index, group in enumerate(groups)
|
|
],
|
|
)
|
|
return len(groups)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Build a non-active Voucher shadow projection and invariant report.")
|
|
parser.add_argument("--start-year", type=int, default=2025)
|
|
parser.add_argument("--end-year", type=int, default=2025)
|
|
parser.add_argument("--output", default="")
|
|
args = parser.parse_args()
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
signatures = {
|
|
status: latest_signature(conn, args.start_year, args.end_year, status)
|
|
for status in FINAL_STATUSES
|
|
}
|
|
voucher_signature = signatures["voucher_matched"]
|
|
if not voucher_signature:
|
|
raise SystemExit("No voucher_matched query projection found.")
|
|
invariants = status_invariants(conn, args.start_year, args.end_year, signatures)
|
|
shadow_groups, shadow_summary = build_shadow(
|
|
conn,
|
|
args.start_year,
|
|
args.end_year,
|
|
voucher_signature,
|
|
)
|
|
stored = store_shadow(
|
|
conn,
|
|
args.start_year,
|
|
args.end_year,
|
|
voucher_signature,
|
|
shadow_groups,
|
|
shadow_summary,
|
|
)
|
|
conn.commit()
|
|
|
|
payload = {
|
|
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
|
"db": str(DB_PATH),
|
|
"shadow_version": SHADOW_VERSION,
|
|
"start_year": args.start_year,
|
|
"end_year": args.end_year,
|
|
"source_signatures": signatures,
|
|
"invariants": invariants,
|
|
"shadow_summary": shadow_summary,
|
|
"stored_shadow_groups": stored,
|
|
"samples": shadow_groups[:50],
|
|
}
|
|
output = Path(args.output) if args.output else (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "reports"
|
|
/ f"wehago_voucher_shadow_diagnostic_{args.start_year}_{args.end_year}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
|
)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps({"output": str(output), **shadow_summary, "stored_shadow_groups": stored}, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|