50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
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)
|