diff --git a/netx_api/app_startup.py b/netx_api/app_startup.py index 8acc5f5..beb5497 100644 --- a/netx_api/app_startup.py +++ b/netx_api/app_startup.py @@ -159,17 +159,6 @@ 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 57bcbe5..6d82adb 100644 --- a/netx_api/biz_state/compare_service.py +++ b/netx_api/biz_state/compare_service.py @@ -453,53 +453,6 @@ 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 []) @@ -1719,40 +1672,6 @@ 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, *, @@ -1762,7 +1681,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[..., None] | None = None, + on_load_progress: Callable[[str, int], None] | None = None, ) -> dict[str, Any]: key_fields = list(sheet.get("key_fields") or []) iface_fields = list(sheet.get("iface_fields") or []) @@ -1785,77 +1704,13 @@ 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: - _emit_load("before", n, engine="python", note=skip_reason or "python", phase="loading") + if on_load_progress: + on_load_progress("before", n) def _after_chunk(n: int) -> None: - _emit_load("after", n, engine="python", note=skip_reason or "python", phase="loading") + if on_load_progress: + on_load_progress("after", n) before_raw = _load_metric_rows( db, batch_id=before_batch_id, metric_id=mid, on_chunk=_before_chunk @@ -1887,21 +1742,23 @@ def _run_sheet( summary["after_raw_count"] = len(after_raw) summary["row_filters"] = len(row_filters) summary["unchanged_policy"] = policy - 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 {}), - ) + 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"], + } def _validate_compare_job( @@ -1927,12 +1784,6 @@ 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, @@ -2017,17 +1868,13 @@ def _set_run_progress( progress.update(extra) prev["progress"] = progress run.summary_json = prev - # Re-read status from DB — cancel may have been committed by another session - # (UI cancel / startup recovery) while this worker still holds a stale "running". - db.expire(run, ["status", "message"]) - if str(run.status or "") != "cancelled": - run.status = "running" - title = progress["sheet_title"] or progress["sheet_id"] or "" - run.message = ( - f"{phase} {sheet_index}/{sheet_total}" - + (f" · {title}" if title else "") - + f" · {elapsed_ms // 1000}s" - )[:1024] + 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() @@ -2059,12 +1906,6 @@ 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)) @@ -2086,30 +1927,20 @@ 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; 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 - ] + # 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] 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: @@ -2118,43 +1949,27 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: break _publish_sheets() - _load_pub = {"t": 0.0, "n": -1, "engine": ""} + _load_pub = {"t": 0.0, "n": -1} - def _on_load( - side: str, - n: int, - *, - engine: str = "python", - note: str = "", - phase: str | None = None, - _idx: int = idx, - _sheet: dict = sheet, - ) -> None: + def _on_load(side: str, n: int, _idx: int = idx, _sheet: dict = sheet) -> None: now = time.monotonic() - 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 + # Avoid committing every 5k on million-row BGP loads + 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=str(phase or ("loading" if eng != "sql" else "sql_count")), + phase="loading", sheet_index=_idx, sheet_total=total, sheet=_sheet, started_mono=started_mono, - extra=extra, + extra={ + "load_side": side, + "rows_loaded": n, + }, ) _set_run_progress( @@ -2165,7 +1980,6 @@ 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, @@ -2277,13 +2091,6 @@ 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( @@ -2310,9 +2117,6 @@ 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 @@ -2325,123 +2129,34 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: db.commit() return get_run(db, run.id) except HTTPException as exc: - run = db.get(BizCompareRun, run_id) or run - if str(run.status or "") != "cancelled": - run.status = "failed" - run.message = str(getattr(exc, "detail", "") or exc)[:1024] - prev = dict(run.summary_json or {}) - 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.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 = 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 {}) + run.status = "failed" + run.message = str(exc)[:1024] + prev = dict(run.summary_json or {}) prog = dict(prev.get("progress") or {}) - prog["phase"] = "cancelled" - prog["elapsed_ms"] = int(prog.get("elapsed_ms") or 0) + prog["phase"] = "failed" + prog["elapsed_ms"] = int((time.monotonic() - started_mono) * 1000) 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: + prev["duration_ms"] = prog["elapsed_ms"] + run.summary_json = prev 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) + raise 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") @@ -2468,8 +2183,6 @@ 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 deleted file mode 100644 index 7df648b..0000000 --- a/netx_api/biz_state/compare_sql.py +++ /dev/null @@ -1,646 +0,0 @@ -"""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 48af725..7920526 100644 --- a/netx_api/biz_state_router.py +++ b/netx_api/biz_state_router.py @@ -746,12 +746,6 @@ 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 deleted file mode 100644 index e24ea40..0000000 --- a/tests/test_biz_state_compare_recovery.py +++ /dev/null @@ -1,98 +0,0 @@ -"""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 deleted file mode 100644 index e61f83c..0000000 --- a/tests/test_biz_state_compare_sql.py +++ /dev/null @@ -1,218 +0,0 @@ -"""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 1ee7ef0..12f7171 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -501,10 +501,6 @@ 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.", @@ -524,20 +520,10 @@ 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 f6f364a..4b14657 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -500,10 +500,6 @@ 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: "成功行仅统计数量未落库。可在任务配置将「成功行保存」改为抽样后重新比对（抽查用）。", @@ -523,20 +519,10 @@ 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 fdd13d3..21fc4eb 100644 --- a/web/src/pages/network/BizComparePage.tsx +++ b/web/src/pages/network/BizComparePage.tsx @@ -7,7 +7,6 @@ import { useDebouncedValue } from "../../hooks/useDebouncedValue"; import { useToast } from "../../hooks/useToast"; import { useI18n } from "../../i18n"; import { - bizCompareCancelRun, bizCompareCreateJob, bizCompareCreateMapping, bizCompareCreateTemplate, @@ -36,7 +35,7 @@ import { cutoverCachedGet, cutoverCachedGetSWR, invalidateCutoverCache } from ". import { jobChipColor, NmStatusChip } from "./nmChips"; type PageTab = "templates" | "jobs"; -type JobDetailTab = "config" | "runs" | "result"; +type JobDetailTab = "config" | "result"; type KindFilter = "diff" | "all" | "added" | "removed" | "changed" | "unchanged"; type CreateJobStep = 0 | 1 | 2 | 3; const CREATE_JOB_STEPS = 4; @@ -138,8 +137,6 @@ 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[]; }; @@ -1063,10 +1060,14 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage useEffect(() => { const runId = String(runDetail?.id || ""); const mid = resultSheetId || sheetIdentity(activeRunSheet) || ""; - 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) { + const st = String(runDetail?.status || ""); + if ( + !runId || + !mid || + jobDetailTab !== "result" || + st === "running" || + st === "queued" + ) { setPagedDiffs([]); setResultTotal(0); return; @@ -1108,7 +1109,6 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage runDetail?.status, resultSheetId, activeRunSheet?.metric_id, - activeRunSheet?.status, kindFilter, debouncedResultKw, resultPage, @@ -1802,14 +1802,12 @@ 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, 50); - const items = r.items || []; - setRuns(items); - if (items.length) { - const latest = await bizCompareGetRun(String((items as any[])[0].id)); + const r = await bizCompareListRuns(id); + setRuns(r.items || []); + if ((r.items || []).length) { + const latest = await bizCompareGetRun(String((r.items as any[])[0].id)); setRunDetail(latest); - // Land on batch list so stuck/cancelled runs are visible and actionable - setJobDetailTab("runs"); + setJobDetailTab("result"); } } catch (e) { showError(formatErr(e)); @@ -1845,42 +1843,32 @@ 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("runs"); + setJobDetailTab("result"); showOk(t("bizCompare.runStarted")); - const r = await bizCompareListRuns(jobId, 50); + const r = await bizCompareListRuns(jobId); 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; @@ -1893,10 +1881,7 @@ 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(() => { @@ -1921,9 +1906,6 @@ 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)); @@ -1937,31 +1919,6 @@ 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); @@ -1974,25 +1931,6 @@ 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; @@ -2002,16 +1940,19 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage await bizCompareDeleteRun(runId); let nextRuns: typeof runs = []; if (jobId) { - const r = await bizCompareListRuns(jobId, 50); + const r = await bizCompareListRuns(jobId); nextRuns = r.items || []; } else { nextRuns = (runs || []).filter((r) => String(r.id) !== runId); } setRuns(nextRuns); if (wasCurrent) { - setRunDetail(null); - setResultSheetId(""); - setJobDetailTab("runs"); + if (nextRuns.length) { + await loadRun(String(nextRuns[0].id)); + } else { + setRunDetail(null); + setResultSheetId(""); + } } showOk(t("bizCompare.runDeleted")); } catch (e) { @@ -2923,18 +2864,6 @@ 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 || "…"), @@ -3281,15 +3035,10 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage {Number(runProgress.rows_loaded || 0) > 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), - })} + {t("bizCompare.runRowsLoaded", { + side: String(runProgress.load_side || "—"), + n: String(runProgress.rows_loaded || 0), + })} ) : null} {runDetail.message ? ( @@ -3746,7 +3495,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage !Number(runDetail?.summary?.unchanged_listed || 0) ? t("bizCompare.unchangedNotStored") : t("bizCompare.resultEmpty") - : activeSheetPending + : runIsActive ? t("bizCompare.runStatusRunning") : t("bizCompare.resultEmpty")}
@@ -3787,10 +3536,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 d3b22f8..be84ad7 100644 --- a/web/src/services/api.ts +++ b/web/src/services/api.ts @@ -2154,12 +2154,6 @@ 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)}`,