fix(skill-addresses): 补提交被 gitignore 漏掉的 address_scan 引擎模块
skills/ 忽略规则误伤了 packages/harness/deerflow/skills/address_scan.py， 导致部署环境缺模块、启动报 ModuleNotFoundError: deerflow.skills.address_scan。 强制纳入跟踪。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.8 committed
1 parent
6b480fdbe9
commit
f8228847ab
1 file changed
+255
@@ -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 (``(?<![\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
|
||||
Reference in new issue
Block a user