176 lines
8.7 KiB
Python
176 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Conservative offline cleanup. The default only prints a plan."""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import uuid
|
||
import zipfile
|
||
from collections import Counter
|
||
from pathlib import Path
|
||
|
||
PACKAGES = ("hockey_data", "ui_builder", "khl_site", "broadcast_settings", "tests", "__pycache__")
|
||
SNAPSHOT_NAME = re.compile(r"^(?:\d{8}_\d{6}_\d{6}_[A-Za-z0-9_-]+|[a-f0-9]{64}_[A-Za-z0-9_-]*)\.xml$")
|
||
BACKUP_NAME = re.compile(r"^ui_builder(?:_draft|_published)?_\d{8}_\d{6}_\d{6}\.json$")
|
||
# Only this obsolete, unmounted version is eligible. A modified file is preserved.
|
||
OLD_UI_SHA256 = "1dcd302b4b6ef74616833a4a90bb37fccee1c92478432b0d5fa6ca0f20b3aa2d"
|
||
# Exact historical root files from this project's archived documentation.
|
||
# Current README, arbitrary archive paths and executable/source files are never eligible.
|
||
HISTORICAL_ROOT_NAMES = frozenset("""
|
||
EDITOR_ACCESS.md HOCKEY_DASHBOARD.md HOCKEY_DATA_ARCHITECTURE.md HOCKEY_TIMERS.md
|
||
HOCKEY_WORKFLOW.md HOTFIX_MANIFEST.json KEEP_LOCAL_CREDENTIALS.txt
|
||
NORMALIZED_STATISTICS_DB.md PREVIOUS_40_FAILURES_RU.md README_BUILD172_RU.md
|
||
README_BUILD173_RU.md README_BUILD174_RU.md README_BUILD175_RU.md README_BUILD176_RU.md
|
||
README_BUILD177_RU.md README_BUILD178.txt README_BUILD183.txt README_BUILD184.txt
|
||
README_BUILD186_RU.txt README_BUILD187_RU.txt README_BUILD188_RU.txt README_BUILD189_RU.txt
|
||
README_UPDATE180_RU.txt README_UPDATE181_RU.txt README_UPDATE182_RU.txt README_build179_RU.txt
|
||
REMOTE_DATABASE.md SHORTCUTS.md STAGE1_MANIFEST.json STAGE2_MANIFEST.json STAT2TV_SCHEDULE.md
|
||
TIMERS.md TRIGGERS.md UPDATE_ANALYTICS_RU.md UPDATE_HOTFIX_RU.md UPDATE_KHL_CAREER_RU.md
|
||
UPDATE_KHL_READING_RU.md UPDATE_STAGE1_RU.md UPDATE_STAGE2_RU.md UPDATE_TABS_RU.md
|
||
UPDATE_WITHOUT_UI_CONFIG.txt VALIDATION_ANALYTICS_RU.md VALIDATION_HOTFIX_RU.md
|
||
VALIDATION_KHL_CAREER_RU.md VALIDATION_STAGE2_RU.md
|
||
""".split())
|
||
|
||
|
||
def safe_file(root: Path, path: Path) -> bool:
|
||
try:
|
||
relative = path.relative_to(root)
|
||
except ValueError:
|
||
return False
|
||
return path.is_file() and all(not (root / Path(*relative.parts[:i])).is_symlink()
|
||
for i in range(1, len(relative.parts) + 1))
|
||
|
||
|
||
def digest(path: Path) -> str:
|
||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||
|
||
|
||
def archived_document_hashes(root: Path) -> dict[str, str]:
|
||
"""Read only allowlisted root docs; never extract or trust arbitrary ZIP paths."""
|
||
archive = root / "docs/history.zip"
|
||
if not safe_file(root, archive):
|
||
return {}
|
||
try:
|
||
with zipfile.ZipFile(archive) as history:
|
||
infos = history.infolist()
|
||
counts = Counter(info.filename for info in infos)
|
||
return {info.filename: hashlib.sha256(history.read(info)).hexdigest()
|
||
for info in infos if info.filename in HISTORICAL_ROOT_NAMES
|
||
and counts[info.filename] == 1 and not info.is_dir()
|
||
and info.file_size <= 256 * 1024}
|
||
except (OSError, ValueError, RuntimeError, zipfile.BadZipFile):
|
||
return {}
|
||
|
||
|
||
def plan_cleanup(root: Path) -> list[dict]:
|
||
root = root.resolve()
|
||
if not all((root / name).is_file() for name in
|
||
("app.py", "hockey_data/service.py", "ui_builder/integration.py")):
|
||
raise ValueError("Укажите корень проекта Hockey с app.py")
|
||
plan: list[dict] = []
|
||
for name, sha in archived_document_hashes(root).items():
|
||
path = root / name
|
||
if safe_file(root, path) and digest(path) == sha:
|
||
plan.append({"action": "delete", "path": name, "sha256": sha,
|
||
"bytes": path.stat().st_size, "reason": "archived_historical_doc"})
|
||
for package in PACKAGES:
|
||
for path in (root / package).rglob("*.pyc"):
|
||
if "__pycache__" in path.relative_to(root).parts and safe_file(root, path):
|
||
plan.append({"action": "delete", "path": str(path.relative_to(root)),
|
||
"bytes": path.stat().st_size, "sha256": digest(path), "reason": "bytecode"})
|
||
snapshots: dict[str, Path] = {}
|
||
for path in sorted((root / "hockey_data/snapshots").glob("*.xml")):
|
||
if not SNAPSHOT_NAME.fullmatch(path.name) or not safe_file(root, path):
|
||
continue
|
||
sha = digest(path)
|
||
previous = snapshots.setdefault(sha, path)
|
||
if previous != path and not os.path.samefile(previous, path):
|
||
plan.append({"action": "hardlink", "path": str(path.relative_to(root)),
|
||
"source": str(previous.relative_to(root)), "sha256": sha,
|
||
"bytes": path.stat().st_size, "reason": "identical_snapshot_keep_path"})
|
||
# Never touch live settings/local. Only old copies already migrated exactly
|
||
# are removable; the marker prevents old versions resurrecting them at start.
|
||
marker = root / "settings/local/.ui_backups_migrated_v1.json"
|
||
try:
|
||
migrated = safe_file(root, marker) and json.loads(marker.read_text()).get("completed") is True
|
||
except (OSError, ValueError, AttributeError):
|
||
migrated = False
|
||
if migrated:
|
||
for path in (root / "settings/backups").glob("*.json"):
|
||
local = root / "settings/local/backups" / path.name
|
||
if BACKUP_NAME.fullmatch(path.name) and safe_file(root, path) and safe_file(root, local):
|
||
sha = digest(path)
|
||
if digest(local) == sha:
|
||
plan.append({"action": "delete", "path": str(path.relative_to(root)),
|
||
"bytes": path.stat().st_size, "sha256": sha,
|
||
"source": str(local.relative_to(root)), "reason": "migrated_backup_copy"})
|
||
orphan = root / "hockey_data/ui_builder/static/app.js"
|
||
active = root / "ui_builder/static/app.js"
|
||
if safe_file(root, orphan) and safe_file(root, active) and digest(orphan) == OLD_UI_SHA256:
|
||
plan.append({"action": "delete", "path": str(orphan.relative_to(root)),
|
||
"bytes": orphan.stat().st_size, "sha256": OLD_UI_SHA256, "reason": "unmounted_old_ui"})
|
||
# The active agent.log is intentionally excluded.
|
||
old_log = root / "agent/agent.log.1"
|
||
if safe_file(root, old_log):
|
||
plan.append({"action": "delete", "path": str(old_log.relative_to(root)),
|
||
"bytes": old_log.stat().st_size, "sha256": digest(old_log), "reason": "rotated_log"})
|
||
return plan
|
||
|
||
|
||
def apply_plan(root: Path, plan: list[dict]) -> list[dict]:
|
||
root = root.resolve()
|
||
def key(step):
|
||
return tuple(step.get(name) for name in ("action", "path", "source", "sha256", "reason"))
|
||
allowed = {key(step) for step in plan_cleanup(root)}
|
||
results = []
|
||
for step in plan:
|
||
path = root / step["path"]
|
||
result = {**step, "status": "skipped"}
|
||
temporary = None
|
||
try:
|
||
if key(step) not in allowed:
|
||
result["detail"] = "not_allowlisted"
|
||
elif not safe_file(root, path) or digest(path) != step["sha256"]:
|
||
result["detail"] = "file_changed"
|
||
elif step["action"] == "hardlink":
|
||
source = root / step["source"]
|
||
if safe_file(root, source) and digest(source) == step["sha256"]:
|
||
temporary = path.with_name(path.name + ".cleanup-" + uuid.uuid4().hex)
|
||
os.link(source, temporary)
|
||
os.replace(temporary, path)
|
||
result["status"] = "done"
|
||
elif step["action"] == "delete":
|
||
source = root / step["source"] if "source" in step else None
|
||
if source is None or (safe_file(root, source) and digest(source) == step["sha256"]):
|
||
path.unlink()
|
||
result["status"] = "done"
|
||
except OSError as error:
|
||
result["detail"] = type(error).__name__
|
||
finally:
|
||
if temporary is not None:
|
||
temporary.unlink(missing_ok=True)
|
||
results.append(result)
|
||
return results
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parent)
|
||
modes = parser.add_mutually_exclusive_group()
|
||
modes.add_argument("--apply", action="store_true", help="Apply after stopping Hockey and Agent")
|
||
modes.add_argument("--dry-run", action="store_true", help="Print only (default)")
|
||
args = parser.parse_args()
|
||
plan = plan_cleanup(args.root)
|
||
results = apply_plan(args.root, plan) if args.apply else plan
|
||
summary = Counter(step["reason"] for step in results)
|
||
print(json.dumps({"mode": "apply" if args.apply else "dry-run", "counts": dict(summary),
|
||
"potential_bytes": sum(step["bytes"] for step in plan), "items": results},
|
||
ensure_ascii=False, indent=2))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|