diff --git a/netx_api/app_startup.py b/netx_api/app_startup.py index beb5497..8acc5f5 100644 --- a/netx_api/app_startup.py +++ b/netx_api/app_startup.py @@ -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 diff --git a/netx_api/biz_state/compare_service.py b/netx_api/biz_state/compare_service.py index 6d82adb..57bcbe5 100644 --- a/netx_api/biz_state/compare_service.py +++ b/netx_api/biz_state/compare_service.py @@ -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,13 +2017,17 @@ def _set_run_progress( progress.update(extra) prev["progress"] = progress run.summary_json = prev - run.status = "running" - title = progress["sheet_title"] or progress["sheet_id"] or "" - run.message = ( - f"{phase} {sheet_index}/{sheet_total}" - + (f" · {title}" if title else "") - + f" · {elapsed_ms // 1000}s" - )[:1024] + # 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 = ( + f"{phase} {sheet_index}/{sheet_total}" + + (f" · {title}" if title else "") + + f" · {elapsed_ms // 1000}s" + )[:1024] db.commit() @@ -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 - if n - _load_pub["n"] < 25_000 and now - _load_pub["t"] < 2.0: - return + 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,34 +2325,123 @@ 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.status = "failed" - run.message = str(getattr(exc, "detail", "") or exc)[:1024] - prev = dict(run.summary_json or {}) - prog = dict(prev.get("progress") or {}) - prog["phase"] = "failed" - prog["elapsed_ms"] = int((time.monotonic() - started_mono) * 1000) - prev["progress"] = prog - prev["duration_ms"] = prog["elapsed_ms"] - run.summary_json = prev - db.commit() + 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 {}) + prog = dict(prev.get("progress") or {}) + prog["phase"] = "failed" + prog["elapsed_ms"] = int((time.monotonic() - started_mono) * 1000) + prev["progress"] = prog + prev["duration_ms"] = prog["elapsed_ms"] + run.summary_json = prev + db.commit() raise except Exception as exc: _log.exception("compare run failed run=%s job=%s", run_id, j.id) - run.status = "failed" - run.message = str(exc)[:1024] - prev = dict(run.summary_json or {}) - prog = dict(prev.get("progress") or {}) - prog["phase"] = "failed" - prog["elapsed_ms"] = int((time.monotonic() - started_mono) * 1000) - prev["progress"] = prog - prev["duration_ms"] = prog["elapsed_ms"] - run.summary_json = prev - db.commit() + 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 {}) + prog = dict(prev.get("progress") or {}) + prog["phase"] = "failed" + prog["elapsed_ms"] = int((time.monotonic() - started_mono) * 1000) + prev["progress"] = prog + prev["duration_ms"] = prog["elapsed_ms"] + run.summary_json = prev + db.commit() 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") diff --git a/netx_api/biz_state/compare_sql.py b/netx_api/biz_state/compare_sql.py new file mode 100644 index 0000000..7df648b --- /dev/null +++ b/netx_api/biz_state/compare_sql.py @@ -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} diff --git a/netx_api/biz_state_router.py b/netx_api/biz_state_router.py index 7920526..48af725 100644 --- a/netx_api/biz_state_router.py +++ b/netx_api/biz_state_router.py @@ -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) diff --git a/tests/test_biz_state_compare_recovery.py b/tests/test_biz_state_compare_recovery.py new file mode 100644 index 0000000..e24ea40 --- /dev/null +++ b/tests/test_biz_state_compare_recovery.py @@ -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() diff --git a/tests/test_biz_state_compare_sql.py b/tests/test_biz_state_compare_sql.py new file mode 100644 index 0000000..e61f83c --- /dev/null +++ b/tests/test_biz_state_compare_sql.py @@ -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() diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 12f7171..1ee7ef0 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -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", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index 4b14657..f6f364a 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -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}} 次", diff --git a/web/src/pages/network/BizComparePage.tsx b/web/src/pages/network/BizComparePage.tsx index 21fc4eb..fdd13d3 100644 --- a/web/src/pages/network/BizComparePage.tsx +++ b/web/src/pages/network/BizComparePage.tsx @@ -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; 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(""); - } + 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")} + + + {!runs.length ? ( +
{t("bizCompare.noRuns")}
+ ) : ( +
+ + + + + + + + + + + + {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; + 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 ( + + + + + + + + ); + })} + +
{t("bizCompare.colStatus")}{t("bizCompare.colTime")}{t("bizCompare.sidesTitle")}{t("bizCompare.colResult")}{t("bizCompare.colActions")}
+ {stLabel} + {active && (prog.phase || eng) ? ( +
+ {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}` + : ""} +
+ ) : null} + {(r as any).message ? ( +
+ {String((r as any).message).slice(0, 80)} +
+ ) : null} + 		{when} + {sideDeviceName(before)} {sideCollectTime(before)} + {" → "} + {sideDeviceName(after)} {sideCollectTime(after)} + + {active + ? "…" + : t("bizCompare.runBatchSummary", { + fail: String(fail), + ok: String(ok), + added: String(Number(sum.added || 0)), + })} + +
+ + {active ? ( + + ) : ( + + )} +
+
+
+ )} + ) : (
{t("bizCompare.runStatusRunning")} + {runEngine === "sql" || runEngine === "python" ? ( + + {runEngine === "sql" + ? t("bizCompare.runEngineSql") + : t("bizCompare.runEnginePython")} + {runProgress.engine_note + ? ` · ${t("bizCompare.runEngineNote", { + note: String(runProgress.engine_note), + })}` + : ""} + + ) : null} {t("bizCompare.runProgress", { phase: String(runProgress.phase || "…"), @@ -3035,10 +3281,15 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage {Number(runProgress.rows_loaded || 0) > 0 ? ( - {t("bizCompare.runRowsLoaded", { - side: String(runProgress.load_side || "—"), - n: String(runProgress.rows_loaded || 0), - })} + {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), + })} ) : null} {runDetail.message ? ( @@ -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")}
@@ -3536,10 +3787,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage {jobId ? ( ) : null} + ) : jobDetailTab === "runs" ? ( + ) : ( <> )} diff --git a/web/src/services/api.ts b/web/src/services/api.ts index be84ad7..d3b22f8 100644 --- a/web/src/services/api.ts +++ b/web/src/services/api.ts @@ -2154,6 +2154,12 @@ export const bizCompareListRuns = (jobId: string, limit = 20) => export const bizCompareGetRun = (runId: string) => apiGet>(`/v1/biz-state/compare/runs/${encodeURIComponent(runId)}`); +export const bizCompareCancelRun = (runId: string) => + apiPost>( + `/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)}`,