507 lines
20 KiB
Python
507 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from runtime_config import DB_PATH # noqa: E402
|
|
from scripts.project_export_cache_ranges import ( # noqa: E402
|
|
current_ready_export_signature,
|
|
latest_export_signature,
|
|
project_range,
|
|
projection_signature,
|
|
)
|
|
from wehago_compare import ( # noqa: E402
|
|
QUERY_PROJECTION_VERSION,
|
|
_parse_row_date_with_year,
|
|
_raw_erp_entry_date_values,
|
|
_raw_erp_trace_prefilter,
|
|
_raw_erp_trace_prefilter_score,
|
|
_raw_erp_trace_score,
|
|
clean,
|
|
parse_amount,
|
|
)
|
|
|
|
|
|
TRACE_LOGIC_VERSION = "raw-erp-trace-candidate-v1"
|
|
SOURCE_STATUSES = ("voucher_unmatched", "voucher_recheck")
|
|
|
|
|
|
def _connect(db_path: Path = DB_PATH) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(db_path, timeout=30)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA busy_timeout = 30000")
|
|
return conn
|
|
|
|
|
|
def _ensure_schema(conn: sqlite3.Connection) -> None:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS wehago_raw_erp_trace_candidate_cache (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
candidate_key TEXT NOT NULL DEFAULT '',
|
|
fiscal_year INTEGER NOT NULL,
|
|
source_mode TEXT NOT NULL DEFAULT '',
|
|
source_signature TEXT NOT NULL DEFAULT '',
|
|
logic_version TEXT NOT NULL DEFAULT '',
|
|
status_key TEXT NOT NULL DEFAULT '',
|
|
group_index INTEGER NOT NULL DEFAULT 0,
|
|
row_index INTEGER NOT NULL DEFAULT 0,
|
|
ledger_date TEXT NOT NULL DEFAULT '',
|
|
voucher_no TEXT NOT NULL DEFAULT '',
|
|
ledger_account_name TEXT NOT NULL DEFAULT '',
|
|
ledger_vendor TEXT NOT NULL DEFAULT '',
|
|
ledger_desc TEXT NOT NULL DEFAULT '',
|
|
ledger_amount REAL NOT NULL DEFAULT 0,
|
|
erp_draft_no TEXT NOT NULL DEFAULT '',
|
|
erp_confirmed_no TEXT NOT NULL DEFAULT '',
|
|
erp_account_name TEXT NOT NULL DEFAULT '',
|
|
erp_vendor TEXT NOT NULL DEFAULT '',
|
|
erp_desc TEXT NOT NULL DEFAULT '',
|
|
erp_amount REAL NOT NULL DEFAULT 0,
|
|
amount_field TEXT NOT NULL DEFAULT '',
|
|
score REAL NOT NULL DEFAULT 0,
|
|
matched_case TEXT NOT NULL DEFAULT '',
|
|
review_reason TEXT NOT NULL DEFAULT '',
|
|
candidate_json TEXT NOT NULL DEFAULT '{}',
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
columns = {
|
|
str(row["name"] or "")
|
|
for row in conn.execute("PRAGMA table_info(wehago_raw_erp_trace_candidate_cache)").fetchall()
|
|
}
|
|
if "candidate_key" not in columns:
|
|
conn.execute(
|
|
"ALTER TABLE wehago_raw_erp_trace_candidate_cache ADD COLUMN candidate_key TEXT NOT NULL DEFAULT ''"
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE INDEX IF NOT EXISTS idx_wehago_raw_trace_candidate_source
|
|
ON wehago_raw_erp_trace_candidate_cache(
|
|
fiscal_year, source_mode, source_signature, logic_version, status_key, group_index
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE INDEX IF NOT EXISTS idx_wehago_raw_trace_candidate_voucher
|
|
ON wehago_raw_erp_trace_candidate_cache(fiscal_year, ledger_date, voucher_no)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE INDEX IF NOT EXISTS idx_wehago_raw_trace_candidate_score
|
|
ON wehago_raw_erp_trace_candidate_cache(fiscal_year, score DESC)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE UNIQUE INDEX IF NOT EXISTS idx_wehago_raw_trace_candidate_key
|
|
ON wehago_raw_erp_trace_candidate_cache(candidate_key)
|
|
WHERE candidate_key <> ''
|
|
"""
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _source_signature(conn: sqlite3.Connection, year: int, source_mode: str) -> str:
|
|
if source_mode == "current":
|
|
yearly_signature = current_ready_export_signature(conn, year)
|
|
return projection_signature({year: yearly_signature}, year, year, allow_stale=False)
|
|
if source_mode == "stale-diagnostic":
|
|
yearly_signature = latest_export_signature(conn, year, allow_stale=True)
|
|
return projection_signature({year: yearly_signature}, year, year, allow_stale=True)
|
|
raise ValueError(f"Unknown source mode: {source_mode}")
|
|
|
|
|
|
def _ensure_projection(conn: sqlite3.Connection, year: int, source_mode: str) -> str:
|
|
signature = _source_signature(conn, year, source_mode)
|
|
exists = conn.execute(
|
|
"""
|
|
SELECT 1
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND signature = ?
|
|
LIMIT 1
|
|
""",
|
|
(year, year, signature),
|
|
).fetchone()
|
|
if exists is not None:
|
|
return signature
|
|
project_range(conn, year, year, allow_stale=(source_mode == "stale-diagnostic"))
|
|
return signature
|
|
|
|
|
|
def _entry_amount_index(conn: sqlite3.Connection, year: int) -> dict[float, list[dict[str, Any]]]:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT fiscal_year, proof_date, confirmed_no, draft_no, account_code, account_name,
|
|
debit_supply, debit_tax, credit_supply, credit_tax,
|
|
support_dept_name, cost_dept_name, desc1, desc2, vendor_name, management_item
|
|
FROM wehago_voucher_rows
|
|
WHERE fiscal_year = ?
|
|
""",
|
|
(year,),
|
|
)
|
|
indexed: dict[float, list[dict[str, Any]]] = defaultdict(list)
|
|
fields = (
|
|
("debit_supply", "debit", False),
|
|
("credit_supply", "credit", False),
|
|
("debit_tax", "debit", True),
|
|
("credit_tax", "credit", True),
|
|
)
|
|
for row in rows:
|
|
raw = dict(row)
|
|
for field, side, tax_evidence in fields:
|
|
amount = parse_amount(raw.get(field))
|
|
if abs(amount) < 0.5:
|
|
continue
|
|
entry = dict(raw)
|
|
entry["_raw_amount_field"] = field
|
|
entry["_raw_amount_side"] = side
|
|
entry["_raw_amount_value"] = amount
|
|
entry["_raw_tax_evidence"] = tax_evidence
|
|
entry["_raw_entry_dates"] = tuple(_raw_erp_entry_date_values(entry))
|
|
indexed[round(abs(amount), 2)].append(entry)
|
|
return dict(indexed)
|
|
|
|
|
|
def _group_filters(args: argparse.Namespace) -> tuple[str, list[Any]]:
|
|
filters: list[str] = []
|
|
params: list[Any] = []
|
|
if args.voucher_no:
|
|
placeholders = ",".join("?" for _ in args.voucher_no)
|
|
filters.append(f"g.voucher_no IN ({placeholders})")
|
|
params.extend(args.voucher_no)
|
|
if args.ledger_date:
|
|
placeholders = ",".join("?" for _ in args.ledger_date)
|
|
filters.append(f"g.ledger_date IN ({placeholders})")
|
|
params.extend(args.ledger_date)
|
|
if not filters:
|
|
return "", []
|
|
return " AND " + " AND ".join(filters), params
|
|
|
|
|
|
def _load_source_groups(
|
|
conn: sqlite3.Connection,
|
|
year: int,
|
|
signature: str,
|
|
args: argparse.Namespace,
|
|
) -> list[dict[str, Any]]:
|
|
filter_sql, filter_params = _group_filters(args)
|
|
limit_sql = " LIMIT ?" if args.limit_groups else ""
|
|
offset_sql = " OFFSET ?" if args.limit_groups and args.group_offset else ""
|
|
params: list[Any] = [year, year, signature, *SOURCE_STATUSES, *filter_params]
|
|
if args.limit_groups:
|
|
params.append(int(args.limit_groups))
|
|
if args.group_offset:
|
|
params.append(int(args.group_offset))
|
|
groups = conn.execute(
|
|
f"""
|
|
SELECT g.status_key, g.group_index, g.fiscal_year, g.ledger_date, g.voucher_no, g.draft_no
|
|
FROM wehago_compare_query_groups AS g
|
|
WHERE g.start_year = ?
|
|
AND g.end_year = ?
|
|
AND g.signature = ?
|
|
AND g.status_key IN ({','.join('?' for _ in SOURCE_STATUSES)})
|
|
{filter_sql}
|
|
ORDER BY g.status_key ASC, g.group_index ASC
|
|
{limit_sql}
|
|
{offset_sql}
|
|
""",
|
|
params,
|
|
).fetchall()
|
|
if not groups:
|
|
return []
|
|
|
|
result: list[dict[str, Any]] = []
|
|
for group_row in groups:
|
|
row_items = conn.execute(
|
|
"""
|
|
SELECT *
|
|
FROM wehago_compare_query_rows
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND signature = ?
|
|
AND status_key = ?
|
|
AND group_index = ?
|
|
ORDER BY row_index ASC
|
|
""",
|
|
(year, year, signature, group_row["status_key"], group_row["group_index"]),
|
|
).fetchall()
|
|
rows = []
|
|
for row in row_items:
|
|
data = dict(row)
|
|
data.setdefault("group_voucher_no", group_row["voucher_no"])
|
|
data.setdefault("group_draft_no", group_row["draft_no"])
|
|
if not clean(data.get("voucher_no")):
|
|
data["voucher_no"] = group_row["voucher_no"]
|
|
if not clean(data.get("draft_no")):
|
|
data["draft_no"] = group_row["draft_no"]
|
|
rows.append(data)
|
|
result.append({"group": dict(group_row), "rows": rows})
|
|
return result
|
|
|
|
|
|
def _ledger_amount(row: dict[str, Any]) -> float:
|
|
return round(max(abs(parse_amount(row.get("ledger_debit"))), abs(parse_amount(row.get("ledger_credit")))), 2)
|
|
|
|
|
|
def _entries_in_date_window(
|
|
ledger_row: dict[str, Any],
|
|
entries: list[dict[str, Any]],
|
|
window_days: int,
|
|
) -> list[dict[str, Any]]:
|
|
if window_days < 0:
|
|
return entries
|
|
ledger_dates = [
|
|
value
|
|
for value in (
|
|
_parse_row_date_with_year(ledger_row, "ledger_date"),
|
|
_parse_row_date_with_year(ledger_row, "proof_date"),
|
|
)
|
|
if value
|
|
]
|
|
if not ledger_dates:
|
|
return entries
|
|
filtered: list[dict[str, Any]] = []
|
|
for entry in entries:
|
|
entry_dates = entry.get("_raw_entry_dates") or ()
|
|
if any(abs((ledger_date - entry_date).days) <= window_days for ledger_date in ledger_dates for entry_date in entry_dates):
|
|
filtered.append(entry)
|
|
return filtered
|
|
|
|
|
|
def _matched_case(score: float, candidate: dict[str, Any]) -> str:
|
|
existing = clean(candidate.get("matched_case"))
|
|
if existing:
|
|
return existing
|
|
if score >= 105:
|
|
return "RAW_ERP_HIGH_CONFIDENCE_TRACE_CANDIDATE"
|
|
return "RAW_ERP_SOURCE_TRACE_CANDIDATE"
|
|
|
|
|
|
def _candidate_identity(
|
|
year: int,
|
|
source_mode: str,
|
|
source_signature: str,
|
|
logic_version: str,
|
|
status_key: str,
|
|
group_index: int,
|
|
row_index: int,
|
|
candidate: dict[str, Any],
|
|
) -> str:
|
|
raw = "|".join(
|
|
clean(part)
|
|
for part in (
|
|
year,
|
|
source_mode,
|
|
source_signature,
|
|
logic_version,
|
|
status_key,
|
|
group_index,
|
|
row_index,
|
|
candidate.get("draft_no"),
|
|
candidate.get("voucher_confirmed_no"),
|
|
candidate.get("voucher_account_name"),
|
|
candidate.get("voucher_debit"),
|
|
candidate.get("voucher_credit"),
|
|
)
|
|
)
|
|
return hashlib.sha1(raw.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _insert_candidates(
|
|
conn: sqlite3.Connection,
|
|
year: int,
|
|
source_mode: str,
|
|
source_signature: str,
|
|
rows: list[dict[str, Any]],
|
|
) -> None:
|
|
conn.executemany(
|
|
"""
|
|
INSERT OR REPLACE INTO wehago_raw_erp_trace_candidate_cache (
|
|
candidate_key, fiscal_year, source_mode, source_signature, logic_version, status_key,
|
|
group_index, row_index, ledger_date, voucher_no,
|
|
ledger_account_name, ledger_vendor, ledger_desc, ledger_amount,
|
|
erp_draft_no, erp_confirmed_no, erp_account_name, erp_vendor, erp_desc,
|
|
erp_amount, amount_field, score, matched_case, review_reason, candidate_json
|
|
)
|
|
VALUES (
|
|
:candidate_key, :fiscal_year, :source_mode, :source_signature, :logic_version, :status_key,
|
|
:group_index, :row_index, :ledger_date, :voucher_no,
|
|
:ledger_account_name, :ledger_vendor, :ledger_desc, :ledger_amount,
|
|
:erp_draft_no, :erp_confirmed_no, :erp_account_name, :erp_vendor, :erp_desc,
|
|
:erp_amount, :amount_field, :score, :matched_case, :review_reason, :candidate_json
|
|
)
|
|
""",
|
|
rows,
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def build_candidates(args: argparse.Namespace) -> dict[str, Any]:
|
|
conn = _connect(args.db)
|
|
try:
|
|
_ensure_schema(conn)
|
|
source_signature = _ensure_projection(conn, args.year, args.source)
|
|
if args.reset:
|
|
conn.execute(
|
|
"""
|
|
DELETE FROM wehago_raw_erp_trace_candidate_cache
|
|
WHERE fiscal_year = ?
|
|
AND source_mode = ?
|
|
AND source_signature = ?
|
|
AND logic_version = ?
|
|
""",
|
|
(args.year, args.source, source_signature, TRACE_LOGIC_VERSION),
|
|
)
|
|
conn.commit()
|
|
|
|
amount_index = _entry_amount_index(conn, args.year)
|
|
source_groups = _load_source_groups(conn, args.year, source_signature, args)
|
|
candidates: list[dict[str, Any]] = []
|
|
scored_rows = 0
|
|
skipped_common_amounts = 0
|
|
|
|
for source_group in source_groups:
|
|
group_meta = source_group["group"]
|
|
status_key = clean(group_meta.get("status_key"))
|
|
group_index = int(group_meta.get("group_index") or 0)
|
|
for ledger_row in source_group["rows"]:
|
|
amount = _ledger_amount(ledger_row)
|
|
if amount <= 0:
|
|
continue
|
|
entries = list(amount_index.get(amount, []) or [])
|
|
if not entries:
|
|
continue
|
|
if len(entries) > args.date_prefilter_threshold:
|
|
date_entries = _entries_in_date_window(ledger_row, entries, args.date_window_days)
|
|
if date_entries:
|
|
entries = date_entries
|
|
if len(entries) > args.prefilter_threshold:
|
|
entries = [entry for entry in entries if _raw_erp_trace_prefilter(ledger_row, entry)]
|
|
skipped_common_amounts += 1
|
|
if len(entries) > args.max_candidates_per_row:
|
|
ranked = [
|
|
(_raw_erp_trace_prefilter_score(ledger_row, entry), entry)
|
|
for entry in entries
|
|
]
|
|
ranked = [item for item in ranked if item[0] > 0]
|
|
ranked.sort(key=lambda item: item[0], reverse=True)
|
|
entries = [entry for _score, entry in ranked[: args.max_candidates_per_row]]
|
|
|
|
row_candidates: list[tuple[float, dict[str, Any], dict[str, Any]]] = []
|
|
for entry in entries:
|
|
score, candidate = _raw_erp_trace_score(ledger_row, entry)
|
|
scored_rows += 1
|
|
if score < args.min_score:
|
|
continue
|
|
row_candidates.append((score, entry, candidate))
|
|
row_candidates.sort(key=lambda item: item[0], reverse=True)
|
|
for score, entry, candidate in row_candidates[: args.top_per_row]:
|
|
row_index = int(ledger_row.get("row_index") or 0)
|
|
matched_case = _matched_case(score, candidate)
|
|
payload = {
|
|
"candidate_id": _candidate_identity(
|
|
args.year,
|
|
args.source,
|
|
source_signature,
|
|
TRACE_LOGIC_VERSION,
|
|
status_key,
|
|
group_index,
|
|
row_index,
|
|
candidate,
|
|
),
|
|
"ledger_row": ledger_row,
|
|
"erp_entry": entry,
|
|
"candidate_row": candidate,
|
|
}
|
|
candidates.append(
|
|
{
|
|
"candidate_key": payload["candidate_id"],
|
|
"fiscal_year": args.year,
|
|
"source_mode": args.source,
|
|
"source_signature": source_signature,
|
|
"logic_version": TRACE_LOGIC_VERSION,
|
|
"status_key": status_key,
|
|
"group_index": group_index,
|
|
"row_index": row_index,
|
|
"ledger_date": clean(ledger_row.get("ledger_date")),
|
|
"voucher_no": clean(ledger_row.get("voucher_no") or group_meta.get("voucher_no")),
|
|
"ledger_account_name": clean(ledger_row.get("ledger_account_name")),
|
|
"ledger_vendor": clean(ledger_row.get("ledger_vendor")),
|
|
"ledger_desc": clean(ledger_row.get("ledger_desc")),
|
|
"ledger_amount": amount,
|
|
"erp_draft_no": clean(candidate.get("draft_no")),
|
|
"erp_confirmed_no": clean(candidate.get("voucher_confirmed_no")),
|
|
"erp_account_name": clean(candidate.get("voucher_account_name")),
|
|
"erp_vendor": clean(candidate.get("voucher_vendor")),
|
|
"erp_desc": clean(candidate.get("voucher_desc")),
|
|
"erp_amount": max(
|
|
abs(parse_amount(candidate.get("voucher_debit"))),
|
|
abs(parse_amount(candidate.get("voucher_credit"))),
|
|
),
|
|
"amount_field": clean(entry.get("_raw_amount_field")),
|
|
"score": float(score),
|
|
"matched_case": matched_case,
|
|
"review_reason": matched_case,
|
|
"candidate_json": json.dumps(payload, ensure_ascii=False, default=str),
|
|
}
|
|
)
|
|
if candidates:
|
|
_insert_candidates(conn, args.year, args.source, source_signature, candidates)
|
|
return {
|
|
"year": args.year,
|
|
"source_mode": args.source,
|
|
"source_signature": source_signature,
|
|
"group_offset": args.group_offset,
|
|
"limit_groups": args.limit_groups,
|
|
"groups": len(source_groups),
|
|
"amount_buckets": len(amount_index),
|
|
"scored_rows": scored_rows,
|
|
"common_amount_prefilters": skipped_common_amounts,
|
|
"inserted_candidates": len(candidates),
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Build bounded raw ERP trace candidate cache for voucher compare review.")
|
|
parser.add_argument("--year", type=int, required=True)
|
|
parser.add_argument(
|
|
"--source",
|
|
choices=("current", "stale-diagnostic"),
|
|
default="current",
|
|
help="current는 현재 로직 ready projection만 사용합니다. stale-diagnostic은 진단용 후보 산출에만 사용하세요.",
|
|
)
|
|
parser.add_argument("--db", type=Path, default=DB_PATH)
|
|
parser.add_argument("--reset", action="store_true")
|
|
parser.add_argument("--limit-groups", type=int, default=0)
|
|
parser.add_argument("--group-offset", type=int, default=0)
|
|
parser.add_argument("--ledger-date", action="append", default=[])
|
|
parser.add_argument("--voucher-no", action="append", default=[])
|
|
parser.add_argument("--min-score", type=float, default=70)
|
|
parser.add_argument("--top-per-row", type=int, default=5)
|
|
parser.add_argument("--max-candidates-per-row", type=int, default=30)
|
|
parser.add_argument("--date-prefilter-threshold", type=int, default=30)
|
|
parser.add_argument("--date-window-days", type=int, default=62)
|
|
parser.add_argument("--prefilter-threshold", type=int, default=30)
|
|
args = parser.parse_args()
|
|
print(json.dumps(build_candidates(args), ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|