260 lines
8.6 KiB
Python
260 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from runtime_config import DB_PATH
|
|
|
|
|
|
DEFAULT_SAMPLES = (
|
|
"2025-02-25-00029",
|
|
"2025-01-10-00006",
|
|
"2025-01-15-50027",
|
|
"2025-01-15-50028",
|
|
"2025-01-21-50197",
|
|
"2025-01-21-50198",
|
|
)
|
|
|
|
|
|
def run_command(command: list[str]) -> str:
|
|
proc = subprocess.run(command, cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True)
|
|
if proc.returncode != 0:
|
|
payload = {
|
|
"command": command,
|
|
"returncode": proc.returncode,
|
|
"stdout": proc.stdout,
|
|
"stderr": proc.stderr,
|
|
}
|
|
raise SystemExit(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
return proc.stdout
|
|
|
|
|
|
def parse_sample(value: str, default_year: int) -> tuple[int, str, str]:
|
|
text = value.strip()
|
|
parts = text.replace("/", "-").split("-")
|
|
if len(parts) == 4:
|
|
year, month, day, voucher = parts
|
|
elif len(parts) == 3:
|
|
year = str(default_year)
|
|
month, day, voucher = parts
|
|
else:
|
|
raise ValueError(f"Invalid sample format: {value}")
|
|
return int(year), f"{int(month):02d}-{int(day):02d}", f"{int(voucher):05d}" if voucher.isdigit() else voucher
|
|
|
|
|
|
def active_signature(conn: sqlite3.Connection, year: int) -> str:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT setting_json
|
|
FROM wehago_compare_settings
|
|
WHERE setting_key = ?
|
|
LIMIT 1
|
|
""",
|
|
(f"wehago_active_query_projection:{year}:{year}",),
|
|
).fetchone()
|
|
if not row:
|
|
return ""
|
|
try:
|
|
payload = json.loads(row[0] or "{}")
|
|
except Exception:
|
|
return ""
|
|
return str(payload.get("signature") or "") if isinstance(payload, dict) else ""
|
|
|
|
|
|
def validate_projection(conn: sqlite3.Connection, year: int, signature: str) -> dict[str, Any]:
|
|
counts = {
|
|
row[0]: int(row[1] or 0)
|
|
for row in conn.execute(
|
|
"""
|
|
SELECT status_key, COUNT(*)
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ? AND end_year = ? AND signature = ?
|
|
AND status_key IN ('voucher_matched', 'voucher_unmatched', 'voucher_recheck', 'voucher_excepted')
|
|
GROUP BY status_key
|
|
""",
|
|
(year, year, signature),
|
|
).fetchall()
|
|
}
|
|
final_counts = {
|
|
row[0]: int(row[1] or 0)
|
|
for row in conn.execute(
|
|
"""
|
|
SELECT final_status, COUNT(*)
|
|
FROM wehago_compare_final_status_projection
|
|
WHERE start_year = ? AND end_year = ? AND signature = ?
|
|
GROUP BY final_status
|
|
""",
|
|
(year, year, signature),
|
|
).fetchall()
|
|
}
|
|
raw_total = int(
|
|
conn.execute(
|
|
"""
|
|
SELECT COUNT(DISTINCT compare_voucher_no)
|
|
FROM wehago_ledger_rows
|
|
WHERE fiscal_year = ?
|
|
AND COALESCE(compare_voucher_no, '') <> ''
|
|
""",
|
|
(year,),
|
|
).fetchone()[0]
|
|
or 0
|
|
)
|
|
recheck_without_erp = int(
|
|
conn.execute(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM wehago_compare_query_groups g
|
|
WHERE g.start_year = ? AND g.end_year = ? AND g.signature = ?
|
|
AND g.status_key = 'voucher_recheck'
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM wehago_compare_query_rows r
|
|
WHERE r.start_year = g.start_year
|
|
AND r.end_year = g.end_year
|
|
AND r.signature = g.signature
|
|
AND r.status_key = g.status_key
|
|
AND r.group_index = g.group_index
|
|
AND COALESCE(r.voucher_account_name, '') <> ''
|
|
AND (
|
|
ABS(COALESCE(r.voucher_debit, 0)) > 0.0001
|
|
OR ABS(COALESCE(r.voucher_credit, 0)) > 0.0001
|
|
)
|
|
)
|
|
""",
|
|
(year, year, signature),
|
|
).fetchone()[0]
|
|
or 0
|
|
)
|
|
unmatched_with_erp = int(
|
|
conn.execute(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM wehago_compare_query_groups g
|
|
WHERE g.start_year = ? AND g.end_year = ? AND g.signature = ?
|
|
AND g.status_key = 'voucher_unmatched'
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM wehago_compare_query_rows r
|
|
WHERE r.start_year = g.start_year
|
|
AND r.end_year = g.end_year
|
|
AND r.signature = g.signature
|
|
AND r.status_key = g.status_key
|
|
AND r.group_index = g.group_index
|
|
AND COALESCE(r.voucher_account_name, '') <> ''
|
|
AND (
|
|
ABS(COALESCE(r.voucher_debit, 0)) > 0.0001
|
|
OR ABS(COALESCE(r.voucher_credit, 0)) > 0.0001
|
|
)
|
|
)
|
|
""",
|
|
(year, year, signature),
|
|
).fetchone()[0]
|
|
or 0
|
|
)
|
|
return {
|
|
"raw_total": raw_total,
|
|
"counts": counts,
|
|
"final_counts": final_counts,
|
|
"classified_total": sum(counts.values()),
|
|
"difference": raw_total - sum(counts.values()),
|
|
"recheck_without_erp": recheck_without_erp,
|
|
"unmatched_with_erp": unmatched_with_erp,
|
|
}
|
|
|
|
|
|
def sample_statuses(conn: sqlite3.Connection, year: int, signature: str, samples: list[str]) -> list[dict[str, Any]]:
|
|
result: list[dict[str, Any]] = []
|
|
for sample in samples:
|
|
sample_year, ledger_date, voucher_no = parse_sample(sample, year)
|
|
rows = [
|
|
{
|
|
"status_key": row[0],
|
|
"ledger_date": row[1],
|
|
"voucher_no": row[2],
|
|
"draft_no": row[3],
|
|
"review_reason": row[4],
|
|
}
|
|
for row in conn.execute(
|
|
"""
|
|
SELECT status_key, ledger_date, voucher_no, draft_no, review_reason
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ? AND end_year = ? AND signature = ?
|
|
AND fiscal_year = ?
|
|
AND ledger_date = ?
|
|
AND voucher_no = ?
|
|
ORDER BY status_key, group_index
|
|
""",
|
|
(year, year, signature, sample_year, ledger_date, voucher_no),
|
|
).fetchall()
|
|
]
|
|
result.append({"sample": sample, "rows": rows})
|
|
return result
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Run a fast WEHAGO logic iteration: compile, reconcile, validate, sample-check, optionally prune.")
|
|
parser.add_argument("--year", type=int, default=2025)
|
|
parser.add_argument("--sample", action="append", default=[])
|
|
parser.add_argument("--skip-compile", action="store_true")
|
|
parser.add_argument("--skip-reconcile", action="store_true")
|
|
parser.add_argument("--prune", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
repo_root = Path(__file__).resolve().parents[1]
|
|
python = str(repo_root / ".venv" / "bin" / "python")
|
|
if not args.skip_compile:
|
|
run_command(
|
|
[
|
|
python,
|
|
"-m",
|
|
"py_compile",
|
|
"scripts/reconcile_wehago_projection_to_db.py",
|
|
"scripts/prune_wehago_projection_history.py",
|
|
"wehago_compare.py",
|
|
"main.py",
|
|
]
|
|
)
|
|
reconcile_output = ""
|
|
if not args.skip_reconcile:
|
|
reconcile_output = run_command([python, "scripts/reconcile_wehago_projection_to_db.py", "--year", str(args.year)])
|
|
prune_output = ""
|
|
if args.prune:
|
|
prune_output = run_command(
|
|
[
|
|
python,
|
|
"scripts/prune_wehago_projection_history.py",
|
|
"--start-year",
|
|
str(args.year),
|
|
"--end-year",
|
|
str(args.year),
|
|
"--execute",
|
|
]
|
|
)
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
try:
|
|
signature = active_signature(conn, args.year)
|
|
samples = args.sample or list(DEFAULT_SAMPLES)
|
|
payload = {
|
|
"year": args.year,
|
|
"active_signature": signature,
|
|
"validation": validate_projection(conn, args.year, signature) if signature else {},
|
|
"samples": sample_statuses(conn, args.year, signature, samples) if signature else [],
|
|
"reconcile_output_tail": reconcile_output.strip().splitlines()[-8:],
|
|
"prune_output": json.loads(prune_output) if prune_output.strip().startswith("{") else prune_output,
|
|
}
|
|
finally:
|
|
conn.close()
|
|
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|