2131 lines
78 KiB
Python
2131 lines
78 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only architecture inventory for AirShelf.
|
|
|
|
The scanner intentionally has no apply/write mode. It emits deterministic JSON to
|
|
stdout and limits all content reads to architecture-relevant, non-sensitive files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import base64
|
|
import fnmatch
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import zlib
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any, Callable, Iterable, Sequence
|
|
from urllib.parse import unquote
|
|
|
|
|
|
SCHEMA_VERSION = 1
|
|
MAX_TEXT_BYTES = 1_000_000
|
|
MAX_HASH_BYTES = 10_000_000
|
|
TARGET_DOCUMENTS = (
|
|
"core/backend/ARCHITECTURE.md",
|
|
"core/frontend/ARCHITECTURE.md",
|
|
)
|
|
GIT_SCOPES = ("core/backend", "core/frontend", "k8s/core")
|
|
RELEVANT_PATTERNS = (
|
|
"core/backend/airshelf/settings/*.py",
|
|
"core/backend/airshelf/urls.py",
|
|
"core/backend/apps/*/apps.py",
|
|
"core/backend/apps/*/models.py",
|
|
"core/backend/apps/*/urls.py",
|
|
"core/backend/apps/*/tasks.py",
|
|
"core/backend/apps/ai/providers/*.py",
|
|
"core/backend/requirements.txt",
|
|
"core/backend/Dockerfile",
|
|
"core/backend/docker-entrypoint.sh",
|
|
"k8s/core/*",
|
|
"k8s/core/**/*",
|
|
"core/frontend/package.json",
|
|
"core/frontend/src/App.tsx",
|
|
"core/frontend/src/api.ts",
|
|
"core/frontend/src/types.ts",
|
|
"core/frontend/src/routes/*",
|
|
"core/frontend/src/routes/**/*",
|
|
"core/frontend/src/components/*",
|
|
"core/frontend/src/components/**/*",
|
|
"core/frontend/vite.config.ts",
|
|
"core/frontend/Dockerfile",
|
|
"core/frontend/nginx.conf",
|
|
)
|
|
EXCLUDED_PARTS = frozenset(
|
|
{
|
|
".git",
|
|
".venv",
|
|
"venv",
|
|
"node_modules",
|
|
"dist",
|
|
"build",
|
|
".next",
|
|
"coverage",
|
|
"__pycache__",
|
|
".pytest_cache",
|
|
".mypy_cache",
|
|
".ruff_cache",
|
|
".cache",
|
|
"cache",
|
|
"media",
|
|
"uploads",
|
|
"user_uploads",
|
|
"staticfiles",
|
|
"ab_results",
|
|
}
|
|
)
|
|
SENSITIVE_NAMES = frozenset(
|
|
{
|
|
"account.md",
|
|
"credentials.json",
|
|
"service-account.json",
|
|
"secrets.json",
|
|
"id_rsa",
|
|
"id_ed25519",
|
|
}
|
|
)
|
|
SENSITIVE_SUFFIXES = (
|
|
".pem",
|
|
".key",
|
|
".p12",
|
|
".pfx",
|
|
".sqlite",
|
|
".sqlite3",
|
|
".db",
|
|
)
|
|
SYNC_MARKER_RE = re.compile(
|
|
r"<!--\s*architecture-sync-commit:\s*([0-9a-fA-F]{7,40})\s*-->"
|
|
)
|
|
HEADING_RE = re.compile(r"^(#{1,6})\s+(.+?)\s*$", re.MULTILINE)
|
|
TS_EXPORT_RE = re.compile(
|
|
r"\bexport\s+(?:default\s+)?(?:async\s+)?"
|
|
r"(?:const|let|var|function|class|type|interface|enum)\s+([A-Za-z_$][\w$]*)"
|
|
)
|
|
QUOTED_PATH_RE = re.compile(r"[\"'`](/[^\"'`\s]*)[\"'`]")
|
|
MARKDOWN_LINK_RE = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)")
|
|
MARKDOWN_REFERENCE_LINK_RE = re.compile(r"(?m)^\s*\[[^\]]+\]:\s*(\S+)")
|
|
ABSOLUTE_LOCAL_PATH_RE = re.compile(
|
|
r"(?i)(?:\b[A-Z]:[\\/]|file://|/(?:Users|home|var/folders|private/tmp)/)"
|
|
)
|
|
SECRET_CONTENT_PATTERNS = (
|
|
re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
|
|
re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b"),
|
|
re.compile(
|
|
r"(?i)\b(?:api[_-]?key|secret|password)\b\s*[:=]\s*[\"']?[A-Za-z0-9_./+=-]{12,}"
|
|
),
|
|
)
|
|
BACKEND_DOCUMENT = "core/backend/ARCHITECTURE.md"
|
|
FRONTEND_DOCUMENT = "core/frontend/ARCHITECTURE.md"
|
|
BACKEND_APP_SECTIONS = {
|
|
"common": "4.1 `common`",
|
|
"accounts": "4.2 `accounts`",
|
|
"products": "4.3 `products`",
|
|
"assets": "4.4 `assets`",
|
|
"projects": "4.5 `projects`",
|
|
"ai": "4.6 `ai`",
|
|
"billing": "4.7 `billing`",
|
|
"ops": "4.8 `ops`",
|
|
"adminpanel": "4.9 `adminpanel`",
|
|
}
|
|
|
|
|
|
class ScanError(RuntimeError):
|
|
"""A safe, user-reportable scanner failure."""
|
|
|
|
|
|
def normalize_path(value: str) -> str:
|
|
path = value.replace("\\", "/").removeprefix("./")
|
|
pure = PurePosixPath(path)
|
|
if not path or pure.is_absolute() or ".." in pure.parts:
|
|
return ""
|
|
return pure.as_posix()
|
|
|
|
|
|
def is_forbidden(path: str) -> bool:
|
|
normalized = normalize_path(path)
|
|
if not normalized:
|
|
return True
|
|
parts = tuple(part.lower() for part in PurePosixPath(normalized).parts)
|
|
name = parts[-1]
|
|
if any(part in EXCLUDED_PARTS for part in parts):
|
|
return True
|
|
if name == ".env" or name.startswith(".env."):
|
|
return True
|
|
if name in SENSITIVE_NAMES or name.endswith(SENSITIVE_SUFFIXES):
|
|
return True
|
|
return False
|
|
|
|
|
|
def is_relevant(path: str) -> bool:
|
|
normalized = normalize_path(path)
|
|
return bool(normalized) and not is_forbidden(normalized) and any(
|
|
fnmatch.fnmatchcase(normalized, pattern) for pattern in RELEVANT_PATTERNS
|
|
)
|
|
|
|
|
|
def selected_documents(scope: str) -> tuple[str, ...]:
|
|
if scope == "backend":
|
|
return (BACKEND_DOCUMENT,)
|
|
if scope == "frontend":
|
|
return (FRONTEND_DOCUMENT,)
|
|
return TARGET_DOCUMENTS
|
|
|
|
|
|
def path_in_scope(path: str, scope: str) -> bool:
|
|
if scope == "backend":
|
|
return path.startswith("core/backend/") or path.startswith("k8s/core/")
|
|
if scope == "frontend":
|
|
return path.startswith("core/frontend/")
|
|
return path.startswith(("core/backend/", "core/frontend/", "k8s/core/"))
|
|
|
|
|
|
def filter_entries(entries: Iterable[dict[str, str]], scope: str) -> list[dict[str, str]]:
|
|
return [entry for entry in entries if path_in_scope(entry["path"], scope)]
|
|
|
|
|
|
def safe_candidate(root: Path, relative: str) -> Path | None:
|
|
normalized = normalize_path(relative)
|
|
if not normalized or is_forbidden(normalized):
|
|
return None
|
|
candidate = root.joinpath(*PurePosixPath(normalized).parts)
|
|
try:
|
|
resolved_root = root.resolve(strict=True)
|
|
resolved = candidate.resolve(strict=False)
|
|
resolved.relative_to(resolved_root)
|
|
except (OSError, ValueError):
|
|
return None
|
|
current = candidate
|
|
while current != root:
|
|
if current.is_symlink():
|
|
return None
|
|
current = current.parent
|
|
return candidate
|
|
|
|
|
|
def read_bytes(root: Path, relative: str, warnings: set[str]) -> bytes | None:
|
|
candidate = safe_candidate(root, relative)
|
|
if candidate is None or not candidate.is_file():
|
|
return None
|
|
try:
|
|
size = candidate.stat().st_size
|
|
if size > MAX_TEXT_BYTES:
|
|
warnings.add(f"file_too_large:{normalize_path(relative)}")
|
|
return None
|
|
return candidate.read_bytes()
|
|
except OSError:
|
|
warnings.add(f"file_unreadable:{normalize_path(relative)}")
|
|
return None
|
|
|
|
|
|
def read_text(root: Path, relative: str, warnings: set[str]) -> str | None:
|
|
data = read_bytes(root, relative, warnings)
|
|
if data is None:
|
|
return None
|
|
try:
|
|
return data.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
warnings.add(f"file_not_utf8:{normalize_path(relative)}")
|
|
return None
|
|
|
|
|
|
def file_fingerprint(root: Path, relative: str, warnings: set[str]) -> str | None:
|
|
candidate = safe_candidate(root, relative)
|
|
if candidate is None or not candidate.is_file():
|
|
return None
|
|
try:
|
|
if candidate.stat().st_size > MAX_HASH_BYTES:
|
|
warnings.add(f"fingerprint_file_too_large:{normalize_path(relative)}")
|
|
return None
|
|
digest = hashlib.sha256()
|
|
with candidate.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(131_072), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
except OSError:
|
|
warnings.add(f"fingerprint_unavailable:{normalize_path(relative)}")
|
|
return None
|
|
|
|
|
|
def git(root: Path, arguments: Sequence[str], *, required: bool = True) -> bytes:
|
|
environment = os.environ.copy()
|
|
environment["GIT_OPTIONAL_LOCKS"] = "0"
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "-c", "core.quotepath=false", *arguments],
|
|
cwd=root,
|
|
env=environment,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
except OSError as error:
|
|
raise ScanError("git_unavailable") from error
|
|
if result.returncode and required:
|
|
raise ScanError(f"git_command_failed:{arguments[0]}")
|
|
return result.stdout if result.returncode == 0 else b""
|
|
|
|
|
|
def locate_repository(start: Path) -> Path:
|
|
output = git(start, ("rev-parse", "--show-toplevel"))
|
|
try:
|
|
root = Path(output.decode("utf-8").strip()).resolve(strict=True)
|
|
except (UnicodeDecodeError, OSError) as error:
|
|
raise ScanError("repository_root_unavailable") from error
|
|
if not root.is_dir():
|
|
raise ScanError("repository_root_unavailable")
|
|
return root
|
|
|
|
|
|
def decode_git_token(token: bytes) -> str:
|
|
return token.decode("utf-8", errors="replace")
|
|
|
|
|
|
def parse_name_status(
|
|
data: bytes, path_filter: Callable[[str], bool] = is_relevant
|
|
) -> list[dict[str, str]]:
|
|
tokens = [decode_git_token(token) for token in data.split(b"\0") if token]
|
|
entries: list[dict[str, str]] = []
|
|
index = 0
|
|
while index < len(tokens):
|
|
status = tokens[index]
|
|
index += 1
|
|
if index >= len(tokens):
|
|
break
|
|
if status.startswith(("R", "C")):
|
|
old_path = normalize_path(tokens[index])
|
|
index += 1
|
|
if index >= len(tokens):
|
|
break
|
|
new_path = normalize_path(tokens[index])
|
|
index += 1
|
|
if path_filter(new_path):
|
|
entry = {"path": new_path, "status": status}
|
|
if path_filter(old_path):
|
|
entry["old_path"] = old_path
|
|
entries.append(entry)
|
|
elif path_filter(old_path):
|
|
entries.append({"path": old_path, "status": "D"})
|
|
continue
|
|
path = normalize_path(tokens[index])
|
|
index += 1
|
|
if path_filter(path):
|
|
entries.append({"path": path, "status": status})
|
|
return sorted(entries, key=lambda item: (item["path"], item["status"]))
|
|
|
|
|
|
def diff_entries(root: Path, arguments: Sequence[str]) -> list[dict[str, str]]:
|
|
output = git(
|
|
root,
|
|
(*arguments, "--name-status", "--find-renames", "-z", "--", *GIT_SCOPES),
|
|
)
|
|
return parse_name_status(output)
|
|
|
|
|
|
def untracked_entries(root: Path) -> list[dict[str, str]]:
|
|
output = git(
|
|
root,
|
|
("ls-files", "--others", "--exclude-standard", "-z", "--", *GIT_SCOPES),
|
|
)
|
|
entries = []
|
|
for token in output.split(b"\0"):
|
|
if not token:
|
|
continue
|
|
path = normalize_path(decode_git_token(token))
|
|
if is_relevant(path):
|
|
entries.append({"path": path, "status": "??"})
|
|
return sorted(entries, key=lambda item: item["path"])
|
|
|
|
|
|
def all_diff_entries(root: Path, arguments: Sequence[str]) -> list[dict[str, str]]:
|
|
output = git(
|
|
root,
|
|
(*arguments, "--name-status", "--find-renames", "-z", "--"),
|
|
)
|
|
return parse_name_status(output, lambda path: bool(normalize_path(path)))
|
|
|
|
|
|
def all_untracked_entries(root: Path) -> list[dict[str, str]]:
|
|
output = git(root, ("ls-files", "--others", "--exclude-standard", "-z", "--"))
|
|
entries = []
|
|
for token in output.split(b"\0"):
|
|
if not token:
|
|
continue
|
|
path = normalize_path(decode_git_token(token))
|
|
if path:
|
|
entries.append({"path": path, "status": "??"})
|
|
return sorted(entries, key=lambda item: item["path"])
|
|
|
|
|
|
def worktree_guard(root: Path) -> dict[str, Any]:
|
|
groups = (
|
|
("staged", all_diff_entries(root, ("diff", "--cached"))),
|
|
("unstaged", all_diff_entries(root, ("diff",))),
|
|
("untracked", all_untracked_entries(root)),
|
|
)
|
|
states: dict[str, set[str]] = {}
|
|
forbidden_paths: set[str] = set()
|
|
for source, entries in groups:
|
|
for entry in entries:
|
|
paths = [entry["path"]]
|
|
if entry.get("old_path"):
|
|
paths.append(entry["old_path"])
|
|
for path in paths:
|
|
if is_forbidden(path):
|
|
forbidden_paths.add(path)
|
|
continue
|
|
states.setdefault(path, set()).add(f"{source}:{entry['status']}")
|
|
return {
|
|
"files": [
|
|
{"path": path, "states": sorted(values)}
|
|
for path, values in sorted(states.items())
|
|
],
|
|
"forbidden_change_count": len(forbidden_paths),
|
|
}
|
|
|
|
|
|
def verification_error(
|
|
code: str, message: str, suggestion: str, **details: Any
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"code": code,
|
|
"message": message,
|
|
"suggestion": suggestion,
|
|
**details,
|
|
}
|
|
|
|
|
|
def compare_worktree_guards(
|
|
before: dict[str, Any], current: dict[str, Any], allowed_documents: set[str]
|
|
) -> list[dict[str, Any]]:
|
|
errors: list[dict[str, Any]] = []
|
|
before_map = {item["path"]: item["states"] for item in before.get("files", [])}
|
|
current_map = {item["path"]: item["states"] for item in current.get("files", [])}
|
|
for path in sorted(set(before_map) | set(current_map)):
|
|
if path in allowed_documents:
|
|
continue
|
|
if before_map.get(path) != current_map.get(path):
|
|
errors.append(
|
|
verification_error(
|
|
"write_outside_allowlist",
|
|
"检测到计划外文件状态发生变化。",
|
|
"撤销本流程对该文件的修改，或重新生成包含正确范围的新计划。",
|
|
path=path,
|
|
)
|
|
)
|
|
if before.get("forbidden_change_count", 0) != current.get(
|
|
"forbidden_change_count", 0
|
|
):
|
|
errors.append(
|
|
verification_error(
|
|
"sensitive_or_excluded_path_changed",
|
|
"检测到敏感或排除路径的工作区状态发生变化。",
|
|
"停止文档同步并由用户单独检查敏感或排除文件。",
|
|
)
|
|
)
|
|
return errors
|
|
|
|
|
|
def current_head(root: Path) -> str | None:
|
|
output = git(root, ("rev-parse", "--verify", "HEAD"), required=False)
|
|
value = output.decode("ascii", errors="ignore").strip().lower()
|
|
return value if re.fullmatch(r"[0-9a-f]{40}", value) else None
|
|
|
|
|
|
def marker_is_usable(root: Path, marker: str, head: str | None) -> bool:
|
|
if head is None:
|
|
return False
|
|
exists = subprocess.run(
|
|
["git", "cat-file", "-e", f"{marker}^{{commit}}"],
|
|
cwd=root,
|
|
env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
).returncode == 0
|
|
if not exists:
|
|
return False
|
|
return subprocess.run(
|
|
["git", "merge-base", "--is-ancestor", marker, head],
|
|
cwd=root,
|
|
env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
).returncode == 0
|
|
|
|
|
|
def document_states(
|
|
root: Path, warnings: set[str], documents: Sequence[str] = TARGET_DOCUMENTS
|
|
) -> tuple[dict[str, dict[str, Any]], dict[str, str | None]]:
|
|
states: dict[str, dict[str, Any]] = {}
|
|
markers: dict[str, str | None] = {}
|
|
for path in documents:
|
|
data = read_bytes(root, path, warnings)
|
|
if data is None:
|
|
states[path] = {"exists": False, "headings": [], "sha256": None}
|
|
markers[path] = None
|
|
warnings.add(f"target_document_missing:{path}")
|
|
continue
|
|
try:
|
|
text = data.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
text = ""
|
|
warnings.add(f"target_document_not_utf8:{path}")
|
|
marker_match = SYNC_MARKER_RE.search(text)
|
|
marker = marker_match.group(1).lower() if marker_match else None
|
|
markers[path] = marker
|
|
headings = [
|
|
{"level": len(match.group(1)), "title": match.group(2).strip()}
|
|
for match in HEADING_RE.finditer(text)
|
|
]
|
|
states[path] = {
|
|
"exists": True,
|
|
"headings": headings,
|
|
"sha256": hashlib.sha256(data).hexdigest(),
|
|
"sync_commit": marker,
|
|
}
|
|
return states, markers
|
|
|
|
|
|
def walk_relevant_files(root: Path) -> list[str]:
|
|
roots = (
|
|
"core/backend/airshelf",
|
|
"core/backend/apps",
|
|
"core/backend/requirements.txt",
|
|
"core/backend/Dockerfile",
|
|
"core/backend/docker-entrypoint.sh",
|
|
"core/frontend/package.json",
|
|
"core/frontend/src",
|
|
"core/frontend/vite.config.ts",
|
|
"core/frontend/Dockerfile",
|
|
"core/frontend/nginx.conf",
|
|
"k8s/core",
|
|
)
|
|
paths: set[str] = set()
|
|
for relative in roots:
|
|
candidate = safe_candidate(root, relative)
|
|
if candidate is None or not candidate.exists():
|
|
continue
|
|
candidates: Iterable[Path] = (candidate,) if candidate.is_file() else candidate.rglob("*")
|
|
for item in candidates:
|
|
if not item.is_file() or item.is_symlink():
|
|
continue
|
|
try:
|
|
rel = item.relative_to(root).as_posix()
|
|
except ValueError:
|
|
continue
|
|
if is_relevant(rel):
|
|
paths.add(rel)
|
|
return sorted(paths)
|
|
|
|
|
|
def scope_content_fingerprint(
|
|
root: Path, scope: str, warnings: set[str]
|
|
) -> tuple[str, int]:
|
|
paths = [path for path in walk_relevant_files(root) if path_in_scope(path, scope)]
|
|
fingerprints = {path: file_fingerprint(root, path, warnings) for path in paths}
|
|
canonical = json.dumps(
|
|
fingerprints, ensure_ascii=True, sort_keys=True, separators=(",", ":")
|
|
)
|
|
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:20], len(paths)
|
|
|
|
|
|
def parse_python(root: Path, relative: str, warnings: set[str]) -> ast.Module | None:
|
|
text = read_text(root, relative, warnings)
|
|
if text is None:
|
|
return None
|
|
try:
|
|
return ast.parse(text, filename=relative)
|
|
except SyntaxError:
|
|
warnings.add(f"python_parse_failed:{relative}")
|
|
return None
|
|
|
|
|
|
def dotted_name(node: ast.AST) -> str:
|
|
if isinstance(node, ast.Name):
|
|
return node.id
|
|
if isinstance(node, ast.Attribute):
|
|
parent = dotted_name(node.value)
|
|
return f"{parent}.{node.attr}" if parent else node.attr
|
|
return ""
|
|
|
|
|
|
def model_classes(root: Path, relative: str, warnings: set[str]) -> list[str]:
|
|
tree = parse_python(root, relative, warnings)
|
|
if tree is None:
|
|
return []
|
|
classes = []
|
|
for node in tree.body:
|
|
if not isinstance(node, ast.ClassDef):
|
|
continue
|
|
bases = [dotted_name(base).split(".")[-1] for base in node.bases]
|
|
if any(base.lower().endswith("model") or base == "AbstractUser" for base in bases):
|
|
classes.append(node.name)
|
|
return sorted(classes)
|
|
|
|
|
|
def django_routes(root: Path, relative: str, warnings: set[str]) -> list[dict[str, str | None]]:
|
|
tree = parse_python(root, relative, warnings)
|
|
if tree is None:
|
|
return []
|
|
routes: list[dict[str, str | None]] = []
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call) or not node.args:
|
|
continue
|
|
if dotted_name(node.func).split(".")[-1] not in {"path", "re_path"}:
|
|
continue
|
|
first = node.args[0]
|
|
if not isinstance(first, ast.Constant) or not isinstance(first.value, str):
|
|
continue
|
|
included: str | None = None
|
|
if len(node.args) > 1 and isinstance(node.args[1], ast.Call):
|
|
call = node.args[1]
|
|
if dotted_name(call.func).split(".")[-1] == "include" and call.args:
|
|
value = call.args[0]
|
|
if isinstance(value, ast.Constant) and isinstance(value.value, str):
|
|
included = value.value
|
|
routes.append({"include": included, "path": first.value})
|
|
return sorted(routes, key=lambda item: (item["path"], item["include"] or ""))
|
|
|
|
|
|
def celery_tasks(root: Path, relative: str, warnings: set[str]) -> list[str]:
|
|
tree = parse_python(root, relative, warnings)
|
|
if tree is None:
|
|
return []
|
|
tasks = []
|
|
for node in tree.body:
|
|
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
continue
|
|
decorator_names = []
|
|
for decorator in node.decorator_list:
|
|
target = decorator.func if isinstance(decorator, ast.Call) else decorator
|
|
decorator_names.append(dotted_name(target).split(".")[-1])
|
|
if any(name in {"shared_task", "task", "periodic_task"} for name in decorator_names):
|
|
tasks.append(node.name)
|
|
return sorted(tasks)
|
|
|
|
|
|
def top_level_classes(root: Path, relative: str, warnings: set[str]) -> list[str]:
|
|
tree = parse_python(root, relative, warnings)
|
|
if tree is None:
|
|
return []
|
|
return sorted(node.name for node in tree.body if isinstance(node, ast.ClassDef))
|
|
|
|
|
|
def assignment_names(tree: ast.Module) -> set[str]:
|
|
names: set[str] = set()
|
|
for node in tree.body:
|
|
targets: list[ast.AST] = []
|
|
if isinstance(node, ast.Assign):
|
|
targets.extend(node.targets)
|
|
elif isinstance(node, ast.AnnAssign):
|
|
targets.append(node.target)
|
|
for target in targets:
|
|
if isinstance(target, ast.Name):
|
|
names.add(target.id)
|
|
return names
|
|
|
|
|
|
def requirement_names(root: Path, warnings: set[str]) -> list[str]:
|
|
text = read_text(root, "core/backend/requirements.txt", warnings)
|
|
if text is None:
|
|
return []
|
|
names = set()
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith(("#", "-")):
|
|
continue
|
|
match = re.match(r"([A-Za-z0-9_.-]+)", stripped)
|
|
if match:
|
|
names.add(match.group(1).lower())
|
|
return sorted(names)
|
|
|
|
|
|
def collect_backend(root: Path, warnings: set[str]) -> dict[str, Any]:
|
|
apps_root = root / "core" / "backend" / "apps"
|
|
apps = []
|
|
if apps_root.is_dir() and not apps_root.is_symlink():
|
|
apps = sorted(
|
|
item.name
|
|
for item in apps_root.iterdir()
|
|
if item.is_dir()
|
|
and not item.is_symlink()
|
|
and not is_forbidden(f"core/backend/apps/{item.name}")
|
|
and ((item / "apps.py").is_file() or (item / "__init__.py").is_file())
|
|
)
|
|
|
|
models: dict[str, list[str]] = {}
|
|
app_routes: dict[str, list[dict[str, str | None]]] = {}
|
|
tasks: dict[str, list[str]] = {}
|
|
for app in apps:
|
|
model_path = f"core/backend/apps/{app}/models.py"
|
|
route_path = f"core/backend/apps/{app}/urls.py"
|
|
task_path = f"core/backend/apps/{app}/tasks.py"
|
|
if safe_candidate(root, model_path) and (root / model_path).is_file():
|
|
models[app] = model_classes(root, model_path, warnings)
|
|
if safe_candidate(root, route_path) and (root / route_path).is_file():
|
|
app_routes[app] = django_routes(root, route_path, warnings)
|
|
if safe_candidate(root, task_path) and (root / task_path).is_file():
|
|
tasks[app] = celery_tasks(root, task_path, warnings)
|
|
|
|
provider_root = root / "core" / "backend" / "apps" / "ai" / "providers"
|
|
providers = []
|
|
if provider_root.is_dir() and not provider_root.is_symlink():
|
|
for file in sorted(provider_root.glob("*.py"), key=lambda item: item.name):
|
|
if file.name == "__init__.py" or file.is_symlink():
|
|
continue
|
|
relative = file.relative_to(root).as_posix()
|
|
providers.append(
|
|
{"classes": top_level_classes(root, relative, warnings), "path": relative}
|
|
)
|
|
|
|
settings = {}
|
|
settings_root = root / "core" / "backend" / "airshelf" / "settings"
|
|
if settings_root.is_dir() and not settings_root.is_symlink():
|
|
important = {
|
|
"AUTHENTICATION_BACKENDS",
|
|
"AUTH_USER_MODEL",
|
|
"CACHES",
|
|
"DATABASES",
|
|
"INSTALLED_APPS",
|
|
"MIDDLEWARE",
|
|
"REST_FRAMEWORK",
|
|
"STORAGES",
|
|
}
|
|
prefixes = ("AI_", "ARK_", "CELERY_", "MODEL_", "REDIS_", "TOKENSSR_", "TOS_", "YUNQI_")
|
|
for file in sorted(settings_root.glob("*.py"), key=lambda item: item.name):
|
|
if file.name == "__init__.py" or file.is_symlink():
|
|
continue
|
|
relative = file.relative_to(root).as_posix()
|
|
tree = parse_python(root, relative, warnings)
|
|
symbols = [] if tree is None else sorted(
|
|
name for name in assignment_names(tree) if name in important or name.startswith(prefixes)
|
|
)
|
|
settings[file.stem] = symbols
|
|
|
|
base_settings = read_text(root, "core/backend/airshelf/settings/base.py", warnings) or ""
|
|
configured_apps = sorted(set(re.findall(r"[\"']apps\.([a-zA-Z0-9_]+)", base_settings)))
|
|
deployment_candidates = (
|
|
"core/backend/Dockerfile",
|
|
"core/backend/docker-entrypoint.sh",
|
|
)
|
|
deployment_files = [
|
|
path for path in deployment_candidates if (safe_candidate(root, path) or Path()).is_file()
|
|
]
|
|
k8s_files = [path for path in walk_relevant_files(root) if path.startswith("k8s/core/")]
|
|
return {
|
|
"api_routes": django_routes(root, "core/backend/airshelf/urls.py", warnings),
|
|
"app_routes": app_routes,
|
|
"apps": apps,
|
|
"celery_tasks": tasks,
|
|
"configured_apps": configured_apps,
|
|
"deployment_files": deployment_files + k8s_files,
|
|
"models": models,
|
|
"providers": providers,
|
|
"requirements": requirement_names(root, warnings),
|
|
"settings_symbols": settings,
|
|
}
|
|
|
|
|
|
def relative_source_files(root: Path, base: str, suffixes: tuple[str, ...]) -> list[str]:
|
|
directory = safe_candidate(root, base)
|
|
if directory is None or not directory.is_dir() or directory.is_symlink():
|
|
return []
|
|
files = []
|
|
for item in directory.rglob("*"):
|
|
if not item.is_file() or item.is_symlink() or item.suffix.lower() not in suffixes:
|
|
continue
|
|
relative = item.relative_to(root).as_posix()
|
|
if not is_forbidden(relative):
|
|
files.append(relative)
|
|
return sorted(files)
|
|
|
|
|
|
def typescript_exports(root: Path, relative: str, warnings: set[str]) -> list[str]:
|
|
text = read_text(root, relative, warnings)
|
|
return [] if text is None else sorted(set(TS_EXPORT_RE.findall(text)))
|
|
|
|
|
|
def quoted_paths(root: Path, relative: str, warnings: set[str]) -> list[str]:
|
|
text = read_text(root, relative, warnings)
|
|
return [] if text is None else sorted(set(QUOTED_PATH_RE.findall(text)))
|
|
|
|
|
|
def stage_order(root: Path, warnings: set[str]) -> list[str]:
|
|
text = read_text(root, "core/frontend/src/routes/stage-config.ts", warnings)
|
|
if text is None:
|
|
return []
|
|
match = re.search(r"\bstageOrder\s*=\s*\[([^\]]*)\]", text, re.DOTALL)
|
|
if not match:
|
|
return []
|
|
return re.findall(r"[\"']([A-Za-z0-9_-]+)[\"']", match.group(1))
|
|
|
|
|
|
def frontend_dependencies(root: Path, warnings: set[str]) -> dict[str, list[str]]:
|
|
text = read_text(root, "core/frontend/package.json", warnings)
|
|
if text is None:
|
|
return {"dependencies": [], "devDependencies": []}
|
|
try:
|
|
package = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
warnings.add("json_parse_failed:core/frontend/package.json")
|
|
return {"dependencies": [], "devDependencies": []}
|
|
return {
|
|
group: sorted(str(name) for name in package.get(group, {}).keys())
|
|
for group in ("dependencies", "devDependencies")
|
|
}
|
|
|
|
|
|
def collect_frontend(root: Path, warnings: set[str]) -> dict[str, Any]:
|
|
deployment_candidates = (
|
|
"core/frontend/Dockerfile",
|
|
"core/frontend/nginx.conf",
|
|
"core/frontend/vite.config.ts",
|
|
)
|
|
deployment_files = [
|
|
path for path in deployment_candidates if (safe_candidate(root, path) or Path()).is_file()
|
|
]
|
|
return {
|
|
"api_endpoints": quoted_paths(root, "core/frontend/src/api.ts", warnings),
|
|
"api_exports": typescript_exports(root, "core/frontend/src/api.ts", warnings),
|
|
"component_files": relative_source_files(
|
|
root, "core/frontend/src/components", (".ts", ".tsx")
|
|
),
|
|
"dependencies": frontend_dependencies(root, warnings),
|
|
"deployment_files": deployment_files,
|
|
"route_files": relative_source_files(root, "core/frontend/src/routes", (".ts", ".tsx")),
|
|
"route_paths": quoted_paths(root, "core/frontend/src/routes/route-config.ts", warnings),
|
|
"stage_order": stage_order(root, warnings),
|
|
"type_exports": typescript_exports(root, "core/frontend/src/types.ts", warnings),
|
|
}
|
|
|
|
|
|
def changed_line_text(root: Path, base: str, path: str, warnings: set[str]) -> list[str]:
|
|
if not is_relevant(path):
|
|
return []
|
|
output = git(
|
|
root,
|
|
("diff", "--no-ext-diff", "--unified=0", base, "--", path),
|
|
required=False,
|
|
)
|
|
if not output:
|
|
current = read_text(root, path, warnings)
|
|
return [] if current is None else [f"+{line}" for line in current.splitlines()]
|
|
if len(output) > 2_000_000:
|
|
warnings.add(f"diff_too_large:{path}")
|
|
output = output[:2_000_000]
|
|
text = output.decode("utf-8", errors="replace")
|
|
return [
|
|
line
|
|
for line in text.splitlines()
|
|
if line.startswith(("+", "-")) and not line.startswith(("+++", "---"))
|
|
]
|
|
|
|
|
|
def text_at_commit(root: Path, commit: str, path: str, warnings: set[str]) -> str | None:
|
|
if not is_relevant(path):
|
|
return None
|
|
output = git(root, ("show", f"{commit}:{path}"), required=False)
|
|
if not output:
|
|
return None
|
|
if len(output) > MAX_TEXT_BYTES:
|
|
warnings.add(f"base_file_too_large:{path}")
|
|
return None
|
|
try:
|
|
return output.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
warnings.add(f"base_file_not_utf8:{path}")
|
|
return None
|
|
|
|
|
|
def change_index(entries: Iterable[dict[str, str]]) -> dict[str, set[str]]:
|
|
indexed: dict[str, set[str]] = {}
|
|
for entry in entries:
|
|
status = entry["status"]
|
|
code = "A" if status == "??" else status[:1]
|
|
path = entry["path"]
|
|
if path:
|
|
indexed.setdefault(path, set()).add(code)
|
|
old_path = entry.get("old_path")
|
|
if old_path:
|
|
indexed.setdefault(old_path, set()).add("D")
|
|
indexed[path].add("A")
|
|
return indexed
|
|
|
|
|
|
def dependency_names_from_package(text: str | None) -> set[str] | None:
|
|
if text is None:
|
|
return set()
|
|
try:
|
|
package = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
names: set[str] = set()
|
|
for group in ("dependencies", "devDependencies"):
|
|
values = package.get(group, {})
|
|
if isinstance(values, dict):
|
|
names.update(str(name) for name in values)
|
|
return names
|
|
|
|
|
|
def dependency_names_from_requirements(text: str | None) -> set[str]:
|
|
if text is None:
|
|
return set()
|
|
names = set()
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith(("#", "-")):
|
|
continue
|
|
match = re.match(r"([A-Za-z0-9_.-]+)", stripped)
|
|
if match:
|
|
names.add(match.group(1).lower())
|
|
return names
|
|
|
|
|
|
def dependency_set_changed(
|
|
root: Path, base: str, path: str, warnings: set[str]
|
|
) -> bool:
|
|
before = text_at_commit(root, base, path, warnings)
|
|
after = read_text(root, path, warnings)
|
|
if path.endswith("package.json"):
|
|
before_names = dependency_names_from_package(before)
|
|
after_names = dependency_names_from_package(after)
|
|
if before_names is None or after_names is None:
|
|
warnings.add(f"dependency_parse_failed:{path}")
|
|
return True
|
|
return before_names != after_names
|
|
return dependency_names_from_requirements(before) != dependency_names_from_requirements(after)
|
|
|
|
|
|
def stage_order_from_text(text: str | None) -> list[str]:
|
|
if text is None:
|
|
return []
|
|
match = re.search(r"\bstageOrder\s*=\s*\[([^\]]*)\]", text, re.DOTALL)
|
|
if not match:
|
|
return []
|
|
return re.findall(r"[\"']([A-Za-z0-9_-]+)[\"']", match.group(1))
|
|
|
|
|
|
def stage_order_changed(root: Path, base: str, warnings: set[str]) -> bool:
|
|
path = "core/frontend/src/routes/stage-config.ts"
|
|
before = text_at_commit(root, base, path, warnings)
|
|
after = read_text(root, path, warnings)
|
|
return stage_order_from_text(before) != stage_order_from_text(after)
|
|
|
|
|
|
def frontend_page_section(path: str) -> str:
|
|
if "/admin/" in path:
|
|
return "8.7 平台超管后台"
|
|
stem = PurePosixPath(path).stem
|
|
sections = {
|
|
"products": "8.1 商品",
|
|
"product-detail": "8.1 商品",
|
|
"product-create": "8.1 商品",
|
|
"projects": "8.2 视频项目",
|
|
"pipeline": "8.3 生产管线",
|
|
"ai-tools": "8.4 图片创作",
|
|
"models": "8.4 图片创作",
|
|
"free-create": "8.5 自由视频创作",
|
|
"library": "8.6 资产库与垃圾桶",
|
|
"trash": "8.6 资产库与垃圾桶",
|
|
}
|
|
return sections.get(stem, "8. 页面模块")
|
|
|
|
|
|
def impact_spec(
|
|
path: str,
|
|
statuses: set[str],
|
|
changed_lines: Sequence[str],
|
|
*,
|
|
dependency_changed: bool = False,
|
|
stages_changed: bool = False,
|
|
) -> dict[str, str] | None:
|
|
changed = "\n".join(changed_lines)
|
|
added_or_deleted = bool(statuses & {"A", "D", "R", "C"})
|
|
|
|
if path.startswith("core/backend/") or path.startswith("k8s/core/"):
|
|
document = BACKEND_DOCUMENT
|
|
app_match = re.match(r"core/backend/apps/([^/]+)/([^/]+\.py)$", path)
|
|
app = app_match.group(1) if app_match else ""
|
|
filename = app_match.group(2) if app_match else ""
|
|
app_section = BACKEND_APP_SECTIONS.get(app, "2. 目录结构、4. 应用模块")
|
|
|
|
if path.startswith("core/backend/airshelf/settings/") and re.search(
|
|
r"(?i)(INSTALLED_APPS|MIDDLEWARE|DATABASE|DB_ENGINE|\bENGINE\b|CACHE|REDIS|CELERY|"
|
|
r"REST_FRAMEWORK|AUTH_|Authentication|Permission|STORAGES|TOS_|ARK_|"
|
|
r"YUNQI_|TOKENSSR_|MODEL_|AI_|apps\.)",
|
|
changed,
|
|
):
|
|
return {
|
|
"change_type": "backend.settings_boundary",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "后端框架、基础设施或鉴权配置边界发生变化",
|
|
"section": "3.1 Settings、3.3 鉴权",
|
|
}
|
|
if path == "core/backend/airshelf/urls.py" and (
|
|
added_or_deleted or re.search(r"\b(path|re_path|include)\s*\(", changed)
|
|
):
|
|
return {
|
|
"change_type": "backend.root_routes",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "后端一级路由入口发生变化",
|
|
"section": "3.2 路由",
|
|
}
|
|
if filename == "apps.py" and (
|
|
added_or_deleted or re.search(r"(?m)^[+-].*\bname\s*=", changed)
|
|
):
|
|
return {
|
|
"change_type": "backend.django_apps",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": f"Django 应用 `{app}` 的注册结构发生变化",
|
|
"section": "2. 目录结构、4. 应用模块",
|
|
}
|
|
if filename == "models.py" and (
|
|
added_or_deleted
|
|
or re.search(r"(?m)^[+-]\s*class\s+\w+\s*\([^\n]*(?:Model|AbstractUser)", changed)
|
|
):
|
|
return {
|
|
"change_type": "backend.model_classes",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": f"`{app}` 核心模型类型发生增删或继承变化",
|
|
"section": app_section,
|
|
}
|
|
if filename == "urls.py" and (
|
|
added_or_deleted or re.search(r"\b(path|re_path|include)\s*\(", changed)
|
|
):
|
|
return {
|
|
"change_type": "backend.app_routes",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": f"`{app}` 应用接口边界发生变化",
|
|
"section": app_section,
|
|
}
|
|
if filename == "tasks.py" and (
|
|
added_or_deleted or re.search(r"(shared_task|periodic_task|\.task\s*\()", changed)
|
|
):
|
|
return {
|
|
"change_type": "backend.celery_tasks",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": f"`{app}` 的 Celery 任务入口发生变化",
|
|
"section": "7.1 Celery",
|
|
}
|
|
if path.startswith("core/backend/apps/ai/providers/") and (
|
|
added_or_deleted
|
|
or re.search(r"(?m)^[+-]\s*class\s+", changed)
|
|
or (
|
|
PurePosixPath(path).name == "__init__.py"
|
|
and re.search(r"(?m)^[+-]\s*(?:from|import)\s+", changed)
|
|
)
|
|
):
|
|
return {
|
|
"change_type": "backend.ai_providers",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "AI Provider 类型或适配层结构发生变化",
|
|
"section": "6. AI Provider 架构",
|
|
}
|
|
if path == "core/backend/requirements.txt" and dependency_changed:
|
|
return {
|
|
"change_type": "backend.dependencies",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "后端依赖集合发生变化",
|
|
"section": "10. 部署",
|
|
}
|
|
if path in {"core/backend/Dockerfile", "core/backend/docker-entrypoint.sh"} or path.startswith(
|
|
"k8s/core/"
|
|
):
|
|
if changed_lines or added_or_deleted:
|
|
return {
|
|
"change_type": "backend.deployment",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "后端容器或部署拓扑发生变化",
|
|
"section": "10. 部署",
|
|
}
|
|
return None
|
|
|
|
if path.startswith("core/frontend/"):
|
|
document = FRONTEND_DOCUMENT
|
|
if path == "core/frontend/package.json" and dependency_changed:
|
|
return {
|
|
"change_type": "frontend.dependencies",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "前端依赖集合发生变化",
|
|
"section": "2. 技术栈",
|
|
}
|
|
if path == "core/frontend/src/routes/route-config.ts" and (
|
|
added_or_deleted
|
|
or re.search(r"(?m)^[+-].*(?:[\"'`]/|type\s+Page|mainNav|supplementNav|pathForPage)", changed)
|
|
):
|
|
return {
|
|
"change_type": "frontend.routing",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "前端页面路由或导航边界发生变化",
|
|
"section": "5. 路由架构",
|
|
}
|
|
if path == "core/frontend/src/routes/stage-config.ts" and stages_changed:
|
|
return {
|
|
"change_type": "frontend.pipeline_stages",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "五阶段顺序或阶段标识发生变化",
|
|
"section": "8.3 生产管线",
|
|
}
|
|
if path.startswith("core/frontend/src/routes/") and PurePosixPath(path).suffix in {
|
|
".ts",
|
|
".tsx",
|
|
} and added_or_deleted:
|
|
return {
|
|
"change_type": "frontend.pages",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "正式业务页面模块发生增删或移动",
|
|
"section": frontend_page_section(path),
|
|
}
|
|
if path == "core/frontend/src/App.tsx" and re.search(
|
|
r"(?i)(createContext|Context\.Provider|useReducer|redux|zustand|RouterProvider|BrowserRouter)",
|
|
changed,
|
|
):
|
|
return {
|
|
"change_type": "frontend.app_controller",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "应用总控制器或全局状态边界发生变化",
|
|
"section": "4.2 `App.tsx`",
|
|
}
|
|
if path == "core/frontend/src/api.ts" and re.search(
|
|
r"(?i)(VITE_API_BASE_URL|Authorization|localStorage|sessionStorage|fetch\s*\(|"
|
|
r"axios|(?:const|function)\s+request\b|export\s+const\s+\w*Api\b)",
|
|
changed,
|
|
):
|
|
return {
|
|
"change_type": "frontend.api_boundary",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "前端请求、鉴权或 API SDK 边界发生变化",
|
|
"section": "6. API 层、7. 登录、存储与权限",
|
|
}
|
|
if path.startswith("core/frontend/src/components/") and added_or_deleted and re.search(
|
|
r"(?i)(shell|layout|router|provider|store|context)", PurePosixPath(path).stem
|
|
):
|
|
return {
|
|
"change_type": "frontend.shared_infrastructure",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "前端共享基础组件发生增删或移动",
|
|
"section": "3. 目录结构、4. 应用入口",
|
|
}
|
|
if path in {
|
|
"core/frontend/Dockerfile",
|
|
"core/frontend/nginx.conf",
|
|
"core/frontend/vite.config.ts",
|
|
} and (changed_lines or added_or_deleted):
|
|
return {
|
|
"change_type": "frontend.deployment",
|
|
"classification": "architecture",
|
|
"document": document,
|
|
"reason": "前端构建、代理或容器部署边界发生变化",
|
|
"section": "11. 构建与部署",
|
|
}
|
|
return None
|
|
|
|
|
|
def baseline_impacts(document: str, baseline_files: Sequence[str]) -> list[dict[str, Any]]:
|
|
if document == BACKEND_DOCUMENT:
|
|
groups = (
|
|
(
|
|
"backend.baseline.apps",
|
|
"2. 目录结构、4. 应用模块",
|
|
"缺少同步标记，需要核对 Django 应用与核心模型基线",
|
|
lambda path: "/apps/" in path and path.endswith(("apps.py", "models.py", "urls.py")),
|
|
),
|
|
(
|
|
"backend.baseline.entry",
|
|
"3. Django 工程入口",
|
|
"缺少同步标记，需要核对 Settings 与一级路由基线",
|
|
lambda path: "/airshelf/settings/" in path or path.endswith("airshelf/urls.py"),
|
|
),
|
|
(
|
|
"backend.baseline.execution",
|
|
"6. AI Provider 架构、7. 任务执行模型",
|
|
"缺少同步标记，需要核对 Provider 与 Celery 任务基线",
|
|
lambda path: "/providers/" in path or path.endswith("tasks.py"),
|
|
),
|
|
(
|
|
"backend.baseline.deployment",
|
|
"10. 部署",
|
|
"缺少同步标记，需要核对依赖、容器与部署基线",
|
|
lambda path: path.endswith(("requirements.txt", "Dockerfile", "docker-entrypoint.sh"))
|
|
or path.startswith("k8s/core/"),
|
|
),
|
|
)
|
|
else:
|
|
groups = (
|
|
(
|
|
"frontend.baseline.stack",
|
|
"2. 技术栈",
|
|
"缺少同步标记，需要核对前端技术依赖基线",
|
|
lambda path: path.endswith("package.json"),
|
|
),
|
|
(
|
|
"frontend.baseline.structure",
|
|
"3. 目录结构、5. 路由架构、8. 页面模块",
|
|
"缺少同步标记，需要核对页面、路由与共享组件基线",
|
|
lambda path: "/src/routes/" in path or "/src/components/" in path,
|
|
),
|
|
(
|
|
"frontend.baseline.boundaries",
|
|
"4. 应用入口、6. API 层",
|
|
"缺少同步标记，需要核对应用入口与 API SDK 基线",
|
|
lambda path: path.endswith(("src/App.tsx", "src/api.ts", "src/types.ts")),
|
|
),
|
|
(
|
|
"frontend.baseline.deployment",
|
|
"11. 构建与部署",
|
|
"缺少同步标记，需要核对构建、代理与容器基线",
|
|
lambda path: path.endswith(("Dockerfile", "nginx.conf", "vite.config.ts")),
|
|
),
|
|
)
|
|
impacts = []
|
|
for change_type, section, reason, matches in groups:
|
|
evidence = sorted(path for path in baseline_files if matches(path))[:24]
|
|
if evidence:
|
|
impacts.append(
|
|
{
|
|
"change_type": change_type,
|
|
"classification": "baseline_review",
|
|
"document": document,
|
|
"evidence": evidence,
|
|
"reason": reason,
|
|
"section": section,
|
|
}
|
|
)
|
|
return impacts
|
|
|
|
|
|
def map_impacts(
|
|
root: Path,
|
|
head: str | None,
|
|
documents: Sequence[str],
|
|
usable_markers: dict[str, str],
|
|
baseline_files: Sequence[str],
|
|
staged: Sequence[dict[str, str]],
|
|
unstaged: Sequence[dict[str, str]],
|
|
untracked: Sequence[dict[str, str]],
|
|
warnings: set[str],
|
|
) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
|
|
impacts: list[dict[str, Any]] = []
|
|
ordinary: dict[str, str] = {}
|
|
for document in documents:
|
|
base = usable_markers.get(document)
|
|
if base is None or head is None:
|
|
impacts.extend(baseline_impacts(document, baseline_files))
|
|
continue
|
|
committed = diff_entries(root, ("diff", f"{base}..{head}"))
|
|
indexed = change_index([*committed, *staged, *unstaged, *untracked])
|
|
if document == BACKEND_DOCUMENT:
|
|
domain_paths = sorted(
|
|
path
|
|
for path in indexed
|
|
if path.startswith("core/backend/") or path.startswith("k8s/core/")
|
|
)
|
|
else:
|
|
domain_paths = sorted(
|
|
path for path in indexed if path.startswith("core/frontend/")
|
|
)
|
|
for path in domain_paths:
|
|
lines = changed_line_text(root, base, path, warnings)
|
|
dependency_changed = path in {
|
|
"core/backend/requirements.txt",
|
|
"core/frontend/package.json",
|
|
} and dependency_set_changed(root, base, path, warnings)
|
|
stages_changed = path == "core/frontend/src/routes/stage-config.ts" and stage_order_changed(
|
|
root, base, warnings
|
|
)
|
|
spec = impact_spec(
|
|
path,
|
|
indexed[path],
|
|
lines,
|
|
dependency_changed=dependency_changed,
|
|
stages_changed=stages_changed,
|
|
)
|
|
if spec is None:
|
|
ordinary[path] = "未发现会改变架构文档事实的结构信号"
|
|
continue
|
|
impacts.append({**spec, "evidence": [path]})
|
|
|
|
buckets: dict[tuple[str, str, str, str, str], set[str]] = {}
|
|
for impact in impacts:
|
|
key = (
|
|
impact["document"],
|
|
impact["section"],
|
|
impact["change_type"],
|
|
impact["classification"],
|
|
impact["reason"],
|
|
)
|
|
buckets.setdefault(key, set()).update(impact["evidence"])
|
|
merged = [
|
|
{
|
|
"change_type": key[2],
|
|
"classification": key[3],
|
|
"document": key[0],
|
|
"evidence": sorted(evidence),
|
|
"reason": key[4],
|
|
"section": key[1],
|
|
}
|
|
for key, evidence in buckets.items()
|
|
]
|
|
merged.sort(key=lambda item: (item["document"], item["section"], item["change_type"]))
|
|
non_architecture = [
|
|
{"path": path, "reason": reason} for path, reason in sorted(ordinary.items())
|
|
]
|
|
return merged, non_architecture
|
|
|
|
|
|
def encode_guard(payload: dict[str, Any]) -> str:
|
|
raw = json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode(
|
|
"utf-8"
|
|
)
|
|
encoded = base64.urlsafe_b64encode(zlib.compress(raw, level=9)).decode("ascii").rstrip("=")
|
|
checksum = hashlib.sha256(raw).hexdigest()[:16]
|
|
return f"{encoded}.{checksum}"
|
|
|
|
|
|
def decode_guard(token: str) -> dict[str, Any]:
|
|
try:
|
|
encoded, checksum = token.rsplit(".", 1)
|
|
padding = "=" * (-len(encoded) % 4)
|
|
raw = zlib.decompress(base64.urlsafe_b64decode(encoded + padding))
|
|
if hashlib.sha256(raw).hexdigest()[:16] != checksum:
|
|
raise ValueError("checksum")
|
|
payload = json.loads(raw.decode("utf-8"))
|
|
except (ValueError, UnicodeDecodeError, json.JSONDecodeError, zlib.error) as error:
|
|
raise ScanError("verification_guard_invalid") from error
|
|
if not isinstance(payload, dict) or payload.get("guard_version") != 1:
|
|
raise ScanError("verification_guard_invalid")
|
|
return payload
|
|
|
|
|
|
def headings_from_text(text: str, maximum_level: int = 3) -> list[dict[str, Any]]:
|
|
return [
|
|
{"level": len(match.group(1)), "title": match.group(2).strip()}
|
|
for match in HEADING_RE.finditer(text)
|
|
if len(match.group(1)) <= maximum_level
|
|
]
|
|
|
|
|
|
def section_records(text: str) -> list[dict[str, Any]]:
|
|
matches = [
|
|
match for match in HEADING_RE.finditer(text) if len(match.group(1)) <= 3
|
|
]
|
|
records: list[dict[str, Any]] = []
|
|
stack: list[tuple[int, str]] = []
|
|
key_counts: dict[str, int] = {}
|
|
for index, match in enumerate(matches):
|
|
level = len(match.group(1))
|
|
title = match.group(2).strip()
|
|
while stack and stack[-1][0] >= level:
|
|
stack.pop()
|
|
ancestors = [item[1] for item in stack]
|
|
stack.append((level, title))
|
|
end = len(text)
|
|
for candidate in matches[index + 1 :]:
|
|
if len(candidate.group(1)) <= level:
|
|
end = candidate.start()
|
|
break
|
|
base_key = f"{level}:{title}"
|
|
key_counts[base_key] = key_counts.get(base_key, 0) + 1
|
|
key = (
|
|
base_key
|
|
if key_counts[base_key] == 1
|
|
else f"{base_key}#{key_counts[base_key]}"
|
|
)
|
|
records.append(
|
|
{
|
|
"ancestors": ancestors,
|
|
"end": end,
|
|
"hash": hashlib.sha256(text[match.start() : end].encode("utf-8")).hexdigest(),
|
|
"key": key,
|
|
"level": level,
|
|
"start": match.start(),
|
|
"title": title,
|
|
}
|
|
)
|
|
return records
|
|
|
|
|
|
def protected_section_hashes(text: str, planned_sections: Sequence[str]) -> dict[str, str]:
|
|
records = section_records(text)
|
|
planned = set(planned_sections)
|
|
protected: dict[str, str] = {}
|
|
for record in records:
|
|
descendant_planned = any(
|
|
other["title"] in planned
|
|
and other["start"] > record["start"]
|
|
and other["start"] < record["end"]
|
|
for other in records
|
|
)
|
|
affected = (
|
|
record["title"] in planned
|
|
or any(ancestor in planned for ancestor in record["ancestors"])
|
|
or descendant_planned
|
|
)
|
|
if not affected:
|
|
protected[record["key"]] = record["hash"]
|
|
return protected
|
|
|
|
|
|
def document_header_hash(text: str) -> str:
|
|
second_level = re.search(r"(?m)^##\s+", text)
|
|
header = text[: second_level.start()] if second_level else text
|
|
kept_lines = []
|
|
for line in header.splitlines():
|
|
stripped = line.strip()
|
|
if re.fullmatch(r">\s*最后核对：\d{4}-\d{2}-\d{2}。", stripped):
|
|
continue
|
|
if SYNC_MARKER_RE.fullmatch(stripped):
|
|
continue
|
|
kept_lines.append(line.rstrip())
|
|
normalized = "\n".join(kept_lines).strip()
|
|
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def is_ordered_subsequence(before: Sequence[str], current: Sequence[str]) -> bool:
|
|
iterator = iter(current)
|
|
return all(any(candidate == expected for candidate in iterator) for expected in before)
|
|
|
|
|
|
def validate_headings(
|
|
document: str,
|
|
before: Sequence[dict[str, Any]],
|
|
current: Sequence[dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
errors: list[dict[str, Any]] = []
|
|
before_primary = [item["title"] for item in before if item["level"] <= 2]
|
|
current_primary = [item["title"] for item in current if item["level"] <= 2]
|
|
if before_primary != current_primary:
|
|
errors.append(
|
|
verification_error(
|
|
"primary_headings_changed",
|
|
"原有一级或二级标题被删除、改名、增加或调整顺序。",
|
|
"恢复原有一级和二级标题；内容变化应限制在已批准章节正文内。",
|
|
document=document,
|
|
)
|
|
)
|
|
before_tertiary = [item["title"] for item in before if item["level"] == 3]
|
|
current_tertiary = [item["title"] for item in current if item["level"] == 3]
|
|
if not is_ordered_subsequence(before_tertiary, current_tertiary):
|
|
errors.append(
|
|
verification_error(
|
|
"existing_subheadings_removed_or_reordered",
|
|
"原有三级标题被删除或调整顺序。",
|
|
"恢复原有三级标题及顺序；如需新增标题，只能追加到已批准章节。",
|
|
document=document,
|
|
)
|
|
)
|
|
return errors
|
|
|
|
|
|
def markdown_targets(text: str) -> list[str]:
|
|
values = MARKDOWN_LINK_RE.findall(text) + MARKDOWN_REFERENCE_LINK_RE.findall(text)
|
|
targets = []
|
|
for value in values:
|
|
target = value.strip().strip("<>")
|
|
if " " in target and not target.startswith(("http://", "https://")):
|
|
target = target.split(" ", 1)[0]
|
|
targets.append(target)
|
|
return targets
|
|
|
|
|
|
def validate_relative_links(root: Path, document: str, text: str) -> list[dict[str, Any]]:
|
|
errors: list[dict[str, Any]] = []
|
|
document_path = root.joinpath(*PurePosixPath(document).parts)
|
|
for raw_target in markdown_targets(text):
|
|
if not raw_target or raw_target.startswith("#"):
|
|
continue
|
|
if raw_target.lower().startswith("file://"):
|
|
errors.append(
|
|
verification_error(
|
|
"absolute_link_forbidden",
|
|
"文档包含本机绝对链接。",
|
|
"改为仓库内相对链接。",
|
|
document=document,
|
|
)
|
|
)
|
|
continue
|
|
target = unquote(raw_target.split("#", 1)[0].split("?", 1)[0])
|
|
if not target:
|
|
continue
|
|
if target.startswith(("/", "\\")) or re.match(r"^[A-Za-z]:[\\/]", target):
|
|
errors.append(
|
|
verification_error(
|
|
"absolute_link_forbidden",
|
|
"文档包含本机或根路径链接。",
|
|
"改为仓库内相对链接。",
|
|
document=document,
|
|
)
|
|
)
|
|
continue
|
|
if re.match(r"^[A-Za-z][A-Za-z0-9+.-]*:", raw_target):
|
|
continue
|
|
candidate = (document_path.parent / target).resolve(strict=False)
|
|
try:
|
|
relative = candidate.relative_to(root.resolve(strict=True)).as_posix()
|
|
except (OSError, ValueError):
|
|
errors.append(
|
|
verification_error(
|
|
"link_outside_repository",
|
|
"相对链接解析到了仓库外部。",
|
|
"将链接改为仓库内部存在的相对路径。",
|
|
document=document,
|
|
)
|
|
)
|
|
continue
|
|
if is_forbidden(relative) or not candidate.exists():
|
|
errors.append(
|
|
verification_error(
|
|
"relative_link_invalid",
|
|
"文档包含不存在或不允许引用的相对链接。",
|
|
"修正链接目标，或移除无法由仓库内容证明的链接。",
|
|
document=document,
|
|
target=raw_target,
|
|
)
|
|
)
|
|
return errors
|
|
|
|
|
|
def validate_document_safety(document: str, text: str) -> list[dict[str, Any]]:
|
|
errors: list[dict[str, Any]] = []
|
|
if ABSOLUTE_LOCAL_PATH_RE.search(text):
|
|
errors.append(
|
|
verification_error(
|
|
"absolute_local_path_detected",
|
|
"文档包含本机绝对路径。",
|
|
"删除本机路径并改用仓库相对路径。",
|
|
document=document,
|
|
)
|
|
)
|
|
if any(pattern.search(text) for pattern in SECRET_CONTENT_PATTERNS):
|
|
errors.append(
|
|
verification_error(
|
|
"sensitive_content_detected",
|
|
"文档包含疑似密钥、密码或私钥内容。",
|
|
"立即删除敏感值，仅保留环境变量名称或安全配置说明。",
|
|
document=document,
|
|
)
|
|
)
|
|
return errors
|
|
|
|
|
|
def referenced_code_paths(document: str, text: str) -> list[str]:
|
|
candidates = set()
|
|
for value in re.findall(r"`([^`\n]+)`", text):
|
|
value = value.strip().replace("\\", "/")
|
|
if not value or any(char in value for char in "*?{}<>") or " " in value:
|
|
continue
|
|
if value.startswith("core/"):
|
|
candidate = value.rstrip("/.,;:")
|
|
elif document == BACKEND_DOCUMENT and value.startswith(("apps/", "airshelf/", "skills/")):
|
|
candidate = f"core/backend/{value}".rstrip("/.,;:")
|
|
elif document == FRONTEND_DOCUMENT and value.startswith(("src/", "public/")):
|
|
candidate = f"core/frontend/{value}".rstrip("/.,;:")
|
|
else:
|
|
continue
|
|
candidates.add(normalize_path(candidate))
|
|
return sorted(path for path in candidates if path)
|
|
|
|
|
|
def validate_code_references(
|
|
root: Path, document: str, text: str
|
|
) -> tuple[list[dict[str, Any]], list[str]]:
|
|
errors: list[dict[str, Any]] = []
|
|
checked: list[str] = []
|
|
for path in referenced_code_paths(document, text):
|
|
if is_forbidden(path):
|
|
errors.append(
|
|
verification_error(
|
|
"forbidden_code_reference",
|
|
"文档引用了敏感或排除路径。",
|
|
"移除该引用，仅保留允许公开的架构事实。",
|
|
document=document,
|
|
)
|
|
)
|
|
continue
|
|
candidate = safe_candidate(root, path)
|
|
if candidate is None or not candidate.exists():
|
|
errors.append(
|
|
verification_error(
|
|
"code_reference_missing",
|
|
"文档中的代码路径无法在仓库中验证。",
|
|
"修正路径，或删除没有代码依据的描述。",
|
|
document=document,
|
|
path=path,
|
|
)
|
|
)
|
|
continue
|
|
checked.append(path)
|
|
return errors, checked
|
|
|
|
|
|
def expand_sections(section: str) -> list[str]:
|
|
return [part.strip() for part in section.split("、") if part.strip()]
|
|
|
|
|
|
def planned_updates(impacts: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
sections_by_document: dict[str, set[str]] = {}
|
|
for impact in impacts:
|
|
sections_by_document.setdefault(impact["document"], set()).update(
|
|
expand_sections(impact["section"])
|
|
)
|
|
return [
|
|
{"document": document, "sections": sorted(sections)}
|
|
for document, sections in sorted(sections_by_document.items())
|
|
]
|
|
|
|
|
|
def build_snapshot(root: Path, mode: str, scope: str = "all") -> dict[str, Any]:
|
|
warnings: set[str] = set()
|
|
targets = selected_documents(scope)
|
|
head = current_head(root)
|
|
if head is None:
|
|
warnings.add("head_commit_unavailable")
|
|
|
|
documents, markers = document_states(root, warnings, targets)
|
|
usable_markers: dict[str, str] = {}
|
|
for document, marker in markers.items():
|
|
if marker is None:
|
|
warnings.add(f"sync_marker_missing:{document}")
|
|
elif marker_is_usable(root, marker, head):
|
|
usable_markers[document] = marker
|
|
else:
|
|
warnings.add(f"sync_marker_unusable:{document}")
|
|
|
|
baseline_required = len(usable_markers) != len(targets)
|
|
committed = []
|
|
for marker in sorted(set(usable_markers.values())):
|
|
committed.append(
|
|
{
|
|
"base_commit": marker,
|
|
"files": filter_entries(
|
|
diff_entries(root, ("diff", f"{marker}..{head}")), scope
|
|
),
|
|
}
|
|
)
|
|
baseline_files = (
|
|
[path for path in walk_relevant_files(root) if path_in_scope(path, scope)]
|
|
if baseline_required
|
|
else []
|
|
)
|
|
staged = filter_entries(diff_entries(root, ("diff", "--cached")), scope)
|
|
unstaged = filter_entries(diff_entries(root, ("diff",)), scope)
|
|
untracked = filter_entries(untracked_entries(root), scope)
|
|
|
|
changed_paths = set(baseline_files)
|
|
for group in committed:
|
|
changed_paths.update(item["path"] for item in group["files"])
|
|
for group in (staged, unstaged, untracked):
|
|
changed_paths.update(item["path"] for item in group)
|
|
|
|
common_marker: str | None = None
|
|
marker_values = set(usable_markers.values())
|
|
if len(usable_markers) == len(targets) and len(marker_values) == 1:
|
|
common_marker = next(iter(marker_values))
|
|
|
|
impacts, non_architecture = map_impacts(
|
|
root,
|
|
head,
|
|
targets,
|
|
usable_markers,
|
|
baseline_files,
|
|
staged,
|
|
unstaged,
|
|
untracked,
|
|
warnings,
|
|
)
|
|
architecture: dict[str, Any] = {}
|
|
if BACKEND_DOCUMENT in targets:
|
|
architecture["backend"] = collect_backend(root, warnings)
|
|
if FRONTEND_DOCUMENT in targets:
|
|
architecture["frontend"] = collect_frontend(root, warnings)
|
|
content_fingerprint, fingerprinted_file_count = scope_content_fingerprint(
|
|
root, scope, warnings
|
|
)
|
|
snapshot_core = {
|
|
"architecture": architecture,
|
|
"changed_files": sorted(changed_paths),
|
|
"changes": {
|
|
"baseline": baseline_files,
|
|
"committed": committed,
|
|
"staged": staged,
|
|
"unstaged": unstaged,
|
|
"untracked": untracked,
|
|
},
|
|
"content_fingerprint": content_fingerprint,
|
|
"documents": documents,
|
|
"fingerprinted_file_count": fingerprinted_file_count,
|
|
"repository": {
|
|
"base_commit": common_marker,
|
|
"baseline_documents": sorted(set(targets) - set(usable_markers)),
|
|
"baseline_required": baseline_required,
|
|
"current_commit": head,
|
|
"sync_commits": markers,
|
|
},
|
|
"scope": scope,
|
|
}
|
|
canonical = json.dumps(snapshot_core, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
|
|
snapshot_id = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:20]
|
|
warnings_list = sorted(warnings)
|
|
update_plan = planned_updates(impacts)
|
|
requires_update = bool(impacts)
|
|
plan_material = {
|
|
"impacts": impacts,
|
|
"non_architecture_changes": non_architecture,
|
|
"planned_updates": update_plan,
|
|
"scope": scope,
|
|
"snapshot_id": snapshot_id,
|
|
"warnings": warnings_list,
|
|
}
|
|
plan_canonical = json.dumps(
|
|
plan_material, ensure_ascii=True, sort_keys=True, separators=(",", ":")
|
|
)
|
|
plan_id = f"arch-{hashlib.sha256(plan_canonical.encode('utf-8')).hexdigest()[:20]}"
|
|
evidence_allowlist = sorted(
|
|
{path for impact in impacts for path in impact.get("evidence", [])}
|
|
)
|
|
document_preconditions = [
|
|
{
|
|
"document": item["document"],
|
|
"before_sha256": documents[item["document"]]["sha256"],
|
|
"sections": item["sections"],
|
|
}
|
|
for item in update_plan
|
|
]
|
|
working_tree_evidence = sorted(
|
|
{
|
|
entry["path"]
|
|
for entries in (staged, unstaged, untracked)
|
|
for entry in entries
|
|
}
|
|
)
|
|
control_paths = (
|
|
"tools/architecture-sync/WORKFLOW.md",
|
|
"tools/architecture-sync/scan_architecture.py",
|
|
".agents/skills/sync-architecture-docs/SKILL.md",
|
|
".claude/skills/sync-architecture-docs/SKILL.md",
|
|
)
|
|
guard_documents: dict[str, Any] = {}
|
|
for item in document_preconditions:
|
|
document_text = read_text(root, item["document"], warnings) or ""
|
|
guard_documents[item["document"]] = {
|
|
"before_sha256": item["before_sha256"],
|
|
"header_hash": document_header_hash(document_text),
|
|
"headings": [
|
|
heading
|
|
for heading in documents[item["document"]]["headings"]
|
|
if heading["level"] <= 3
|
|
],
|
|
"protected_sections": protected_section_hashes(
|
|
document_text, item["sections"]
|
|
),
|
|
"sections": item["sections"],
|
|
}
|
|
guard_payload = {
|
|
"allowed_documents": [item["document"] for item in update_plan],
|
|
"content_fingerprint": content_fingerprint,
|
|
"current_commit": head,
|
|
"documents": guard_documents,
|
|
"evidence_sample": [
|
|
{
|
|
"exists": bool(
|
|
(safe_candidate(root, path) or Path("__missing__")).exists()
|
|
),
|
|
"path": path,
|
|
}
|
|
for path in evidence_allowlist[:24]
|
|
],
|
|
"guard_version": 1,
|
|
"initial_worktree": worktree_guard(root),
|
|
"plan_id": plan_id,
|
|
"protected_control_files": {
|
|
path: file_fingerprint(root, path, warnings) for path in control_paths
|
|
},
|
|
"scope": scope,
|
|
}
|
|
verification_guard = encode_guard(guard_payload)
|
|
return {
|
|
"analysis_ready": True,
|
|
**snapshot_core,
|
|
"confirmation": {
|
|
"accepted_phrase": "确认更新项目文档",
|
|
"plan_id": plan_id,
|
|
"required": requires_update,
|
|
"scanner_write_capability": False,
|
|
"short_phrase": "确认",
|
|
"short_phrase_requires_same_pending_context": True,
|
|
},
|
|
"impact_mapping_version": 1,
|
|
"impacts": impacts,
|
|
"mode": mode,
|
|
"next_action": "await_confirmation" if requires_update else "none",
|
|
"non_architecture_changes": non_architecture,
|
|
"plan_id": plan_id,
|
|
"planned_updates": update_plan,
|
|
"requires_update": requires_update,
|
|
"schema_version": SCHEMA_VERSION,
|
|
"snapshot_id": snapshot_id,
|
|
"update_context": {
|
|
"document_preconditions": document_preconditions,
|
|
"evidence_allowlist": evidence_allowlist,
|
|
"git_actions_allowed": False,
|
|
"patch_method": "apply_patch",
|
|
"plan_id": plan_id,
|
|
"read_policy": "minimum_needed_from_allowlist",
|
|
"sync_metadata": {
|
|
"checked_date_format": "YYYY-MM-DD",
|
|
"commit": head,
|
|
"has_uncommitted_evidence": bool(working_tree_evidence),
|
|
"marker": (
|
|
f"<!-- architecture-sync-commit: {head} -->" if head else None
|
|
),
|
|
},
|
|
"verification_guard": verification_guard,
|
|
"working_tree_evidence": working_tree_evidence,
|
|
},
|
|
"warnings": warnings_list,
|
|
"write_policy": {
|
|
"current_plan_documents": list(targets),
|
|
"never_modify": [
|
|
"core/ARCHITECTURE.md",
|
|
"business_code",
|
|
"configuration",
|
|
"migrations",
|
|
"tests",
|
|
"deployment_files",
|
|
"git_index",
|
|
"git_commits",
|
|
"git_remote",
|
|
],
|
|
"write_authorized": False,
|
|
},
|
|
}
|
|
|
|
|
|
def verify_snapshot(root: Path, token: str, scope: str) -> dict[str, Any]:
|
|
errors: list[dict[str, Any]] = []
|
|
warnings: list[str] = []
|
|
checked_code_paths: set[str] = set()
|
|
guard = decode_guard(token)
|
|
plan_id = str(guard.get("plan_id", ""))
|
|
guard_scope = guard.get("scope")
|
|
if guard_scope != scope:
|
|
errors.append(
|
|
verification_error(
|
|
"scope_mismatch",
|
|
"校验范围与已确认计划不一致。",
|
|
"使用原计划相同的 --scope 重新执行 verify。",
|
|
)
|
|
)
|
|
allowed_documents = set(guard.get("allowed_documents", []))
|
|
if not allowed_documents or not allowed_documents.issubset(set(TARGET_DOCUMENTS)):
|
|
errors.append(
|
|
verification_error(
|
|
"allowed_documents_invalid",
|
|
"校验令牌没有有效的目标文档白名单。",
|
|
"丢弃当前令牌并重新运行 plan。",
|
|
)
|
|
)
|
|
|
|
head = current_head(root)
|
|
if head != guard.get("current_commit"):
|
|
errors.append(
|
|
verification_error(
|
|
"head_commit_changed",
|
|
"计划生成后 HEAD 已发生变化。",
|
|
"停止写入并重新运行 plan 获取新确认。",
|
|
)
|
|
)
|
|
|
|
fingerprint_warnings: set[str] = set()
|
|
current_fingerprint, fingerprinted_count = scope_content_fingerprint(
|
|
root, str(guard_scope), fingerprint_warnings
|
|
)
|
|
if current_fingerprint != guard.get("content_fingerprint"):
|
|
errors.append(
|
|
verification_error(
|
|
"code_state_changed",
|
|
"计划生成后相关代码事实发生变化。",
|
|
"停止当前更新，重新扫描并重新确认。",
|
|
)
|
|
)
|
|
warnings.extend(sorted(fingerprint_warnings))
|
|
|
|
for path, expected in guard.get("protected_control_files", {}).items():
|
|
current = file_fingerprint(root, path, set())
|
|
if current != expected:
|
|
errors.append(
|
|
verification_error(
|
|
"control_file_changed",
|
|
"同步工作流、扫描器或 Skill 在执行期间发生变化。",
|
|
"停止验证，恢复控制文件或重新生成计划。",
|
|
path=path,
|
|
)
|
|
)
|
|
|
|
errors.extend(
|
|
compare_worktree_guards(
|
|
guard.get("initial_worktree", {}), worktree_guard(root), allowed_documents
|
|
)
|
|
)
|
|
|
|
document_results = []
|
|
for document in sorted(allowed_documents):
|
|
guard_document = guard.get("documents", {}).get(document)
|
|
if not isinstance(guard_document, dict):
|
|
errors.append(
|
|
verification_error(
|
|
"document_guard_missing",
|
|
"目标文档缺少修改前保护信息。",
|
|
"丢弃当前令牌并重新运行 plan。",
|
|
document=document,
|
|
)
|
|
)
|
|
continue
|
|
read_warnings: set[str] = set()
|
|
data = read_bytes(root, document, read_warnings)
|
|
if data is None:
|
|
errors.append(
|
|
verification_error(
|
|
"target_document_missing",
|
|
"目标架构文档不存在或无法读取。",
|
|
"恢复目标文档后重新生成计划。",
|
|
document=document,
|
|
)
|
|
)
|
|
continue
|
|
warnings.extend(sorted(read_warnings))
|
|
try:
|
|
text = data.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
errors.append(
|
|
verification_error(
|
|
"target_document_not_utf8",
|
|
"目标架构文档不是有效 UTF-8。",
|
|
"将文档恢复为 UTF-8 编码后重新验证。",
|
|
document=document,
|
|
)
|
|
)
|
|
continue
|
|
current_sha = hashlib.sha256(data).hexdigest()
|
|
changed = current_sha != guard_document.get("before_sha256")
|
|
if not changed:
|
|
warnings.append(f"planned_document_unchanged:{document}")
|
|
current_headings = headings_from_text(text)
|
|
errors.extend(
|
|
validate_headings(document, guard_document.get("headings", []), current_headings)
|
|
)
|
|
if document_header_hash(text) != guard_document.get("header_hash"):
|
|
errors.append(
|
|
verification_error(
|
|
"document_header_changed",
|
|
"文档头部除核对日期和同步标记外发生了计划外变化。",
|
|
"恢复文档标题和说明，只保留允许更新的日期与同步标记。",
|
|
document=document,
|
|
)
|
|
)
|
|
current_sections = {record["key"]: record["hash"] for record in section_records(text)}
|
|
changed_protected_sections = [
|
|
key
|
|
for key, expected_hash in guard_document.get("protected_sections", {}).items()
|
|
if current_sections.get(key) != expected_hash
|
|
]
|
|
if changed_protected_sections:
|
|
errors.append(
|
|
verification_error(
|
|
"unplanned_section_changed",
|
|
"计划外章节正文发生变化。",
|
|
"恢复计划外章节，只保留 planned_updates 指定范围内的局部补丁。",
|
|
document=document,
|
|
sections=changed_protected_sections,
|
|
)
|
|
)
|
|
errors.extend(validate_relative_links(root, document, text))
|
|
errors.extend(validate_document_safety(document, text))
|
|
reference_errors, checked = validate_code_references(root, document, text)
|
|
errors.extend(reference_errors)
|
|
checked_code_paths.update(checked)
|
|
|
|
if changed:
|
|
markers = [value.lower() for value in SYNC_MARKER_RE.findall(text)]
|
|
if len(markers) != 1 or markers[0] != guard.get("current_commit"):
|
|
errors.append(
|
|
verification_error(
|
|
"sync_marker_invalid",
|
|
"已修改文档缺少唯一且正确的同步 commit 标记。",
|
|
"在文首保留一条指向计划 HEAD 的 architecture-sync-commit 标记。",
|
|
document=document,
|
|
)
|
|
)
|
|
if not re.search(r"(?m)^>\s*最后核对：\d{4}-\d{2}-\d{2}。\s*$", text):
|
|
errors.append(
|
|
verification_error(
|
|
"checked_date_missing",
|
|
"已修改文档缺少格式正确的最后核对日期。",
|
|
"在文首使用“> 最后核对：YYYY-MM-DD。”格式更新日期。",
|
|
document=document,
|
|
)
|
|
)
|
|
document_results.append(
|
|
{
|
|
"changed": changed,
|
|
"document": document,
|
|
"headings_checked": len(current_headings),
|
|
}
|
|
)
|
|
|
|
evidence_checked = []
|
|
for item in guard.get("evidence_sample", []):
|
|
path = item.get("path", "")
|
|
if not path or is_forbidden(path):
|
|
errors.append(
|
|
verification_error(
|
|
"evidence_path_invalid",
|
|
"校验令牌包含越界或敏感证据路径。",
|
|
"丢弃当前令牌并重新运行 plan。",
|
|
)
|
|
)
|
|
continue
|
|
current_exists = bool((safe_candidate(root, path) or Path("__missing__")).exists())
|
|
if current_exists != bool(item.get("exists")):
|
|
errors.append(
|
|
verification_error(
|
|
"evidence_presence_changed",
|
|
"计划证据文件的存在状态发生变化。",
|
|
"停止当前更新，重新扫描并重新确认。",
|
|
path=path,
|
|
)
|
|
)
|
|
elif current_exists:
|
|
evidence_checked.append(path)
|
|
|
|
status = "passed" if not errors else "failed"
|
|
return {
|
|
"mode": "verify",
|
|
"plan_id": plan_id,
|
|
"schema_version": SCHEMA_VERSION,
|
|
"verification": {
|
|
"checks": {
|
|
"allowed_documents": sorted(allowed_documents),
|
|
"code_references_checked": len(checked_code_paths),
|
|
"evidence_files_checked": len(evidence_checked),
|
|
"fingerprinted_files_checked": fingerprinted_count,
|
|
"head_commit": head,
|
|
"scope": guard_scope,
|
|
},
|
|
"documents": document_results,
|
|
"errors": errors,
|
|
"status": status,
|
|
"warnings": sorted(set(warnings)),
|
|
},
|
|
}
|
|
|
|
|
|
def parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Read-only AirShelf architecture scanner")
|
|
parser.add_argument("mode", choices=("plan", "verify"))
|
|
parser.add_argument(
|
|
"--scope",
|
|
choices=("all", "backend", "frontend"),
|
|
default="all",
|
|
help="Limit the current plan to both documents or one architecture domain",
|
|
)
|
|
parser.add_argument(
|
|
"--guard",
|
|
help="Verification guard emitted by plan mode; required by verify mode",
|
|
)
|
|
parser.add_argument(
|
|
"--repo-root",
|
|
type=Path,
|
|
default=Path.cwd(),
|
|
help="Any path inside the Git repository (default: current directory)",
|
|
)
|
|
return parser.parse_args(arguments)
|
|
|
|
|
|
def main(arguments: Sequence[str] | None = None) -> int:
|
|
args = parse_arguments(arguments)
|
|
try:
|
|
root = locate_repository(args.repo_root)
|
|
if args.mode == "verify":
|
|
if not args.guard:
|
|
payload = {
|
|
"mode": "verify",
|
|
"schema_version": SCHEMA_VERSION,
|
|
"verification": {
|
|
"errors": [
|
|
verification_error(
|
|
"verification_guard_required",
|
|
"verify 模式缺少 plan 生成的校验令牌。",
|
|
"先运行 plan，确认后传入 update_context.verification_guard。",
|
|
)
|
|
],
|
|
"status": "failed",
|
|
"warnings": [],
|
|
},
|
|
}
|
|
print(json.dumps(payload, ensure_ascii=True, sort_keys=True, indent=2))
|
|
return 3
|
|
payload = verify_snapshot(root, args.guard, args.scope)
|
|
else:
|
|
payload = build_snapshot(root, args.mode, args.scope)
|
|
except ScanError as error:
|
|
if args.mode == "verify":
|
|
payload = {
|
|
"mode": "verify",
|
|
"schema_version": SCHEMA_VERSION,
|
|
"verification": {
|
|
"errors": [
|
|
verification_error(
|
|
str(error),
|
|
"校验令牌无效或只读校验无法继续。",
|
|
"丢弃当前令牌，重新运行 plan 并重新确认。",
|
|
)
|
|
],
|
|
"status": "failed",
|
|
"warnings": [],
|
|
},
|
|
}
|
|
else:
|
|
payload = {
|
|
"error": str(error),
|
|
"mode": args.mode,
|
|
"schema_version": SCHEMA_VERSION,
|
|
}
|
|
print(json.dumps(payload, ensure_ascii=True, sort_keys=True, indent=2))
|
|
return 2
|
|
print(json.dumps(payload, ensure_ascii=True, sort_keys=True, indent=2))
|
|
if args.mode == "verify" and payload["verification"]["status"] != "passed":
|
|
return 3
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace")
|
|
raise SystemExit(main())
|