You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
58 lines
2.0 KiB
58 lines
2.0 KiB
import os
|
|
import logging
|
|
from checkers.base import BaseChecker
|
|
from exporter import SafetyExporter
|
|
from models import SafetyStatus, SafetyLevel
|
|
|
|
logger = logging.getLogger("Checker.FileExist")
|
|
|
|
|
|
class FileExistChecker(BaseChecker):
|
|
"""检查指定路径文件是否存在。"""
|
|
|
|
def __init__(self, config: dict, exporter: SafetyExporter):
|
|
super().__init__("FileExistChecker", config, exporter)
|
|
self.items = config.get("items", [])
|
|
|
|
def run_once(self) -> SafetyStatus:
|
|
if not self.items:
|
|
return SafetyStatus(
|
|
name="FileExistCheck", source=self.name, level=SafetyLevel.OK,
|
|
message="No files configured",
|
|
)
|
|
|
|
results = []
|
|
for item in self.items:
|
|
path = item.get("path", "")
|
|
name = item.get("name", os.path.basename(path))
|
|
level_str = item.get("level", "ERROR")
|
|
level = SafetyLevel[level_str]
|
|
|
|
exists = os.path.isfile(path)
|
|
results.append({
|
|
"name": name,
|
|
"path": path,
|
|
"exists": exists,
|
|
"level": level,
|
|
})
|
|
|
|
if not exists:
|
|
logger.warning(f"File missing: {name} ({path})")
|
|
|
|
missing = [r for r in results if not r["exists"]]
|
|
if missing:
|
|
worst = max(missing, key=lambda r: r["level"])
|
|
msgs = [f"{r['name']}({r['path']})" for r in missing]
|
|
return SafetyStatus(
|
|
name="FileExistCheck", source=self.name, level=worst["level"],
|
|
message=f"Missing files: {'; '.join(msgs)}",
|
|
details={"missing": [r["name"] for r in missing]},
|
|
)
|
|
|
|
logger.info(f"FileExistCheck: all {len(self.items)} files exist")
|
|
return SafetyStatus(
|
|
name="FileExistCheck", source=self.name, level=SafetyLevel.OK,
|
|
message=f"All {len(self.items)} files exist",
|
|
details={"checked": len(self.items)},
|
|
)
|