66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
"""
|
|
Applies a reviewed batch of FileRecords to disk.
|
|
|
|
Safety rules, non-negotiable regardless of which front end calls this:
|
|
- dry_run=True by default. Nothing is touched unless you explicitly
|
|
pass dry_run=False.
|
|
- Every run produces a log (same shape for dry-run and real runs), so
|
|
a dry-run's log IS the preview of exactly what a real run would do.
|
|
- A file is only ever touched if its status is "confirmed" — "pending"
|
|
or "skip" rows are left completely alone.
|
|
- If the computed target path already exists, the file is skipped with
|
|
a clear reason rather than silently overwritten.
|
|
"""
|
|
|
|
import csv
|
|
import shutil
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from .model import FileRecord
|
|
|
|
|
|
def plan_new_path(rec: FileRecord) -> Path:
|
|
return rec.root / rec.dest_folder / rec.new_filename_preview
|
|
|
|
|
|
def apply_batch(records: List[FileRecord], dry_run: bool = True,
|
|
log_path: Optional[Path] = None) -> List[dict]:
|
|
log_rows = []
|
|
|
|
for rec in records:
|
|
if rec.status != "confirmed":
|
|
continue
|
|
|
|
new_path = plan_new_path(rec)
|
|
row = {
|
|
"timestamp": datetime.now().isoformat(timespec="seconds"),
|
|
"old_path": str(rec.path),
|
|
"new_path": str(new_path),
|
|
"dry_run": dry_run,
|
|
"result": "",
|
|
}
|
|
|
|
if new_path == rec.path:
|
|
row["result"] = "unchanged"
|
|
elif new_path.exists():
|
|
row["result"] = "SKIPPED - target already exists"
|
|
elif dry_run:
|
|
row["result"] = "would move"
|
|
else:
|
|
new_path.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(rec.path), str(new_path))
|
|
rec.path = new_path # the record now represents the file at its NEW location —
|
|
row["result"] = "moved" # without this, anything using rec.path afterwards
|
|
rec.status = "done" # (identity, re-display, a second move) points at nothing.
|
|
|
|
log_rows.append(row)
|
|
|
|
if log_path and log_rows:
|
|
with open(log_path, "w", newline="", encoding="utf-8") as f:
|
|
writer = csv.DictWriter(f, fieldnames=list(log_rows[0].keys()))
|
|
writer.writeheader()
|
|
writer.writerows(log_rows)
|
|
|
|
return log_rows
|