From f8228847ab6afedddc0584bdcefee7b8a4edb85b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Mon, 22 Jun 2026 17:29:10 +0800 Subject: [PATCH] =?UTF-8?q?fix(skill-addresses):=20=E8=A1=A5=E6=8F=90?= =?UTF-8?q?=E4=BA=A4=E8=A2=AB=20gitignore=20=E6=BC=8F=E6=8E=89=E7=9A=84=20?= =?UTF-8?q?address=5Fscan=20=E5=BC=95=E6=93=8E=E6=A8=A1=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit skills/ 忽略规则误伤了 packages/harness/deerflow/skills/address_scan.py， 导致部署环境缺模块、启动报 ModuleNotFoundError: deerflow.skills.address_scan。 强制纳入跟踪。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../harness/deerflow/skills/address_scan.py | 255 ++++++++++++++++++ 1 file changed, 255 insertions(+) create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/skills/address_scan.py diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/skills/address_scan.py b/offline-backend-20260512/backend/packages/harness/deerflow/skills/address_scan.py new file mode 100644 index 00000000..473b3de5 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/skills/address_scan.py @@ -0,0 +1,255 @@ +"""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