121 lines
3.6 KiB
Python
121 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from main import engine
|
|
from wehago_compare import (
|
|
_apply_highpass_bundle_trace_review,
|
|
_status_projection_signature,
|
|
_store_export_compact_status_projection,
|
|
clean,
|
|
)
|
|
|
|
|
|
PATCH_STATUSES = (
|
|
"voucher_matched",
|
|
"erp_voucher_matched",
|
|
"voucher_unmatched",
|
|
"voucher_recheck",
|
|
"erp_voucher_unmatched",
|
|
"voucher_excepted",
|
|
)
|
|
|
|
|
|
def _setting_key(status_key: str, start_year: int, end_year: int) -> str:
|
|
return f"wehago_active_status_projection:{status_key}:{start_year}:{end_year}"
|
|
|
|
|
|
def _load_active_signature(conn, status_key: str, start_year: int, end_year: int) -> str:
|
|
row = conn.exec_driver_sql(
|
|
"""
|
|
SELECT setting_json
|
|
FROM wehago_compare_settings
|
|
WHERE setting_key = ?
|
|
LIMIT 1
|
|
""",
|
|
(_setting_key(status_key, start_year, end_year),),
|
|
).first()
|
|
if not row or not row[0]:
|
|
return ""
|
|
try:
|
|
payload = json.loads(str(row[0] or "{}"))
|
|
except Exception:
|
|
return ""
|
|
if not isinstance(payload, dict):
|
|
return ""
|
|
return clean(payload.get("signature"))
|
|
|
|
|
|
def _load_projection_groups(conn, status_key: str, start_year: int, end_year: int) -> list[dict]:
|
|
signature = _load_active_signature(conn, status_key, start_year, end_year)
|
|
if not signature:
|
|
return []
|
|
rows = conn.exec_driver_sql(
|
|
"""
|
|
SELECT summary_json, rows_json
|
|
FROM wehago_status_projection_groups
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = ?
|
|
AND signature = ?
|
|
ORDER BY group_index
|
|
""",
|
|
(int(start_year), int(end_year), status_key, signature),
|
|
).fetchall()
|
|
groups: list[dict] = []
|
|
for row in rows:
|
|
try:
|
|
summary = json.loads(str(row[0] or "{}"))
|
|
except Exception:
|
|
summary = {}
|
|
try:
|
|
group_rows = json.loads(str(row[1] or "[]"))
|
|
except Exception:
|
|
group_rows = []
|
|
if isinstance(summary, dict) and isinstance(group_rows, list):
|
|
groups.append({"summary": summary, "rows": group_rows})
|
|
return groups
|
|
|
|
|
|
def apply_patch_for_range(start_year: int, end_year: int) -> dict[str, int]:
|
|
with engine.begin() as conn:
|
|
sections = {
|
|
status_key: _load_projection_groups(conn, status_key, start_year, end_year)
|
|
for status_key in PATCH_STATUSES
|
|
}
|
|
before = {status_key: len(sections.get(status_key) or []) for status_key in PATCH_STATUSES}
|
|
sections.setdefault("hanmac_unconnected", [])
|
|
|
|
patched = _apply_highpass_bundle_trace_review(sections)
|
|
after = {status_key: len(patched.get(status_key) or []) for status_key in PATCH_STATUSES}
|
|
|
|
for status_key in PATCH_STATUSES:
|
|
signature = _status_projection_signature(conn, status_key, start_year, end_year)
|
|
_store_export_compact_status_projection(
|
|
conn,
|
|
start_year,
|
|
end_year,
|
|
status_key,
|
|
signature,
|
|
list(patched.get(status_key) or []),
|
|
)
|
|
return {
|
|
**{f"before_{key}": value for key, value in before.items()},
|
|
**{f"after_{key}": value for key, value in after.items()},
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
start_year = int(sys.argv[1]) if len(sys.argv) > 1 else 2025
|
|
end_year = int(sys.argv[2]) if len(sys.argv) > 2 else start_year
|
|
result = apply_patch_for_range(start_year, end_year)
|
|
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|