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:
|
except Exception:
|
||||||
_log.exception("startup: biz_state collect recovery failed")
|
_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:
|
try:
|
||||||
from .port_traffic_migrate import backfill_port_traffic_series
|
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]:
|
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."""
|
"""Placeholder meta so the UI lists all check items while a run is in flight."""
|
||||||
key_fields = list(sheet.get("key_fields") or [])
|
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 ""
|
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(
|
def _run_sheet(
|
||||||
db: Session,
|
db: Session,
|
||||||
*,
|
*,
|
||||||
|
|
@ -1681,7 +1762,7 @@ def _run_sheet(
|
||||||
port_map: dict[str, str],
|
port_map: dict[str, str],
|
||||||
iface_normalize_rules: list[dict[str, str]] | None = None,
|
iface_normalize_rules: list[dict[str, str]] | None = None,
|
||||||
store_unchanged: str = "auto",
|
store_unchanged: str = "auto",
|
||||||
on_load_progress: Callable[[str, int], None] | None = None,
|
on_load_progress: Callable[..., None] | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
key_fields = list(sheet.get("key_fields") or [])
|
key_fields = list(sheet.get("key_fields") or [])
|
||||||
iface_fields = list(sheet.get("iface_fields") or [])
|
iface_fields = list(sheet.get("iface_fields") or [])
|
||||||
|
|
@ -1704,13 +1785,77 @@ def _run_sheet(
|
||||||
ignore_ports = bool(ignore_ports)
|
ignore_ports = bool(ignore_ports)
|
||||||
mid = sheet["metric_id"]
|
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:
|
def _before_chunk(n: int) -> None:
|
||||||
if on_load_progress:
|
_emit_load("before", n, engine="python", note=skip_reason or "python", phase="loading")
|
||||||
on_load_progress("before", n)
|
|
||||||
|
|
||||||
def _after_chunk(n: int) -> None:
|
def _after_chunk(n: int) -> None:
|
||||||
if on_load_progress:
|
_emit_load("after", n, engine="python", note=skip_reason or "python", phase="loading")
|
||||||
on_load_progress("after", n)
|
|
||||||
|
|
||||||
before_raw = _load_metric_rows(
|
before_raw = _load_metric_rows(
|
||||||
db, batch_id=before_batch_id, metric_id=mid, on_chunk=_before_chunk
|
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["after_raw_count"] = len(after_raw)
|
||||||
summary["row_filters"] = len(row_filters)
|
summary["row_filters"] = len(row_filters)
|
||||||
summary["unchanged_policy"] = policy
|
summary["unchanged_policy"] = policy
|
||||||
return {
|
summary.setdefault("engine", "python")
|
||||||
"sheet_id": sheet_key(sheet),
|
return _sheet_result_envelope(
|
||||||
"title": sheet_title(sheet),
|
sheet,
|
||||||
"metric_id": sheet["metric_id"],
|
key_fields=key_fields,
|
||||||
"key_fields": key_fields,
|
iface_fields=iface_fields,
|
||||||
"iface_fields": iface_fields,
|
compare_fields=compare_fields,
|
||||||
"compare_fields": compare_fields,
|
display_fields=display_fields,
|
||||||
"display_fields": display_fields,
|
row_filters=row_filters,
|
||||||
"row_filters": row_filters,
|
field_rules=field_rules,
|
||||||
"field_rules": field_rules,
|
ignore_ports=ignore_ports,
|
||||||
"ignore_port_changes": ignore_ports,
|
mode=mode,
|
||||||
"mode": mode,
|
summary=summary,
|
||||||
"status": "done",
|
diffs=list(result.get("diffs") or []),
|
||||||
"summary": summary,
|
mapping_stats=dict(result.get("mapping_stats") or {}),
|
||||||
"diffs": result["diffs"],
|
)
|
||||||
"mapping_stats": result["mapping_stats"],
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def _validate_compare_job(
|
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))
|
sheets_cfg = _filter_enabled_sheets(sheets_cfg, getattr(j, "enabled_sheet_ids", None))
|
||||||
if not sheets_cfg:
|
if not sheets_cfg:
|
||||||
raise HTTPException(status_code=400, detail="no_enabled_sheets")
|
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 {
|
return {
|
||||||
"job": j,
|
"job": j,
|
||||||
"template": tpl,
|
"template": tpl,
|
||||||
|
|
@ -1868,6 +2017,10 @@ def _set_run_progress(
|
||||||
progress.update(extra)
|
progress.update(extra)
|
||||||
prev["progress"] = progress
|
prev["progress"] = progress
|
||||||
run.summary_json = prev
|
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"
|
run.status = "running"
|
||||||
title = progress["sheet_title"] or progress["sheet_id"] or ""
|
title = progress["sheet_title"] or progress["sheet_id"] or ""
|
||||||
run.message = (
|
run.message = (
|
||||||
|
|
@ -1906,6 +2059,12 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
||||||
run.message = "no_enabled_sheets"
|
run.message = "no_enabled_sheets"
|
||||||
db.commit()
|
db.commit()
|
||||||
raise HTTPException(status_code=400, detail="no_enabled_sheets")
|
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()
|
started_mono = time.monotonic()
|
||||||
store_mode = normalize_store_unchanged(getattr(j, "store_unchanged", None))
|
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] = {}
|
field_counts: dict[str, int] = {}
|
||||||
total = len(sheets_cfg)
|
total = len(sheets_cfg)
|
||||||
|
|
||||||
# Prefer seeded pending sheets from create; rebuild if missing
|
# Prefer seeded pending sheets from create; realign to small-first order
|
||||||
sheet_metas: list[dict[str, Any]] = list((run.summary_json or {}).get("sheets") or [])
|
prev_metas = list((run.summary_json or {}).get("sheets") or [])
|
||||||
if len(sheet_metas) != total:
|
by_key = {sheet_key(m): m for m in prev_metas}
|
||||||
sheet_metas = [_pending_sheet_meta(s) for s in sheets_cfg]
|
sheet_metas = [
|
||||||
|
by_key.get(sheet_key(s)) or _pending_sheet_meta(s) for s in sheets_cfg
|
||||||
|
]
|
||||||
|
|
||||||
def _publish_sheets() -> None:
|
def _publish_sheets() -> None:
|
||||||
prev = dict(run.summary_json or {})
|
prev = dict(run.summary_json or {})
|
||||||
prev["sheets"] = list(sheet_metas)
|
prev["sheets"] = list(sheet_metas)
|
||||||
prev.update({k: agg[k] for k in agg})
|
prev.update({k: agg[k] for k in agg})
|
||||||
run.summary_json = prev
|
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()
|
db.commit()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
for idx, sheet in enumerate(sheets_cfg, start=1):
|
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)
|
sid = sheet_key(sheet)
|
||||||
# Mark current sheet running in the sidebar list
|
# Mark current sheet running in the sidebar list
|
||||||
for meta in sheet_metas:
|
for meta in sheet_metas:
|
||||||
|
|
@ -1949,27 +2118,43 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
||||||
break
|
break
|
||||||
_publish_sheets()
|
_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()
|
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:
|
if n - _load_pub["n"] < 25_000 and now - _load_pub["t"] < 2.0:
|
||||||
return
|
return
|
||||||
_load_pub["t"] = now
|
_load_pub["t"] = now
|
||||||
_load_pub["n"] = n
|
_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(
|
_set_run_progress(
|
||||||
db,
|
db,
|
||||||
run,
|
run,
|
||||||
phase="loading",
|
phase=str(phase or ("loading" if eng != "sql" else "sql_count")),
|
||||||
sheet_index=_idx,
|
sheet_index=_idx,
|
||||||
sheet_total=total,
|
sheet_total=total,
|
||||||
sheet=_sheet,
|
sheet=_sheet,
|
||||||
started_mono=started_mono,
|
started_mono=started_mono,
|
||||||
extra={
|
extra=extra,
|
||||||
"load_side": side,
|
|
||||||
"rows_loaded": n,
|
|
||||||
},
|
|
||||||
)
|
)
|
||||||
|
|
||||||
_set_run_progress(
|
_set_run_progress(
|
||||||
|
|
@ -1980,6 +2165,7 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
||||||
sheet_total=total,
|
sheet_total=total,
|
||||||
sheet=sheet,
|
sheet=sheet,
|
||||||
started_mono=started_mono,
|
started_mono=started_mono,
|
||||||
|
extra={"engine": "", "engine_note": ""},
|
||||||
)
|
)
|
||||||
one = _run_sheet(
|
one = _run_sheet(
|
||||||
db,
|
db,
|
||||||
|
|
@ -2091,6 +2277,13 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
||||||
fail_diffs.clear()
|
fail_diffs.clear()
|
||||||
ok_diffs.clear()
|
ok_diffs.clear()
|
||||||
db.commit()
|
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)
|
duration_ms = int((time.monotonic() - started_mono) * 1000)
|
||||||
top_fields = sorted(
|
top_fields = sorted(
|
||||||
|
|
@ -2117,6 +2310,9 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
||||||
},
|
},
|
||||||
"sheets": sheet_metas,
|
"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.status = "success"
|
||||||
run.summary_json = summary_payload
|
run.summary_json = summary_payload
|
||||||
run.mapping_stats_json = mapping_by_metric
|
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()
|
db.commit()
|
||||||
return get_run(db, run.id)
|
return get_run(db, run.id)
|
||||||
except HTTPException as exc:
|
except HTTPException as exc:
|
||||||
|
run = db.get(BizCompareRun, run_id) or run
|
||||||
|
if str(run.status or "") != "cancelled":
|
||||||
run.status = "failed"
|
run.status = "failed"
|
||||||
run.message = str(getattr(exc, "detail", "") or exc)[:1024]
|
run.message = str(getattr(exc, "detail", "") or exc)[:1024]
|
||||||
prev = dict(run.summary_json or {})
|
prev = dict(run.summary_json or {})
|
||||||
|
|
@ -2142,6 +2340,8 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
||||||
raise
|
raise
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
_log.exception("compare run failed run=%s job=%s", run_id, j.id)
|
_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.status = "failed"
|
||||||
run.message = str(exc)[:1024]
|
run.message = str(exc)[:1024]
|
||||||
prev = dict(run.summary_json or {})
|
prev = dict(run.summary_json or {})
|
||||||
|
|
@ -2155,8 +2355,93 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]:
|
||||||
raise
|
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]:
|
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."""
|
"""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)
|
lock = _job_compare_lock(job_id)
|
||||||
if not lock.acquire(blocking=False):
|
if not lock.acquire(blocking=False):
|
||||||
raise HTTPException(status_code=409, detail="compare_already_running")
|
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
|
Returns immediately so the HTTP worker / UI stay responsive. Poll
|
||||||
``GET /compare/runs/{id}`` for progress (``summary.progress``).
|
``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)
|
lock = _job_compare_lock(job_id)
|
||||||
if not lock.acquire(blocking=False):
|
if not lock.acquire(blocking=False):
|
||||||
raise HTTPException(status_code=409, detail="compare_already_running")
|
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)
|
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}")
|
@router.delete("/compare/runs/{run_id}")
|
||||||
def api_delete_run(run_id: str, db: Session = Depends(get_db)) -> dict[str, Any]:
|
def api_delete_run(run_id: str, db: Session = Depends(get_db)) -> dict[str, Any]:
|
||||||
return cmp_svc.delete_run(db, run_id)
|
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",
|
runStatusDoneSec: "Done {{s}}s",
|
||||||
runProgress: "{{phase}} · sheet {{i}}/{{n}} · {{sheet}} · {{s}}s elapsed",
|
runProgress: "{{phase}} · sheet {{i}}/{{n}} · {{sheet}} · {{s}}s elapsed",
|
||||||
runRowsLoaded: "Loaded {{side}} {{n}} rows",
|
runRowsLoaded: "Loaded {{side}} {{n}} rows",
|
||||||
|
runRowsSqlCount: "DB count {{side}} {{n}} rows",
|
||||||
|
runEngineSql: "engine SQL",
|
||||||
|
runEnginePython: "engine Python",
|
||||||
|
runEngineNote: "reason {{note}}",
|
||||||
sheetPending: "Pending",
|
sheetPending: "Pending",
|
||||||
ranWithDuration: "Compare finished ({{s}}s)",
|
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.",
|
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",
|
edit: "Edit",
|
||||||
delete: "Delete",
|
delete: "Delete",
|
||||||
tabConfig: "Job config",
|
tabConfig: "Job config",
|
||||||
|
tabRuns: "Compare batches",
|
||||||
tabResult: "Result",
|
tabResult: "Result",
|
||||||
runs: "Run history",
|
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",
|
noRuns: "No runs yet — run a compare first",
|
||||||
result: "Result",
|
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",
|
pickBatchRun: "Select compare run",
|
||||||
pickRun: "Select run…",
|
pickRun: "Select run…",
|
||||||
runCount: "{{n}} runs",
|
runCount: "{{n}} runs",
|
||||||
|
|
|
||||||
|
|
@ -500,6 +500,10 @@ const zh = {
|
||||||
runStatusDoneSec: "完成 {{s}}s",
|
runStatusDoneSec: "完成 {{s}}s",
|
||||||
runProgress: "{{phase}} · 表 {{i}}/{{n}} · {{sheet}} · 已用 {{s}}s",
|
runProgress: "{{phase}} · 表 {{i}}/{{n}} · {{sheet}} · 已用 {{s}}s",
|
||||||
runRowsLoaded: "已加载 {{side}} {{n}} 行",
|
runRowsLoaded: "已加载 {{side}} {{n}} 行",
|
||||||
|
runRowsSqlCount: "库内统计 {{side}} {{n}} 行",
|
||||||
|
runEngineSql: "引擎 SQL",
|
||||||
|
runEnginePython: "引擎 Python",
|
||||||
|
runEngineNote: "原因 {{note}}",
|
||||||
sheetPending: "等待中",
|
sheetPending: "等待中",
|
||||||
ranWithDuration: "比对完成(耗时 {{s}} 秒)",
|
ranWithDuration: "比对完成(耗时 {{s}} 秒)",
|
||||||
unchangedNotStored: "成功行仅统计数量未落库。可在任务配置将「成功行保存」改为抽样后重新比对(抽查用)。",
|
unchangedNotStored: "成功行仅统计数量未落库。可在任务配置将「成功行保存」改为抽样后重新比对(抽查用)。",
|
||||||
|
|
@ -519,10 +523,20 @@ const zh = {
|
||||||
edit: "编辑",
|
edit: "编辑",
|
||||||
delete: "删除",
|
delete: "删除",
|
||||||
tabConfig: "任务配置",
|
tabConfig: "任务配置",
|
||||||
|
tabRuns: "对比批次",
|
||||||
tabResult: "比对结果",
|
tabResult: "比对结果",
|
||||||
runs: "历史比对",
|
runs: "历史比对",
|
||||||
|
runsHint: "每次「立即比对」生成一条批次。重启后卡住的「比对中」会自动标为已取消;也可手动终止后重跑。",
|
||||||
noRuns: "尚无比对记录,请先执行比对",
|
noRuns: "尚无比对记录,请先执行比对",
|
||||||
result: "比对结果",
|
result: "比对结果",
|
||||||
|
viewResult: "查看结果",
|
||||||
|
cancelRun: "终止",
|
||||||
|
confirmCancelRun: "确定终止该比对批次?终止后可重新发起比对。",
|
||||||
|
runCancelled: "比对已取消",
|
||||||
|
runStatusCancelled: "已取消",
|
||||||
|
compareAlreadyRunning: "已有比对在进行中,请先在「对比批次」中终止或等待完成",
|
||||||
|
runBatchSummary: "失败 {{fail}} · 成功 {{ok}} · 新增 {{added}}",
|
||||||
|
colTime: "时间",
|
||||||
pickBatchRun: "选择比对记录",
|
pickBatchRun: "选择比对记录",
|
||||||
pickRun: "选择比对记录…",
|
pickRun: "选择比对记录…",
|
||||||
runCount: "{{n}} 次",
|
runCount: "{{n}} 次",
|
||||||
|
|
|
||||||
|
|
@ -7,6 +7,7 @@ import { useDebouncedValue } from "../../hooks/useDebouncedValue";
|
||||||
import { useToast } from "../../hooks/useToast";
|
import { useToast } from "../../hooks/useToast";
|
||||||
import { useI18n } from "../../i18n";
|
import { useI18n } from "../../i18n";
|
||||||
import {
|
import {
|
||||||
|
bizCompareCancelRun,
|
||||||
bizCompareCreateJob,
|
bizCompareCreateJob,
|
||||||
bizCompareCreateMapping,
|
bizCompareCreateMapping,
|
||||||
bizCompareCreateTemplate,
|
bizCompareCreateTemplate,
|
||||||
|
|
@ -35,7 +36,7 @@ import { cutoverCachedGet, cutoverCachedGetSWR, invalidateCutoverCache } from ".
|
||||||
import { jobChipColor, NmStatusChip } from "./nmChips";
|
import { jobChipColor, NmStatusChip } from "./nmChips";
|
||||||
|
|
||||||
type PageTab = "templates" | "jobs";
|
type PageTab = "templates" | "jobs";
|
||||||
type JobDetailTab = "config" | "result";
|
type JobDetailTab = "config" | "runs" | "result";
|
||||||
type KindFilter = "diff" | "all" | "added" | "removed" | "changed" | "unchanged";
|
type KindFilter = "diff" | "all" | "added" | "removed" | "changed" | "unchanged";
|
||||||
type CreateJobStep = 0 | 1 | 2 | 3;
|
type CreateJobStep = 0 | 1 | 2 | 3;
|
||||||
const CREATE_JOB_STEPS = 4;
|
const CREATE_JOB_STEPS = 4;
|
||||||
|
|
@ -137,6 +138,8 @@ type RunSheet = {
|
||||||
display_fields?: string[];
|
display_fields?: string[];
|
||||||
field_rules?: FieldRule[];
|
field_rules?: FieldRule[];
|
||||||
mode?: string;
|
mode?: string;
|
||||||
|
/** pending|running|queued|done|cancelled — set while overall run is active */
|
||||||
|
status?: string;
|
||||||
summary?: Record<string, number>;
|
summary?: Record<string, number>;
|
||||||
diffs?: DiffRow[];
|
diffs?: DiffRow[];
|
||||||
};
|
};
|
||||||
|
|
@ -1060,14 +1063,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
const runId = String(runDetail?.id || "");
|
const runId = String(runDetail?.id || "");
|
||||||
const mid = resultSheetId || sheetIdentity(activeRunSheet) || "";
|
const mid = resultSheetId || sheetIdentity(activeRunSheet) || "";
|
||||||
const st = String(runDetail?.status || "");
|
const sheetSt = String(activeRunSheet?.status || "");
|
||||||
if (
|
// Block only while *this* sheet is still in flight — done sheets are readable mid-run
|
||||||
!runId ||
|
const sheetStillRunning = ["pending", "running", "queued"].includes(sheetSt);
|
||||||
!mid ||
|
if (!runId || !mid || jobDetailTab !== "result" || sheetStillRunning) {
|
||||||
jobDetailTab !== "result" ||
|
|
||||||
st === "running" ||
|
|
||||||
st === "queued"
|
|
||||||
) {
|
|
||||||
setPagedDiffs([]);
|
setPagedDiffs([]);
|
||||||
setResultTotal(0);
|
setResultTotal(0);
|
||||||
return;
|
return;
|
||||||
|
|
@ -1109,6 +1108,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
runDetail?.status,
|
runDetail?.status,
|
||||||
resultSheetId,
|
resultSheetId,
|
||||||
activeRunSheet?.metric_id,
|
activeRunSheet?.metric_id,
|
||||||
|
activeRunSheet?.status,
|
||||||
kindFilter,
|
kindFilter,
|
||||||
debouncedResultKw,
|
debouncedResultKw,
|
||||||
resultPage,
|
resultPage,
|
||||||
|
|
@ -1802,12 +1802,14 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
const job = jobs.find((x) => x.id === id);
|
const job = jobs.find((x) => x.id === id);
|
||||||
if (job) resetJobForm(job);
|
if (job) resetJobForm(job);
|
||||||
try {
|
try {
|
||||||
const r = await bizCompareListRuns(id);
|
const r = await bizCompareListRuns(id, 50);
|
||||||
setRuns(r.items || []);
|
const items = r.items || [];
|
||||||
if ((r.items || []).length) {
|
setRuns(items);
|
||||||
const latest = await bizCompareGetRun(String((r.items as any[])[0].id));
|
if (items.length) {
|
||||||
|
const latest = await bizCompareGetRun(String((items as any[])[0].id));
|
||||||
setRunDetail(latest);
|
setRunDetail(latest);
|
||||||
setJobDetailTab("result");
|
// Land on batch list so stuck/cancelled runs are visible and actionable
|
||||||
|
setJobDetailTab("runs");
|
||||||
}
|
}
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
showError(formatErr(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 () => {
|
const runNow = async () => {
|
||||||
if (!jobId) return;
|
if (!jobId) return;
|
||||||
if (!enabledJobSheetCount) {
|
if (!enabledJobSheetCount) {
|
||||||
showError(t("bizCompare.needSheets"));
|
showError(t("bizCompare.needSheets"));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
if (jobHasActiveRun) {
|
||||||
|
showError(t("bizCompare.compareAlreadyRunning"));
|
||||||
|
setJobDetailTab("runs");
|
||||||
|
return;
|
||||||
|
}
|
||||||
setBusy(true);
|
setBusy(true);
|
||||||
try {
|
try {
|
||||||
await bizCompareUpdateJob(jobId, jobConfigBody());
|
await bizCompareUpdateJob(jobId, jobConfigBody());
|
||||||
// Async enqueue — returns immediately with status=running; poll below.
|
// Async enqueue — returns immediately with status=running; poll below.
|
||||||
const run = await bizCompareRunJob(jobId);
|
const run = await bizCompareRunJob(jobId);
|
||||||
setRunDetail(run);
|
setRunDetail(run);
|
||||||
setJobDetailTab("result");
|
setJobDetailTab("runs");
|
||||||
showOk(t("bizCompare.runStarted"));
|
showOk(t("bizCompare.runStarted"));
|
||||||
const r = await bizCompareListRuns(jobId);
|
const r = await bizCompareListRuns(jobId, 50);
|
||||||
setRuns(r.items || []);
|
setRuns(r.items || []);
|
||||||
await refresh({ force: true });
|
await refresh({ force: true });
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
showError(formatErr(e));
|
showError(formatErr(e));
|
||||||
|
setJobDetailTab("runs");
|
||||||
} finally {
|
} finally {
|
||||||
setBusy(false);
|
setBusy(false);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
const runStatus = String(runDetail?.status || "");
|
|
||||||
const runIsActive = runStatus === "running" || runStatus === "queued";
|
|
||||||
const runProgress = (runDetail?.summary?.progress || {}) as {
|
const runProgress = (runDetail?.summary?.progress || {}) as {
|
||||||
phase?: string;
|
phase?: string;
|
||||||
sheet_index?: number;
|
sheet_index?: number;
|
||||||
|
|
@ -1881,7 +1893,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
diff_rows?: number;
|
diff_rows?: number;
|
||||||
load_side?: string;
|
load_side?: string;
|
||||||
rows_loaded?: number;
|
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.
|
// Poll active compare runs so the modal can be closed and reopened safely.
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
|
|
@ -1906,6 +1921,9 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
} else if (!notified && st === "failed") {
|
} else if (!notified && st === "failed") {
|
||||||
notified = true;
|
notified = true;
|
||||||
showError(String(d.message || t("bizCompare.runFailed")));
|
showError(String(d.message || t("bizCompare.runFailed")));
|
||||||
|
} else if (!notified && st === "cancelled") {
|
||||||
|
notified = true;
|
||||||
|
showError(String(d.message || t("bizCompare.runCancelled")));
|
||||||
}
|
}
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
if (!cancelled) showError(formatErr(e));
|
if (!cancelled) showError(formatErr(e));
|
||||||
|
|
@ -1919,6 +1937,31 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
};
|
};
|
||||||
}, [runDetail?.id, runIsActive, jobId, showOk, showError, t]);
|
}, [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) => {
|
const loadRun = async (runId: string) => {
|
||||||
try {
|
try {
|
||||||
const d = await bizCompareGetRun(runId);
|
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) => {
|
const removeRun = async (runId: string) => {
|
||||||
if (!runId) return;
|
if (!runId) return;
|
||||||
if (!window.confirm(t("bizCompare.confirmDeleteRun"))) return;
|
if (!window.confirm(t("bizCompare.confirmDeleteRun"))) return;
|
||||||
|
|
@ -1940,19 +2002,16 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
await bizCompareDeleteRun(runId);
|
await bizCompareDeleteRun(runId);
|
||||||
let nextRuns: typeof runs = [];
|
let nextRuns: typeof runs = [];
|
||||||
if (jobId) {
|
if (jobId) {
|
||||||
const r = await bizCompareListRuns(jobId);
|
const r = await bizCompareListRuns(jobId, 50);
|
||||||
nextRuns = r.items || [];
|
nextRuns = r.items || [];
|
||||||
} else {
|
} else {
|
||||||
nextRuns = (runs || []).filter((r) => String(r.id) !== runId);
|
nextRuns = (runs || []).filter((r) => String(r.id) !== runId);
|
||||||
}
|
}
|
||||||
setRuns(nextRuns);
|
setRuns(nextRuns);
|
||||||
if (wasCurrent) {
|
if (wasCurrent) {
|
||||||
if (nextRuns.length) {
|
|
||||||
await loadRun(String(nextRuns[0].id));
|
|
||||||
} else {
|
|
||||||
setRunDetail(null);
|
setRunDetail(null);
|
||||||
setResultSheetId("");
|
setResultSheetId("");
|
||||||
}
|
setJobDetailTab("runs");
|
||||||
}
|
}
|
||||||
showOk(t("bizCompare.runDeleted"));
|
showOk(t("bizCompare.runDeleted"));
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
|
|
@ -2864,6 +2923,18 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
>
|
>
|
||||||
{t("bizCompare.tabConfig")}
|
{t("bizCompare.tabConfig")}
|
||||||
</Button>
|
</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
|
<Button
|
||||||
size="sm"
|
size="sm"
|
||||||
variant={jobDetailTab === "result" ? "primary" : "secondary"}
|
variant={jobDetailTab === "result" ? "primary" : "secondary"}
|
||||||
|
|
@ -2876,6 +2947,169 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
|
|
||||||
{jobDetailTab === "config" ? (
|
{jobDetailTab === "config" ? (
|
||||||
renderJobForm()
|
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
|
<div
|
||||||
ref={boardRef}
|
ref={boardRef}
|
||||||
|
|
@ -3022,6 +3256,18 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
{runDetail && runIsActive ? (
|
{runDetail && runIsActive ? (
|
||||||
<div className="bs-cmp-progress" role="status" aria-live="polite">
|
<div className="bs-cmp-progress" role="status" aria-live="polite">
|
||||||
<strong>{t("bizCompare.runStatusRunning")}</strong>
|
<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">
|
<span className="muted">
|
||||||
{t("bizCompare.runProgress", {
|
{t("bizCompare.runProgress", {
|
||||||
phase: String(runProgress.phase || "…"),
|
phase: String(runProgress.phase || "…"),
|
||||||
|
|
@ -3035,7 +3281,12 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
</span>
|
</span>
|
||||||
{Number(runProgress.rows_loaded || 0) > 0 ? (
|
{Number(runProgress.rows_loaded || 0) > 0 ? (
|
||||||
<span className="muted">
|
<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 || "—"),
|
side: String(runProgress.load_side || "—"),
|
||||||
n: String(runProgress.rows_loaded || 0),
|
n: String(runProgress.rows_loaded || 0),
|
||||||
})}
|
})}
|
||||||
|
|
@ -3495,7 +3746,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
!Number(runDetail?.summary?.unchanged_listed || 0)
|
!Number(runDetail?.summary?.unchanged_listed || 0)
|
||||||
? t("bizCompare.unchangedNotStored")
|
? t("bizCompare.unchangedNotStored")
|
||||||
: t("bizCompare.resultEmpty")
|
: t("bizCompare.resultEmpty")
|
||||||
: runIsActive
|
: activeSheetPending
|
||||||
? t("bizCompare.runStatusRunning")
|
? t("bizCompare.runStatusRunning")
|
||||||
: t("bizCompare.resultEmpty")}
|
: t("bizCompare.resultEmpty")}
|
||||||
</div>
|
</div>
|
||||||
|
|
@ -3536,10 +3787,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
<Button
|
<Button
|
||||||
size="sm"
|
size="sm"
|
||||||
variant="primary"
|
variant="primary"
|
||||||
isDisabled={busy || runIsActive}
|
isDisabled={busy || jobHasActiveRun}
|
||||||
onPress={() => void runNow()}
|
onPress={() => void runNow()}
|
||||||
>
|
>
|
||||||
{runIsActive ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
{jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
||||||
</Button>
|
</Button>
|
||||||
{jobId ? (
|
{jobId ? (
|
||||||
<Button size="sm" variant="danger" isDisabled={busy} onPress={() => void removeJob(jobId)}>
|
<Button size="sm" variant="danger" isDisabled={busy} onPress={() => void removeJob(jobId)}>
|
||||||
|
|
@ -3547,12 +3798,21 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
</Button>
|
</Button>
|
||||||
) : null}
|
) : null}
|
||||||
</>
|
</>
|
||||||
|
) : jobDetailTab === "runs" ? (
|
||||||
|
<Button
|
||||||
|
size="sm"
|
||||||
|
variant="primary"
|
||||||
|
isDisabled={busy || jobHasActiveRun}
|
||||||
|
onPress={() => void runNow()}
|
||||||
|
>
|
||||||
|
{jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
||||||
|
</Button>
|
||||||
) : (
|
) : (
|
||||||
<>
|
<>
|
||||||
<Button
|
<Button
|
||||||
size="sm"
|
size="sm"
|
||||||
variant="secondary"
|
variant="secondary"
|
||||||
isDisabled={busy || !runDetail?.id}
|
isDisabled={busy || runIsActive || !runDetail?.id}
|
||||||
onPress={() => void downloadRunTables()}
|
onPress={() => void downloadRunTables()}
|
||||||
>
|
>
|
||||||
{t("bizCompare.exportTables")}
|
{t("bizCompare.exportTables")}
|
||||||
|
|
@ -3568,10 +3828,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage
|
||||||
<Button
|
<Button
|
||||||
size="sm"
|
size="sm"
|
||||||
variant="primary"
|
variant="primary"
|
||||||
isDisabled={busy || runIsActive}
|
isDisabled={busy || jobHasActiveRun}
|
||||||
onPress={() => void runNow()}
|
onPress={() => void runNow()}
|
||||||
>
|
>
|
||||||
{runIsActive ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
{jobHasActiveRun ? t("bizCompare.runStatusRunning") : t("bizCompare.runNow")}
|
||||||
</Button>
|
</Button>
|
||||||
</>
|
</>
|
||||||
)}
|
)}
|
||||||
|
|
|
||||||
|
|
@ -2154,6 +2154,12 @@ export const bizCompareListRuns = (jobId: string, limit = 20) =>
|
||||||
export const bizCompareGetRun = (runId: string) =>
|
export const bizCompareGetRun = (runId: string) =>
|
||||||
apiGet<Record<string, unknown>>(`/v1/biz-state/compare/runs/${encodeURIComponent(runId)}`);
|
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) =>
|
export const bizCompareDeleteRun = (runId: string) =>
|
||||||
apiDelete<{ ok: boolean; job_id?: string; run_id?: string }>(
|
apiDelete<{ ok: boolean; job_id?: string; run_id?: string }>(
|
||||||
`/v1/biz-state/compare/runs/${encodeURIComponent(runId)}`,
|
`/v1/biz-state/compare/runs/${encodeURIComponent(runId)}`,
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue