714 lines
26 KiB
Python
714 lines
26 KiB
Python
"""技能地址(URL/IP)批量替换 API (技能地址管理).
|
||
|
||
Scans the **running** skills directory (``skills/{public,custom}``) — the files
|
||
the agent actually reads — for ``http(s)://`` URLs and IPv4 (``ip[:port]``)
|
||
addresses, aggregates each unique address with every place it appears (across
|
||
``.md`` and ``.py``, across files), and lets an admin rewrite them in bulk.
|
||
|
||
Why this shape: the same address routinely lives in both a skill's ``SKILL.md``
|
||
(documented endpoint) and its ``scripts/*.py`` (the actual call), so the natural
|
||
unit is *"one address → replace all of its occurrences at once"*. The pure
|
||
scan/replace logic lives in ``deerflow.skills.address_scan`` (harness layer,
|
||
substring-safe span replacement); this router only walks the tree, enforces
|
||
admin auth + path safety, backs up before writing, and refreshes the skills
|
||
system-prompt cache so edits take effect immediately.
|
||
|
||
**Scale / availability (150+ skills):** every filesystem walk / read / write is
|
||
offloaded to a worker thread via ``asyncio.to_thread`` so the event loop is
|
||
never blocked while scanning or rewriting — the Gateway stays responsive. The
|
||
streaming endpoints additionally emit NDJSON progress (``已处理 X / 共 N``) so the
|
||
UI can render a progress bar for both the scan and the apply phases.
|
||
|
||
Routes (prefix ``/api/skill-addresses``, **admin only**):
|
||
GET /scan?kinds=url,ip&skill=<name> read-only aggregated scan (one shot)
|
||
GET /scan/stream?kinds=...&skill=... scan with per-skill NDJSON progress
|
||
POST /preview dry-run diff of a replacement set
|
||
POST /apply write changes (one shot, with backup)
|
||
POST /apply/stream write changes with per-file progress
|
||
GET /backups list restorable backups (newest first)
|
||
POST /rollback/{backup_id} restore a backup
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import shutil
|
||
from collections.abc import AsyncIterator
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
|
||
from fastapi import APIRouter, HTTPException, Query, Request
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, ConfigDict, Field
|
||
|
||
from app.gateway.deps import get_optional_user_from_request
|
||
from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
|
||
from deerflow.skills.address_scan import (
|
||
VALID_KINDS,
|
||
apply_replacements,
|
||
infer_kind,
|
||
is_valid_replacement,
|
||
normalize_kinds,
|
||
scan_occurrences,
|
||
)
|
||
from deerflow.skills.storage import get_or_new_skill_storage
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/skill-addresses", tags=["skill-addresses"])
|
||
|
||
_BACKUP_DIRNAME = ".address-edits"
|
||
_SCAN_SUBDIRS = ("public", "custom")
|
||
_SCAN_EXTS = (".md", ".py")
|
||
|
||
|
||
# ── schemas ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class OccurrenceModel(BaseModel):
|
||
skill: str
|
||
category: str # "public" | "custom"
|
||
file: str # POSIX path relative to the skills root
|
||
line: int
|
||
col: int
|
||
snippet: str
|
||
|
||
|
||
class AddressGroup(BaseModel):
|
||
address: str
|
||
kind: str # "url" | "ip"
|
||
count: int
|
||
occurrences: list[OccurrenceModel]
|
||
|
||
|
||
class ScanResponse(BaseModel):
|
||
groups: list[AddressGroup]
|
||
scanned_files: int
|
||
scanned_skills: int
|
||
kinds: list[str]
|
||
scanned_at: str
|
||
|
||
|
||
class Replacement(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
from_: str = Field(alias="from", min_length=1)
|
||
to: str = Field(min_length=1)
|
||
kind: str | None = None
|
||
|
||
|
||
class PreviewRequest(BaseModel):
|
||
replacements: list[Replacement] = Field(default_factory=list)
|
||
|
||
|
||
class PreviewFileHit(BaseModel):
|
||
file: str
|
||
line: int
|
||
before: str
|
||
after: str
|
||
|
||
|
||
class PreviewItem(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
from_: str = Field(serialization_alias="from")
|
||
to: str
|
||
kind: str
|
||
count: int
|
||
files: list[PreviewFileHit]
|
||
|
||
|
||
class PreviewResponse(BaseModel):
|
||
items: list[PreviewItem]
|
||
total: int
|
||
warnings: list[str]
|
||
expected_total: int
|
||
|
||
|
||
class ApplyRequest(BaseModel):
|
||
replacements: list[Replacement] = Field(default_factory=list)
|
||
expected_total: int | None = None
|
||
backup: bool = True
|
||
|
||
|
||
class ChangedFile(BaseModel):
|
||
file: str
|
||
count: int
|
||
|
||
|
||
class ApplyResponse(BaseModel):
|
||
changed_files: list[ChangedFile]
|
||
replaced_total: int
|
||
backup_id: str | None
|
||
warnings: list[str]
|
||
|
||
|
||
class ReplacementSummary(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
from_: str = Field(serialization_alias="from")
|
||
to: str
|
||
kind: str
|
||
|
||
|
||
class BackupEntry(BaseModel):
|
||
backup_id: str
|
||
files: int
|
||
created_at: str
|
||
replaced_total: int = 0
|
||
replacements: list[ReplacementSummary] = Field(default_factory=list)
|
||
|
||
|
||
class BackupListResponse(BaseModel):
|
||
backups: list[BackupEntry]
|
||
|
||
|
||
class RollbackResponse(BaseModel):
|
||
backup_id: str
|
||
restored_files: list[str]
|
||
|
||
|
||
# ── auth / path helpers ──────────────────────────────────────────────────────
|
||
|
||
|
||
async def _require_admin(request: Request) -> None:
|
||
"""Admin-only. Lets the call through when auth is disabled (no user)."""
|
||
|
||
user = await get_optional_user_from_request(request)
|
||
if user is None:
|
||
return
|
||
if getattr(user, "system_role", None) != "admin":
|
||
raise HTTPException(status_code=403, detail="技能地址管理仅限管理员")
|
||
|
||
|
||
def _skills_root() -> Path:
|
||
return Path(get_or_new_skill_storage().get_skills_root_path()).resolve()
|
||
|
||
|
||
def _is_within(root: Path, p: Path) -> bool:
|
||
try:
|
||
p.resolve().relative_to(root)
|
||
return True
|
||
except (ValueError, OSError):
|
||
return False
|
||
|
||
|
||
def _skill_of(root: Path, path: Path) -> tuple[str, str, str]:
|
||
"""Return ``(category, skill_name, posix_rel_path)`` for a scanned file."""
|
||
|
||
rel = path.relative_to(root)
|
||
parts = rel.parts
|
||
category = parts[0] if parts else ""
|
||
skill_name = parts[1] if len(parts) > 1 else ""
|
||
return category, skill_name, rel.as_posix()
|
||
|
||
|
||
def _read_text(path: Path) -> str | None:
|
||
"""Read UTF-8 text preserving original line endings; skip binary/bad files."""
|
||
|
||
try:
|
||
with open(path, "r", encoding="utf-8", newline="") as fh:
|
||
return fh.read()
|
||
except (UnicodeDecodeError, OSError):
|
||
return None
|
||
|
||
|
||
def _write_text(path: Path, text: str) -> None:
|
||
with open(path, "w", encoding="utf-8", newline="") as fh:
|
||
fh.write(text)
|
||
|
||
|
||
# ── synchronous workers (run via asyncio.to_thread; never block the loop) ─────
|
||
|
||
|
||
def _iter_skill_files(root: Path, skill: str | None = None) -> list[Path]:
|
||
"""All ``*.md`` / ``*.py`` under ``{root}/{public,custom}``.
|
||
|
||
Hidden directories (``.address-edits`` backups, ``.history``) are skipped.
|
||
When *skill* is given, only that skill's directory is walked.
|
||
"""
|
||
|
||
files: list[Path] = []
|
||
for sub in _SCAN_SUBDIRS:
|
||
base = root / sub
|
||
if skill:
|
||
base = base / skill
|
||
if not base.is_dir():
|
||
continue
|
||
for p in base.rglob("*"):
|
||
if not p.is_file() or p.suffix.lower() not in _SCAN_EXTS:
|
||
continue
|
||
try:
|
||
rel_parts = p.relative_to(root).parts
|
||
except ValueError:
|
||
continue
|
||
if any(part.startswith(".") for part in rel_parts):
|
||
continue
|
||
files.append(p)
|
||
return files
|
||
|
||
|
||
def _group_files_by_skill(root: Path, files: list[Path]) -> dict[tuple[str, str], list[Path]]:
|
||
"""Bucket scanned files under their owning ``(category, skill_name)``."""
|
||
|
||
buckets: dict[tuple[str, str], list[Path]] = {}
|
||
for p in files:
|
||
category, skill_name, _ = _skill_of(root, p)
|
||
buckets.setdefault((category, skill_name), []).append(p)
|
||
return buckets
|
||
|
||
|
||
def _occurrence_kind(value: str) -> str:
|
||
return infer_kind(value) or "url"
|
||
|
||
|
||
def _merge_occurrences(groups: dict[str, AddressGroup], pairs: list[tuple[str, OccurrenceModel]]) -> None:
|
||
"""Merge ``(address, occurrence)`` pairs into the aggregate group map."""
|
||
|
||
for address, occ in pairs:
|
||
g = groups.get(address)
|
||
if g is None:
|
||
g = AddressGroup(address=address, kind=_occurrence_kind(address), count=0, occurrences=[])
|
||
groups[address] = g
|
||
g.count += 1
|
||
g.occurrences.append(occ)
|
||
|
||
|
||
def _plan_apply(root: Path, mapping: dict[str, str]) -> tuple[list[tuple[Path, str, int, str]], int]:
|
||
"""Compute the rewrite plan without touching disk → (plan, current_total)."""
|
||
|
||
plan: list[tuple[Path, str, int, str]] = []
|
||
current_total = 0
|
||
for path in _iter_skill_files(root):
|
||
text = _read_text(path)
|
||
if text is None:
|
||
continue
|
||
new_text, cnt = apply_replacements(text, mapping, VALID_KINDS)
|
||
if cnt <= 0:
|
||
continue
|
||
if not _is_within(root, path):
|
||
raise HTTPException(status_code=400, detail=f"拒绝写入技能目录之外的文件：{path}")
|
||
_, _, rel = _skill_of(root, path)
|
||
plan.append((path, new_text, cnt, rel))
|
||
current_total += cnt
|
||
return plan, current_total
|
||
|
||
|
||
def _backup_file(root: Path, backup_id: str, path: Path, rel: str) -> None:
|
||
dest = root / _BACKUP_DIRNAME / backup_id / rel
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copy2(path, dest)
|
||
|
||
|
||
_MANIFEST_SUFFIX = ".manifest.json"
|
||
|
||
|
||
def _manifest_path(root: Path, backup_id: str) -> Path:
|
||
"""Manifest lives as a *sibling* of the backup dir (``{id}.manifest.json``),
|
||
not inside it — so it is neither restored back into the skills tree on
|
||
rollback nor counted as a backed-up file."""
|
||
|
||
return root / _BACKUP_DIRNAME / f"{backup_id}{_MANIFEST_SUFFIX}"
|
||
|
||
|
||
def _write_manifest(
|
||
root: Path,
|
||
backup_id: str,
|
||
norm: list[tuple[str, str, str]],
|
||
changed: list["ChangedFile"],
|
||
replaced_total: int,
|
||
created_at: str,
|
||
) -> None:
|
||
"""Record what this edit changed so the UI can show a meaningful, per-edit
|
||
history (原地址 → 新地址, 文件数, 处数) and roll a single record back."""
|
||
|
||
data = {
|
||
"created_at": created_at,
|
||
"replaced_total": replaced_total,
|
||
"replacements": [{"from": frm, "to": to, "kind": kind} for frm, to, kind in norm],
|
||
"changed_files": [{"file": c.file, "count": c.count} for c in changed],
|
||
}
|
||
try: # best-effort: a metadata write failure must not fail the apply itself
|
||
path = _manifest_path(root, backup_id)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
json.dump(data, fh, ensure_ascii=False)
|
||
except OSError:
|
||
logger.warning("Failed to write skill-address backup manifest for %s", backup_id, exc_info=True)
|
||
|
||
|
||
def _read_manifest(root: Path, backup_id: str) -> dict | None:
|
||
try:
|
||
with open(_manifest_path(root, backup_id), "r", encoding="utf-8") as fh:
|
||
return json.load(fh)
|
||
except (OSError, json.JSONDecodeError):
|
||
return None
|
||
|
||
|
||
# ── scan extraction shared by one-shot + stream ───────────────────────────────
|
||
|
||
|
||
def _scan_files_with_addresses(root: Path, files: list[Path], wanted: tuple[str, ...]) -> tuple[list[tuple[str, OccurrenceModel]], int]:
|
||
"""Scan files → ([(address, occurrence)], files_read)."""
|
||
|
||
pairs: list[tuple[str, OccurrenceModel]] = []
|
||
read = 0
|
||
for path in files:
|
||
text = _read_text(path)
|
||
if text is None:
|
||
continue
|
||
read += 1
|
||
category, skill_name, rel = _skill_of(root, path)
|
||
for o in scan_occurrences(text, wanted):
|
||
pairs.append(
|
||
(
|
||
o.value,
|
||
OccurrenceModel(skill=skill_name, category=category, file=rel, line=o.line, col=o.col, snippet=o.snippet),
|
||
)
|
||
)
|
||
return pairs, read
|
||
|
||
|
||
def _ordered_groups(groups: dict[str, AddressGroup]) -> list[AddressGroup]:
|
||
return sorted(groups.values(), key=lambda g: (-g.count, g.address))
|
||
|
||
|
||
def _build_mapping(replacements: list[Replacement]) -> tuple[dict[str, str], list[tuple[str, str, str]], list[str]]:
|
||
"""Validate replacements → ``(mapping, [(from,to,kind)], warnings)``.
|
||
|
||
Drops (with a warning) any entry that is empty, duplicated, not a recognized
|
||
URL/IP, a no-op (``from == to``), or whose ``to`` is malformed for its kind.
|
||
"""
|
||
|
||
mapping: dict[str, str] = {}
|
||
norm: list[tuple[str, str, str]] = []
|
||
warnings: list[str] = []
|
||
seen: set[str] = set()
|
||
for r in replacements:
|
||
frm = r.from_.strip()
|
||
to = r.to.strip()
|
||
if not frm:
|
||
continue
|
||
if frm in seen:
|
||
warnings.append(f"重复的原始地址已忽略：{frm}")
|
||
continue
|
||
kind = (r.kind or "").strip().lower() or infer_kind(frm)
|
||
if kind not in VALID_KINDS:
|
||
warnings.append(f"无法识别为 URL/IP，已忽略：{frm}")
|
||
continue
|
||
if to == frm:
|
||
warnings.append(f"替换值与原值相同，已忽略：{frm}")
|
||
continue
|
||
if not is_valid_replacement(kind, to):
|
||
warnings.append(f"替换值不是合法的 {kind}，已忽略：{to}")
|
||
continue
|
||
seen.add(frm)
|
||
mapping[frm] = to
|
||
norm.append((frm, to, kind))
|
||
return mapping, norm, warnings
|
||
|
||
|
||
def _ndjson(obj: dict) -> bytes:
|
||
return (json.dumps(obj, ensure_ascii=False) + "\n").encode("utf-8")
|
||
|
||
|
||
# ── routes: scan ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
@router.get("/scan", response_model=ScanResponse)
|
||
async def scan_addresses(
|
||
request: Request,
|
||
kinds: str | None = Query(default=None, description="逗号分隔：url,ip（默认全部）"),
|
||
skill: str | None = Query(default=None, description="仅扫描指定技能目录"),
|
||
) -> ScanResponse:
|
||
await _require_admin(request)
|
||
root = _skills_root()
|
||
wanted = normalize_kinds(kinds)
|
||
files = await asyncio.to_thread(_iter_skill_files, root, skill)
|
||
buckets = _group_files_by_skill(root, files)
|
||
groups: dict[str, AddressGroup] = {}
|
||
scanned = 0
|
||
for paths in buckets.values():
|
||
pairs, read = await asyncio.to_thread(_scan_files_with_addresses, root, paths, wanted)
|
||
_merge_occurrences(groups, pairs)
|
||
scanned += read
|
||
return ScanResponse(
|
||
groups=_ordered_groups(groups),
|
||
scanned_files=scanned,
|
||
scanned_skills=len(buckets),
|
||
kinds=list(wanted),
|
||
scanned_at=datetime.now(UTC).isoformat(),
|
||
)
|
||
|
||
|
||
@router.get("/scan/stream")
|
||
async def scan_addresses_stream(
|
||
request: Request,
|
||
kinds: str | None = Query(default=None),
|
||
skill: str | None = Query(default=None),
|
||
) -> StreamingResponse:
|
||
"""Scan with per-skill NDJSON progress.
|
||
|
||
Emits ``{"type":"start","total":N}`` then one
|
||
``{"type":"progress","done":i,"total":N,"skill":name}`` per skill, finally
|
||
``{"type":"result", ...ScanResponse...}``.
|
||
"""
|
||
|
||
await _require_admin(request)
|
||
root = _skills_root()
|
||
wanted = normalize_kinds(kinds)
|
||
|
||
async def gen() -> AsyncIterator[bytes]:
|
||
files = await asyncio.to_thread(_iter_skill_files, root, skill)
|
||
buckets = _group_files_by_skill(root, files)
|
||
total = len(buckets)
|
||
yield _ndjson({"type": "start", "total": total})
|
||
groups: dict[str, AddressGroup] = {}
|
||
scanned = 0
|
||
done = 0
|
||
for (category, skill_name), paths in buckets.items():
|
||
pairs, read = await asyncio.to_thread(_scan_files_with_addresses, root, paths, wanted)
|
||
_merge_occurrences(groups, pairs)
|
||
scanned += read
|
||
done += 1
|
||
yield _ndjson({"type": "progress", "done": done, "total": total, "skill": skill_name or category})
|
||
result = ScanResponse(
|
||
groups=_ordered_groups(groups),
|
||
scanned_files=scanned,
|
||
scanned_skills=total,
|
||
kinds=list(wanted),
|
||
scanned_at=datetime.now(UTC).isoformat(),
|
||
)
|
||
yield _ndjson({"type": "result", **result.model_dump()})
|
||
|
||
return StreamingResponse(gen(), media_type="application/x-ndjson")
|
||
|
||
|
||
# ── routes: preview ──────────────────────────────────────────────────────────
|
||
|
||
|
||
def _compute_preview(root: Path, mapping: dict[str, str], norm: list[tuple[str, str, str]]) -> tuple[list[PreviewItem], int]:
|
||
counts: dict[str, int] = {frm: 0 for frm, _, _ in norm}
|
||
files: dict[str, list[PreviewFileHit]] = {frm: [] for frm, _, _ in norm}
|
||
total = 0
|
||
if mapping:
|
||
for path in _iter_skill_files(root):
|
||
text = _read_text(path)
|
||
if text is None:
|
||
continue
|
||
_, _, rel = _skill_of(root, path)
|
||
for o in scan_occurrences(text, VALID_KINDS):
|
||
if o.value not in mapping:
|
||
continue
|
||
to = mapping[o.value]
|
||
counts[o.value] += 1
|
||
total += 1
|
||
files[o.value].append(PreviewFileHit(file=rel, line=o.line, before=o.snippet, after=o.snippet.replace(o.value, to)))
|
||
items = [PreviewItem(from_=frm, to=to, kind=kind, count=counts[frm], files=files[frm]) for frm, to, kind in norm]
|
||
return items, total
|
||
|
||
|
||
@router.post("/preview", response_model=PreviewResponse)
|
||
async def preview_replacements(request: Request, body: PreviewRequest) -> PreviewResponse:
|
||
await _require_admin(request)
|
||
root = _skills_root()
|
||
mapping, norm, warnings = _build_mapping(body.replacements)
|
||
items, total = await asyncio.to_thread(_compute_preview, root, mapping, norm)
|
||
return PreviewResponse(items=items, total=total, warnings=warnings, expected_total=total)
|
||
|
||
|
||
# ── routes: apply ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _new_backup_id() -> str:
|
||
return datetime.now(UTC).strftime("%Y%m%d-%H%M%S-%f")
|
||
|
||
|
||
@router.post("/apply", response_model=ApplyResponse)
|
||
async def apply_replacements_route(request: Request, body: ApplyRequest) -> ApplyResponse:
|
||
await _require_admin(request)
|
||
root = _skills_root()
|
||
mapping, norm, warnings = _build_mapping(body.replacements)
|
||
if not mapping:
|
||
return ApplyResponse(changed_files=[], replaced_total=0, backup_id=None, warnings=warnings)
|
||
|
||
plan, current_total = await asyncio.to_thread(_plan_apply, root, mapping)
|
||
if body.expected_total is not None and body.expected_total != current_total:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail=f"文件自上次扫描后已变化（预期 {body.expected_total} 处，现为 {current_total} 处），请重新扫描后再应用。",
|
||
)
|
||
if not plan:
|
||
return ApplyResponse(changed_files=[], replaced_total=0, backup_id=None, warnings=warnings)
|
||
|
||
backup_id: str | None = None
|
||
if body.backup:
|
||
backup_id = _new_backup_id()
|
||
for path, _new, _cnt, rel in plan:
|
||
await asyncio.to_thread(_backup_file, root, backup_id, path, rel)
|
||
|
||
changed: list[ChangedFile] = []
|
||
replaced_total = 0
|
||
for path, new_text, cnt, rel in plan:
|
||
await asyncio.to_thread(_write_text, path, new_text)
|
||
changed.append(ChangedFile(file=rel, count=cnt))
|
||
replaced_total += cnt
|
||
|
||
if backup_id:
|
||
await asyncio.to_thread(_write_manifest, root, backup_id, norm, changed, replaced_total, datetime.now(UTC).isoformat())
|
||
|
||
await _refresh_cache_safe()
|
||
logger.info("Skill address apply: %d files, %d replacements, backup=%s", len(changed), replaced_total, backup_id)
|
||
return ApplyResponse(changed_files=changed, replaced_total=replaced_total, backup_id=backup_id, warnings=warnings)
|
||
|
||
|
||
@router.post("/apply/stream")
|
||
async def apply_replacements_stream(request: Request, body: ApplyRequest) -> StreamingResponse:
|
||
"""Apply with per-file NDJSON progress.
|
||
|
||
Validation + the optimistic-lock check happen *before* streaming (so a 409
|
||
is a normal HTTP error). Then: ``{"type":"start","total_files":M,...}`` →
|
||
one ``{"type":"progress","done":k,...}`` per written file →
|
||
``{"type":"result", ...ApplyResponse...}``.
|
||
"""
|
||
|
||
await _require_admin(request)
|
||
root = _skills_root()
|
||
mapping, norm, warnings = _build_mapping(body.replacements)
|
||
|
||
plan: list[tuple[Path, str, int, str]] = []
|
||
current_total = 0
|
||
if mapping:
|
||
plan, current_total = await asyncio.to_thread(_plan_apply, root, mapping)
|
||
if body.expected_total is not None and body.expected_total != current_total:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail=f"文件自上次扫描后已变化（预期 {body.expected_total} 处，现为 {current_total} 处），请重新扫描后再应用。",
|
||
)
|
||
|
||
async def gen() -> AsyncIterator[bytes]:
|
||
if not plan:
|
||
result = ApplyResponse(changed_files=[], replaced_total=0, backup_id=None, warnings=warnings)
|
||
yield _ndjson({"type": "result", **result.model_dump()})
|
||
return
|
||
|
||
backup_id: str | None = None
|
||
if body.backup:
|
||
backup_id = _new_backup_id()
|
||
|
||
yield _ndjson({"type": "start", "total_files": len(plan), "total_replacements": current_total})
|
||
|
||
changed: list[ChangedFile] = []
|
||
replaced_total = 0
|
||
done = 0
|
||
for path, new_text, cnt, rel in plan:
|
||
if backup_id:
|
||
await asyncio.to_thread(_backup_file, root, backup_id, path, rel)
|
||
await asyncio.to_thread(_write_text, path, new_text)
|
||
changed.append(ChangedFile(file=rel, count=cnt))
|
||
replaced_total += cnt
|
||
done += 1
|
||
yield _ndjson({"type": "progress", "done": done, "total_files": len(plan), "file": rel, "count": cnt})
|
||
|
||
if backup_id:
|
||
await asyncio.to_thread(_write_manifest, root, backup_id, norm, changed, replaced_total, datetime.now(UTC).isoformat())
|
||
|
||
await _refresh_cache_safe()
|
||
logger.info("Skill address apply(stream): %d files, %d replacements, backup=%s", len(changed), replaced_total, backup_id)
|
||
result = ApplyResponse(changed_files=changed, replaced_total=replaced_total, backup_id=backup_id, warnings=warnings)
|
||
yield _ndjson({"type": "result", **result.model_dump()})
|
||
|
||
return StreamingResponse(gen(), media_type="application/x-ndjson")
|
||
|
||
|
||
# ── routes: backups / rollback ───────────────────────────────────────────────
|
||
|
||
|
||
def _list_backups(root: Path) -> list[BackupEntry]:
|
||
backup_root = root / _BACKUP_DIRNAME
|
||
entries: list[BackupEntry] = []
|
||
if backup_root.is_dir():
|
||
for d in backup_root.iterdir():
|
||
if not d.is_dir():
|
||
continue
|
||
file_count = sum(1 for p in d.rglob("*") if p.is_file())
|
||
manifest = _read_manifest(root, d.name)
|
||
created = ""
|
||
replaced_total = 0
|
||
replacements: list[ReplacementSummary] = []
|
||
if manifest:
|
||
created = str(manifest.get("created_at") or "")
|
||
replaced_total = int(manifest.get("replaced_total") or 0)
|
||
replacements = [
|
||
ReplacementSummary(from_=str(r.get("from", "")), to=str(r.get("to", "")), kind=str(r.get("kind", "")))
|
||
for r in manifest.get("replacements", [])
|
||
if isinstance(r, dict)
|
||
]
|
||
if not created:
|
||
try:
|
||
created = datetime.fromtimestamp(d.stat().st_mtime, UTC).isoformat()
|
||
except OSError:
|
||
created = ""
|
||
entries.append(
|
||
BackupEntry(
|
||
backup_id=d.name,
|
||
files=file_count,
|
||
created_at=created,
|
||
replaced_total=replaced_total,
|
||
replacements=replacements,
|
||
)
|
||
)
|
||
entries.sort(key=lambda e: e.backup_id, reverse=True)
|
||
return entries
|
||
|
||
|
||
@router.get("/backups", response_model=BackupListResponse)
|
||
async def list_backups(request: Request) -> BackupListResponse:
|
||
await _require_admin(request)
|
||
root = _skills_root()
|
||
entries = await asyncio.to_thread(_list_backups, root)
|
||
return BackupListResponse(backups=entries)
|
||
|
||
|
||
def _restore_backup(root: Path, backup_root: Path) -> list[str]:
|
||
restored: list[str] = []
|
||
for src in backup_root.rglob("*"):
|
||
if not src.is_file():
|
||
continue
|
||
rel = src.relative_to(backup_root)
|
||
dest = root / rel
|
||
if not _is_within(root, dest):
|
||
continue
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copy2(src, dest)
|
||
restored.append(rel.as_posix())
|
||
return restored
|
||
|
||
|
||
@router.post("/rollback/{backup_id}", response_model=RollbackResponse)
|
||
async def rollback(request: Request, backup_id: str) -> RollbackResponse:
|
||
await _require_admin(request)
|
||
root = _skills_root()
|
||
if "/" in backup_id or "\\" in backup_id or backup_id in (".", ".."):
|
||
raise HTTPException(status_code=400, detail="非法的备份编号")
|
||
backup_root = (root / _BACKUP_DIRNAME / backup_id).resolve()
|
||
if not _is_within(root, backup_root) or not backup_root.is_dir():
|
||
raise HTTPException(status_code=404, detail="备份不存在")
|
||
|
||
restored = await asyncio.to_thread(_restore_backup, root, backup_root)
|
||
await _refresh_cache_safe()
|
||
logger.info("Skill address rollback: backup=%s, restored %d files", backup_id, len(restored))
|
||
return RollbackResponse(backup_id=backup_id, restored_files=restored)
|
||
|
||
|
||
# ── misc ─────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
async def _refresh_cache_safe() -> None:
|
||
try:
|
||
await refresh_skills_system_prompt_cache_async()
|
||
except Exception: # noqa: BLE001 — cache refresh is best-effort
|
||
logger.warning("Failed to refresh skills prompt cache", exc_info=True)
|