task-148: 临时目录配置盘点(16 配置项/322 清理调用点扫描器 + 9 条测试 + 报告快照,不改生产代码)

This commit is contained in:
2026-09-02 06:02:14 +08:00
parent dc90ead263
commit 383ec544e1
3 changed files with 1914 additions and 0 deletions
+113
View File
@@ -0,0 +1,113 @@
"""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()