diff --git a/netx_api/biz_state/compare_engine.py b/netx_api/biz_state/compare_engine.py index 3871cc4..b5e1251 100644 --- a/netx_api/biz_state/compare_engine.py +++ b/netx_api/biz_state/compare_engine.py @@ -2,7 +2,6 @@ from __future__ import annotations -from collections import defaultdict, deque from typing import Any, Mapping, Sequence from .compare_rules import field_rule_map, values_equal, explain_diff @@ -12,53 +11,6 @@ from .iface_normalize import ( resolve_mapped_iface, ) -# Prefer these fields when stratifying success samples (BGP multipath / multi-cmd). -_STRATUM_FIELDS = ("neighbor", "direction", "afi", "vrf") - - -def stratum_key(row: Mapping[str, Any] | None) -> str: - """Bucket key for stratified unchanged sampling.""" - if not isinstance(row, Mapping): - return "_" - parts: list[str] = [] - for f in _STRATUM_FIELDS: - v = str(row.get(f) or "").strip() - if v: - parts.append(f"{f}={v}") - return "|".join(parts) if parts else "_" - - -def stratify_take(items: list[Any], limit: int, *, key_fn) -> list[Any]: - """Round-robin across strata so one neighbor/direction cannot consume the whole sample.""" - lim = max(0, int(limit)) - if lim <= 0 or not items: - return [] - if len(items) <= lim: - return list(items) - buckets: dict[str, deque[Any]] = defaultdict(deque) - order: list[str] = [] - for it in items: - sk = str(key_fn(it) or "_") - if sk not in buckets: - order.append(sk) - buckets[sk].append(it) - out: list[Any] = [] - while len(out) < lim and buckets: - drained: list[str] = [] - for sk in order: - q = buckets.get(sk) - if not q: - drained.append(sk) - continue - out.append(q.popleft()) - if len(out) >= lim: - break - for sk in drained: - buckets.pop(sk, None) - if sk in order: - order = [x for x in order if x != sk] - return out - def apply_port_map( row: dict[str, Any], @@ -159,14 +111,9 @@ def compare_rows( ) -> dict[str, Any]: """Return summary + diffs list. - Diff kinds: added | removed | changed | unchanged + Diff kinds: added | removed | changed | unchanged | duplicate - Pipeline: iface normalize (both sides) → port map (before) → ordered - same-key pairing (load / list order within each match key). - - Same match key with N before and M after rows: zip by index - ``0..min(N,M)-1`` for field compare; extras become ``removed`` (before) - or ``added`` (after). Example: 5 vs 2 → 2 compared + 3 removed. + Pipeline: iface normalize (both sides) → port map (before) → match. ``include_unchanged``: when False, matching rows still increment ``summary.unchanged`` but are omitted from ``diffs``. @@ -182,8 +129,8 @@ def compare_rows( - ``True``: force drop iface from match key when a non-empty candidate exists. - ``False``: never drop iface from match key. - ``summary.duplicate`` is always 0. ``duplicate_keys_*`` count match keys - that appear more than once on a side (diagnostic only). + Duplicate match keys are not silently discarded: extras become ``duplicate`` + diffs and ``summary.duplicate_key_list`` lists the colliding keys. ``field_rules`` drives normalize / numeric tolerance / per-field compare mode (template-driven; no metric-specific branches here). @@ -230,22 +177,26 @@ def compare_rows( apply_port_map(r, iface_fields=iface_list, port_map=pmap) for r in before_norm ] - before_groups: dict[tuple[str, ...], list[tuple[dict[str, Any], dict[str, Any]]]] = ( - defaultdict(list) - ) - after_groups: dict[tuple[str, ...], list[dict[str, Any]]] = defaultdict(list) - for orig, mapped in zip(before_rows, before_mapped): - before_groups[row_key(mapped, match_keys)].append((orig, mapped)) + after_index: dict[tuple[str, ...], dict[str, Any]] = {} + after_dup = 0 + after_dup_keys: list[tuple[str, ...]] = [] + after_dup_rows: list[tuple[tuple[str, ...], dict[str, Any]]] = [] for r in after_norm: - after_groups[row_key(r, match_keys)].append(r) + k = row_key(r, match_keys) + if k in after_index: + after_dup += 1 + after_dup_keys.append(k) + after_dup_rows.append((k, r)) + continue # first wins — do not overwrite + after_index[k] = r + before_keys: set[tuple[str, ...]] = set() + before_dup = 0 + before_dup_keys: list[tuple[str, ...]] = [] diffs: list[dict[str, Any]] = [] - added = removed = changed = unchanged = 0 + added = removed = changed = unchanged = duplicate = 0 unchanged_listed = 0 limit_n = None if unchanged_limit is None else max(0, int(unchanged_limit)) - multi_before_keys: list[tuple[str, ...]] = [] - multi_after_keys: list[tuple[str, ...]] = [] - unchanged_candidates: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]] = [] def _key_obj(row: dict[str, Any]) -> dict[str, Any]: return {f: row.get(f, "") for f in key_fields} @@ -258,10 +209,12 @@ def compare_rows( return str(netx.get("row_id") or "") return "" - def _append_unchanged_diff( - orig: dict[str, Any], mapped: dict[str, Any], after_row: dict[str, Any] - ) -> None: + def _emit_unchanged(orig: dict[str, Any], mapped: dict[str, Any], after_row: dict[str, Any]) -> None: nonlocal unchanged_listed + if not include_unchanged: + return + if limit_n is not None and unchanged_listed >= limit_n: + return unchanged_listed += 1 before_rid = _row_id(orig) after_rid = _row_id(after_row) @@ -293,103 +246,96 @@ def compare_rows( } ) - # Stable key order: before encounter order, then after-only keys - seen_keys: set[tuple[str, ...]] = set() - ordered_keys: list[tuple[str, ...]] = [] for orig, mapped in zip(before_rows, before_mapped): k = row_key(mapped, match_keys) - if k not in seen_keys: - seen_keys.add(k) - ordered_keys.append(k) - for r in after_norm: - k = row_key(r, match_keys) - if k not in seen_keys: - seen_keys.add(k) - ordered_keys.append(k) - for k in ordered_keys: - b_list = before_groups.get(k) or [] - a_list = after_groups.get(k) or [] - if len(b_list) > 1: - multi_before_keys.append(k) - if len(a_list) > 1: - multi_after_keys.append(k) - n = max(len(b_list), len(a_list)) - for i in range(n): - if i >= len(b_list): - after = a_list[i] - added += 1 - diffs.append( - { - "kind": "added", - "key": _key_obj(after), - "before": None, - "after": after, - "mapped_before": None, - "changes": {}, - "before_row_id": "", - "after_row_id": _row_id(after), - } - ) - continue - if i >= len(a_list): - orig, mapped = b_list[i] - removed += 1 - diffs.append( - { - "kind": "removed", - "key": _key_obj(mapped), - "before": orig, - "after": None, - "mapped_before": mapped, - "changes": {}, - "before_row_id": _row_id(orig), - "after_row_id": "", - } - ) - continue - orig, mapped = b_list[i] - after = a_list[i] - field_changes: dict[str, dict[str, Any]] = {} - for f in compare_fields: - bv = mapped.get(f, "") - av = after.get(f, "") - rule = rules.get(f) - if not values_equal(bv, av, rule=rule): - entry: dict[str, Any] = {"before": bv, "after": av} - reason = explain_diff(bv, av, rule=rule) - if reason: - entry["reason"] = reason - field_changes[f] = entry - if field_changes: - changed += 1 - diffs.append( - { - "kind": "changed", - "key": _key_obj(mapped), - "before": orig, - "after": after, - "mapped_before": mapped, - "changes": field_changes, - "before_row_id": _row_id(orig), - "after_row_id": _row_id(after), - } - ) - else: - unchanged += 1 - if include_unchanged: - unchanged_candidates.append((orig, mapped, after)) - - if include_unchanged and unchanged_candidates: - if limit_n is None: - picked = unchanged_candidates - else: - picked = stratify_take( - unchanged_candidates, - limit_n, - key_fn=lambda t: stratum_key(t[1]), + if k in before_keys: + before_dup += 1 + before_dup_keys.append(k) + duplicate += 1 + diffs.append( + { + "kind": "duplicate", + "side": "before", + "key": _key_obj(mapped), + "before": orig, + "after": after_index.get(k), + "mapped_before": mapped, + "changes": {}, + } ) - for orig, mapped, after_row in picked: - _append_unchanged_diff(orig, mapped, after_row) + continue # only first before row participates in match + before_keys.add(k) + after = after_index.get(k) + if after is None: + removed += 1 + diffs.append( + { + "kind": "removed", + "key": _key_obj(mapped), + "before": orig, + "after": None, + "mapped_before": mapped, + "changes": {}, + } + ) + continue + # Empty compare_fields = presence-only: keyed rows that exist on both + # sides are unchanged (no value checks). + field_changes: dict[str, dict[str, Any]] = {} + for f in compare_fields: + bv = mapped.get(f, "") + av = after.get(f, "") + rule = rules.get(f) + if not values_equal(bv, av, rule=rule): + entry: dict[str, Any] = {"before": bv, "after": av} + reason = explain_diff(bv, av, rule=rule) + if reason: + entry["reason"] = reason + field_changes[f] = entry + if field_changes: + changed += 1 + diffs.append( + { + "kind": "changed", + "key": _key_obj(mapped), + "before": orig, + "after": after, + "mapped_before": mapped, + "changes": field_changes, + } + ) + else: + unchanged += 1 + _emit_unchanged(orig, mapped, after) + + for k, after in after_index.items(): + if k in before_keys: + continue + added += 1 + diffs.append( + { + "kind": "added", + "key": _key_obj(after), + "before": None, + "after": after, + "mapped_before": None, + "changes": {}, + } + ) + + for k, after in after_dup_rows: + duplicate += 1 + diffs.append( + { + "kind": "duplicate", + "side": "after", + "key": _key_obj(after), + "before": None, + "after": after, + "mapped_before": None, + "changes": {}, + } + ) def _fmt_keys(keys: list[tuple[str, ...]]) -> list[str]: seen: set[str] = set() @@ -399,7 +345,7 @@ def compare_rows( if s not in seen: seen.add(s) out.append(s) - return out[:64] + return out stats = mapping_stats( before_rows=before_norm, @@ -417,21 +363,16 @@ def compare_rows( "removed": removed, "changed": changed, "unchanged": unchanged, - "duplicate": 0, + "duplicate": duplicate, "match_key_fields": match_keys, - "duplicate_keys_before": len(multi_before_keys), - "duplicate_keys_after": len(multi_after_keys), - "duplicate_key_list": _fmt_keys(multi_before_keys + multi_after_keys), + "duplicate_keys_before": before_dup, + "duplicate_keys_after": after_dup, + "duplicate_key_list": _fmt_keys(before_dup_keys + after_dup_keys), "unchanged_listed": unchanged_listed, "unchanged_truncated": bool( include_unchanged and limit_n is not None and unchanged > unchanged_listed ), "unchanged_compact": bool(compact_unchanged and unchanged_listed > 0), - "unchanged_sample_mode": ( - "stratified" - if include_unchanged and limit_n is not None and unchanged > unchanged_listed - else ("full" if include_unchanged else "none") - ), }, "diffs": diffs, "mapping_stats": stats, diff --git a/netx_api/biz_state/compare_service.py b/netx_api/biz_state/compare_service.py index b1fa66c..d6d1553 100644 --- a/netx_api/biz_state/compare_service.py +++ b/netx_api/biz_state/compare_service.py @@ -186,13 +186,6 @@ def _compare_side( _DIFF_CHUNK = 2000 _LOAD_YIELD_PER = 5000 _SEARCH_TEXT_MAX = 4000 -# Heartbeat while bulk-inserting large fail/ok diff sets (vpnv4-scale). -_PERSIST_PROGRESS_EVERY = 10_000 -# Above this, store fail diffs as key + row_id + changes (hydrate sides on read). -_FAIL_COMPACT_MIN = 50_000 -# Live search (kw): load at most this many matching rows per side, return ≤ this many pairs. -_LIVE_SEARCH_LOAD_CAP = 2_000 -_LIVE_SEARCH_RESULT_CAP = 200 # Success-row persist policy (see resolve_unchanged_policy) _STORE_UNCHANGED_MODES = frozenset({"auto", "always", "never", "sample", "keys"}) _UNCHANGED_FULL_MAX = 20_000 @@ -280,41 +273,15 @@ def _persist_sheet_diffs( metric_id: str, diffs: list[dict[str, Any]], seq_start: int = 0, - on_progress: Callable[[int, int], None] | None = None, ) -> int: """Bulk-insert diff rows already selected by the engine policy. Compact success rows carry key + before/after_row_id; JSON sides stay empty and are hydrated from metric tables on read. - - ``on_progress(written, total)`` fires periodically so UI elapsed time moves - during multi-minute inserts (e.g. large vpnv4 fail sets). """ buf: list[dict[str, Any]] = [] seq = int(seq_start or 0) written = 0 - total = len(diffs) - last_prog = 0 - last_prog_t = time.monotonic() - - def _maybe_prog(force: bool = False) -> None: - nonlocal last_prog, last_prog_t - if not on_progress: - return - now = time.monotonic() - if ( - not force - and written - last_prog < _PERSIST_PROGRESS_EVERY - and now - last_prog_t < 2.0 - ): - return - last_prog = written - last_prog_t = now - try: - on_progress(written, total) - except Exception: - _log.exception("persist progress callback failed run=%s metric=%s", run_id, metric_id) - for d in diffs: kind = str(d.get("kind") or "") before = _strip_netx(d.get("before")) @@ -328,14 +295,10 @@ def _persist_sheet_diffs( "mapped_before": mapped, "changes": dict(d.get("changes") or {}), } - # Compact rows: search_text = kind + key (+ changes) only — no fat sides + # Compact success: search_text = kind + key only (no fat sides) search_src = ( - { - "kind": kind, - "key": payload["key"], - **({"changes": payload["changes"]} if payload["changes"] else {}), - } - if bool(d.get("compact")) + {"kind": kind, "key": payload["key"]} + if kind == "unchanged" and bool(d.get("compact")) else payload ) buf.append( @@ -360,14 +323,8 @@ def _persist_sheet_diffs( if len(buf) >= _DIFF_CHUNK: db.bulk_insert_mappings(BizCompareDiff, buf) buf.clear() - # Commit chunks so progress/UI can see mid-write fail rows and - # elapsed_ms advances (otherwise persisting_* looks frozen). - db.commit() - _maybe_prog() if buf: db.bulk_insert_mappings(BizCompareDiff, buf) - db.commit() - _maybe_prog(force=True) return written @@ -557,7 +514,6 @@ def _pending_sheet_meta(sheet: dict[str, Any]) -> dict[str, Any]: "compare_fields": compare_fields, "display_fields": list(sheet.get("display_fields") or []), "field_rules": list(sheet.get("field_rules") or []), - "row_filters": list(sheet.get("row_filters") or []), "ignore_port_changes": sheet.get("ignore_port_changes"), "mode": "presence" if not compare_fields else "fields", "status": "pending", @@ -2467,39 +2423,6 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: fail_diffs = [d for d in diffs if str(d.get("kind") or "") != "unchanged"] ok_diffs = [d for d in diffs if str(d.get("kind") or "") == "unchanged"] mid = sheet_key(one) - # Million-row vpnv4 with many diffs: keep key/row_id/changes only. - if len(fail_diffs) >= _FAIL_COMPACT_MIN: - for d in fail_diffs: - d["before"] = {} - d["after"] = {} - d["mapped_before"] = {} - d["compact"] = True - - def _on_persist( - written: int, - total_n: int, - *, - phase: str, - kind_key: str, - ) -> None: - _set_run_progress( - db, - run, - phase=phase, - sheet_index=idx, - sheet_total=total, - sheet=sheet, - started_mono=started_mono, - extra={ - kind_key: total_n, - "persisted": written, - "persist_total": total_n, - }, - # Separate session so chunk commits in persist do not race - # with progress JSON writes on the worker session. - detach=True, - ) - _set_run_progress( db, run, @@ -2508,22 +2431,12 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: sheet_total=total, sheet=sheet, started_mono=started_mono, - extra={ - "fail_rows": len(fail_diffs), - "persisted": 0, - "persist_total": len(fail_diffs), - }, + extra={"fail_rows": len(fail_diffs)}, ) n_fail = _persist_sheet_diffs( - db, - run_id=run.id, - metric_id=mid, - diffs=fail_diffs, - seq_start=0, - on_progress=lambda w, n: _on_persist( - w, n, phase="persisting_fail", kind_key="fail_rows" - ), + db, run_id=run.id, metric_id=mid, diffs=fail_diffs, seq_start=0 ) + db.commit() if ok_diffs: _set_run_progress( db, @@ -2533,11 +2446,7 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: sheet_total=total, sheet=sheet, started_mono=started_mono, - extra={ - "ok_rows": len(ok_diffs), - "persisted": 0, - "persist_total": len(ok_diffs), - }, + extra={"ok_rows": len(ok_diffs)}, ) _persist_sheet_diffs( db, @@ -2545,9 +2454,6 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: metric_id=mid, diffs=ok_diffs, seq_start=n_fail, - on_progress=lambda w, n: _on_persist( - w, n, phase="persisting_ok", kind_key="ok_rows" - ), ) for k in agg: agg[k] += int(s.get(k) or 0) @@ -2561,7 +2467,6 @@ def _execute_compare_into_run(db: Session, run_id: str) -> dict[str, Any]: "compare_fields": one["compare_fields"], "display_fields": one.get("display_fields") or [], "field_rules": one.get("field_rules") or [], - "row_filters": one.get("row_filters") or [], "ignore_port_changes": one.get("ignore_port_changes"), "mode": one["mode"], "status": "done", @@ -3043,270 +2948,6 @@ def _lookup_sheet(sheets: list[dict[str, Any]], key: str) -> dict[str, Any] | No return None -def _kind_allows(kind_n: str, diff_kind: str) -> bool: - dk = str(diff_kind or "") - if kind_n == "all": - return True - if kind_n == "diff": - return dk in ("removed", "changed") - return dk == kind_n - - -def _kw_match_sql(key_fields: list[str], *, param: str = "kw") -> str: - """OR of ILIKE on key fields (and data_json::text fallback).""" - from .compare_sql import _FIELD_RE, _safe_field - - parts: list[str] = [] - for f in key_fields: - name = str(f or "").strip() - if not name or not _FIELD_RE.match(name): - continue - sf = _safe_field(name) - parts.append(f"lower(trim(both from coalesce(data_json->>'{sf}', ''))) LIKE :{param}") - # Broad fallback so free-text still hits non-key columns (path, next_hop, …) - parts.append(f"lower(data_json::text) LIKE :{param}") - return "(" + " OR ".join(parts) + ")" if parts else f"(lower(data_json::text) LIKE :{param})" - - -def _load_metric_rows_for_search( - db: Session, - *, - batch_id: str, - metric_id: str, - row_filters: list[dict[str, Any]] | None, - key_fields: list[str], - kw: str, - cap: int = _LIVE_SEARCH_LOAD_CAP, -) -> tuple[list[dict[str, Any]], bool]: - """Load rows matching sheet filters + kw. Returns (rows, truncated).""" - from .compare_sql import ( - _dialect_is_postgres, - _filters_sql_compatible, - compile_row_filters_sql, - ) - from ..models import BizStateMetricRow - from sqlalchemy import text as sql_text - - bid = str(batch_id or "").strip() - mid = str(metric_id or "").strip() - needle = str(kw or "").strip() - if not bid or not mid or not needle: - return [], False - lim = max(1, min(int(cap), _LIVE_SEARCH_LOAD_CAP)) - filters = [f for f in (row_filters or []) if isinstance(f, dict)] - like = f"%{needle.lower()}%" - - if _dialect_is_postgres(db): - filter_sql, filter_params = ("TRUE", {}) - if filters and _filters_sql_compatible(filters): - filter_sql, filter_params = compile_row_filters_sql(filters) - kw_sql = _kw_match_sql(key_fields) - # Fetch lim+1 to detect truncation - params = { - "bid": bid, - "mid": mid, - "kw": like, - "lim": lim + 1, - **filter_params, - } - rows = db.execute( - sql_text( - f""" - SELECT id, batch_command_id, task_id, ne_id, seq, data_json, collected_at - FROM biz_state_metric_row - WHERE batch_id = :bid - AND metric_id = :mid - AND ({filter_sql}) - AND ({kw_sql}) - ORDER BY seq ASC, id ASC - LIMIT :lim - """ - ), - params, - ).mappings().all() - truncated = len(rows) > lim - rows = rows[:lim] - out: list[dict[str, Any]] = [] - for r in rows: - data = dict(r["data_json"] or {}) - collected = r["collected_at"] - out.append( - { - **data, - "_netx": { - "batch_id": bid, - "batch_command_id": str(r["batch_command_id"] or ""), - "task_id": str(r["task_id"] or ""), - "ne_id": str(r["ne_id"] or ""), - "collected_at": collected.isoformat() + "Z" - if collected is not None - else None, - "row_id": str(r["id"]), - }, - } - ) - return out, truncated - - # Non-PG / fallback: scan with early stop (OK for tests / small sheets) - q = ( - db.query(BizStateMetricRow) - .filter( - BizStateMetricRow.batch_id == bid, - BizStateMetricRow.metric_id == mid, - ) - .order_by(BizStateMetricRow.seq.asc(), BizStateMetricRow.id.asc()) - ) - out = [] - truncated = False - needle_l = needle.lower() - key_set = [str(k).strip() for k in key_fields if str(k).strip()] - for r in q.yield_per(500): - data = dict(r.data_json or {}) - row = { - **data, - "_netx": { - "batch_id": bid, - "batch_command_id": r.batch_command_id or "", - "task_id": r.task_id or "", - "ne_id": r.ne_id or "", - "collected_at": r.collected_at.isoformat() + "Z" - if r.collected_at - else None, - "row_id": r.id, - }, - } - if filters and not all(row_matches_filter(row, f) for f in filters): - db.expunge(r) - continue - hit = False - for kf in key_set: - if needle_l in str(row.get(kf) or "").lower(): - hit = True - break - if not hit: - blob = json.dumps(data, ensure_ascii=False, default=str).lower() - hit = needle_l in blob - if not hit: - db.expunge(r) - continue - out.append(row) - db.expunge(r) - if len(out) >= lim: - # peek one more? - truncated = True - break - return out, truncated - - -def _live_search_sheet_diffs( - db: Session, - run: BizCompareRun, - sheet: dict[str, Any], - *, - kind: str, - kw: str, - page: int, - page_size: int, -) -> dict[str, Any]: - """Search before/after metric tables, zip-compare, filter by kind tab.""" - mid_src = str(sheet.get("metric_id") or "").strip() - sid = sheet_key(sheet) - key_fields = list(sheet.get("key_fields") or []) - iface_fields = list(sheet.get("iface_fields") or []) - field_rules = list(sheet.get("field_rules") or []) - compare_fields = effective_compare_fields( - list(sheet.get("compare_fields") or []), - field_rules, - ) - row_filters = list(sheet.get("row_filters") or []) - # Older runs may lack row_filters on sheet meta — fall back to template - if not row_filters and run.template_id: - tpl = db.get(BizCompareTemplate, run.template_id) - if tpl: - for s in template_metrics(tpl): - if sheet_key(s) == sid: - row_filters = list(s.get("row_filters") or []) - if not key_fields: - key_fields = list(s.get("key_fields") or []) - if not field_rules: - field_rules = list(s.get("field_rules") or []) - compare_fields = effective_compare_fields( - list(s.get("compare_fields") or compare_fields), - field_rules, - ) - break - - if not key_fields or not mid_src: - return { - "total": 0, - "page": page, - "page_size": page_size, - "metric_id": sid, - "items": [], - "source": "live", - "truncated": False, - } - - ignore_ports = sheet.get("ignore_port_changes") - if ignore_ports is not None: - ignore_ports = bool(ignore_ports) - pmap = _port_map_dict(db, str(run.mapping_id or "")) - tpl = db.get(BizCompareTemplate, run.template_id) if run.template_id else None - norm_rules = template_iface_normalize(tpl) - - before_rows, trunc_b = _load_metric_rows_for_search( - db, - batch_id=str(run.before_batch_id or ""), - metric_id=mid_src, - row_filters=row_filters, - key_fields=key_fields, - kw=kw, - ) - after_rows, trunc_a = _load_metric_rows_for_search( - db, - batch_id=str(run.after_batch_id or ""), - metric_id=mid_src, - row_filters=row_filters, - key_fields=key_fields, - kw=kw, - ) - result = compare_rows( - before_rows=before_rows, - after_rows=after_rows, - key_fields=key_fields, - iface_fields=iface_fields, - compare_fields=compare_fields, - port_map=pmap, - field_rules=field_rules, - iface_normalize_rules=norm_rules, - ignore_port_changes=ignore_ports, - include_unchanged=True, - unchanged_limit=None, - compact_unchanged=False, - ) - kind_n = (kind or "diff").strip().lower() - filtered = [ - d for d in list(result.get("diffs") or []) if _kind_allows(kind_n, str(d.get("kind") or "")) - ] - # Cap pairs returned to keep UI snappy - truncated = bool(trunc_b or trunc_a or len(filtered) > _LIVE_SEARCH_RESULT_CAP) - filtered = filtered[:_LIVE_SEARCH_RESULT_CAP] - total = len(filtered) - start = (page - 1) * page_size - page_items = filtered[start : start + page_size] - return { - "total": total, - "page": page, - "page_size": page_size, - "metric_id": sid, - "items": page_items, - "source": "live", - "truncated": truncated, - "live_before_matched": len(before_rows), - "live_after_matched": len(after_rows), - } - - def list_run_diffs( db: Session, run_id: str, @@ -3332,18 +2973,6 @@ def list_run_diffs( sheet = _lookup_sheet(sheets, asked) if asked else (sheets[0] if sheets else None) mid = sheet_key(sheet) if sheet else (asked or str(r.metric_id or "")) - # Unified search: any kw → live source lookup; kind tab only filters. - if kw_n and sheet: - return _live_search_sheet_diffs( - db, - r, - sheet, - kind=kind_n, - kw=kw_n, - page=page_n, - page_size=size_n, - ) - if _run_has_diff_rows(db, run_id): q = db.query(BizCompareDiff).filter( BizCompareDiff.run_id == run_id, @@ -3351,12 +2980,10 @@ def list_run_diffs( ) if kind_n == "diff": q = q.filter(BizCompareDiff.kind.in_(("removed", "changed"))) - elif kind_n == "removed": - q = q.filter(BizCompareDiff.kind == "removed") - elif kind_n == "changed": - q = q.filter(BizCompareDiff.kind == "changed") elif kind_n != "all": q = q.filter(BizCompareDiff.kind == kind_n) + if kw_n: + q = q.filter(BizCompareDiff.search_text.ilike(f"%{kw_n}%")) total = q.count() rows = ( q.order_by(BizCompareDiff.seq.asc(), BizCompareDiff.id.asc()) @@ -3371,8 +2998,6 @@ def list_run_diffs( "page_size": size_n, "metric_id": mid, "items": items, - "source": "stored", - "truncated": False, } # Legacy: diffs embedded in summary_json / diffs_json @@ -3382,7 +3007,7 @@ def list_run_diffs( inline = list((sheet or {}).get("diffs") or []) if not inline and mid == r.metric_id: inline = list(r.diffs_json or []) - filtered = _filter_inline_diffs(inline, kind=kind_n, kw="") + filtered = _filter_inline_diffs(inline, kind=kind_n, kw=kw_n) total = len(filtered) start = (page_n - 1) * size_n page_items = filtered[start : start + size_n] @@ -3392,8 +3017,6 @@ def list_run_diffs( "page_size": size_n, "metric_id": mid, "items": page_items, - "source": "stored", - "truncated": False, } diff --git a/netx_api/biz_state/compare_sql.py b/netx_api/biz_state/compare_sql.py index b7ba9d3..ae42ed8 100644 --- a/netx_api/biz_state/compare_sql.py +++ b/netx_api/biz_state/compare_sql.py @@ -406,9 +406,6 @@ def run_sql_sheet_compare( ) -> dict[str, Any]: """Compare one sheet via PostgreSQL TEMP tables + FULL OUTER JOIN. - Same-key multipath: rows get ``dup_rn`` by ``seq,id`` and join on - ``(rk, dup_rn)`` (ordered zip). Extras are added/removed, not duplicate. - Returns the same ``{summary, diffs, mapping_stats}`` shape as ``compare_rows``. ``on_progress(side, n, *, engine=\"sql\", note=..., phase=...)``. """ @@ -506,7 +503,9 @@ def run_sql_sheet_compare( ), params, ) - db.execute(text(f"CREATE INDEX IF NOT EXISTS {tname}_rk ON {tname} (rk, dup_rn)")) + db.execute( + text(f"CREATE INDEX IF NOT EXISTS {tname}_rk ON {tname} (rk) WHERE dup_rn = 1") + ) before_n = int( db.execute(text(f"SELECT count(*) FROM {tb}")).scalar() or 0 @@ -526,14 +525,14 @@ def run_sql_sheet_compare( END """ - # Ordered same-key pairing: join on (rk, dup_rn) so 5 vs 2 → 2 compared + 3 removed + # Aggregate primary match kinds (first-wins keys only) agg_rows = db.execute( text( f""" SELECT {kind_expr} AS kind, count(*)::bigint AS n - FROM {tb} b - FULL OUTER JOIN {ta} a - ON b.rk = a.rk AND b.dup_rn = a.dup_rn + 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 """ ) @@ -544,19 +543,13 @@ def run_sql_sheet_compare( changed = counts.get("changed", 0) unchanged = counts.get("unchanged", 0) - # Diagnostic: count of match keys with >1 row (not fail rows) - multi_b = int( - db.execute( - text(f"SELECT count(*) FROM (SELECT rk FROM {tb} GROUP BY rk HAVING count(*) > 1) t") - ).scalar() - or 0 + dup_b = int( + db.execute(text(f"SELECT count(*) FROM {tb} WHERE dup_rn > 1")).scalar() or 0 ) - multi_a = int( - db.execute( - text(f"SELECT count(*) FROM (SELECT rk FROM {ta} GROUP BY rk HAVING count(*) > 1) t") - ).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 @@ -566,7 +559,7 @@ def run_sql_sheet_compare( ) diffs: list[dict[str, Any]] = [] - # Fail rows (stream into Python — should be << million) + # Fail + duplicate rows (stream into Python — should be << million) fail_sql = text( f""" SELECT @@ -576,14 +569,12 @@ def run_sql_sheet_compare( b.data AS before_data, a.data AS after_data, COALESCE(b.rk, a.rk) AS rk - FROM {tb} b - FULL OUTER JOIN {ta} a - ON b.rk = a.rk AND b.dup_rn = a.dup_rn + 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') """ ) - fail_n = 0 - _prog("after", after_n, phase="sql_fail_fetch", note="fetch_fails") 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 @@ -602,77 +593,54 @@ def run_sql_sheet_compare( if kind == "changed": item["changes"] = _changes_from_rows(before, after, compare_fields, rules) diffs.append(item) - fail_n += 1 - if fail_n == 1 or fail_n % 25_000 == 0: - _prog("fail", fail_n, phase="sql_fail_fetch", note="fetch_fails") - if fail_n: - _prog("fail", fail_n, phase="sql_fail_fetch", note="fetch_fails") + + # 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] = {} - # Stratified sample: round-robin by neighbor|direction|afi|vrf so one - # BGP command cannot consume the whole success sample. - stratum_expr = """ - concat_ws('|', - coalesce(nullif(trim(both from coalesce(b.data->>'neighbor','')), ''), '_'), - coalesce(nullif(trim(both from coalesce(b.data->>'direction','')), ''), '_'), - coalesce(nullif(trim(both from coalesce(b.data->>'afi','')), ''), '_'), - coalesce(nullif(trim(both from coalesce(b.data->>'vrf','')), ''), '_') - ) - """ if limit_n is not None: + lim_sql = " LIMIT :lim" params_u["lim"] = max(0, int(limit_n)) - u_sql = text( - f""" - WITH u AS ( - SELECT - b.id AS before_row_id, - a.id AS after_row_id, - b.data AS before_data, - a.data AS after_data, - {stratum_expr} AS stratum - FROM {tb} b - INNER JOIN {ta} a - ON b.rk = a.rk AND b.dup_rn = a.dup_rn - WHERE NOT ({changed_pred}) - ), - ranked AS ( - SELECT *, - row_number() OVER ( - PARTITION BY stratum ORDER BY before_row_id ASC - ) AS rn_in - FROM u - ), - picked AS ( - SELECT *, - row_number() OVER ( - ORDER BY rn_in ASC, stratum ASC, before_row_id ASC - ) AS pick_ord - FROM ranked - ) - SELECT before_row_id, after_row_id, before_data, after_data - FROM picked - WHERE pick_ord <= :lim - """ - ) - else: - 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 {tb} b - INNER JOIN {ta} a - ON b.rk = a.rk AND b.dup_rn = a.dup_rn - WHERE NOT ({changed_pred}) - """ - ) + 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 {}) @@ -709,6 +677,7 @@ def run_sql_sheet_compare( 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, @@ -716,21 +685,16 @@ def run_sql_sheet_compare( "removed": removed, "changed": changed, "unchanged": unchanged, - "duplicate": 0, + "duplicate": duplicate, "match_key_fields": list(key_fields), - "duplicate_keys_before": multi_b, - "duplicate_keys_after": multi_a, - "duplicate_key_list": [], + "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), - "unchanged_sample_mode": ( - "stratified" - if include_u and limit_n is not None and unchanged > unchanged_listed - else ("full" if include_u else "none") - ), "engine": "sql", "before_raw_count": raw_b, "after_raw_count": raw_a, diff --git a/tests/test_biz_state_compare.py b/tests/test_biz_state_compare.py index dcb07e1..79a5f86 100644 --- a/tests/test_biz_state_compare.py +++ b/tests/test_biz_state_compare.py @@ -5,8 +5,7 @@ from __future__ import annotations import copy import unittest -from netx_api.biz_state.compare_engine import compare_rows, mapping_stats, stratify_take -from netx_api.biz_state.compare_service import _kind_allows +from netx_api.biz_state.compare_engine import compare_rows, mapping_stats from netx_api.biz_state.compare_rules import ( apply_row_filters, arp_dynamic_row_filters, @@ -199,70 +198,12 @@ class CompareEngineTests(unittest.TestCase): self.assertEqual(out["summary"]["unchanged_listed"], 2) self.assertTrue(out["summary"]["unchanged_truncated"]) self.assertTrue(out["summary"]["unchanged_compact"]) - self.assertEqual(out["summary"]["unchanged_sample_mode"], "stratified") self.assertEqual(len(out["diffs"]), 2) self.assertEqual(out["diffs"][0]["before"], {}) self.assertEqual(out["diffs"][0]["after"], {}) self.assertEqual(out["diffs"][0]["before_row_id"], "b0") self.assertEqual(out["diffs"][0]["after_row_id"], "a0") - def test_stratify_take_round_robin(self) -> None: - items = [("a", i) for i in range(5)] + [("b", i) for i in range(5)] - got = stratify_take(items, 4, key_fn=lambda x: x[0]) - self.assertEqual([x[0] for x in got], ["a", "b", "a", "b"]) - - def test_kind_allows_tabs(self) -> None: - self.assertTrue(_kind_allows("all", "unchanged")) - self.assertTrue(_kind_allows("diff", "removed")) - self.assertTrue(_kind_allows("diff", "changed")) - self.assertFalse(_kind_allows("diff", "added")) - self.assertFalse(_kind_allows("diff", "unchanged")) - self.assertTrue(_kind_allows("unchanged", "unchanged")) - self.assertFalse(_kind_allows("unchanged", "removed")) - self.assertTrue(_kind_allows("removed", "removed")) - self.assertTrue(_kind_allows("changed", "changed")) - - def test_unchanged_sample_stratified_across_neighbors(self) -> None: - """Sample must cover multiple neighbors, not only the first command's rows.""" - before = [] - after = [] - for n_i, neigh in enumerate(("1.1.1.1", "2.2.2.2", "3.3.3.3")): - for j in range(10): - net = f"10.{n_i}.{j}.0/24" - before.append( - { - "neighbor": neigh, - "direction": "in", - "network": net, - "v": "1", - "_netx": {"row_id": f"b-{neigh}-{j}"}, - } - ) - after.append( - { - "neighbor": neigh, - "direction": "in", - "network": net, - "v": "1", - "_netx": {"row_id": f"a-{neigh}-{j}"}, - } - ) - out = compare_rows( - before_rows=before, - after_rows=after, - key_fields=["neighbor", "direction", "network"], - iface_fields=[], - compare_fields=["v"], - port_map={}, - include_unchanged=True, - unchanged_limit=6, - compact_unchanged=True, - ) - self.assertEqual(out["summary"]["unchanged"], 30) - self.assertEqual(out["summary"]["unchanged_listed"], 6) - keys = [d["key"]["neighbor"] for d in out["diffs"]] - self.assertEqual(set(keys), {"1.1.1.1", "2.2.2.2", "3.3.3.3"}) - def test_mac_normalize_via_field_rules(self) -> None: before = [{"ip": "1.1.1.1", "mac": "00:11:22:33:44:55", "iface": "gei-0/1"}] after = [{"ip": "1.1.1.1", "mac": "0011.2233.4455", "iface": "gei-0/1"}] @@ -534,13 +475,6 @@ class CompareSheetDefaultsTests(unittest.TestCase): for r in (route4.get("field_rules") or []) ) ) - route_vpn = next(s for s in sheets if sheet_key(s) == "bgp_route.vpnv4") - self.assertTrue( - any( - f.get("field") == "afi" and f.get("value") == "vpnv4" - for f in (route_vpn.get("row_filters") or []) - ) - ) vpnv4 = next(s for s in sheets if sheet_key(s) == "bgp_peer.vpnv4") self.assertEqual( vpnv4["row_filters"], @@ -549,8 +483,7 @@ class CompareSheetDefaultsTests(unittest.TestCase): isis4 = next(s for s in sheets if sheet_key(s) == "isis_adjacency.ipv4") self.assertEqual(isis4["row_filters"][0]["op"], "contains") - def test_ordered_same_key_pairing_2v2(self) -> None: - """Same key, two rows each: zip by order (not first-wins + duplicate).""" + def test_duplicate_match_keys_are_reported(self) -> None: before = [ {"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "1"}, {"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "9"}, @@ -567,47 +500,15 @@ class CompareSheetDefaultsTests(unittest.TestCase): compare_fields=["remote_ip"], port_map={"a": "a"}, ) - self.assertEqual(out["summary"]["before_count"], 2) - self.assertEqual(out["summary"]["after_count"], 2) - self.assertEqual(out["summary"]["duplicate"], 0) self.assertEqual(out["summary"]["duplicate_keys_before"], 1) self.assertEqual(out["summary"]["duplicate_keys_after"], 1) + self.assertEqual(out["summary"]["duplicate"], 2) self.assertIn("a|X|1", out["summary"]["duplicate_key_list"]) kinds = [d["kind"] for d in out["diffs"]] - self.assertNotIn("duplicate", kinds) - # Pair 0: 1==1 unchanged; pair 1: 9!=2 changed + self.assertEqual(kinds.count("duplicate"), 2) + # First before wins → matches first after → unchanged (same remote_ip) self.assertEqual(out["summary"]["unchanged"], 1) - self.assertEqual(out["summary"]["changed"], 1) - self.assertEqual(out["summary"]["added"], 0) - self.assertEqual(out["summary"]["removed"], 0) - - def test_ordered_multipath_5_vs_2(self) -> None: - """5 before / 2 after same key → 2 compared + 3 removed failures.""" - key = {"local_if": "a", "remote_sys": "X", "remote_if": "1"} - before = [{**key, "remote_ip": str(i)} for i in range(5)] - after = [{**key, "remote_ip": "0"}, {**key, "remote_ip": "1"}] - out = compare_rows( - before_rows=before, - after_rows=after, - key_fields=["local_if", "remote_sys", "remote_if"], - iface_fields=["local_if"], - compare_fields=["remote_ip"], - port_map={"a": "a"}, - ) - self.assertEqual(out["summary"]["before_count"], 5) - self.assertEqual(out["summary"]["after_count"], 2) - self.assertEqual(out["summary"]["removed"], 3) - self.assertEqual(out["summary"]["added"], 0) - self.assertEqual( - out["summary"]["unchanged"] + out["summary"]["changed"], - 2, - ) - self.assertEqual(out["summary"]["unchanged"], 2) self.assertEqual(out["summary"]["changed"], 0) - self.assertEqual(out["summary"]["duplicate"], 0) - kinds = [d["kind"] for d in out["diffs"]] - self.assertEqual(kinds.count("removed"), 3) - self.assertNotIn("duplicate", kinds) def test_ignore_port_changes_false_keeps_iface(self) -> None: before = [ diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 1367d79..1ee7ef0 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -502,23 +502,20 @@ const en = { runProgress: "{{phase}} · sheet {{i}}/{{n}} · {{sheet}} · {{s}}s elapsed", runRowsLoaded: "Loaded {{side}} {{n}} rows", runRowsSqlCount: "DB count {{side}} {{n}} rows", - runPersisting: "Wrote {{done}}/{{total}} 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.", - unchangedSampleHint: "{{total}} success rows total; browsing a stratified sample of {{listed}}. Search any route for live source lookup. Pass rate uses all {{total}}.", - liveSearchHint: "Search: live verdict from source batches (tabs only filter kind). Works for fail, success, and added.", - liveSearchTruncatedHint: "Search results truncated — narrow the query (full prefix / neighbor).", + unchangedSampleHint: "{{total}} success rows total; browsing a sample of {{listed}} (hydrated from source tables for cutover spot-check). Pass rate uses all {{total}}.", storeUnchanged: "Store success rows", storeUnchangedAuto: "Auto (full if small / sample 5k + hydrate if large)", storeUnchangedSample: "Sample only (up to 5k keys + hydrate)", storeUnchangedKeys: "All success keys (full browse; large write still slow)", storeUnchangedAlways: "Always full JSON (slow on huge sheets)", storeUnchangedNever: "Never (counts only)", - storeUnchangedHint: "Cutover focuses on fails and pass rate; success defaults to a stratified sample. Type a filter to look up any route from source tables. Pass rate uses the full success count.", + storeUnchangedHint: "Cutover focuses on fails and pass rate; success defaults to a sample. Pass rate always uses the full success count.", saveJob: "Save config", jobSaved: "Job config saved", jobDeleted: "Compare job deleted", @@ -545,7 +542,7 @@ const en = { pickRun: "Select run…", runCount: "{{n}} runs", resultEmpty: "No matching diff rows", - resultFilterPh: "Search prefix / neighbor / RD… (live source lookup)", + resultFilterPh: "Filter key / values…", kindAll: "All", kindDiff: "Fail", kindAdded: "Added", @@ -577,7 +574,7 @@ const en = { filterFailField: "Failed compare field", filterFailFieldAll: "Any failed field", resultEmpty: "No matching rows", - resultFilterPh: "Search prefix / neighbor / RD… (live source lookup)", + resultFilterPh: "Filter identity / values…", diffCount: "{{n}} failed", passRate: "Pass rate", passOk: "All passed", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index bd7c3e9..f6f364a 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -501,23 +501,20 @@ const zh = { runProgress: "{{phase}} · 表 {{i}}/{{n}} · {{sheet}} · 已用 {{s}}s", runRowsLoaded: "已加载 {{side}} {{n}} 行", runRowsSqlCount: "库内统计 {{side}} {{n}} 行", - runPersisting: "已写入 {{done}}/{{total}} 行", runEngineSql: "引擎 SQL", runEnginePython: "引擎 Python", runEngineNote: "原因 {{note}}", sheetPending: "等待中", ranWithDuration: "比对完成（耗时 {{s}} 秒）", unchangedNotStored: "成功行仅统计数量未落库。可在任务配置将「成功行保存」改为抽样后重新比对（抽查用）。", - unchangedSampleHint: "成功共 {{total}} 条，明细抽样 {{listed}} 条（分层抽查）。要查任意路由请输入筛选条件——将按原表即时判定。通过率按全部 {{total}} 计。", - liveSearchHint: "搜索：原表即时判定（页签只过滤种类）。失败/成功/新增均可查到。", - liveSearchTruncatedHint: "搜索结果已截断，请收窄条件（如完整前缀 / neighbor）。", + unchangedSampleHint: "成功共 {{total}} 条，明细抽样 {{listed}} 条（从原表补全显示，割接抽查用）。通过率按全部 {{total}} 计。", storeUnchanged: "成功行保存", storeUnchangedAuto: "自动（小表全量 / 大表抽样 5000+原表补全）", storeUnchangedSample: "仅抽样（最多 5000，身份键+原表补全）", storeUnchangedKeys: "全量身份键（可翻完全部成功，大表写入仍较久）", storeUnchangedAlways: "始终全量 JSON（大表很慢，慎用）", storeUnchangedNever: "不保存（只计数量）", - storeUnchangedHint: "割接优先看失败与通过率；成功默认分层抽样。输入筛选可回查原表（不依赖抽样）。通过率按全部成功计数。", + storeUnchangedHint: "割接优先看失败与通过率；成功默认抽样抽查。通过率始终按全部成功计数。", saveJob: "保存配置", jobSaved: "任务配置已保存", jobDeleted: "比对任务已删除", @@ -544,7 +541,7 @@ const zh = { pickRun: "选择比对记录…", runCount: "{{n}} 次", resultEmpty: "无匹配失败/结果行", - resultFilterPh: "搜索前缀 / neighbor / RD…（回查原表）", + resultFilterPh: "筛选身份 / 字段值…", kindAll: "全部", kindDiff: "失败", kindAdded: "新增", diff --git a/web/src/pages/network/BizComparePage.tsx b/web/src/pages/network/BizComparePage.tsx index af63c82..fdd13d3 100644 --- a/web/src/pages/network/BizComparePage.tsx +++ b/web/src/pages/network/BizComparePage.tsx @@ -38,7 +38,6 @@ import { jobChipColor, NmStatusChip } from "./nmChips"; type PageTab = "templates" | "jobs"; type JobDetailTab = "config" | "runs" | "result"; type KindFilter = "diff" | "all" | "added" | "removed" | "changed" | "unchanged"; -type DiffListSource = "stored" | "live" | ""; type CreateJobStep = 0 | 1 | 2 | 3; const CREATE_JOB_STEPS = 4; @@ -848,8 +847,6 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage const [resultTotal, setResultTotal] = useState(0); const [pagedDiffs, setPagedDiffs] = useState([]); const [diffsLoading, setDiffsLoading] = useState(false); - const [diffsSource, setDiffsSource] = useState(""); - const [diffsTruncated, setDiffsTruncated] = useState(false); const boardRef = useRef(null); const tableScrollRef = useRef(null); const tableScrollPosRef = useRef({ top: 0, left: 0 }); @@ -1072,8 +1069,6 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage if (!runId || !mid || jobDetailTab !== "result" || sheetStillRunning) { setPagedDiffs([]); setResultTotal(0); - setDiffsSource(""); - setDiffsTruncated(false); return; } let cancelled = false; @@ -1091,10 +1086,6 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage if (cancelled) return; setPagedDiffs((res.items || []) as DiffRow[]); setResultTotal(Number(res.total || 0)); - setDiffsSource( - String((res as any).source || "").toLowerCase() === "live" ? "live" : "stored", - ); - setDiffsTruncated(Boolean((res as any).truncated)); const pages = Math.max( 1, Math.ceil(Number(res.total || 0) / Number(res.page_size || resultPageSize)), @@ -1266,15 +1257,8 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage String(activeSheetCard?.status || ""), ); const activeAdded = Number(activeSheetCard?.added || 0); - const activeRemoved = Number(activeSheetCard?.removed || 0); - const activeChanged = Number(activeSheetCard?.changed || 0); const showFailCol = - kindFilter === "diff" || - kindFilter === "all" || - kindFilter === "added" || - kindFilter === "removed" || - kindFilter === "changed"; - const isLiveSearch = diffsSource === "live" && Boolean(debouncedResultKw.trim()); + kindFilter === "diff" || kindFilter === "all" || kindFilter === "added"; const resultEmptyColSpan = 1 + (showFailCol ? 1 : 0) + @@ -1911,10 +1895,6 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage rows_loaded?: number; engine?: string; engine_note?: string; - persisted?: number; - persist_total?: number; - fail_rows?: number; - ok_rows?: number; }; const runEngine = String(runProgress.engine || "").toLowerCase(); @@ -3312,15 +3292,6 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage })} ) : null} - {String(runProgress.phase || "").startsWith("persisting") && - Number(runProgress.persist_total || 0) > 0 ? ( - - {t("bizCompare.runPersisting", { - done: String(runProgress.persisted || 0), - total: String(runProgress.persist_total || 0), - })} - - ) : null} {runDetail.message ? ( {String(runDetail.message)} ) : null} @@ -3520,8 +3491,6 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage {( [ ["diff", activeFail, "diff"], - ["removed", activeRemoved, "removed"], - ["changed", activeChanged, "changed"], ["added", activeAdded, "added"], ["unchanged", activeSuccess, "unchanged"], ["all", null, "all"], @@ -3541,11 +3510,7 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage ? t("bizCompare.kindAll") : id === "added" ? t("bizCompare.kindAddedShort") - : id === "removed" - ? t("bizCompare.kindRemovedShort") - : id === "changed" - ? t("bizCompare.kindChangedShort") - : t("bizCompare.kindSuccess")} + : t("bizCompare.kindSuccess")} {n !== null ? ( <> {" "} @@ -3565,31 +3530,11 @@ export function BizComparePage({ pageMode = "all" }: { pageMode?: BizComparePage {diffsLoading ? "…" : `${pagedDiffs.length}/${resultTotal}`} - {isLiveSearch ? ( -

- {diffsTruncated - ? t("bizCompare.liveSearchTruncatedHint") - : t("bizCompare.liveSearchHint")} -

- ) : null} - {!isLiveSearch && - kindFilter === "unchanged" && + {kindFilter === "unchanged" && (runDetail?.summary?.unchanged_truncated || - (Number( - (activeRunSheet?.summary as any)?.unchanged || - runDetail?.summary?.unchanged || - 0, - ) > - Number( - (activeRunSheet?.summary as any)?.unchanged_listed ?? - runDetail?.summary?.unchanged_listed ?? - 0, - ) && - Number( - (activeRunSheet?.summary as any)?.unchanged_listed ?? - runDetail?.summary?.unchanged_listed ?? - 0, - ) > 0)) ? ( + (Number(runDetail?.summary?.unchanged || 0) > + Number(runDetail?.summary?.unchanged_listed || 0) && + Number(runDetail?.summary?.unchanged_listed || 0) > 0)) ? (

{t("bizCompare.unchangedSampleHint", { listed: String( diff --git a/web/src/services/api.ts b/web/src/services/api.ts index 6facb74..d3b22f8 100644 --- a/web/src/services/api.ts +++ b/web/src/services/api.ts @@ -2185,10 +2185,6 @@ export const bizCompareListRunDiffs = (params: { page_size: number; metric_id: string; items: Record[]; - source?: string; - truncated?: boolean; - live_before_matched?: number; - live_after_matched?: number; }>(`/v1/biz-state/compare/runs/${encodeURIComponent(params.runId)}/diffs?${p.toString()}`); };