feat: allow admins to update audit checklist configuration
This commit is contained in:
@@ -0,0 +1,49 @@
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import zipfile
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
|
||||
|
||||
def validate_xlsx(content: bytes) -> bytes:
|
||||
"""Validate that content is an XLSX-compatible ZIP containing workbook metadata."""
|
||||
try:
|
||||
with zipfile.ZipFile(BytesIO(content)) as archive:
|
||||
names = set(archive.namelist())
|
||||
except (zipfile.BadZipFile, OSError) as exc:
|
||||
raise ValueError("Checklist không phải file XLSX hợp lệ") from exc
|
||||
if "[Content_Types].xml" not in names or "xl/workbook.xml" not in names:
|
||||
raise ValueError("Checklist không phải file XLSX hợp lệ")
|
||||
return content
|
||||
|
||||
|
||||
def normalize_json(content: bytes) -> bytes:
|
||||
"""Parse and normalize an object/array JSON configuration."""
|
||||
try:
|
||||
parsed = json.loads(content.decode("utf-8-sig"))
|
||||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise ValueError(f"Rule config JSON không hợp lệ: {exc}") from exc
|
||||
if not isinstance(parsed, (dict, list)):
|
||||
raise ValueError("Rule config JSON phải là object hoặc array")
|
||||
return json.dumps(parsed, ensure_ascii=False, indent=2).encode("utf-8") + b"\n"
|
||||
|
||||
|
||||
def install_config_files(config_root: str | Path, files: Iterable[tuple[str, bytes]]) -> None:
|
||||
"""Atomically replace allowlisted config files and keep one .bak copy."""
|
||||
root = Path(config_root)
|
||||
root.mkdir(parents=True, exist_ok=True)
|
||||
for target_name, content in files:
|
||||
if Path(target_name).name != target_name:
|
||||
raise ValueError("Tên file đích không hợp lệ")
|
||||
target = root / target_name
|
||||
temporary = root / f"{target_name}.tmp"
|
||||
backup = root / f"{target_name}.bak"
|
||||
with temporary.open("wb") as output:
|
||||
output.write(content)
|
||||
output.flush()
|
||||
os.fsync(output.fileno())
|
||||
if target.exists():
|
||||
shutil.copy2(target, backup)
|
||||
os.replace(temporary, target)
|
||||
Reference in New Issue
Block a user