mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 00:50:46 +08:00
Compare commits
3 commits
9adf9f2764
...
2df2f50510
| Author | SHA1 | Date | |
|---|---|---|---|
| 2df2f50510 | |||
| ba16167c8e | |||
| 958942c5b6 |
10 changed files with 1659 additions and 99 deletions
|
|
@ -159,6 +159,17 @@ def run_api_startup() -> None:
|
|||
)
|
||||
except Exception:
|
||||
_log.exception("startup: biz_state collect recovery failed")
|
||||
try:
|
||||
from .biz_state.compare_service import recover_interrupted_compares_on_startup
|
||||
|
||||
cmp_rec = recover_interrupted_compares_on_startup(db)
|
||||
if cmp_rec.get("runs"):
|
||||
_log.info(
|
||||
"startup: cancelled %s interrupted biz compare run(s)",
|
||||
cmp_rec.get("runs"),
|
||||
)
|
||||
except Exception:
|
||||
_log.exception("startup: biz compare recovery failed")
|
||||
try:
|
||||
from .port_traffic_migrate import backfill_port_traffic_series
|
||||
|
||||
|
|
|
|||
|
|
@ -453,6 +453,53 @@ def _sheet_meta_from_summary(summary: dict[str, Any], run: BizCompareRun, tpl: A
|
|||
]
|
||||
|
||||
|
||||
def _metric_row_estimate(
|
||||
db: Session, batch_ids: list[str], metric_id: str
|
||||
) -> int:
|
||||
"""Cheap size hint from batch command row_count (max across sides)."""
|
||||
mid = str(metric_id or "").strip()
|
||||
if not mid:
|
||||
return 0
|
||||
best = 0
|
||||
for bid in batch_ids:
|
||||
bid = str(bid or "").strip()
|
||||
if not bid:
|
||||
continue
|
||||
rows = (
|
||||
db.query(BizStateBatchCommand)
|
||||
.filter(
|
||||
BizStateBatchCommand.batch_id == bid,
|
||||
BizStateBatchCommand.metric_id == mid,
|
||||
)
|
||||
.all()
|
||||
)
|
||||
if not rows:
|
||||
continue
|
||||
n = sum(int(c.row_count or 0) for c in rows)
|
||||
if n > best:
|
||||
best = n
|
||||
return best
|
||||
|
||||
|
||||
def _order_sheets_small_first(
|
||||
db: Session,
|
||||
sheets: list[dict[str, Any]],
|
||||
*,
|
||||
before_batch_id: str,
|
||||
after_batch_id: str,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Run smaller metrics first so field engineers can review early results."""
|
||||
if len(sheets) <= 1:
|
||||
return list(sheets)
|
||||
batches = [before_batch_id, after_batch_id]
|
||||
scored: list[tuple[int, int, dict[str, Any]]] = []
|
||||
for i, sheet in enumerate(sheets):
|
||||
n = _metric_row_estimate(db, batches, str(sheet.get("metric_id") or ""))
|
||||
scored.append((n, i, sheet))
|
||||
scored.sort(key=lambda x: (x[0], x[1]))
|
||||
return [s for _, _, s in scored]
|
||||
|
||||
|
||||
def _pending_sheet_meta(sheet: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Placeholder meta so the UI lists all check items while a run is in flight."""
|
||||
key_fields = list(sheet.get("key_fields") or [])
|
||||
|
|
@ -1672,6 +1719,40 @@ def _resolve_after_batch(db: Session, job: BizCompareJob) -> str:
|
|||
return str(latest.id) if latest else ""
|
||||
|
||||
|
||||
def _sheet_result_envelope(
|
||||
sheet: dict[str, Any],
|
||||
*,
|
||||
key_fields: list[str],
|
||||
iface_fields: list[str],
|
||||
compare_fields: list[str],
|
||||
display_fields: list[str],
|
||||
row_filters: list[Any],
|
||||
field_rules: list[Any],
|
||||
ignore_ports: bool | None,
|
||||
mode: str,
|
||||
summary: dict[str, Any],
|
||||
diffs: list[Any],
|
||||
mapping_stats: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"sheet_id": sheet_key(sheet),
|
||||
"title": sheet_title(sheet),
|
||||
"metric_id": sheet["metric_id"],
|
||||
"key_fields": key_fields,
|
||||
"iface_fields": iface_fields,
|
||||
"compare_fields": compare_fields,
|
||||
"display_fields": display_fields,
|
||||
"row_filters": row_filters,
|
||||
"field_rules": field_rules,
|
||||
"ignore_port_changes": ignore_ports,
|
||||
"mode": mode,
|
||||
"status": "done",
|
||||
"summary": summary,
|
||||
"diffs": diffs,
|
||||
"mapping_stats": mapping_stats,
|
||||
}
|
||||
|
||||
|
||||
def _run_sheet(
|
||||
db: Session,
|
||||
*,
|
||||
|
|
@ -1681,7 +1762,7 @@ def _run_sheet(
|
|||
port_map: dict[str, str],
|
||||
iface_normalize_rules: list[dict[str, str]] | None = None,
|
||||
store_unchanged: str = "auto",
|
||||
on_load_progress: Callable[[str, int], None] | None = None,
|
||||
on_load_progress: Callable[..., None] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
key_fields = list(sheet.get("key_fields") or [])
|
||||
iface_fields = list(sheet.get("iface_fields") or [])
|
||||
|
|
@ -1704,13 +1785,77 @@ def _run_sheet(
|
|||
ignore_ports = bool(ignore_ports)
|
||||
mid = sheet["metric_id"]
|
||||
|
||||
def _emit_load(side: str, n: int, **meta: Any) -> None:
|
||||
if not on_load_progress:
|
||||
return
|
||||
try:
|
||||
on_load_progress(side, n, **meta)
|
||||
except TypeError:
|
||||
on_load_progress(side, n)
|
||||
|
||||
# PostgreSQL path: pushdown-safe sheets join in-DB (BGP-scale).
|
||||
from .compare_sql import run_sql_sheet_compare, sql_compare_skip_reason
|
||||
|
||||
skip_reason = sql_compare_skip_reason(
|
||||
db,
|
||||
sheet,
|
||||
port_map=port_map,
|
||||
iface_normalize_rules=iface_normalize_rules,
|
||||
)
|
||||
if not skip_reason:
|
||||
try:
|
||||
result = run_sql_sheet_compare(
|
||||
db,
|
||||
sheet=sheet,
|
||||
before_batch_id=before_batch_id,
|
||||
after_batch_id=after_batch_id,
|
||||
store_unchanged=store_unchanged,
|
||||
on_progress=_emit_load,
|
||||
)
|
||||
summary = dict(result["summary"])
|
||||
# Engine already sets raw counts / policy; keep keys stable
|
||||
if "unchanged_policy" not in summary:
|
||||
summary["unchanged_policy"] = resolve_unchanged_policy(
|
||||
store_unchanged,
|
||||
before_n=int(summary.get("before_count") or 0),
|
||||
after_n=int(summary.get("after_count") or 0),
|
||||
)
|
||||
return _sheet_result_envelope(
|
||||
sheet,
|
||||
key_fields=key_fields,
|
||||
iface_fields=iface_fields,
|
||||
compare_fields=compare_fields,
|
||||
display_fields=display_fields,
|
||||
row_filters=row_filters,
|
||||
field_rules=field_rules,
|
||||
ignore_ports=ignore_ports,
|
||||
mode=mode,
|
||||
summary=summary,
|
||||
diffs=list(result.get("diffs") or []),
|
||||
mapping_stats=dict(result.get("mapping_stats") or {}),
|
||||
)
|
||||
except Exception:
|
||||
_log.exception(
|
||||
"sql compare fallback sheet=%s metric=%s — using Python engine",
|
||||
sheet_key(sheet),
|
||||
mid,
|
||||
)
|
||||
skip_reason = "sql_error_fallback"
|
||||
else:
|
||||
_log.info(
|
||||
"python compare sheet=%s metric=%s skip_sql=%s",
|
||||
sheet_key(sheet),
|
||||
mid,
|
||||
skip_reason,
|
||||
)
|
||||
|
||||
_emit_load("before", 0, engine="python", note=skip_reason or "python", phase="loading")
|
||||
|
||||
def _before_chunk(n: int) -> None:
|
||||
if on_load_progress:
|
||||
on_load_progress("before", n)
|
||||
_emit_load("before", n, engine="python", note=skip_reason or "python", phase="loading")
|
||||
|
||||
def _after_chunk(n: int) -> None:
|
||||
if on_load_progress:
|
||||
on_load_progress("after", n)
|
||||
_emit_load("after", n, engine="python", note=skip_reason or "python", phase="loading")
|
||||
|
||||
before_raw = _load_metric_rows(
|
||||
db, batch_id=before_batch_id, metric_id=mid, on_chunk=_before_chunk
|
||||
|
|
@ -1742,23 +1887,21 @@ def _run_sheet(
|
|||
summary["after_raw_count"] = len(after_raw)
|
||||
summary["row_filters"] = len(row_filters)
|
||||
summary["unchanged_policy"] = policy
|
||||
return {
|
||||
"sheet_id": sheet_key(sheet),
|
||||
"title": sheet_title(sheet),
|
||||
"metric_id": sheet["metric_id"],
|
||||
"key_fields": key_fields,
|
||||
"iface_fields": iface_fields,
|
||||
"compare_fields": compare_fields,
|
||||
"display_fields": display_fields,
|
||||
"row_filters": row_filters,
|
||||
"field_rules": field_rules,
|
||||
"ignore_port_changes": ignore_ports,
|
||||
"mode": mode,
|
||||
"status": "done",
|
||||
"summary": summary,
|
||||
"diffs": result["diffs"],
|
||||
"mapping_stats": result["mapping_stats"],
|
||||
}
|
||||
summary.setdefault("engine", "python")
|
||||
return _sheet_result_envelope(
|
||||
sheet,
|
||||
key_fields=key_fields,
|
||||
iface_fields=iface_fields,
|
||||
compare_fields=compare_fields,
|
||||
display_fields=display_fields,
|
||||
row_filters=row_filters,
|
||||
field_rules=field_rules,
|
||||
ignore_ports=ignore_ports,
|
||||
mode=mode,
|
||||
summary=summary,
|
||||
diffs=list(result.get("diffs") or []),
|
||||
mapping_stats=dict(result.get("mapping_stats") or {}),
|
||||
)
|
||||
|
||||
|
||||
def _validate_compare_job(
|
||||
|
|
@ -1784,6 +1927,12 @@ def _validate_compare_job(
|
|||
sheets_cfg = _filter_enabled_sheets(sheets_cfg, getattr(j, "enabled_sheet_ids", None))
|
||||
if not sheets_cfg:
|
||||
raise HTTPException(status_code=400, detail="no_enabled_sheets")
|
||||
sheets_cfg = _order_sheets_small_first(
|
||||
db,
|
||||
sheets_cfg,
|
||||
before_batch_id=before_batch_id,
|
||||
after_batch_id=after_batch_id,
|
||||
)
|
||||
return {
|
||||
"job": j,
|
||||
"template": tpl,
|
||||
|
|
@ -1868,6 +2017,10 @@ def _set_run_progress(
|
|||
progress.update(extra)
|
||||
prev["progress"] = progress
|
||||
run.summary_json = prev
|
||||
# Re-read status from DB — cancel may have been committed by another session
|
||||
# (UI cancel / startup recovery) while this worker still holds a stale "running".
|
||||
db.expire(run, ["status", "message"])
|
||||
if str(run.status or "") != "cancelled":
|
||||
run.status = "running"
|
||||
title = progress["sheet_title"] or progress["sheet_id"] or ""
|
||||
run.message = (
|
||||
|
|
@ -1906,6 +2059,12 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
|||
run.message = "no_enabled_sheets"
|
||||
db.commit()
|
||||
raise HTTPException(status_code=400, detail="no_enabled_sheets")
|
||||
sheets_cfg = _order_sheets_small_first(
|
||||
db,
|
||||
sheets_cfg,
|
||||
before_batch_id=before_batch_id,
|
||||
after_batch_id=after_batch_id,
|
||||
)
|
||||
|
||||
started_mono = time.monotonic()
|
||||
store_mode = normalize_store_unchanged(getattr(j, "store_unchanged", None))
|
||||
|
|
@ -1927,20 +2086,30 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
|||
field_counts: dict[str, int] = {}
|
||||
total = len(sheets_cfg)
|
||||
|
||||
# Prefer seeded pending sheets from create; rebuild if missing
|
||||
sheet_metas: list[dict[str, Any]] = list((run.summary_json or {}).get("sheets") or [])
|
||||
if len(sheet_metas) != total:
|
||||
sheet_metas = [_pending_sheet_meta(s) for s in sheets_cfg]
|
||||
# Prefer seeded pending sheets from create; realign to small-first order
|
||||
prev_metas = list((run.summary_json or {}).get("sheets") or [])
|
||||
by_key = {sheet_key(m): m for m in prev_metas}
|
||||
sheet_metas = [
|
||||
by_key.get(sheet_key(s)) or _pending_sheet_meta(s) for s in sheets_cfg
|
||||
]
|
||||
|
||||
def _publish_sheets() -> None:
|
||||
prev = dict(run.summary_json or {})
|
||||
prev["sheets"] = list(sheet_metas)
|
||||
prev.update({k: agg[k] for k in agg})
|
||||
run.summary_json = prev
|
||||
db.expire(run, ["status"])
|
||||
# Never resurrect cancelled while publishing incremental sheet metas
|
||||
if str(run.status or "") == "cancelled":
|
||||
db.commit()
|
||||
return
|
||||
db.commit()
|
||||
|
||||
try:
|
||||
for idx, sheet in enumerate(sheets_cfg, start=1):
|
||||
if _run_is_cancelled(db, run_id):
|
||||
run = db.get(BizCompareRun, run_id) or run
|
||||
return get_run(db, run.id)
|
||||
sid = sheet_key(sheet)
|
||||
# Mark current sheet running in the sidebar list
|
||||
for meta in sheet_metas:
|
||||
|
|
@ -1949,27 +2118,43 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
|||
break
|
||||
_publish_sheets()
|
||||
|
||||
_load_pub = {"t": 0.0, "n": -1}
|
||||
_load_pub = {"t": 0.0, "n": -1, "engine": ""}
|
||||
|
||||
def _on_load(side: str, n: int, _idx: int = idx, _sheet: dict = sheet) -> None:
|
||||
def _on_load(
|
||||
side: str,
|
||||
n: int,
|
||||
*,
|
||||
engine: str = "python",
|
||||
note: str = "",
|
||||
phase: str | None = None,
|
||||
_idx: int = idx,
|
||||
_sheet: dict = sheet,
|
||||
) -> None:
|
||||
now = time.monotonic()
|
||||
# Avoid committing every 5k on million-row BGP loads
|
||||
eng = str(engine or "python")
|
||||
# SQL emits sparse updates; Python still throttle chunk spam
|
||||
if eng != "sql":
|
||||
if n - _load_pub["n"] < 25_000 and now - _load_pub["t"] < 2.0:
|
||||
return
|
||||
_load_pub["t"] = now
|
||||
_load_pub["n"] = n
|
||||
_load_pub["engine"] = eng
|
||||
extra: dict[str, Any] = {
|
||||
"load_side": side,
|
||||
"rows_loaded": n,
|
||||
"engine": eng,
|
||||
}
|
||||
if note:
|
||||
extra["engine_note"] = str(note)[:128]
|
||||
_set_run_progress(
|
||||
db,
|
||||
run,
|
||||
phase="loading",
|
||||
phase=str(phase or ("loading" if eng != "sql" else "sql_count")),
|
||||
sheet_index=_idx,
|
||||
sheet_total=total,
|
||||
sheet=_sheet,
|
||||
started_mono=started_mono,
|
||||
extra={
|
||||
"load_side": side,
|
||||
"rows_loaded": n,
|
||||
},
|
||||
extra=extra,
|
||||
)
|
||||
|
||||
_set_run_progress(
|
||||
|
|
@ -1980,6 +2165,7 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
|||
sheet_total=total,
|
||||
sheet=sheet,
|
||||
started_mono=started_mono,
|
||||
extra={"engine": "", "engine_note": ""},
|
||||
)
|
||||
one = _run_sheet(
|
||||
db,
|
||||
|
|
@ -2091,6 +2277,13 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
|||
fail_diffs.clear()
|
||||
ok_diffs.clear()
|
||||
db.commit()
|
||||
if _run_is_cancelled(db, run_id):
|
||||
run = db.get(BizCompareRun, run_id) or run
|
||||
return get_run(db, run.id)
|
||||
|
||||
if _run_is_cancelled(db, run_id):
|
||||
run = db.get(BizCompareRun, run_id) or run
|
||||
return get_run(db, run.id)
|
||||
|
||||
duration_ms = int((time.monotonic() - started_mono) * 1000)
|
||||
top_fields = sorted(
|
||||
|
|
@ -2117,6 +2310,9 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
|||
},
|
||||
"sheets": sheet_metas,
|
||||
}
|
||||
run = db.get(BizCompareRun, run_id) or run
|
||||
if str(run.status or "") == "cancelled":
|
||||
return get_run(db, run.id)
|
||||
run.status = "success"
|
||||
run.summary_json = summary_payload
|
||||
run.mapping_stats_json = mapping_by_metric
|
||||
|
|
@ -2129,6 +2325,8 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
|||
db.commit()
|
||||
return get_run(db, run.id)
|
||||
except HTTPException as exc:
|
||||
run = db.get(BizCompareRun, run_id) or run
|
||||
if str(run.status or "") != "cancelled":
|
||||
run.status = "failed"
|
||||
run.message = str(getattr(exc, "detail", "") or exc)[:1024]
|
||||
prev = dict(run.summary_json or {})
|
||||
|
|
@ -2142,6 +2340,8 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
|||
raise
|
||||
except Exception as exc:
|
||||
_log.exception("compare run failed run=%s job=%s", run_id, j.id)
|
||||
run = db.get(BizCompareRun, run_id) or run
|
||||
if str(run.status or "") != "cancelled":
|
||||
run.status = "failed"
|
||||
run.message = str(exc)[:1024]
|
||||
prev = dict(run.summary_json or {})
|
||||
|
|
@ -2155,8 +2355,93 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
|||
raise
|
||||
|
||||
|
||||
_INTERRUPT_MARK = "interrupted_by_restart"
|
||||
|
||||
|
||||
def _job_has_active_run(db: Session, job_id: str) -> BizCompareRun | None:
|
||||
return (
|
||||
db.query(BizCompareRun)
|
||||
.filter(
|
||||
BizCompareRun.job_id == str(job_id or ""),
|
||||
BizCompareRun.status.in_(("running", "queued")),
|
||||
)
|
||||
.order_by(BizCompareRun.created_at.desc())
|
||||
.first()
|
||||
)
|
||||
|
||||
|
||||
def _run_is_cancelled(db: Session, run_id: str) -> bool:
|
||||
"""Re-read status so user/startup cancel is visible to the worker thread."""
|
||||
db.expire_all()
|
||||
r = db.get(BizCompareRun, run_id)
|
||||
return bool(r and str(r.status or "") == "cancelled")
|
||||
|
||||
|
||||
def recover_interrupted_compares_on_startup(db: Session) -> dict[str, Any]:
|
||||
"""Mark orphaned running/queued compare runs as cancelled after process restart.
|
||||
|
||||
In-memory job locks die with the process; without this, the UI stays on
|
||||
「比对中」and blocks a new run.
|
||||
"""
|
||||
now = _utcnow()
|
||||
rows = (
|
||||
db.query(BizCompareRun)
|
||||
.filter(BizCompareRun.status.in_(("running", "queued")))
|
||||
.all()
|
||||
)
|
||||
n = 0
|
||||
for r in rows:
|
||||
r.status = "cancelled"
|
||||
msg = str(r.message or "").strip()
|
||||
if _INTERRUPT_MARK not in msg:
|
||||
r.message = f"{msg} | {_INTERRUPT_MARK}".strip(" |")[:1024]
|
||||
prev = dict(r.summary_json or {})
|
||||
prog = dict(prev.get("progress") or {})
|
||||
prog["phase"] = "cancelled"
|
||||
prog["elapsed_ms"] = int(prog.get("elapsed_ms") or 0)
|
||||
prev["progress"] = prog
|
||||
# Mark in-flight sheet placeholders so UI does not show fake pass
|
||||
sheets = list(prev.get("sheets") or [])
|
||||
for sh in sheets:
|
||||
st = str(sh.get("status") or "")
|
||||
if st in ("pending", "running", "queued"):
|
||||
sh["status"] = "cancelled"
|
||||
prev["sheets"] = sheets
|
||||
r.summary_json = prev
|
||||
n += 1
|
||||
if n:
|
||||
db.commit()
|
||||
_log.info("startup: cancelled %s interrupted compare run(s)", n)
|
||||
return {"runs": n, "at": now.isoformat() + "Z"}
|
||||
|
||||
|
||||
def cancel_compare_run(db: Session, run_id: str) -> dict[str, Any]:
|
||||
"""Cancel a running/queued compare so a new run can start."""
|
||||
r = db.get(BizCompareRun, run_id)
|
||||
if not r:
|
||||
raise HTTPException(status_code=404, detail="run_not_found")
|
||||
st = str(r.status or "")
|
||||
if st not in ("running", "queued"):
|
||||
return get_run(db, run_id)
|
||||
r.status = "cancelled"
|
||||
msg = str(r.message or "").strip()
|
||||
r.message = f"{msg} | cancelled_by_user".strip(" |")[:1024]
|
||||
prev = dict(r.summary_json or {})
|
||||
prog = dict(prev.get("progress") or {})
|
||||
prog["phase"] = "cancelled"
|
||||
prev["progress"] = prog
|
||||
for sh in list(prev.get("sheets") or []):
|
||||
if str(sh.get("status") or "") in ("pending", "running", "queued"):
|
||||
sh["status"] = "cancelled"
|
||||
r.summary_json = prev
|
||||
db.commit()
|
||||
return get_run(db, run_id)
|
||||
|
||||
|
||||
def run_compare(db: Session, job_id: str, *, force_after_batch_id: str = "") -> dict[str, Any]:
|
||||
"""Synchronous compare (auto-compare / tests). Blocks the caller until done."""
|
||||
if _job_has_active_run(db, job_id):
|
||||
raise HTTPException(status_code=409, detail="compare_already_running")
|
||||
lock = _job_compare_lock(job_id)
|
||||
if not lock.acquire(blocking=False):
|
||||
raise HTTPException(status_code=409, detail="compare_already_running")
|
||||
|
|
@ -2183,6 +2468,8 @@ def enqueue_compare(
|
|||
Returns immediately so the HTTP worker / UI stay responsive. Poll
|
||||
``GET /compare/runs/{id}`` for progress (``summary.progress``).
|
||||
"""
|
||||
if _job_has_active_run(db, job_id):
|
||||
raise HTTPException(status_code=409, detail="compare_already_running")
|
||||
lock = _job_compare_lock(job_id)
|
||||
if not lock.acquire(blocking=False):
|
||||
raise HTTPException(status_code=409, detail="compare_already_running")
|
||||
|
|
|
|||
646
netx_api/biz_state/compare_sql.py
Normal file
646
netx_api/biz_state/compare_sql.py
Normal file
|
|
@ -0,0 +1,646 @@
|
|||
"""PostgreSQL-backed sheet compare (FULL OUTER JOIN) for pushdown-safe sheets.
|
||||
|
||||
Falls back to the Python engine when port-map / complex rules cannot be expressed
|
||||
in SQL. See ``can_sql_compare``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from typing import Any, Callable, Mapping, Sequence
|
||||
from uuid import uuid4
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from .compare_rules import effective_compare_fields, field_rule_map
|
||||
|
||||
_log = logging.getLogger("netx.biz_state.compare_sql")
|
||||
|
||||
_FIELD_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
||||
_SQL_FILTER_OPS = frozenset(
|
||||
{"eq", "==", "ne", "!=", "in", "not_in", "nin", "contains", "empty", "not_empty", "nonempty", "ci_eq"}
|
||||
)
|
||||
_SQL_NORMALIZE = frozenset({"", "none", "strip", "lower", "upper", "empty_as_blank"})
|
||||
_SQL_COMPARE_MODES = frozenset({"", "eq", "ignore", "skip", "off"})
|
||||
_EMPTY_AS_BLANK = ("n/a", "na", "-", "--", "none", "null")
|
||||
|
||||
|
||||
def _dialect_is_postgres(db: Session) -> bool:
|
||||
bind = db.get_bind()
|
||||
if bind is None:
|
||||
return False
|
||||
return str(getattr(bind.dialect, "name", "") or "").lower() in ("postgresql", "postgres")
|
||||
|
||||
|
||||
def _safe_field(name: str) -> str:
|
||||
f = str(name or "").strip()
|
||||
if not f or not _FIELD_RE.match(f):
|
||||
raise ValueError(f"unsafe_json_field:{name!r}")
|
||||
return f
|
||||
|
||||
|
||||
def _json_text_expr(alias: str, field: str) -> str:
|
||||
"""SQL expression: trimmed text from JSONB column ``alias.data`` / ``data_json``."""
|
||||
f = _safe_field(field)
|
||||
# alias is caller-controlled (_b / data_json) — not user input
|
||||
return f"trim(both from coalesce({alias}->>'{f}', ''))"
|
||||
|
||||
|
||||
def _norm_expr(alias: str, field: str, normalize: str) -> str:
|
||||
base = _json_text_expr(alias, field)
|
||||
mode = str(normalize or "strip").strip().lower() or "strip"
|
||||
if mode in ("", "none", "strip"):
|
||||
return base
|
||||
if mode == "lower":
|
||||
return f"lower({base})"
|
||||
if mode == "upper":
|
||||
return f"upper({base})"
|
||||
if mode == "empty_as_blank":
|
||||
opts = ", ".join(f"'{x}'" for x in _EMPTY_AS_BLANK)
|
||||
return (
|
||||
f"CASE WHEN lower({base}) IN ({opts}) THEN '' ELSE {base} END"
|
||||
)
|
||||
raise ValueError(f"unsupported_normalize:{mode}")
|
||||
|
||||
|
||||
def _filters_sql_compatible(filters: Sequence[Mapping[str, Any]] | None) -> bool:
|
||||
fl = [f for f in (filters or []) if isinstance(f, dict)]
|
||||
return all(_filter_node_compatible(f) for f in fl)
|
||||
|
||||
|
||||
def _filter_node_compatible(filt: Mapping[str, Any]) -> bool:
|
||||
if "any" in filt:
|
||||
kids = filt.get("any") or []
|
||||
return isinstance(kids, list) and all(
|
||||
isinstance(k, dict) and _filter_node_compatible(k) for k in kids
|
||||
)
|
||||
if "all" in filt:
|
||||
kids = filt.get("all") or []
|
||||
return isinstance(kids, list) and all(
|
||||
isinstance(k, dict) and _filter_node_compatible(k) for k in kids
|
||||
)
|
||||
field = str(filt.get("field") or "").strip()
|
||||
op = str(filt.get("op") or "eq").strip().lower()
|
||||
if op not in _SQL_FILTER_OPS:
|
||||
return False
|
||||
if op in ("empty", "not_empty", "nonempty"):
|
||||
return bool(field) and bool(_FIELD_RE.match(field))
|
||||
if not field or not _FIELD_RE.match(field):
|
||||
return False
|
||||
if op in ("in", "not_in", "nin"):
|
||||
vals = filt.get("value")
|
||||
if vals is None:
|
||||
return True
|
||||
if not isinstance(vals, (list, tuple)):
|
||||
vals = [vals]
|
||||
return all(isinstance(x, (str, int, float, bool)) or x is None for x in vals)
|
||||
return True
|
||||
|
||||
|
||||
def _field_rules_sql_compatible(rules: Sequence[Mapping[str, Any]] | None) -> bool:
|
||||
for raw in rules or []:
|
||||
if not isinstance(raw, dict):
|
||||
return False
|
||||
name = str(raw.get("field") or "").strip()
|
||||
if name and not _FIELD_RE.match(name):
|
||||
return False
|
||||
mode = str(raw.get("compare") or "eq").strip().lower() or "eq"
|
||||
if mode not in _SQL_COMPARE_MODES:
|
||||
return False
|
||||
if raw.get("ignore") is True:
|
||||
continue
|
||||
if mode in ("ignore", "skip", "off"):
|
||||
continue
|
||||
norm = str(raw.get("normalize") or "strip").strip().lower() or "strip"
|
||||
if norm not in _SQL_NORMALIZE:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def sql_compare_skip_reason(
|
||||
db: Session,
|
||||
sheet: Mapping[str, Any],
|
||||
*,
|
||||
port_map: Mapping[str, str] | None = None,
|
||||
iface_normalize_rules: Sequence[Mapping[str, str]] | None = None,
|
||||
) -> str:
|
||||
"""Empty string when SQL is allowed; otherwise a short reason for progress/logs.
|
||||
|
||||
Simple ``row_filters`` (eq/in/contains/…) used for BGP sheet splits are fine —
|
||||
they do **not** force Python by themselves.
|
||||
"""
|
||||
if not _dialect_is_postgres(db):
|
||||
return "not_postgres"
|
||||
if port_map:
|
||||
return "port_map"
|
||||
key_fields = [str(k).strip() for k in (sheet.get("key_fields") or []) if str(k).strip()]
|
||||
if not key_fields:
|
||||
return "no_key_fields"
|
||||
if any(not _FIELD_RE.match(k) for k in key_fields):
|
||||
return "unsafe_key_field"
|
||||
iface_fields = [str(f).strip() for f in (sheet.get("iface_fields") or []) if str(f).strip()]
|
||||
iface_set = set(iface_fields)
|
||||
ignore_ports = sheet.get("ignore_port_changes")
|
||||
if ignore_ports is True:
|
||||
return "ignore_port_changes"
|
||||
if ignore_ports is None and iface_set and any(k in iface_set for k in key_fields):
|
||||
return "auto_ignore_ports"
|
||||
field_rules = list(sheet.get("field_rules") or [])
|
||||
if not _field_rules_sql_compatible(field_rules):
|
||||
return "field_rules"
|
||||
compare_fields = effective_compare_fields(
|
||||
list(sheet.get("compare_fields") or []),
|
||||
field_rules,
|
||||
)
|
||||
if any(not _FIELD_RE.match(str(f).strip()) for f in compare_fields if str(f).strip()):
|
||||
return "unsafe_compare_field"
|
||||
used = set(key_fields) | {str(f).strip() for f in compare_fields if str(f).strip()}
|
||||
if iface_normalize_rules and (used & iface_set):
|
||||
return "iface_normalize"
|
||||
if not _filters_sql_compatible(list(sheet.get("row_filters") or [])):
|
||||
return "row_filters"
|
||||
if not str(sheet.get("metric_id") or "").strip():
|
||||
return "no_metric_id"
|
||||
return ""
|
||||
|
||||
|
||||
def can_sql_compare(
|
||||
db: Session,
|
||||
sheet: Mapping[str, Any],
|
||||
*,
|
||||
port_map: Mapping[str, str] | None = None,
|
||||
iface_normalize_rules: Sequence[Mapping[str, str]] | None = None,
|
||||
) -> bool:
|
||||
"""True when this sheet can run entirely as a PostgreSQL JOIN."""
|
||||
return not bool(
|
||||
sql_compare_skip_reason(
|
||||
db,
|
||||
sheet,
|
||||
port_map=port_map,
|
||||
iface_normalize_rules=iface_normalize_rules,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def compile_row_filters_sql(
|
||||
filters: Sequence[Mapping[str, Any]] | None,
|
||||
*,
|
||||
json_col: str = "data_json",
|
||||
param_prefix: str = "f",
|
||||
) -> tuple[str, dict[str, Any]]:
|
||||
"""Compile template row_filters to SQL AND-clause + bind params.
|
||||
|
||||
Returns ``(sql_fragment, params)``. Empty filters → ``(\"TRUE\", {})``.
|
||||
``json_col`` is the JSONB column/expression name (trusted).
|
||||
"""
|
||||
fl = [f for f in (filters or []) if isinstance(f, dict)]
|
||||
if not fl:
|
||||
return "TRUE", {}
|
||||
params: dict[str, Any] = {}
|
||||
counter = {"n": 0}
|
||||
|
||||
def _next(name: str) -> str:
|
||||
counter["n"] += 1
|
||||
return f"{param_prefix}_{counter['n']}_{name}"
|
||||
|
||||
def _node(filt: Mapping[str, Any]) -> str:
|
||||
if "any" in filt:
|
||||
kids = [k for k in (filt.get("any") or []) if isinstance(k, dict)]
|
||||
if not kids:
|
||||
return "TRUE"
|
||||
return "(" + " OR ".join(_node(k) for k in kids) + ")"
|
||||
if "all" in filt:
|
||||
kids = [k for k in (filt.get("all") or []) if isinstance(k, dict)]
|
||||
if not kids:
|
||||
return "TRUE"
|
||||
return "(" + " AND ".join(_node(k) for k in kids) + ")"
|
||||
field = _safe_field(str(filt.get("field") or ""))
|
||||
op = str(filt.get("op") or "eq").strip().lower()
|
||||
expr = _json_text_expr(json_col, field)
|
||||
# _json_text_expr uses alias->> ; for bare column use data_json directly
|
||||
if json_col == "data_json":
|
||||
expr = f"trim(both from coalesce(data_json->>'{field}', ''))"
|
||||
if op in ("empty",):
|
||||
return f"({expr} = '')"
|
||||
if op in ("not_empty", "nonempty"):
|
||||
return f"({expr} <> '')"
|
||||
expect = filt.get("value")
|
||||
if op in ("eq", "==", "ci_eq"):
|
||||
key = _next("v")
|
||||
params[key] = str(expect or "").strip()
|
||||
if op == "ci_eq" or op in ("eq", "=="):
|
||||
# Python eq is case-insensitive via .lower()
|
||||
params[key] = str(expect or "").strip().lower()
|
||||
return f"(lower({expr}) = :{key})"
|
||||
if op in ("ne", "!="):
|
||||
key = _next("v")
|
||||
params[key] = str(expect or "").strip().lower()
|
||||
return f"(lower({expr}) <> :{key})"
|
||||
if op == "contains":
|
||||
key = _next("v")
|
||||
needle = str(expect or "").strip().lower()
|
||||
params[key] = f"%{needle}%"
|
||||
return f"(lower({expr}) LIKE :{key})"
|
||||
if op in ("in", "not_in", "nin"):
|
||||
vals = expect
|
||||
if vals is None:
|
||||
vals = []
|
||||
if not isinstance(vals, (list, tuple)):
|
||||
vals = [vals]
|
||||
clean = [str(x).strip().lower() for x in vals if str(x).strip()]
|
||||
if not clean:
|
||||
return "FALSE" if op == "in" else "TRUE"
|
||||
keys = []
|
||||
for i, v in enumerate(clean):
|
||||
k = _next(f"in{i}")
|
||||
params[k] = v
|
||||
keys.append(f":{k}")
|
||||
inside = f"lower({expr}) IN ({', '.join(keys)})"
|
||||
return f"({inside})" if op == "in" else f"(NOT {inside})"
|
||||
raise ValueError(f"unsupported_filter_op:{op}")
|
||||
|
||||
parts = [_node(f) for f in fl]
|
||||
return "(" + " AND ".join(parts) + ")", params
|
||||
|
||||
|
||||
def _rk_sql(key_fields: list[str], *, json_col: str = "data_json") -> str:
|
||||
parts = []
|
||||
for k in key_fields:
|
||||
f = _safe_field(k)
|
||||
if json_col == "data_json":
|
||||
parts.append(f"trim(both from coalesce(data_json->>'{f}', ''))")
|
||||
else:
|
||||
parts.append(f"trim(both from coalesce({json_col}->>'{f}', ''))")
|
||||
if len(parts) == 1:
|
||||
return parts[0]
|
||||
return "concat_ws('|', " + ", ".join(parts) + ")"
|
||||
|
||||
|
||||
def _changed_predicate(
|
||||
compare_fields: list[str],
|
||||
rules: dict[str, dict[str, Any]],
|
||||
*,
|
||||
before_alias: str = "b",
|
||||
after_alias: str = "a",
|
||||
) -> str:
|
||||
"""SQL boolean: True when any compare field differs (IS DISTINCT FROM)."""
|
||||
if not compare_fields:
|
||||
return "FALSE"
|
||||
clauses: list[str] = []
|
||||
for f in compare_fields:
|
||||
name = str(f).strip()
|
||||
if not name:
|
||||
continue
|
||||
rule = rules.get(name) or {}
|
||||
mode = str(rule.get("compare") or "eq").strip().lower() or "eq"
|
||||
if mode in ("ignore", "skip", "off") or rule.get("ignore") is True:
|
||||
continue
|
||||
norm = str(rule.get("normalize") or "strip").strip().lower() or "strip"
|
||||
bv = _norm_expr(f"{before_alias}.data", name, norm)
|
||||
av = _norm_expr(f"{after_alias}.data", name, norm)
|
||||
clauses.append(f"({bv} IS DISTINCT FROM {av})")
|
||||
if not clauses:
|
||||
return "FALSE"
|
||||
return "(" + " OR ".join(clauses) + ")"
|
||||
|
||||
|
||||
def _key_obj_from_data(data: dict[str, Any] | None, key_fields: list[str]) -> dict[str, Any]:
|
||||
row = data if isinstance(data, dict) else {}
|
||||
return {f: row.get(f, "") for f in key_fields}
|
||||
|
||||
|
||||
def _changes_from_rows(
|
||||
before: dict[str, Any] | None,
|
||||
after: dict[str, Any] | None,
|
||||
compare_fields: list[str],
|
||||
rules: dict[str, dict[str, Any]],
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
from .compare_rules import explain_diff, values_equal
|
||||
|
||||
out: dict[str, dict[str, Any]] = {}
|
||||
b = before or {}
|
||||
a = after or {}
|
||||
for f in compare_fields:
|
||||
rule = rules.get(f)
|
||||
bv = b.get(f, "")
|
||||
av = a.get(f, "")
|
||||
if values_equal(bv, av, rule=rule):
|
||||
continue
|
||||
entry: dict[str, Any] = {"before": bv, "after": av}
|
||||
reason = explain_diff(bv, av, rule=rule)
|
||||
if reason:
|
||||
entry["reason"] = reason
|
||||
out[f] = entry
|
||||
return out
|
||||
|
||||
|
||||
def run_sql_sheet_compare(
|
||||
db: Session,
|
||||
*,
|
||||
sheet: Mapping[str, Any],
|
||||
before_batch_id: str,
|
||||
after_batch_id: str,
|
||||
store_unchanged: str = "auto",
|
||||
on_progress: Callable[..., None] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Compare one sheet via PostgreSQL TEMP tables + FULL OUTER JOIN.
|
||||
|
||||
Returns the same ``{summary, diffs, mapping_stats}`` shape as ``compare_rows``.
|
||||
``on_progress(side, n, *, engine=\"sql\", note=..., phase=...)``.
|
||||
"""
|
||||
if not _dialect_is_postgres(db):
|
||||
raise RuntimeError("sql_compare_requires_postgres")
|
||||
|
||||
def _prog(side: str, n: int, *, phase: str = "sql_count", note: str = "") -> None:
|
||||
if not on_progress:
|
||||
return
|
||||
try:
|
||||
on_progress(side, n, engine="sql", note=note, phase=phase)
|
||||
except TypeError:
|
||||
on_progress(side, n)
|
||||
|
||||
key_fields = [str(k).strip() for k in (sheet.get("key_fields") or []) if str(k).strip()]
|
||||
field_rules = list(sheet.get("field_rules") or [])
|
||||
rules = field_rule_map(field_rules)
|
||||
compare_fields = effective_compare_fields(
|
||||
list(sheet.get("compare_fields") or []),
|
||||
field_rules,
|
||||
)
|
||||
row_filters = list(sheet.get("row_filters") or [])
|
||||
mid = str(sheet.get("metric_id") or "").strip()
|
||||
bid_b = str(before_batch_id or "").strip()
|
||||
bid_a = str(after_batch_id or "").strip()
|
||||
if not mid or not bid_b or not bid_a or not key_fields:
|
||||
raise ValueError("sql_compare_missing_args")
|
||||
|
||||
filter_sql, filter_params = compile_row_filters_sql(row_filters)
|
||||
rk = _rk_sql(key_fields)
|
||||
tag = uuid4().hex[:8]
|
||||
tb = f"_netx_cmp_b_{tag}"
|
||||
ta = f"_netx_cmp_a_{tag}"
|
||||
|
||||
_prog("before", 0, phase="sql_count", note="count")
|
||||
|
||||
# Raw counts (no row_filters)
|
||||
raw_b = int(
|
||||
db.execute(
|
||||
text(
|
||||
"SELECT count(*) FROM biz_state_metric_row "
|
||||
"WHERE batch_id = :bid AND metric_id = :mid"
|
||||
),
|
||||
{"bid": bid_b, "mid": mid},
|
||||
).scalar()
|
||||
or 0
|
||||
)
|
||||
_prog("before", raw_b, phase="sql_count")
|
||||
raw_a = int(
|
||||
db.execute(
|
||||
text(
|
||||
"SELECT count(*) FROM biz_state_metric_row "
|
||||
"WHERE batch_id = :bid AND metric_id = :mid"
|
||||
),
|
||||
{"bid": bid_a, "mid": mid},
|
||||
).scalar()
|
||||
or 0
|
||||
)
|
||||
_prog("after", raw_a, phase="sql_count")
|
||||
|
||||
base_params = {"bid": bid_b, "mid": mid, **filter_params}
|
||||
# Build TEMP sides
|
||||
for tname, batch_id, side in ((tb, bid_b, "before"), (ta, bid_a, "after")):
|
||||
db.execute(text(f"DROP TABLE IF EXISTS {tname}"))
|
||||
params = {**base_params, "bid": batch_id}
|
||||
_prog(side, raw_b if side == "before" else raw_a, phase="sql_project", note="temp")
|
||||
# PRESERVE ROWS: compare progress commits must not drop temps mid-run
|
||||
db.execute(
|
||||
text(
|
||||
f"""
|
||||
CREATE TEMP TABLE {tname} ON COMMIT PRESERVE ROWS AS
|
||||
SELECT
|
||||
id,
|
||||
({rk}) AS rk,
|
||||
data_json AS data,
|
||||
row_number() OVER (
|
||||
PARTITION BY ({rk})
|
||||
ORDER BY seq ASC, id ASC
|
||||
) AS dup_rn
|
||||
FROM biz_state_metric_row
|
||||
WHERE batch_id = :bid
|
||||
AND metric_id = :mid
|
||||
AND ({filter_sql})
|
||||
"""
|
||||
),
|
||||
params,
|
||||
)
|
||||
db.execute(text(f"CREATE INDEX ON {tname} (rk) WHERE dup_rn = 1"))
|
||||
|
||||
before_n = int(
|
||||
db.execute(text(f"SELECT count(*) FROM {tb}")).scalar() or 0
|
||||
)
|
||||
after_n = int(
|
||||
db.execute(text(f"SELECT count(*) FROM {ta}")).scalar() or 0
|
||||
)
|
||||
_prog("after", after_n, phase="sql_join", note="join")
|
||||
|
||||
changed_pred = _changed_predicate(compare_fields, rules)
|
||||
kind_expr = f"""
|
||||
CASE
|
||||
WHEN b.id IS NULL THEN 'added'
|
||||
WHEN a.id IS NULL THEN 'removed'
|
||||
WHEN {changed_pred} THEN 'changed'
|
||||
ELSE 'unchanged'
|
||||
END
|
||||
"""
|
||||
|
||||
# Aggregate primary match kinds (first-wins keys only)
|
||||
agg_rows = db.execute(
|
||||
text(
|
||||
f"""
|
||||
SELECT {kind_expr} AS kind, count(*)::bigint AS n
|
||||
FROM (SELECT * FROM {tb} WHERE dup_rn = 1) b
|
||||
FULL OUTER JOIN (SELECT * FROM {ta} WHERE dup_rn = 1) a
|
||||
ON b.rk = a.rk
|
||||
GROUP BY 1
|
||||
"""
|
||||
)
|
||||
).mappings().all()
|
||||
counts = {str(r["kind"]): int(r["n"] or 0) for r in agg_rows}
|
||||
added = counts.get("added", 0)
|
||||
removed = counts.get("removed", 0)
|
||||
changed = counts.get("changed", 0)
|
||||
unchanged = counts.get("unchanged", 0)
|
||||
|
||||
dup_b = int(
|
||||
db.execute(text(f"SELECT count(*) FROM {tb} WHERE dup_rn > 1")).scalar() or 0
|
||||
)
|
||||
dup_a = int(
|
||||
db.execute(text(f"SELECT count(*) FROM {ta} WHERE dup_rn > 1")).scalar() or 0
|
||||
)
|
||||
duplicate = dup_b + dup_a
|
||||
|
||||
# Lazy import — avoid circular import with compare_service
|
||||
from .compare_service import resolve_unchanged_policy
|
||||
|
||||
policy = resolve_unchanged_policy(
|
||||
store_unchanged, before_n=before_n, after_n=after_n
|
||||
)
|
||||
diffs: list[dict[str, Any]] = []
|
||||
|
||||
# Fail + duplicate rows (stream into Python — should be << million)
|
||||
fail_sql = text(
|
||||
f"""
|
||||
SELECT
|
||||
{kind_expr} AS kind,
|
||||
b.id AS before_row_id,
|
||||
a.id AS after_row_id,
|
||||
b.data AS before_data,
|
||||
a.data AS after_data,
|
||||
COALESCE(b.rk, a.rk) AS rk
|
||||
FROM (SELECT * FROM {tb} WHERE dup_rn = 1) b
|
||||
FULL OUTER JOIN (SELECT * FROM {ta} WHERE dup_rn = 1) a
|
||||
ON b.rk = a.rk
|
||||
WHERE {kind_expr} IN ('added', 'removed', 'changed')
|
||||
"""
|
||||
)
|
||||
for row in db.execute(fail_sql).mappings():
|
||||
kind = str(row["kind"] or "")
|
||||
before = dict(row["before_data"] or {}) if row["before_data"] is not None else None
|
||||
after = dict(row["after_data"] or {}) if row["after_data"] is not None else None
|
||||
key_src = after if kind == "added" else (before or after or {})
|
||||
item: dict[str, Any] = {
|
||||
"kind": kind,
|
||||
"key": _key_obj_from_data(key_src, key_fields),
|
||||
"before": before,
|
||||
"after": after,
|
||||
"mapped_before": before,
|
||||
"changes": {},
|
||||
"before_row_id": str(row["before_row_id"] or ""),
|
||||
"after_row_id": str(row["after_row_id"] or ""),
|
||||
}
|
||||
if kind == "changed":
|
||||
item["changes"] = _changes_from_rows(before, after, compare_fields, rules)
|
||||
diffs.append(item)
|
||||
|
||||
# Duplicate extras
|
||||
for side, tname in (("before", tb), ("after", ta)):
|
||||
q = text(
|
||||
f"""
|
||||
SELECT id, data, rk FROM {tname} WHERE dup_rn > 1
|
||||
"""
|
||||
)
|
||||
for row in db.execute(q).mappings():
|
||||
data = dict(row["data"] or {})
|
||||
diffs.append(
|
||||
{
|
||||
"kind": "duplicate",
|
||||
"side": side,
|
||||
"key": _key_obj_from_data(data, key_fields),
|
||||
"before": data if side == "before" else None,
|
||||
"after": data if side == "after" else None,
|
||||
"mapped_before": data if side == "before" else None,
|
||||
"changes": {},
|
||||
"before_row_id": str(row["id"] or "") if side == "before" else "",
|
||||
"after_row_id": str(row["id"] or "") if side == "after" else "",
|
||||
}
|
||||
)
|
||||
|
||||
unchanged_listed = 0
|
||||
include_u = bool(policy.get("include"))
|
||||
compact = bool(policy.get("compact"))
|
||||
limit_n = policy.get("limit")
|
||||
if include_u and unchanged > 0:
|
||||
lim_sql = ""
|
||||
params_u: dict[str, Any] = {}
|
||||
if limit_n is not None:
|
||||
lim_sql = " LIMIT :lim"
|
||||
params_u["lim"] = max(0, int(limit_n))
|
||||
u_sql = text(
|
||||
f"""
|
||||
SELECT
|
||||
b.id AS before_row_id,
|
||||
a.id AS after_row_id,
|
||||
b.data AS before_data,
|
||||
a.data AS after_data
|
||||
FROM (SELECT * FROM {tb} WHERE dup_rn = 1) b
|
||||
INNER JOIN (SELECT * FROM {ta} WHERE dup_rn = 1) a
|
||||
ON b.rk = a.rk
|
||||
WHERE NOT ({changed_pred})
|
||||
{lim_sql}
|
||||
"""
|
||||
)
|
||||
for row in db.execute(u_sql, params_u).mappings():
|
||||
before = dict(row["before_data"] or {})
|
||||
after = dict(row["after_data"] or {})
|
||||
unchanged_listed += 1
|
||||
if compact:
|
||||
diffs.append(
|
||||
{
|
||||
"kind": "unchanged",
|
||||
"key": _key_obj_from_data(before, key_fields),
|
||||
"before": {},
|
||||
"after": {},
|
||||
"mapped_before": {},
|
||||
"changes": {},
|
||||
"compact": True,
|
||||
"before_row_id": str(row["before_row_id"] or ""),
|
||||
"after_row_id": str(row["after_row_id"] or ""),
|
||||
}
|
||||
)
|
||||
else:
|
||||
diffs.append(
|
||||
{
|
||||
"kind": "unchanged",
|
||||
"key": _key_obj_from_data(before, key_fields),
|
||||
"before": before,
|
||||
"after": after,
|
||||
"mapped_before": before,
|
||||
"changes": {},
|
||||
"before_row_id": str(row["before_row_id"] or ""),
|
||||
"after_row_id": str(row["after_row_id"] or ""),
|
||||
}
|
||||
)
|
||||
|
||||
# Cleanup (also ON COMMIT DROP)
|
||||
db.execute(text(f"DROP TABLE IF EXISTS {tb}"))
|
||||
db.execute(text(f"DROP TABLE IF EXISTS {ta}"))
|
||||
|
||||
dup_key_list: list[str] = []
|
||||
summary = {
|
||||
"before_count": before_n,
|
||||
"after_count": after_n,
|
||||
"added": added,
|
||||
"removed": removed,
|
||||
"changed": changed,
|
||||
"unchanged": unchanged,
|
||||
"duplicate": duplicate,
|
||||
"match_key_fields": list(key_fields),
|
||||
"duplicate_keys_before": dup_b,
|
||||
"duplicate_keys_after": dup_a,
|
||||
"duplicate_key_list": dup_key_list,
|
||||
"unchanged_listed": unchanged_listed,
|
||||
"unchanged_truncated": bool(
|
||||
include_u and limit_n is not None and unchanged > unchanged_listed
|
||||
),
|
||||
"unchanged_compact": bool(compact and unchanged_listed > 0),
|
||||
"engine": "sql",
|
||||
"before_raw_count": raw_b,
|
||||
"after_raw_count": raw_a,
|
||||
"row_filters": len(row_filters),
|
||||
"unchanged_policy": policy,
|
||||
}
|
||||
mapping_stats = {
|
||||
"before_iface_count": 0,
|
||||
"after_iface_count": 0,
|
||||
"map_pairs": 0,
|
||||
"hit_before": [],
|
||||
"miss_before": [],
|
||||
"hit_after": [],
|
||||
"miss_after": [],
|
||||
"unused_before_keys": [],
|
||||
"ok": True,
|
||||
"ignore_port_changes": False,
|
||||
"engine": "sql",
|
||||
}
|
||||
return {"summary": summary, "diffs": diffs, "mapping_stats": mapping_stats}
|
||||
|
|
@ -746,6 +746,12 @@ def api_get_run(run_id: str, db: Session = Depends(get_db)) -> dict[str, Any]:
|
|||
return cmp_svc.get_run(db, run_id)
|
||||
|
||||
|
||||
@router.post("/compare/runs/{run_id}/cancel")
|
||||
def api_cancel_run(run_id: str, db: Session = Depends(get_db)) -> dict[str, Any]:
|
||||
"""Cancel a stuck/running compare so a new run can start."""
|
||||
return cmp_svc.cancel_compare_run(db, run_id)
|
||||
|
||||
|
||||
@router.delete("/compare/runs/{run_id}")
|
||||
def api_delete_run(run_id: str, db: Session = Depends(get_db)) -> dict[str, Any]:
|
||||
return cmp_svc.delete_run(db, run_id)
|
||||
|
|
|
|||
98
tests/test_biz_state_compare_recovery.py
Normal file
98
tests/test_biz_state_compare_recovery.py
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
"""Startup recovery for interrupted biz-state compare runs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from netx_api.biz_state.compare_service import (
|
||||
cancel_compare_run,
|
||||
recover_interrupted_compares_on_startup,
|
||||
)
|
||||
from netx_api.db import Base
|
||||
from netx_api.models import BizCompareJob, BizCompareRun
|
||||
|
||||
|
||||
class BizStateCompareRecoveryTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
|
||||
TestingSession = sessionmaker(
|
||||
bind=engine, autoflush=False, autocommit=False, expire_on_commit=False
|
||||
)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
self.db = TestingSession()
|
||||
self.job = BizCompareJob(
|
||||
id="j1",
|
||||
name="cutover",
|
||||
template_id="tpl1",
|
||||
status="active",
|
||||
)
|
||||
self.db.add(self.job)
|
||||
self.running = BizCompareRun(
|
||||
id="r_run",
|
||||
job_id="j1",
|
||||
status="running",
|
||||
message="loading 1/2 · BGP",
|
||||
summary_json={
|
||||
"progress": {"phase": "loading", "sheet_index": 1, "sheet_total": 2},
|
||||
"sheets": [
|
||||
{"sheet_id": "bgp", "status": "running"},
|
||||
{"sheet_id": "isis", "status": "pending"},
|
||||
],
|
||||
},
|
||||
)
|
||||
self.queued = BizCompareRun(
|
||||
id="r_q",
|
||||
job_id="j1",
|
||||
status="queued",
|
||||
message="queued",
|
||||
summary_json={"sheets": [{"sheet_id": "bgp", "status": "pending"}]},
|
||||
)
|
||||
self.ok = BizCompareRun(
|
||||
id="r_ok",
|
||||
job_id="j1",
|
||||
status="success",
|
||||
message="done",
|
||||
summary_json={"added": 0, "removed": 0, "changed": 0},
|
||||
)
|
||||
self.db.add_all([self.running, self.queued, self.ok])
|
||||
self.db.commit()
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.db.close()
|
||||
|
||||
def test_startup_cancels_running_and_queued(self) -> None:
|
||||
out = recover_interrupted_compares_on_startup(self.db)
|
||||
self.assertEqual(out["runs"], 2)
|
||||
|
||||
self.db.refresh(self.running)
|
||||
self.db.refresh(self.queued)
|
||||
self.db.refresh(self.ok)
|
||||
self.assertEqual(self.running.status, "cancelled")
|
||||
self.assertIn("interrupted_by_restart", self.running.message or "")
|
||||
self.assertEqual(
|
||||
(self.running.summary_json or {}).get("progress", {}).get("phase"),
|
||||
"cancelled",
|
||||
)
|
||||
sheets = list((self.running.summary_json or {}).get("sheets") or [])
|
||||
self.assertEqual(sheets[0].get("status"), "cancelled")
|
||||
self.assertEqual(sheets[1].get("status"), "cancelled")
|
||||
self.assertEqual(self.queued.status, "cancelled")
|
||||
self.assertEqual(self.ok.status, "success")
|
||||
|
||||
def test_cancel_compare_run_user(self) -> None:
|
||||
out = cancel_compare_run(self.db, "r_run")
|
||||
self.assertEqual(out["status"], "cancelled")
|
||||
self.db.refresh(self.running)
|
||||
self.assertEqual(self.running.status, "cancelled")
|
||||
self.assertIn("cancelled_by_user", self.running.message or "")
|
||||
|
||||
def test_cancel_idempotent_on_success(self) -> None:
|
||||
out = cancel_compare_run(self.db, "r_ok")
|
||||
self.assertEqual(out["status"], "success")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
218
tests/test_biz_state_compare_sql.py
Normal file
218
tests/test_biz_state_compare_sql.py
Normal file
|
|
@ -0,0 +1,218 @@
|
|||
"""Unit tests for SQL compare eligibility and filter compilation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from netx_api.biz_state.compare_sql import (
|
||||
can_sql_compare,
|
||||
compile_row_filters_sql,
|
||||
sql_compare_skip_reason,
|
||||
_field_rules_sql_compatible,
|
||||
_filters_sql_compatible,
|
||||
_safe_field,
|
||||
)
|
||||
|
||||
|
||||
class _FakeDialect:
|
||||
def __init__(self, name: str) -> None:
|
||||
self.name = name
|
||||
|
||||
|
||||
class _FakeBind:
|
||||
def __init__(self, name: str) -> None:
|
||||
self.dialect = _FakeDialect(name)
|
||||
|
||||
|
||||
def _db(dialect: str = "postgresql") -> MagicMock:
|
||||
db = MagicMock()
|
||||
db.get_bind.return_value = _FakeBind(dialect)
|
||||
return db
|
||||
|
||||
|
||||
class CompareSqlGateTests(unittest.TestCase):
|
||||
def test_safe_field_rejects_injection(self) -> None:
|
||||
with self.assertRaises(ValueError):
|
||||
_safe_field("a'; drop table x;--")
|
||||
with self.assertRaises(ValueError):
|
||||
_safe_field("x.y")
|
||||
self.assertEqual(_safe_field("network"), "network")
|
||||
|
||||
def test_requires_postgres(self) -> None:
|
||||
sheet = {
|
||||
"metric_id": "bgp_route",
|
||||
"key_fields": ["network", "next_hop"],
|
||||
"compare_fields": ["path"],
|
||||
"iface_fields": [],
|
||||
"field_rules": [],
|
||||
"row_filters": [],
|
||||
}
|
||||
self.assertFalse(can_sql_compare(_db("sqlite"), sheet, port_map={}))
|
||||
self.assertTrue(can_sql_compare(_db("postgresql"), sheet, port_map={}))
|
||||
|
||||
def test_rejects_port_map(self) -> None:
|
||||
sheet = {
|
||||
"metric_id": "lldp_neighbor",
|
||||
"key_fields": ["local_if", "remote_sys"],
|
||||
"compare_fields": [],
|
||||
"iface_fields": ["local_if"],
|
||||
"ignore_port_changes": False,
|
||||
"field_rules": [],
|
||||
"row_filters": [],
|
||||
}
|
||||
self.assertFalse(
|
||||
can_sql_compare(_db(), sheet, port_map={"gei-0/1": "gei-0/2"})
|
||||
)
|
||||
|
||||
def test_rejects_auto_ignore_ports_with_iface_key(self) -> None:
|
||||
sheet = {
|
||||
"metric_id": "lldp_neighbor",
|
||||
"key_fields": ["local_if", "remote_sys"],
|
||||
"compare_fields": [],
|
||||
"iface_fields": ["local_if"],
|
||||
"ignore_port_changes": None,
|
||||
"field_rules": [],
|
||||
"row_filters": [],
|
||||
}
|
||||
self.assertFalse(can_sql_compare(_db(), sheet, port_map={}))
|
||||
|
||||
def test_rejects_mac_normalize(self) -> None:
|
||||
sheet = {
|
||||
"metric_id": "arp",
|
||||
"key_fields": ["ip", "vrf"],
|
||||
"compare_fields": ["mac"],
|
||||
"iface_fields": [],
|
||||
"field_rules": [{"field": "mac", "normalize": "mac"}],
|
||||
"row_filters": [],
|
||||
}
|
||||
self.assertFalse(can_sql_compare(_db(), sheet, port_map={}))
|
||||
|
||||
def test_rejects_age_timer_filter(self) -> None:
|
||||
sheet = {
|
||||
"metric_id": "arp",
|
||||
"key_fields": ["ip"],
|
||||
"compare_fields": [],
|
||||
"iface_fields": [],
|
||||
"field_rules": [],
|
||||
"row_filters": [{"field": "age", "op": "age_timer"}],
|
||||
}
|
||||
self.assertFalse(can_sql_compare(_db(), sheet, port_map={}))
|
||||
|
||||
def test_accepts_bgp_route_style(self) -> None:
|
||||
sheet = {
|
||||
"metric_id": "bgp_route",
|
||||
"key_fields": [
|
||||
"local_as",
|
||||
"afi",
|
||||
"vrf",
|
||||
"neighbor",
|
||||
"direction",
|
||||
"rd",
|
||||
"network",
|
||||
"next_hop",
|
||||
],
|
||||
"compare_fields": ["path", "as_num"],
|
||||
"iface_fields": [],
|
||||
"field_rules": [],
|
||||
"row_filters": [{"field": "afi", "op": "eq", "value": "ipv4"}],
|
||||
}
|
||||
self.assertTrue(can_sql_compare(_db(), sheet, port_map={}))
|
||||
# BGP afi/vrf sheet splits via row_filters must not force Python
|
||||
self.assertEqual(sql_compare_skip_reason(_db(), sheet, port_map={}), "")
|
||||
|
||||
def test_rejects_iface_normalize_when_key_uses_iface(self) -> None:
|
||||
sheet = {
|
||||
"metric_id": "interface_brief",
|
||||
"key_fields": ["interface"],
|
||||
"compare_fields": ["admin"],
|
||||
"iface_fields": ["interface"],
|
||||
"ignore_port_changes": False,
|
||||
"field_rules": [],
|
||||
"row_filters": [],
|
||||
}
|
||||
self.assertFalse(
|
||||
can_sql_compare(
|
||||
_db(),
|
||||
sheet,
|
||||
port_map={},
|
||||
iface_normalize_rules=[{"from": "GE", "to": "gei"}],
|
||||
)
|
||||
)
|
||||
|
||||
def test_accepts_ignore_port_changes_false_without_normalize(self) -> None:
|
||||
sheet = {
|
||||
"metric_id": "interface_brief",
|
||||
"key_fields": ["interface"],
|
||||
"compare_fields": ["admin"],
|
||||
"iface_fields": ["interface"],
|
||||
"ignore_port_changes": False,
|
||||
"field_rules": [],
|
||||
"row_filters": [],
|
||||
}
|
||||
self.assertTrue(can_sql_compare(_db(), sheet, port_map={}, iface_normalize_rules=[]))
|
||||
|
||||
|
||||
class CompareSqlFilterCompileTests(unittest.TestCase):
|
||||
def test_empty_filters(self) -> None:
|
||||
sql, params = compile_row_filters_sql([])
|
||||
self.assertEqual(sql, "TRUE")
|
||||
self.assertEqual(params, {})
|
||||
|
||||
def test_eq_case_insensitive(self) -> None:
|
||||
sql, params = compile_row_filters_sql(
|
||||
[{"field": "afi", "op": "eq", "value": "IPv4"}]
|
||||
)
|
||||
self.assertIn("lower(", sql)
|
||||
self.assertIn("afi", sql)
|
||||
self.assertEqual(list(params.values()), ["ipv4"])
|
||||
|
||||
def test_contains(self) -> None:
|
||||
sql, params = compile_row_filters_sql(
|
||||
[{"field": "af", "op": "contains", "value": "IPv4"}]
|
||||
)
|
||||
self.assertIn("LIKE", sql)
|
||||
self.assertEqual(list(params.values()), ["%ipv4%"])
|
||||
|
||||
def test_any_all_nesting(self) -> None:
|
||||
sql, params = compile_row_filters_sql(
|
||||
[
|
||||
{
|
||||
"any": [
|
||||
{"field": "entry_type", "op": "eq", "value": "dynamic"},
|
||||
{
|
||||
"all": [
|
||||
{"field": "entry_type", "op": "empty"},
|
||||
{"field": "ip", "op": "not_empty"},
|
||||
]
|
||||
},
|
||||
]
|
||||
}
|
||||
]
|
||||
)
|
||||
self.assertIn(" OR ", sql)
|
||||
self.assertIn(" AND ", sql)
|
||||
self.assertTrue(_filters_sql_compatible([{"any": [{"field": "a", "op": "eq", "value": "1"}]}]))
|
||||
|
||||
def test_in_list(self) -> None:
|
||||
sql, params = compile_row_filters_sql(
|
||||
[{"field": "afi", "op": "in", "value": ["ipv4", "ipv6"]}]
|
||||
)
|
||||
self.assertIn(" IN (", sql)
|
||||
self.assertEqual(sorted(params.values()), ["ipv4", "ipv6"])
|
||||
|
||||
def test_field_rules_matrix(self) -> None:
|
||||
self.assertTrue(_field_rules_sql_compatible([{"field": "mac", "normalize": "lower"}]))
|
||||
self.assertFalse(_field_rules_sql_compatible([{"field": "mac", "normalize": "mac"}]))
|
||||
self.assertFalse(
|
||||
_field_rules_sql_compatible(
|
||||
[{"field": "rx", "compare": "numeric", "tolerance": 1}]
|
||||
)
|
||||
)
|
||||
self.assertTrue(
|
||||
_field_rules_sql_compatible([{"field": "x", "compare": "ignore"}])
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -501,6 +501,10 @@ const en = {
|
|||
runStatusDoneSec: "Done {{s}}s",
|
||||
runProgress: "{{phase}} · sheet {{i}}/{{n}} · {{sheet}} · {{s}}s elapsed",
|
||||
runRowsLoaded: "Loaded {{side}} {{n}} rows",
|
||||
runRowsSqlCount: "DB count {{side}} {{n}} rows",
|
||||
runEngineSql: "engine SQL",
|
||||
runEnginePython: "engine Python",
|
||||
runEngineNote: "reason {{note}}",
|
||||
sheetPending: "Pending",
|
||||
ranWithDuration: "Compare finished ({{s}}s)",
|
||||
unchangedNotStored: "Success rows were counted but not stored. Set “Store success rows” to sample and re-run for spot-check.",
|
||||
|
|
@ -520,10 +524,20 @@ const en = {
|
|||
edit: "Edit",
|
||||
delete: "Delete",
|
||||
tabConfig: "Job config",
|
||||
tabRuns: "Compare batches",
|
||||
tabResult: "Result",
|
||||
runs: "Run history",
|
||||
runsHint: "Each “Run now” creates a batch. Stuck “Running” after restart is auto-cancelled; you can also cancel manually and re-run.",
|
||||
noRuns: "No runs yet — run a compare first",
|
||||
result: "Result",
|
||||
viewResult: "View result",
|
||||
cancelRun: "Cancel",
|
||||
confirmCancelRun: "Cancel this compare batch? You can start a new run afterward.",
|
||||
runCancelled: "Compare cancelled",
|
||||
runStatusCancelled: "Cancelled",
|
||||
compareAlreadyRunning: "A compare is already running — cancel it under Compare batches or wait",
|
||||
runBatchSummary: "Fail {{fail}} · Pass {{ok}} · Added {{added}}",
|
||||
colTime: "Time",
|
||||
pickBatchRun: "Select compare run",
|
||||
pickRun: "Select run…",
|
||||
runCount: "{{n}} runs",
|
||||
|
|
|
|||
|
|
@ -500,6 +500,10 @@ const zh = {
|
|||
runStatusDoneSec: "完成 {{s}}s",
|
||||
runProgress: "{{phase}} · 表 {{i}}/{{n}} · {{sheet}} · 已用 {{s}}s",
|
||||
runRowsLoaded: "已加载 {{side}} {{n}} 行",
|
||||
runRowsSqlCount: "库内统计 {{side}} {{n}} 行",
|
||||
runEngineSql: "引擎 SQL",
|
||||
runEnginePython: "引擎 Python",
|
||||
runEngineNote: "原因 {{note}}",
|
||||
sheetPending: "等待中",
|
||||
ranWithDuration: "比对完成(耗时 {{s}} 秒)",
|
||||
unchangedNotStored: "成功行仅统计数量未落库。可在任务配置将「成功行保存」改为抽样后重新比对(抽查用)。",
|
||||
|
|
@ -519,10 +523,20 @@ const zh = {
|
|||
edit: "编辑",
|
||||
delete: "删除",
|
||||
tabConfig: "任务配置",
|
||||
tabRuns: "对比批次",
|
||||
tabResult: "比对结果",
|
||||
runs: "历史比对",
|
||||
runsHint: "每次「立即比对」生成一条批次。重启后卡住的「比对中」会自动标为已取消;也可手动终止后重跑。",
|
||||
noRuns: "尚无比对记录,请先执行比对",
|
||||
result: "比对结果",
|
||||
viewResult: "查看结果",
|
||||
cancelRun: "终止",
|
||||
confirmCancelRun: "确定终止该比对批次?终止后可重新发起比对。",
|
||||
runCancelled: "比对已取消",
|
||||
runStatusCancelled: "已取消",
|
||||
compareAlreadyRunning: "已有比对在进行中,请先在「对比批次」中终止或等待完成",
|
||||
runBatchSummary: "失败 {{fail}} · 成功 {{ok}} · 新增 {{added}}",
|
||||
colTime: "时间",
|
||||
pickBatchRun: "选择比对记录",
|
||||
pickRun: "选择比对记录…",
|
||||
runCount: "{{n}} 次",
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import { useDebouncedValue } from "../../hooks/useDebouncedValue";
|
|||
import { useToast } from "../../hooks/useToast";
|
||||
import { useI18n } from "../../i18n";
|
||||
import {
|
||||
bizCompareCancelRun,
|
||||
bizCompareCreateJob,
|
||||
bizCompareCreateMapping,
|
||||
bizCompareCreateTemplate,
|
||||
|
|
@ -35,7 +36,7 @@ import { cutoverCachedGet, cutoverCachedGetSWR, invalidateCutoverCache } from ".
|
|||
import { jobChipColor, NmStatusChip } from "./nmChips";
|
||||
|
||||
type PageTab = "templates" | "jobs";
|
||||
type JobDetailTab = "config" | "result";
|
||||
type JobDetailTab = "config" | "runs" | "result";
|
||||
type KindFilter = "diff" | "all" | "added" | "removed" | "changed" | "unchanged";
|
||||
type CreateJobStep = 0 | 1 | 2 | 3;
|
||||
const CREATE_JOB_STEPS = 4;
|
||||
|
|
@ -137,6 +138,8 @@ type RunSheet = {
|
|||
display_fields?: string[];
|
||||
field_rules?: FieldRule[];
|
||||
mode?: string;
|
||||
/** pending|running|queued|done|cancelled — set while overall run is active */
|
||||
status?: string;
|
||||
summary?: Record<string, number>;
|
||||
diffs?: DiffRow[];
|
||||
};
|
||||
|
|
@ -1060,14 +1063,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
useEffect(() => {
|
||||
const runId = String(runDetail?.id || "");
|
||||
const mid = resultSheetId || sheetIdentity(activeRunSheet) || "";
|
||||
const st = String(runDetail?.status || "");
|
||||
if (
|
||||
!runId ||
|
||||
!mid ||
|
||||
jobDetailTab !== "result" ||
|
||||
st === "running" ||
|
||||
st === "queued"
|
||||
) {
|
||||
const sheetSt = String(activeRunSheet?.status || "");
|
||||
// Block only while *this* sheet is still in flight — done sheets are readable mid-run
|
||||
const sheetStillRunning = ["pending", "running", "queued"].includes(sheetSt);
|
||||
if (!runId || !mid || jobDetailTab !== "result" || sheetStillRunning) {
|
||||
setPagedDiffs([]);
|
||||
setResultTotal(0);
|
||||
return;
|
||||
|
|
@ -1109,6 +1108,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
runDetail?.status,
|
||||
resultSheetId,
|
||||
activeRunSheet?.metric_id,
|
||||
activeRunSheet?.status,
|
||||
kindFilter,
|
||||
debouncedResultKw,
|
||||
resultPage,
|
||||
|
|
@ -1802,12 +1802,14 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
const job = jobs.find((x) => x.id === id);
|
||||
if (job) resetJobForm(job);
|
||||
try {
|
||||
const r = await bizCompareListRuns(id);
|
||||
setRuns(r.items || []);
|
||||
if ((r.items || []).length) {
|
||||
const latest = await bizCompareGetRun(String((r.items as any[])[0].id));
|
||||
const r = await bizCompareListRuns(id, 50);
|
||||
const items = r.items || [];
|
||||
setRuns(items);
|
||||
if (items.length) {
|
||||
const latest = await bizCompareGetRun(String((items as any[])[0].id));
|
||||
setRunDetail(latest);
|
||||
setJobDetailTab("result");
|
||||
// Land on batch list so stuck/cancelled runs are visible and actionable
|
||||
setJobDetailTab("runs");
|
||||
}
|
||||
} catch (e) {
|
||||
showError(formatErr(e));
|
||||
|
|
@ -1843,32 +1845,42 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
}
|
||||
};
|
||||
|
||||
const runStatus = String(runDetail?.status || "");
|
||||
const runIsActive = runStatus === "running" || runStatus === "queued";
|
||||
const jobHasActiveRun = runs.some((r) => {
|
||||
const st = String((r as any).status || "");
|
||||
return st === "running" || st === "queued";
|
||||
});
|
||||
|
||||
const runNow = async () => {
|
||||
if (!jobId) return;
|
||||
if (!enabledJobSheetCount) {
|
||||
showError(t("bizCompare.needSheets"));
|
||||
return;
|
||||
}
|
||||
if (jobHasActiveRun) {
|
||||
showError(t("bizCompare.compareAlreadyRunning"));
|
||||
setJobDetailTab("runs");
|
||||
return;
|
||||
}
|
||||
setBusy(true);
|
||||
try {
|
||||
await bizCompareUpdateJob(jobId, jobConfigBody());
|
||||
// Async enqueue — returns immediately with status=running; poll below.
|
||||
const run = await bizCompareRunJob(jobId);
|
||||
setRunDetail(run);
|
||||
setJobDetailTab("result");
|
||||
setJobDetailTab("runs");
|
||||
showOk(t("bizCompare.runStarted"));
|
||||
const r = await bizCompareListRuns(jobId);
|
||||
const r = await bizCompareListRuns(jobId, 50);
|
||||
setRuns(r.items || []);
|
||||
await refresh({ force: true });
|
||||
} catch (e) {
|
||||
showError(formatErr(e));
|
||||
setJobDetailTab("runs");
|
||||
} finally {
|
||||
setBusy(false);
|
||||
}
|
||||
};
|
||||
|
||||
const runStatus = String(runDetail?.status || "");
|
||||
const runIsActive = runStatus === "running" || runStatus === "queued";
|
||||
const runProgress = (runDetail?.summary?.progress || {}) as {
|
||||
phase?: string;
|
||||
sheet_index?: number;
|
||||
|
|
@ -1881,7 +1893,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
diff_rows?: number;
|
||||
load_side?: string;
|
||||
rows_loaded?: number;
|
||||
engine?: string;
|
||||
engine_note?: string;
|
||||
};
|
||||
const runEngine = String(runProgress.engine || "").toLowerCase();
|
||||
|
||||
// Poll active compare runs so the modal can be closed and reopened safely.
|
||||
useEffect(() => {
|
||||
|
|
@ -1906,6 +1921,9 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
} else if (!notified && st === "failed") {
|
||||
notified = true;
|
||||
showError(String(d.message || t("bizCompare.runFailed")));
|
||||
} else if (!notified && st === "cancelled") {
|
||||
notified = true;
|
||||
showError(String(d.message || t("bizCompare.runCancelled")));
|
||||
}
|
||||
} catch (e) {
|
||||
if (!cancelled) showError(formatErr(e));
|
||||
|
|
@ -1919,6 +1937,31 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
};
|
||||
}, [runDetail?.id, runIsActive, jobId, showOk, showError, t]);
|
||||
|
||||
// Keep the runs list fresh while any batch on this job is active
|
||||
useEffect(() => {
|
||||
if (!jobId || !jobHasActiveRun || jobDetailTab !== "runs") return;
|
||||
let cancelled = false;
|
||||
const tick = async () => {
|
||||
try {
|
||||
const r = await bizCompareListRuns(jobId, 50);
|
||||
if (!cancelled) setRuns(r.items || []);
|
||||
const curId = String(runDetail?.id || "");
|
||||
if (curId) {
|
||||
const d = await bizCompareGetRun(curId);
|
||||
if (!cancelled) setRunDetail(d);
|
||||
}
|
||||
} catch {
|
||||
/* ignore list poll errors */
|
||||
}
|
||||
};
|
||||
void tick();
|
||||
const id = window.setInterval(() => void tick(), 2500);
|
||||
return () => {
|
||||
cancelled = true;
|
||||
window.clearInterval(id);
|
||||
};
|
||||
}, [jobId, jobHasActiveRun, jobDetailTab, runDetail?.id]);
|
||||
|
||||
const loadRun = async (runId: string) => {
|
||||
try {
|
||||
const d = await bizCompareGetRun(runId);
|
||||
|
|
@ -1931,6 +1974,25 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
}
|
||||
};
|
||||
|
||||
const cancelRun = async (runId: string) => {
|
||||
if (!runId) return;
|
||||
if (!window.confirm(t("bizCompare.confirmCancelRun"))) return;
|
||||
setBusy(true);
|
||||
try {
|
||||
const d = await bizCompareCancelRun(runId);
|
||||
if (String(runDetail?.id || "") === runId) setRunDetail(d);
|
||||
if (jobId) {
|
||||
const r = await bizCompareListRuns(jobId, 50);
|
||||
setRuns(r.items || []);
|
||||
}
|
||||
showOk(t("bizCompare.runCancelled"));
|
||||
} catch (e) {
|
||||
showError(formatErr(e));
|
||||
} finally {
|
||||
setBusy(false);
|
||||
}
|
||||
};
|
||||
|
||||
const removeRun = async (runId: string) => {
|
||||
if (!runId) return;
|
||||
if (!window.confirm(t("bizCompare.confirmDeleteRun"))) return;
|
||||
|
|
@ -1940,19 +2002,16 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
await bizCompareDeleteRun(runId);
|
||||
let nextRuns: typeof runs = [];
|
||||
if (jobId) {
|
||||
const r = await bizCompareListRuns(jobId);
|
||||
const r = await bizCompareListRuns(jobId, 50);
|
||||
nextRuns = r.items || [];
|
||||
} else {
|
||||
nextRuns = (runs || []).filter((r) => String(r.id) !== runId);
|
||||
}
|
||||
setRuns(nextRuns);
|
||||
if (wasCurrent) {
|
||||
if (nextRuns.length) {
|
||||
await loadRun(String(nextRuns[0].id));
|
||||
} else {
|
||||
setRunDetail(null);
|
||||
setResultSheetId("");
|
||||
}
|
||||
setJobDetailTab("runs");
|
||||
}
|
||||
showOk(t("bizCompare.runDeleted"));
|
||||
} catch (e) {
|
||||
|
|
@ -2864,6 +2923,18 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
>
|
||||
{t("bizCompare.tabConfig")}
|
||||
</Button>
|
||||
<Button
|
||||
size="sm"
|
||||
variant={jobDetailTab === "runs" ? "primary" : "secondary"}
|
||||
className={jobDetailTab === "runs" ? "is-active" : undefined}
|
||||
onPress={() => {
|
||||
setJobDetailTab("runs");
|
||||
if (jobId) void bizCompareListRuns(jobId, 50).then((r) => setRuns(r.items || []));
|
||||
}}
|
||||
>
|
||||
{t("bizCompare.tabRuns")}
|
||||
{runs.length ? ` (${runs.length})` : ""}
|
||||
</Button>
|
||||
<Button
|
||||
size="sm"
|
||||
variant={jobDetailTab === "result" ? "primary" : "secondary"}
|
||||
|
|
@ -2876,6 +2947,169 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
|
||||
{jobDetailTab === "config" ? (
|
||||
renderJobForm()
|
||||
) : jobDetailTab === "runs" ? (
|
||||
<div className="bs-cmp-runs">
|
||||
<div className="btn-row" style={{ marginBottom: 8, justifyContent: "space-between" }}>
|
||||
<p className="muted bm-hint" style={{ margin: 0 }}>
|
||||
{t("bizCompare.runsHint")}
|
||||
</p>
|
||||
<Button
|
||||
size="sm"
|
||||
variant="primary"
|
||||
isDisabled={busy || jobHasActiveRun}
|
||||
onPress={() => void runNow()}
|
||||
>
|
||||
{jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
||||
</Button>
|
||||
</div>
|
||||
{!runs.length ? (
|
||||
<div className="pt-list-empty">{t("bizCompare.noRuns")}</div>
|
||||
) : (
|
||||
<div className="table-wrap">
|
||||
<table className="data-table">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>{t("bizCompare.colStatus")}</th>
|
||||
<th>{t("bizCompare.colTime")}</th>
|
||||
<th>{t("bizCompare.sidesTitle")}</th>
|
||||
<th>{t("bizCompare.colResult")}</th>
|
||||
<th>{t("bizCompare.colActions")}</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{runs.map((r) => {
|
||||
const st = String((r as any).status || "");
|
||||
const active = st === "running" || st === "queued";
|
||||
const selected = String(runDetail?.id || "") === String(r.id);
|
||||
const before = enrichSide(
|
||||
(r as any).before,
|
||||
beforeTaskId,
|
||||
tasks,
|
||||
beforeBatches,
|
||||
);
|
||||
const after = enrichSide(
|
||||
(r as any).after,
|
||||
afterTaskId || beforeTaskId,
|
||||
tasks,
|
||||
afterBatches.length ? afterBatches : beforeBatches,
|
||||
);
|
||||
const when = formatSystemTime((r as any).created_at) || "—";
|
||||
const sum = ((r as any).summary || {}) as Record<string, number>;
|
||||
const fail =
|
||||
Number(sum.removed || 0) + Number(sum.changed || 0);
|
||||
const ok = Number(sum.unchanged || 0);
|
||||
const durMs = Number(sum.duration_ms || 0);
|
||||
const prog = ((r as any).progress || {}) as {
|
||||
phase?: string;
|
||||
sheet_index?: number;
|
||||
sheet_total?: number;
|
||||
sheet_title?: string;
|
||||
rows_loaded?: number;
|
||||
load_side?: string;
|
||||
engine?: string;
|
||||
engine_note?: string;
|
||||
};
|
||||
const eng = String(prog.engine || "").toLowerCase();
|
||||
const stLabel =
|
||||
st === "running" || st === "queued"
|
||||
? t("bizCompare.runStatusRunning")
|
||||
: st === "failed"
|
||||
? t("bizCompare.runStatusFailed")
|
||||
: st === "cancelled"
|
||||
? t("bizCompare.runStatusCancelled")
|
||||
: t("bizCompare.runStatusDoneSec", {
|
||||
s: String(durMs > 0 ? Math.round(durMs / 1000) : 0),
|
||||
});
|
||||
return (
|
||||
<tr key={String(r.id)} className={selected ? "is-selected" : undefined}>
|
||||
<td>
|
||||
<NmStatusChip color={jobChipColor(st)}>{stLabel}</NmStatusChip>
|
||||
{active && (prog.phase || eng) ? (
|
||||
<div className="muted" style={{ fontSize: "0.75rem", marginTop: 4 }}>
|
||||
{eng === "sql"
|
||||
? t("bizCompare.runEngineSql")
|
||||
: eng === "python"
|
||||
? t("bizCompare.runEnginePython")
|
||||
: ""}
|
||||
{eng && prog.engine_note
|
||||
? ` · ${prog.engine_note}`
|
||||
: ""}
|
||||
{prog.phase ? `${eng ? " · " : ""}${prog.phase}` : ""}
|
||||
{prog.sheet_total
|
||||
? ` · ${prog.sheet_index || 0}/${prog.sheet_total}`
|
||||
: ""}
|
||||
{prog.sheet_title ? ` · ${prog.sheet_title}` : ""}
|
||||
{Number(prog.rows_loaded || 0) > 0
|
||||
? eng === "sql"
|
||||
? ` · ${prog.load_side || ""} ${prog.rows_loaded}`
|
||||
: ` · ${prog.load_side || ""} ${prog.rows_loaded}`
|
||||
: ""}
|
||||
</div>
|
||||
) : null}
|
||||
{(r as any).message ? (
|
||||
<div
|
||||
className="muted"
|
||||
style={{ fontSize: "0.75rem", marginTop: 2 }}
|
||||
title={String((r as any).message)}
|
||||
>
|
||||
{String((r as any).message).slice(0, 80)}
|
||||
</div>
|
||||
) : null}
|
||||
</td>
|
||||
<td>{when}</td>
|
||||
<td>
|
||||
{sideDeviceName(before)} {sideCollectTime(before)}
|
||||
{" → "}
|
||||
{sideDeviceName(after)} {sideCollectTime(after)}
|
||||
</td>
|
||||
<td>
|
||||
{active
|
||||
? "…"
|
||||
: t("bizCompare.runBatchSummary", {
|
||||
fail: String(fail),
|
||||
ok: String(ok),
|
||||
added: String(Number(sum.added || 0)),
|
||||
})}
|
||||
</td>
|
||||
<td>
|
||||
<div className="btn-row">
|
||||
<Button
|
||||
size="sm"
|
||||
variant="secondary"
|
||||
isDisabled={busy || active}
|
||||
onPress={() => void loadRun(String(r.id))}
|
||||
>
|
||||
{t("bizCompare.viewResult")}
|
||||
</Button>
|
||||
{active ? (
|
||||
<Button
|
||||
size="sm"
|
||||
variant="danger"
|
||||
isDisabled={busy}
|
||||
onPress={() => void cancelRun(String(r.id))}
|
||||
>
|
||||
{t("bizCompare.cancelRun")}
|
||||
</Button>
|
||||
) : (
|
||||
<Button
|
||||
size="sm"
|
||||
variant="danger"
|
||||
isDisabled={busy}
|
||||
onPress={() => void removeRun(String(r.id))}
|
||||
>
|
||||
{t("bizCompare.deleteRun")}
|
||||
</Button>
|
||||
)}
|
||||
</div>
|
||||
</td>
|
||||
</tr>
|
||||
);
|
||||
})}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
) : (
|
||||
<div
|
||||
ref={boardRef}
|
||||
|
|
@ -3022,6 +3256,18 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
{runDetail && runIsActive ? (
|
||||
<div className="bs-cmp-progress" role="status" aria-live="polite">
|
||||
<strong>{t("bizCompare.runStatusRunning")}</strong>
|
||||
{runEngine === "sql" || runEngine === "python" ? (
|
||||
<span className="muted">
|
||||
{runEngine === "sql"
|
||||
? t("bizCompare.runEngineSql")
|
||||
: t("bizCompare.runEnginePython")}
|
||||
{runProgress.engine_note
|
||||
? ` · ${t("bizCompare.runEngineNote", {
|
||||
note: String(runProgress.engine_note),
|
||||
})}`
|
||||
: ""}
|
||||
</span>
|
||||
) : null}
|
||||
<span className="muted">
|
||||
{t("bizCompare.runProgress", {
|
||||
phase: String(runProgress.phase || "…"),
|
||||
|
|
@ -3035,7 +3281,12 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
</span>
|
||||
{Number(runProgress.rows_loaded || 0) > 0 ? (
|
||||
<span className="muted">
|
||||
{t("bizCompare.runRowsLoaded", {
|
||||
{runEngine === "sql"
|
||||
? t("bizCompare.runRowsSqlCount", {
|
||||
side: String(runProgress.load_side || "—"),
|
||||
n: String(runProgress.rows_loaded || 0),
|
||||
})
|
||||
: t("bizCompare.runRowsLoaded", {
|
||||
side: String(runProgress.load_side || "—"),
|
||||
n: String(runProgress.rows_loaded || 0),
|
||||
})}
|
||||
|
|
@ -3495,7 +3746,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
!Number(runDetail?.summary?.unchanged_listed || 0)
|
||||
? t("bizCompare.unchangedNotStored")
|
||||
: t("bizCompare.resultEmpty")
|
||||
: runIsActive
|
||||
: activeSheetPending
|
||||
? t("bizCompare.runStatusRunning")
|
||||
: t("bizCompare.resultEmpty")}
|
||||
</div>
|
||||
|
|
@ -3536,10 +3787,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
<Button
|
||||
size="sm"
|
||||
variant="primary"
|
||||
isDisabled={busy || runIsActive}
|
||||
isDisabled={busy || jobHasActiveRun}
|
||||
onPress={() => void runNow()}
|
||||
>
|
||||
{runIsActive ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
||||
{jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
||||
</Button>
|
||||
{jobId ? (
|
||||
<Button size="sm" variant="danger" isDisabled={busy} onPress={() => void removeJob(jobId)}>
|
||||
|
|
@ -3547,12 +3798,21 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
</Button>
|
||||
) : null}
|
||||
</>
|
||||
) : jobDetailTab === "runs" ? (
|
||||
<Button
|
||||
size="sm"
|
||||
variant="primary"
|
||||
isDisabled={busy || jobHasActiveRun}
|
||||
onPress={() => void runNow()}
|
||||
>
|
||||
{jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
||||
</Button>
|
||||
) : (
|
||||
<>
|
||||
<Button
|
||||
size="sm"
|
||||
variant="secondary"
|
||||
isDisabled={busy || !runDetail?.id}
|
||||
isDisabled={busy || runIsActive || !runDetail?.id}
|
||||
onPress={() => void downloadRunTables()}
|
||||
>
|
||||
{t("bizCompare.exportTables")}
|
||||
|
|
@ -3568,10 +3828,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
|||
<Button
|
||||
size="sm"
|
||||
variant="primary"
|
||||
isDisabled={busy || runIsActive}
|
||||
isDisabled={busy || jobHasActiveRun}
|
||||
onPress={() => void runNow()}
|
||||
>
|
||||
{runIsActive ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
||||
{jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
||||
</Button>
|
||||
</>
|
||||
)}
|
||||
|
|
|
|||
|
|
@ -2154,6 +2154,12 @@ export const bizCompareListRuns = (jobId: string, limit = 20) =>
|
|||
export const bizCompareGetRun = (runId: string) =>
|
||||
apiGet<Record<string, unknown>>(`/v1/biz-state/compare/runs/${encodeURIComponent(runId)}`);
|
||||
|
||||
export const bizCompareCancelRun = (runId: string) =>
|
||||
apiPost<Record<string, unknown>>(
|
||||
`/v1/biz-state/compare/runs/${encodeURIComponent(runId)}/cancel`,
|
||||
{},
|
||||
);
|
||||
|
||||
export const bizCompareDeleteRun = (runId: string) =>
|
||||
apiDelete<{ ok: boolean; job_id?: string; run_id?: string }>(
|
||||
`/v1/biz-state/compare/runs/${encodeURIComponent(runId)}`,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue