136 lines
4.4 KiB
Python
136 lines
4.4 KiB
Python
from dataclasses import dataclass, field
|
|
|
|
from .snapshot import AccountConfigEntry, is_sensitive_config_path
|
|
|
|
_SENTINEL = object()
|
|
|
|
|
|
def _redact_path(path: str) -> str:
|
|
"""对敏感配置路径进行脱敏处理。"""
|
|
if not is_sensitive_config_path(path):
|
|
return path
|
|
parts = path.rsplit(".", 1)
|
|
if len(parts) == 2:
|
|
return f"{parts[0]}.<redacted>"
|
|
return "<redacted>"
|
|
|
|
|
|
@dataclass
|
|
class ConfigDiff:
|
|
added: list[AccountConfigEntry] = field(default_factory=list)
|
|
removed: list[AccountConfigEntry] = field(default_factory=list)
|
|
changed: list[AccountConfigEntry] = field(default_factory=list)
|
|
unchanged: list[AccountConfigEntry] = field(default_factory=list)
|
|
routes_changed: bool = False
|
|
changed_paths: list[str] = field(default_factory=list)
|
|
|
|
@property
|
|
def has_changes(self) -> bool:
|
|
return bool(self.added or self.removed or self.changed or self.routes_changed or self.changed_paths)
|
|
|
|
@property
|
|
def summary(self) -> str:
|
|
parts = []
|
|
if self.added:
|
|
parts.append(f"added={len(self.added)}")
|
|
if self.removed:
|
|
parts.append(f"removed={len(self.removed)}")
|
|
if self.changed:
|
|
parts.append(f"changed={len(self.changed)}")
|
|
if self.routes_changed:
|
|
parts.append("routes_changed")
|
|
if self.changed_paths:
|
|
redacted_paths = [_redact_path(p) for p in self.changed_paths[:5]]
|
|
paths_preview = ", ".join(redacted_paths)
|
|
if len(self.changed_paths) > 5:
|
|
paths_preview += f" (+{len(self.changed_paths) - 5} more)"
|
|
parts.append(f"paths=[{paths_preview}]")
|
|
return f"ConfigDiff({', '.join(parts)})"
|
|
|
|
|
|
def compute_diff(
|
|
prev: list[AccountConfigEntry],
|
|
next_: list[AccountConfigEntry],
|
|
) -> ConfigDiff:
|
|
prev = prev or []
|
|
next_ = next_ or []
|
|
|
|
prev_map = {f"{e.channel_type}:{e.account_id}": e for e in prev}
|
|
next_map = {f"{e.channel_type}:{e.account_id}": e for e in next_}
|
|
|
|
diff = ConfigDiff()
|
|
|
|
for key, entry in next_map.items():
|
|
if key not in prev_map:
|
|
diff.added.append(entry)
|
|
|
|
for key, entry in prev_map.items():
|
|
if key not in next_map:
|
|
diff.removed.append(entry)
|
|
|
|
for key, entry in next_map.items():
|
|
prev_entry = prev_map.get(key)
|
|
if prev_entry is None:
|
|
continue
|
|
if (
|
|
entry.enabled != prev_entry.enabled
|
|
or entry.configured != prev_entry.configured
|
|
or entry.config_hash != prev_entry.config_hash
|
|
):
|
|
diff.changed.append(entry)
|
|
else:
|
|
diff.unchanged.append(entry)
|
|
|
|
return diff
|
|
|
|
|
|
def _deep_equal(a, b) -> bool:
|
|
if a is b:
|
|
return True
|
|
if type(a) is not type(b):
|
|
return False
|
|
if isinstance(a, dict):
|
|
if len(a) != len(b):
|
|
return False
|
|
return all(k in b and _deep_equal(a[k], b[k]) for k in a)
|
|
if isinstance(a, (list, tuple)):
|
|
if len(a) != len(b):
|
|
return False
|
|
return all(_deep_equal(x, y) for x, y in zip(a, b, strict=False))
|
|
return a == b
|
|
|
|
|
|
def diff_config_paths(prev, next_, prefix: str = "") -> list[str]:
|
|
if prev is next_:
|
|
return []
|
|
|
|
if isinstance(prev, dict) and isinstance(next_, dict):
|
|
keys = set(prev.keys()) | set(next_.keys())
|
|
paths: list[str] = []
|
|
for key in keys:
|
|
prev_value = prev.get(key, _SENTINEL)
|
|
next_value = next_.get(key, _SENTINEL)
|
|
if prev_value is _SENTINEL and next_value is _SENTINEL:
|
|
continue
|
|
child_prefix = f"{prefix}.{key}" if prefix else key
|
|
child_paths = diff_config_paths(prev_value, next_value, child_prefix)
|
|
if child_paths:
|
|
paths.extend(child_paths)
|
|
return paths
|
|
|
|
if isinstance(prev, (list, tuple)) and isinstance(next_, (list, tuple)):
|
|
if _deep_equal(prev, next_):
|
|
return []
|
|
max_len = max(len(prev), len(next_))
|
|
paths: list[str] = []
|
|
for i in range(max_len):
|
|
prev_item = prev[i] if i < len(prev) else _SENTINEL
|
|
next_item = next_[i] if i < len(next_) else _SENTINEL
|
|
item_prefix = f"{prefix}[{i}]"
|
|
child_paths = diff_config_paths(prev_item, next_item, item_prefix)
|
|
if child_paths:
|
|
paths.extend(child_paths)
|
|
return paths
|
|
|
|
return [prefix or "<root>"]
|