114 lines
3.6 KiB
Python
114 lines
3.6 KiB
Python
"""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()
|