125 lines
5.3 KiB
Python
125 lines
5.3 KiB
Python
"""Locations and non-destructive retention for local operational reports."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from tools.rag.common import DATA_DIR
|
|
|
|
CORE_JSON = {'documents.json', 'chunks.json', 'index_version.json'}
|
|
|
|
|
|
def run_id() -> str:
|
|
return datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') + '-' + uuid4().hex[:8]
|
|
|
|
|
|
def report_output(requested: Path | None, category: str, *, data_dir: Path = DATA_DIR) -> Path:
|
|
if not category.replace('_', '').isalnum():
|
|
raise ValueError('Invalid report category')
|
|
if requested is None:
|
|
target = data_dir / 'reports' / category / (run_id() + '.json')
|
|
else:
|
|
target = requested.resolve()
|
|
if target.parent == data_dir.resolve():
|
|
if target.name in CORE_JSON or target.name.startswith('.') or target.suffix != '.json':
|
|
raise ValueError('Report cannot overwrite operational data')
|
|
target = data_dir / 'reports' / category / target.name
|
|
if target.exists():
|
|
raise FileExistsError(f'Report already exists: {target}')
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
return target
|
|
|
|
|
|
def report_log(report: Path, *, data_dir: Path = DATA_DIR) -> Path:
|
|
try:
|
|
relative = report.resolve().relative_to((data_dir / 'reports').resolve())
|
|
except ValueError:
|
|
identity = hashlib.sha256(str(report.absolute()).encode()).hexdigest()[:8]
|
|
relative = Path('custom') / (report.stem + '-' + identity + '.json')
|
|
target = data_dir / 'logs' / relative.with_suffix('.pytest.log')
|
|
if target.exists():
|
|
raise FileExistsError(f'Log already exists: {target}')
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
return target
|
|
|
|
|
|
def resolve_report_input(path: Path, *, data_dir: Path = DATA_DIR) -> Path:
|
|
if path.exists():
|
|
return path
|
|
mappings = {}
|
|
for record in sorted((data_dir / 'reports/migrations').glob('*.json')):
|
|
for move in json.loads(record.read_text()).get('moves', []):
|
|
mappings[Path(move['source']).resolve()] = Path(move['destination'])
|
|
visited = set()
|
|
while not path.exists() and path.resolve() in mappings and path.resolve() not in visited:
|
|
visited.add(path.resolve())
|
|
path = mappings[path.resolve()]
|
|
return path
|
|
|
|
|
|
def migration_plan(data_dir: Path = DATA_DIR) -> list[dict]:
|
|
moves = []
|
|
for path in sorted(data_dir.iterdir()):
|
|
if not path.is_file() or path.name.startswith('.') or path.name in CORE_JSON:
|
|
continue
|
|
if path.suffix == '.json':
|
|
target = data_dir / 'reports/archive/legacy' / path.name
|
|
elif path.suffix == '.log':
|
|
target = data_dir / 'logs/archive/legacy' / path.name
|
|
else:
|
|
continue
|
|
moves.append({'source': str(path.absolute()), 'destination': str(target.absolute()),
|
|
'sha256': hashlib.sha256(path.read_bytes()).hexdigest()})
|
|
return moves
|
|
|
|
|
|
def retention_plan(data_dir: Path = DATA_DIR, *, keep: int = 5) -> list[dict]:
|
|
if keep < 1:
|
|
raise ValueError('Keep at least one verification')
|
|
completed = []
|
|
for path in (data_dir / 'reports/verification').glob('*.json'):
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
if isinstance(data, dict) and type(data.get('passed')) is bool:
|
|
completed.append(path)
|
|
except (ValueError, OSError):
|
|
continue
|
|
moves = []
|
|
for path in sorted(completed, key=lambda p: (p.stat().st_mtime_ns, p.name), reverse=True)[keep:]:
|
|
target = data_dir / 'reports/archive/verification' / path.name
|
|
moves.append({'source': str(path.absolute()), 'destination': str(target.absolute()),
|
|
'sha256': hashlib.sha256(path.read_bytes()).hexdigest()})
|
|
log = data_dir / 'logs/verification' / path.with_suffix('.pytest.log').name
|
|
if log.exists():
|
|
target = data_dir / 'logs/archive/verification' / log.name
|
|
moves.append({'source': str(log.absolute()), 'destination': str(target.absolute()),
|
|
'sha256': hashlib.sha256(log.read_bytes()).hexdigest()})
|
|
return moves
|
|
|
|
|
|
def apply_moves(moves: list[dict], *, data_dir: Path = DATA_DIR) -> Path:
|
|
# Validate the entire plan before moving anything. Never replace existing files.
|
|
for move in moves:
|
|
source, destination = Path(move['source']), Path(move['destination'])
|
|
if destination.exists() or hashlib.sha256(source.read_bytes()).hexdigest() != move['sha256']:
|
|
raise RuntimeError('Report migration conflict; no files moved')
|
|
journal = data_dir / 'reports/migrations' / (run_id() + '.json')
|
|
journal.parent.mkdir(parents=True, exist_ok=True)
|
|
record = {'method': 'Move without rewriting content or deleting historical evidence', 'moves': []}
|
|
def write_journal():
|
|
temporary = journal.with_suffix('.tmp')
|
|
temporary.write_text(json.dumps(record, indent=2) + '\n')
|
|
temporary.replace(journal)
|
|
write_journal()
|
|
for move in moves:
|
|
destination = Path(move['destination'])
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(move['source'], destination)
|
|
record['moves'].append(move)
|
|
write_journal()
|
|
return journal
|