564 lines
22 KiB
Python
564 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fcntl
|
|
import json
|
|
import multiprocessing as mp
|
|
import os
|
|
import signal
|
|
import sys
|
|
import time
|
|
import traceback
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
import wehago_compare as wc
|
|
from main import engine
|
|
|
|
|
|
DEFAULT_YEARS = [2025, 2024, 2023, 2022]
|
|
PROGRESS_PATH = Path("/tmp/wehago_year_projection_watchdog_progress.json")
|
|
LOCK_PATH = Path("/tmp/wehago_compare_compute.lock")
|
|
|
|
|
|
def emit(event: str, **payload: Any) -> None:
|
|
data = {
|
|
"event": event,
|
|
"ts": time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
**payload,
|
|
}
|
|
print(json.dumps(data, ensure_ascii=False, sort_keys=True), flush=True)
|
|
|
|
|
|
def load_progress() -> dict[str, Any]:
|
|
if not PROGRESS_PATH.exists():
|
|
return {"years": {}}
|
|
try:
|
|
payload = json.loads(PROGRESS_PATH.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {"years": {}}
|
|
return payload if isinstance(payload, dict) else {"years": {}}
|
|
|
|
|
|
def store_progress(progress: dict[str, Any]) -> None:
|
|
PROGRESS_PATH.write_text(json.dumps(progress, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def update_progress(year: int, status: str, **payload: Any) -> None:
|
|
progress = load_progress()
|
|
years = progress.setdefault("years", {})
|
|
current = dict(years.get(str(year)) or {})
|
|
current.update(
|
|
{
|
|
"year": int(year),
|
|
"status": status,
|
|
"updated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
**payload,
|
|
}
|
|
)
|
|
years[str(year)] = current
|
|
store_progress(progress)
|
|
|
|
|
|
def _execute_with_lock_retry(conn: Any, statement: Any, params: dict[str, Any], *, attempts: int = 12) -> Any:
|
|
last_exc: BaseException | None = None
|
|
for attempt in range(1, max(int(attempts or 1), 1) + 1):
|
|
try:
|
|
return conn.execute(statement, params)
|
|
except OperationalError as exc:
|
|
if "database is locked" not in str(exc).lower():
|
|
raise
|
|
last_exc = exc
|
|
emit("sqlite_lock_wait", attempt=attempt, sleep_sec=5)
|
|
time.sleep(5)
|
|
if last_exc:
|
|
raise last_exc
|
|
return conn.execute(statement, params)
|
|
|
|
|
|
def _delete_year_projection_rows(conn: Any, year: int) -> dict[str, int]:
|
|
params = {"year": int(year)}
|
|
deleted: dict[str, int] = {}
|
|
table_names = [
|
|
"wehago_compare_query_metrics",
|
|
"wehago_compare_query_groups",
|
|
"wehago_compare_query_rows",
|
|
"wehago_compare_query_page_cache",
|
|
"wehago_compare_final_status_projection",
|
|
"wehago_status_projection_groups",
|
|
"wehago_metric_count_cache",
|
|
"wehago_summary_range_cache",
|
|
]
|
|
for table_name in table_names:
|
|
result = _execute_with_lock_retry(
|
|
conn,
|
|
text(
|
|
f"""
|
|
DELETE FROM {table_name}
|
|
WHERE start_year = :year
|
|
AND end_year = :year
|
|
"""
|
|
),
|
|
params,
|
|
)
|
|
deleted[table_name] = int(result.rowcount or 0)
|
|
result = _execute_with_lock_retry(
|
|
conn,
|
|
text(
|
|
"""
|
|
DELETE FROM wehago_compare_settings
|
|
WHERE setting_key = :query_key
|
|
OR setting_key = :run_key
|
|
OR setting_key LIKE :status_like
|
|
"""
|
|
),
|
|
{
|
|
"query_key": f"wehago_active_query_projection:{year}:{year}",
|
|
"run_key": f"wehago_active_status_projection_run:{year}:{year}",
|
|
"status_like": f"wehago_active_status_projection:%:{year}:{year}",
|
|
},
|
|
)
|
|
deleted["wehago_compare_settings"] = int(result.rowcount or 0)
|
|
return deleted
|
|
|
|
|
|
def _mirror_query_groups_to_active_status_projection(conn: Any, year: int) -> dict[str, int]:
|
|
query_signature = wc._load_latest_query_projection_signature_for_range(conn, year, year)
|
|
if not query_signature:
|
|
return {}
|
|
insert_sql = text(
|
|
"""
|
|
INSERT INTO wehago_status_projection_groups (
|
|
start_year, end_year, status_key, signature, group_index,
|
|
fiscal_year, ledger_date, proof_date, voucher_no, draft_no,
|
|
ledger_row_count, voucher_row_count,
|
|
ledger_debit, ledger_credit, voucher_debit, voucher_credit,
|
|
ledger_accounts, voucher_accounts, ledger_vendors, voucher_vendors,
|
|
review_reason, search_text, summary_json, rows_json, created_at, updated_at
|
|
) VALUES (
|
|
:start_year, :end_year, :status_key, :signature, :group_index,
|
|
:fiscal_year, :ledger_date, :proof_date, :voucher_no, :draft_no,
|
|
:ledger_row_count, :voucher_row_count, :ledger_debit, :ledger_credit,
|
|
:voucher_debit, :voucher_credit, :ledger_accounts, :voucher_accounts,
|
|
:ledger_vendors, :voucher_vendors, :review_reason, :search_text,
|
|
:summary_json, :rows_json, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
mirrored: dict[str, int] = {}
|
|
for status_key in wc.WEHAGO_FINAL_STATUS_ORDER:
|
|
status_signature = wc._status_projection_signature(conn, status_key, year, year)
|
|
if not status_signature:
|
|
continue
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
DELETE FROM wehago_status_projection_groups
|
|
WHERE start_year = :year
|
|
AND end_year = :year
|
|
AND status_key = :status_key
|
|
AND signature = :signature
|
|
"""
|
|
),
|
|
{"year": int(year), "status_key": status_key, "signature": status_signature},
|
|
)
|
|
groups = conn.execute(
|
|
text(
|
|
"""
|
|
SELECT *
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = :year
|
|
AND end_year = :year
|
|
AND status_key = :status_key
|
|
AND signature = :query_signature
|
|
ORDER BY group_index
|
|
"""
|
|
),
|
|
{"year": int(year), "status_key": status_key, "query_signature": query_signature},
|
|
).mappings().all()
|
|
payloads: list[dict[str, Any]] = []
|
|
for group in groups:
|
|
group_index = int(group.get("group_index") or 0)
|
|
rows = [
|
|
dict(row)
|
|
for row in conn.execute(
|
|
text(
|
|
"""
|
|
SELECT *
|
|
FROM wehago_compare_query_rows
|
|
WHERE start_year = :year
|
|
AND end_year = :year
|
|
AND status_key = :status_key
|
|
AND signature = :query_signature
|
|
AND group_index = :group_index
|
|
ORDER BY row_index
|
|
"""
|
|
),
|
|
{
|
|
"year": int(year),
|
|
"status_key": status_key,
|
|
"query_signature": query_signature,
|
|
"group_index": group_index,
|
|
},
|
|
).mappings()
|
|
]
|
|
summary = {
|
|
"fiscal_year": int(group.get("fiscal_year") or 0),
|
|
"ledger_date": wc.clean(group.get("ledger_date")),
|
|
"proof_date": wc.clean(group.get("proof_date")),
|
|
"voucher_no": wc.clean(group.get("voucher_no")),
|
|
"draft_no": wc.clean(group.get("draft_no")),
|
|
"ledger_row_count": int(group.get("ledger_row_count") or 0),
|
|
"voucher_row_count": int(group.get("voucher_row_count") or 0),
|
|
"ledger_debit": wc.parse_amount(group.get("ledger_debit")),
|
|
"ledger_credit": wc.parse_amount(group.get("ledger_credit")),
|
|
"voucher_debit": wc.parse_amount(group.get("voucher_debit")),
|
|
"voucher_credit": wc.parse_amount(group.get("voucher_credit")),
|
|
"ledger_accounts": wc.clean(group.get("ledger_accounts")),
|
|
"voucher_accounts": wc.clean(group.get("voucher_accounts")),
|
|
"ledger_vendors": wc.clean(group.get("ledger_vendors")),
|
|
"voucher_vendors": wc.clean(group.get("voucher_vendors")),
|
|
"review_reason": wc.clean(group.get("review_reason")),
|
|
}
|
|
payloads.append(
|
|
{
|
|
"start_year": int(year),
|
|
"end_year": int(year),
|
|
"status_key": status_key,
|
|
"signature": status_signature,
|
|
"group_index": group_index,
|
|
**summary,
|
|
"search_text": wc.clean(group.get("search_text")),
|
|
"summary_json": json.dumps(summary, ensure_ascii=False, separators=(",", ":")),
|
|
"rows_json": json.dumps(rows, ensure_ascii=False, separators=(",", ":")),
|
|
}
|
|
)
|
|
for chunk in wc._chunked(payloads):
|
|
conn.execute(insert_sql, chunk)
|
|
wc._store_active_status_projection_signature(conn, status_key, year, year, status_signature, len(payloads))
|
|
mirrored[status_key] = len(payloads)
|
|
return mirrored
|
|
|
|
|
|
def _count_projection_rows(conn: Any, year: int) -> dict[str, Any]:
|
|
params = {"year": int(year)}
|
|
query_signature = wc._load_latest_query_projection_signature_for_range(conn, year, year)
|
|
status_counts = {
|
|
str(row.get("status_key")): int(row.get("count") or 0)
|
|
for row in conn.execute(
|
|
text(
|
|
"""
|
|
SELECT status_key, COUNT(*) AS count
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = :year
|
|
AND end_year = :year
|
|
GROUP BY status_key
|
|
ORDER BY status_key
|
|
"""
|
|
),
|
|
params,
|
|
).mappings()
|
|
}
|
|
compact_counts = {
|
|
str(row.get("status_key")): int(row.get("count") or 0)
|
|
for row in conn.execute(
|
|
text(
|
|
"""
|
|
SELECT status_key, COUNT(*) AS count
|
|
FROM wehago_status_projection_groups
|
|
WHERE start_year = :year
|
|
AND end_year = :year
|
|
GROUP BY status_key
|
|
ORDER BY status_key
|
|
"""
|
|
),
|
|
params,
|
|
).mappings()
|
|
}
|
|
final_counts = {
|
|
str(row.get("final_status")): int(row.get("count") or 0)
|
|
for row in conn.execute(
|
|
text(
|
|
"""
|
|
SELECT final_status, COUNT(*) AS count
|
|
FROM wehago_compare_final_status_projection
|
|
WHERE start_year = :year
|
|
AND end_year = :year
|
|
AND signature = :signature
|
|
GROUP BY final_status
|
|
ORDER BY final_status
|
|
"""
|
|
),
|
|
{"year": int(year), "signature": query_signature},
|
|
).mappings()
|
|
}
|
|
active_summary = wc._load_cached_active_status_projection_run_summary(conn, year, year)
|
|
raw_total = int(
|
|
conn.execute(
|
|
text(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM (
|
|
SELECT compare_voucher_no
|
|
FROM wehago_ledger_rows
|
|
WHERE fiscal_year = :year
|
|
AND COALESCE(compare_voucher_no, '') <> ''
|
|
GROUP BY compare_voucher_no
|
|
)
|
|
"""
|
|
),
|
|
params,
|
|
).scalar_one()
|
|
or 0
|
|
)
|
|
return {
|
|
"query_signature": query_signature,
|
|
"query_group_counts": status_counts,
|
|
"compact_group_counts": compact_counts,
|
|
"final_counts": final_counts,
|
|
"raw_voucher_total": raw_total,
|
|
"active_summary": {
|
|
"source": (active_summary or {}).get("source"),
|
|
"ready": (active_summary or {}).get("ready"),
|
|
"raw_total": (active_summary or {}).get("raw_total"),
|
|
"classified_total": (active_summary or {}).get("classified_total"),
|
|
},
|
|
}
|
|
|
|
|
|
def _skip_query_page_prewarm() -> None:
|
|
def _noop_prewarm(*_args: Any, **_kwargs: Any) -> None:
|
|
emit("projection_page_prewarm_skipped")
|
|
|
|
wc._prewarm_query_page_projection_cache = _noop_prewarm
|
|
|
|
|
|
def rebuild_one_year(year: int, *, skip_page_prewarm: bool) -> None:
|
|
started = time.monotonic()
|
|
update_progress(year, "running", started_at=time.strftime("%Y-%m-%d %H:%M:%S"))
|
|
if skip_page_prewarm:
|
|
_skip_query_page_prewarm()
|
|
try:
|
|
wc._clear_compare_runtime_caches()
|
|
with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
|
|
conn.execute(text("PRAGMA busy_timeout = 60000"))
|
|
emit("year_purge_start", year=year)
|
|
deleted = _delete_year_projection_rows(conn, year)
|
|
emit("year_purge_done", year=year, deleted=deleted)
|
|
|
|
signature = wc._build_db_state_signature(conn, year, year)
|
|
emit("year_snapshot_start", year=year, signature=signature)
|
|
wc._refresh_year_resolved_sections(conn, year)
|
|
emit("year_snapshot_done", year=year)
|
|
|
|
emit("compact_projection_start", year=year)
|
|
for status_key in sorted(wc.COMPACT_EXPORT_STATUS_KEYS):
|
|
compact_signature = wc._ensure_export_compact_status_projection(conn, year, year, status_key)
|
|
emit(
|
|
"compact_projection_status_done",
|
|
year=year,
|
|
status_key=status_key,
|
|
signature=compact_signature,
|
|
)
|
|
hanmac_signature = wc._ensure_hanmac_unconnected_status_projection(conn, year, year)
|
|
emit("compact_projection_hanmac_done", year=year, signature=hanmac_signature)
|
|
|
|
emit("query_projection_start", year=year)
|
|
counts, snapshot_state = wc._rebuild_compare_query_projection(engine, conn, year, year)
|
|
emit("query_projection_done", year=year, counts=counts, snapshot_state=snapshot_state)
|
|
|
|
emit("active_status_mirror_start", year=year)
|
|
mirrored_counts = _mirror_query_groups_to_active_status_projection(conn, year)
|
|
emit("active_status_mirror_done", year=year, counts=mirrored_counts)
|
|
|
|
emit("active_run_summary_start", year=year)
|
|
active_summary = wc._load_active_status_projection_run_summary(conn, year, year, allow_rebuild=True)
|
|
wc._ensure_active_status_run_final_projection(conn, year, year, active_summary)
|
|
emit(
|
|
"active_run_summary_done",
|
|
year=year,
|
|
source=(active_summary or {}).get("source"),
|
|
raw_total=(active_summary or {}).get("raw_total"),
|
|
classified_total=(active_summary or {}).get("classified_total"),
|
|
ready=(active_summary or {}).get("ready"),
|
|
)
|
|
|
|
wc._clear_compare_runtime_caches()
|
|
verification = _count_projection_rows(conn, year)
|
|
elapsed_sec = round(time.monotonic() - started, 1)
|
|
update_progress(
|
|
year,
|
|
"done",
|
|
elapsed_sec=elapsed_sec,
|
|
verification=verification,
|
|
)
|
|
emit("year_done", year=year, elapsed_sec=elapsed_sec, verification=verification)
|
|
except BaseException as exc:
|
|
elapsed_sec = round(time.monotonic() - started, 1)
|
|
update_progress(
|
|
year,
|
|
"failed",
|
|
elapsed_sec=elapsed_sec,
|
|
error=str(exc),
|
|
traceback=traceback.format_exc(),
|
|
)
|
|
emit("year_failed", year=year, elapsed_sec=elapsed_sec, error=str(exc), traceback=traceback.format_exc())
|
|
raise
|
|
|
|
|
|
def child_entry(year: int, skip_page_prewarm: bool) -> None:
|
|
rebuild_one_year(int(year), skip_page_prewarm=bool(skip_page_prewarm))
|
|
|
|
|
|
def process_cpu_seconds(pid: int) -> float | None:
|
|
try:
|
|
stat = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
|
|
parts = stat.split()
|
|
ticks = int(parts[13]) + int(parts[14])
|
|
hz = os.sysconf(os.sysconf_names["SC_CLK_TCK"])
|
|
return float(ticks) / float(hz)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def terminate_process(proc: mp.Process, *, reason: str) -> None:
|
|
emit("worker_terminate", pid=proc.pid, reason=reason)
|
|
if proc.pid:
|
|
try:
|
|
os.kill(proc.pid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
pass
|
|
proc.join(timeout=20)
|
|
if proc.is_alive() and proc.pid:
|
|
emit("worker_kill", pid=proc.pid, reason=reason)
|
|
try:
|
|
os.kill(proc.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
proc.join(timeout=10)
|
|
|
|
|
|
def run_year_with_watchdog(
|
|
year: int,
|
|
*,
|
|
retries: int,
|
|
per_year_timeout_sec: int,
|
|
idle_timeout_sec: int,
|
|
heartbeat_sec: int,
|
|
skip_page_prewarm: bool,
|
|
deadline_monotonic: float,
|
|
) -> bool:
|
|
attempt = 0
|
|
while attempt <= retries:
|
|
attempt += 1
|
|
if time.monotonic() >= deadline_monotonic:
|
|
update_progress(year, "failed", error="deadline exceeded before start")
|
|
emit("deadline_exceeded_before_year", year=year)
|
|
return False
|
|
emit("worker_start", year=year, attempt=attempt)
|
|
proc = mp.Process(target=child_entry, args=(int(year), bool(skip_page_prewarm)), daemon=False)
|
|
proc.start()
|
|
last_cpu = process_cpu_seconds(proc.pid or 0) or 0.0
|
|
last_cpu_progress_at = time.monotonic()
|
|
started = time.monotonic()
|
|
last_heartbeat_at = 0.0
|
|
timed_out_reason = ""
|
|
while proc.is_alive():
|
|
now = time.monotonic()
|
|
cpu_now = process_cpu_seconds(proc.pid or 0)
|
|
if cpu_now is not None and cpu_now > last_cpu + 0.01:
|
|
last_cpu = cpu_now
|
|
last_cpu_progress_at = now
|
|
if now - last_heartbeat_at >= heartbeat_sec:
|
|
emit(
|
|
"worker_heartbeat",
|
|
year=year,
|
|
attempt=attempt,
|
|
pid=proc.pid,
|
|
elapsed_sec=round(now - started, 1),
|
|
cpu_sec=round(last_cpu, 1),
|
|
idle_sec=round(now - last_cpu_progress_at, 1),
|
|
)
|
|
last_heartbeat_at = now
|
|
if now - started >= per_year_timeout_sec:
|
|
timed_out_reason = f"per-year timeout {per_year_timeout_sec}s"
|
|
break
|
|
if now - last_cpu_progress_at >= idle_timeout_sec:
|
|
timed_out_reason = f"idle timeout {idle_timeout_sec}s"
|
|
break
|
|
if now >= deadline_monotonic:
|
|
timed_out_reason = "total deadline exceeded"
|
|
break
|
|
time.sleep(5)
|
|
if timed_out_reason:
|
|
terminate_process(proc, reason=timed_out_reason)
|
|
update_progress(year, "retrying" if attempt <= retries else "failed", error=timed_out_reason, attempt=attempt)
|
|
else:
|
|
proc.join()
|
|
if proc.exitcode == 0:
|
|
emit("worker_done", year=year, attempt=attempt)
|
|
return True
|
|
emit("worker_failed", year=year, attempt=attempt, exitcode=proc.exitcode, reason=timed_out_reason)
|
|
return False
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Rebuild WEHAGO compare projections year by year with a watchdog.")
|
|
parser.add_argument("--years", nargs="*", type=int, default=DEFAULT_YEARS)
|
|
parser.add_argument("--deadline-hours", type=float, default=5.0)
|
|
parser.add_argument("--per-year-timeout-minutes", type=float, default=70.0)
|
|
parser.add_argument("--idle-timeout-minutes", type=float, default=10.0)
|
|
parser.add_argument("--heartbeat-seconds", type=int, default=60)
|
|
parser.add_argument("--retries", type=int, default=1)
|
|
parser.add_argument("--resume", action="store_true", help="Skip years already marked done in the progress file.")
|
|
parser.add_argument("--with-page-prewarm", action="store_true", help="Prewarm query page cache during rebuild. This is now the default.")
|
|
parser.add_argument("--skip-page-prewarm", action="store_true", help="Skip query page cache prewarm during rebuild.")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
years = [int(year) for year in args.years if int(year or 0) > 0]
|
|
deadline_monotonic = time.monotonic() + max(float(args.deadline_hours or 5.0), 0.1) * 3600.0
|
|
LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
with LOCK_PATH.open("w") as lock_file:
|
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
|
|
emit(
|
|
"watchdog_start",
|
|
years=years,
|
|
deadline_hours=args.deadline_hours,
|
|
per_year_timeout_minutes=args.per_year_timeout_minutes,
|
|
idle_timeout_minutes=args.idle_timeout_minutes,
|
|
retries=args.retries,
|
|
skip_page_prewarm=args.skip_page_prewarm,
|
|
)
|
|
progress = load_progress()
|
|
ok = True
|
|
for year in years:
|
|
if args.resume and (progress.get("years", {}).get(str(year)) or {}).get("status") == "done":
|
|
emit("year_skip_done", year=year)
|
|
continue
|
|
year_ok = run_year_with_watchdog(
|
|
year,
|
|
retries=max(int(args.retries or 0), 0),
|
|
per_year_timeout_sec=int(max(float(args.per_year_timeout_minutes or 70.0), 1.0) * 60),
|
|
idle_timeout_sec=int(max(float(args.idle_timeout_minutes or 10.0), 1.0) * 60),
|
|
heartbeat_sec=max(int(args.heartbeat_seconds or 60), 10),
|
|
skip_page_prewarm=args.skip_page_prewarm,
|
|
deadline_monotonic=deadline_monotonic,
|
|
)
|
|
ok = ok and year_ok
|
|
if not year_ok and time.monotonic() >= deadline_monotonic:
|
|
break
|
|
emit("watchdog_done", ok=ok, progress=load_progress())
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|