"""Scan skill text for URL / IP addresses and apply safe batch replacements. Pure-logic layer for the 技能地址(URL/IP)批量替换 feature. This module knows **only about text** — it never walks the filesystem. The HTTP router (``app/gateway/routers/skill_addresses.py``) walks ``skills/{public,custom}``, reads each ``*.md`` / ``*.py`` file, and feeds the text here. Keeping the extraction / replacement logic here (no ``app.*`` import) makes it unit-testable in isolation and respects the harness/app boundary. Two address kinds are supported (the feature scope): * ``"url"`` — full ``http(s)://…`` URLs. * ``"ip"`` — dotted-quad IPv4, optionally ``:port``. Each octet is range checked (0–255) so version-number noise like ``1.2.3.400`` is not matched. The single most important invariant is **substring-safe replacement**: we never ``str.replace`` raw file text (that would rewrite ``1.2.3.4`` *inside* ``11.2.3.40``). Instead we re-detect the exact token spans and only rewrite a span whose text is *exactly* one of the requested ``from`` values. """ from __future__ import annotations import re from dataclasses import dataclass __all__ = [ "VALID_KINDS", "AddressSpan", "Occurrence", "normalize_kinds", "find_spans", "scan_occurrences", "count_matches", "apply_replacements", "is_valid_replacement", "infer_kind", ] # Supported address kinds (the configured feature scope: URL + IP). VALID_KINDS: tuple[str, ...] = ("url", "ip") # A URL body stops at whitespace, quotes, brackets, angle brackets, backticks # and backslashes — everything that commonly *delimits* a URL in md / py. The # trailing punctuation that often clings to the end (a sentence period, a # closing paren already excluded, a Markdown comma) is stripped afterwards. _URL_RE = re.compile(r"https?://[^\s'\"`<>)\]}\\]+", re.IGNORECASE) # Dotted-quad IPv4 (optionally :port). Guarded so it is not glued to a longer # number run (``(?`" + ")。,;:!?】》、…") @dataclass(frozen=True) class AddressSpan: """An address token located at an exact ``[start, end)`` offset in text.""" kind: str value: str start: int end: int @dataclass(frozen=True) class Occurrence: """A located address enriched with 1-based line/column + the source line.""" kind: str value: str line: int col: int snippet: str def normalize_kinds(kinds: object) -> tuple[str, ...]: """Coerce arbitrary input into an ordered, de-duped subset of ``VALID_KINDS``. ``None`` / empty means "all kinds". Unknown kinds are dropped. """ if kinds is None: return VALID_KINDS if isinstance(kinds, str): raw = [p.strip() for p in kinds.split(",")] else: try: raw = [str(p).strip() for p in kinds] # type: ignore[assignment] except TypeError: return VALID_KINDS seen: list[str] = [] for k in raw: kl = k.lower() if kl in VALID_KINDS and kl not in seen: seen.append(kl) return tuple(seen) if seen else VALID_KINDS def _valid_ipv4(quad: str) -> bool: parts = quad.split(".") if len(parts) != 4: return False for p in parts: if not p.isdigit() or len(p) > 3: return False if int(p) > 255: return False return True def _valid_port(port: str | None) -> bool: if port is None: return True return port.isdigit() and 0 <= int(port) <= 65535 def _url_spans(text: str) -> list[AddressSpan]: spans: list[AddressSpan] = [] for m in _URL_RE.finditer(text): start, end = m.start(), m.end() value = m.group(0) # Strip clinging trailing punctuation. while value and value[-1] in _URL_TRAILING: value = value[:-1] end -= 1 if value: spans.append(AddressSpan(kind="url", value=value, start=start, end=end)) return spans def _ip_spans(text: str, exclude: list[AddressSpan]) -> list[AddressSpan]: spans: list[AddressSpan] = [] for m in _IP_RE.finditer(text): quad = m.group(1) port = m.group(2) if not _valid_ipv4(quad) or not _valid_port(port): continue start, end = m.start(), m.end() # Skip an IP that lives *inside* a URL (e.g. http://1.2.3.4:8080/x) — # replacing just the IP there would corrupt the URL. if any(s.start <= start < s.end for s in exclude): continue spans.append(AddressSpan(kind="ip", value=m.group(0), start=start, end=end)) return spans def find_spans(text: str, kinds: object = None) -> list[AddressSpan]: """Return every address token span in *text*, ordered by position. URL regions are always computed first so that IPs embedded inside a URL are not separately surfaced, even when only ``"ip"`` is requested. """ wanted = normalize_kinds(kinds) url_spans = _url_spans(text) result: list[AddressSpan] = [] if "url" in wanted: result.extend(url_spans) if "ip" in wanted: result.extend(_ip_spans(text, exclude=url_spans)) result.sort(key=lambda s: s.start) return result def scan_occurrences(text: str, kinds: object = None, *, snippet_max: int = 240) -> list[Occurrence]: """Locate addresses with 1-based line/column and a trimmed source line.""" spans = find_spans(text, kinds) if not spans: return [] # Precompute line-start offsets for O(log n) line lookup. line_starts = [0] for i, ch in enumerate(text): if ch == "\n": line_starts.append(i + 1) occ: list[Occurrence] = [] import bisect for s in spans: line_idx = bisect.bisect_right(line_starts, s.start) - 1 line_no = line_idx + 1 col = s.start - line_starts[line_idx] + 1 line_end = line_starts[line_idx + 1] - 1 if line_idx + 1 < len(line_starts) else len(text) raw_line = text[line_starts[line_idx]:line_end].rstrip("\r") snippet = raw_line.strip() if len(snippet) > snippet_max: snippet = snippet[: snippet_max - 1] + "…" occ.append(Occurrence(kind=s.kind, value=s.value, line=line_no, col=col, snippet=snippet)) return occ def count_matches(text: str, mapping: dict[str, str], kinds: object = None) -> int: """How many address spans in *text* would be rewritten by *mapping*.""" if not mapping: return 0 return sum(1 for s in find_spans(text, kinds) if s.value in mapping) def apply_replacements(text: str, mapping: dict[str, str], kinds: object = None) -> tuple[str, int]: """Substring-safe batch replace. Only spans whose text is *exactly* a key of *mapping* are rewritten, applied right-to-left so earlier offsets stay valid. Returns ``(new_text, count)``. A no-op mapping (or no hits) returns the original text and ``0``. """ if not mapping: return text, 0 spans = [s for s in find_spans(text, kinds) if s.value in mapping] if not spans: return text, 0 out = text for s in sorted(spans, key=lambda s: s.start, reverse=True): out = out[: s.start] + mapping[s.value] + out[s.end:] return out, len(spans) def infer_kind(value: str) -> str | None: """Return the address kind of *value* if it is a single, whole address. Used to classify a user-supplied ``from`` value (so the matching ``to`` can be validated against the right rules). Returns ``None`` when *value* is not exactly one URL or IP token. """ if not value: return None spans = find_spans(value, VALID_KINDS) if len(spans) == 1 and spans[0].value == value: return spans[0].kind return None def is_valid_replacement(kind: str, value: str) -> bool: """Validate that a user-supplied replacement is well-formed for its kind. Keeps an admin from turning a URL into garbage (e.g. dropping the scheme) or an IP into a non-IP. Empty replacements are rejected. """ if not value or value.strip() != value or any(c.isspace() for c in value): return False if kind == "url": return bool(re.fullmatch(r"https?://[^\s'\"`<>]+", value, re.IGNORECASE)) if kind == "ip": m = _IP_RE.fullmatch(value) if not m: return False return _valid_ipv4(m.group(1)) and _valid_port(m.group(2)) return False