deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/skills/address_scan.py

"""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 (``(?<![\d.])`` / ``(?![\d.])``); octets are range-checked below.
_IP_RE = re.compile(r"(?<![\d.])(\d{1,3}(?:\.\d{1,3}){3})(?::(\d{1,5}))?(?![\d.])")

# Punctuation (ASCII + CJK) that should not be considered part of a URL token
# when it appears at the very end of the match.
_URL_TRAILING = set(".,;:!?'\")]}>`" + "）。，；：！？】》、…")


@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