Preserve latest comparison functionality
This commit is contained in:
b17301
@@ -0,0 +1,259 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user