"""task-148 临时目录配置盘点工具。 盘点 backend-java 的临时文件相关配置(application.yml 中的路径/前缀/保留期) 与既有清理逻辑调用点,产出机器可读清单供审计文档使用。本工具只读,不改代码。 """ from __future__ import annotations import argparse import json import re import sys from dataclasses import dataclass, field from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent YML_PATH = REPO_ROOT / "src/main/resources/application.yml" MAIN_ROOT = REPO_ROOT / "src/main/java" # 临时文件相关配置键(application.yml 键名匹配) CONFIG_KEY_RE = re.compile( r"(local-temp-dir|retention-hours|retention-days|temp-dir|transient-payload|" r"payload-|upload-|source-retention|result-retention|tmp|buffer-retention)", re.IGNORECASE, ) # 既有清理逻辑调用点(临时文件清理) CLEANUP_RE = re.compile( r"(deleteIfExists|Files\.delete|cleanup|deleteTemp|tempDir|transientPayloadStorageService\.delete|" r"\.del\(|FileUtil\.del|cleanupPrepared|deletePayloadIfPresent)", re.IGNORECASE, ) def extract_configs() -> list[dict]: """从 application.yml 提取临时文件相关配置项。""" if not YML_PATH.is_file(): return [] configs: list[dict] = [] for line_no, line in enumerate( YML_PATH.read_text(encoding="utf-8").splitlines(), start=1 ): stripped = line.strip() if not stripped or stripped.startswith("#"): continue match = re.match(r"^([\w.-]+):\s*(.*)$", stripped) if not match: continue key, value = match.group(1), match.group(2).strip() if CONFIG_KEY_RE.search(key): configs.append( { "key": key, "value": value, "line": line_no, "source": "application.yml", } ) return configs def extract_cleanup_calls() -> list[dict]: """扫描 main 源码中临时文件清理调用点。""" calls: list[dict] = [] for java_file in sorted(MAIN_ROOT.rglob("*.java")): text = java_file.read_text(encoding="utf-8") for line_no, line in enumerate(text.splitlines(), start=1): if CLEANUP_RE.search(line) and "import " not in line: calls.append( { "file": str(java_file.relative_to(REPO_ROOT)).replace( "\\", "/" ), "line": line_no, "snippet": line.strip()[:100], } ) return calls @dataclass class Report: configs: list[dict] = field(default_factory=list) cleanup_calls: list[dict] = field(default_factory=list) def to_dict(self) -> dict: return {"configs": self.configs, "cleanup_calls": self.cleanup_calls} def scan() -> Report: return Report(configs=extract_configs(), cleanup_calls=extract_cleanup_calls()) def main() -> None: parser = argparse.ArgumentParser(description="临时目录配置盘点") parser.add_argument("--json", help="输出 JSON 报告路径(默认 stdout)") args = parser.parse_args() report = scan() payload = report.to_dict() if args.json: Path(args.json).write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8" ) else: print(json.dumps(payload, ensure_ascii=False, indent=2)) print( f"configs: {len(payload['configs'])}, cleanup calls: {len(payload['cleanup_calls'])}", file=sys.stderr, ) if __name__ == "__main__": main()