bbchops/transcription
1
0
Fork 0
generated from john/python-template
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

Continue GC code review and cleanup

Browse Source
This commit is contained in:
Jim Lancaster
2026-08-11 16:42:09 -05:00
parent 8d5aec4301
commit b8be27f0c9
16 changed files with 363 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/run_destructive_tests.py
+136 -60
View File
@@ -3,16 +3,16 @@ from __future__ import annotations
import argparse
import ctypes
import os
from ctypes import wintypes
from datetime import datetime
from pathlib import Path
import shlex
import shutil
import subprocess
import sys
from ctypes import wintypes
from datetime import datetime
from pathlib import Path
RETRY_CANCEL_CHOICES = {"n", "no", "c", "cancel", "a", "abort", "q", "quit"}
ACTIVE_BACKUP_FILENAME = ".active-backup"
def show_phase(title: str) -> None:
@@ -88,11 +88,11 @@ def wait_for_restore_preflight(db_file_path: Path) -> bool:
while not test_file_unlocked(db_file_path):
show_phase("Restore Preflight")
print(f"WARNING: Restore preflight blocked: database appears to be in use: {db_file_path}")
print("Close conflicting applications (for example DB Browser for SQLite, uvicorn, or any process using this DB).")
print(
"Close conflicting applications (for example DB Browser for SQLite, uvicorn, or any process using this DB)."
)
print("Restore is paused and waiting for your input.")
answer = input(
f"Attempt {attempt}: type 'retry' to check again, or 'cancel' to skip restore: "
).strip()
answer = input(f"Attempt {attempt}: type 'retry' to check again, or 'cancel' to skip restore: ").strip()
print(f"Input received: '{answer}'")
if answer.lower() in RETRY_CANCEL_CHOICES:
return False
@@ -115,10 +115,61 @@ def restore_backup(backup_path: Path, data_path: Path) -> None:
print(f"Restored data from backup: {backup_path}")
def active_backup_path(backup_root: Path) -> Path | None:
marker_path = backup_root / ACTIVE_BACKUP_FILENAME
if not marker_path.exists():
return None
backup_name = marker_path.read_text(encoding="utf-8").strip()
if not backup_name or Path(backup_name).name != backup_name:
raise RuntimeError(f"Invalid active backup marker: {marker_path}")
backup_path = backup_root / backup_name
if not backup_path.is_dir():
raise FileNotFoundError(
f"Active backup is missing: {backup_path}. "
f"Restore it or remove {marker_path} only after confirming the original data is safe."
)
return backup_path
def create_or_reuse_backup(data_path: Path, backup_root: Path) -> Path:
current_backup = active_backup_path(backup_root)
if current_backup is not None:
print(f"Reusing active test-cycle backup: {current_backup}")
return current_backup
db_path = data_path / "transcription.db"
if not test_file_unlocked(db_path):
print(f"WARNING: Backup preflight warning: database appears to be in use: {db_path}")
print("WARNING: Proceeding with backup, but hot backups can capture an in-flight state.")
timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S")
backup_path = backup_root / f"data-backup-{timestamp}"
shutil.copytree(data_path, backup_path)
marker_path = backup_root / ACTIVE_BACKUP_FILENAME
temporary_marker_path = marker_path.with_suffix(".tmp")
temporary_marker_path.write_text(backup_path.name, encoding="utf-8")
temporary_marker_path.replace(marker_path)
print(f"Created test-cycle backup: {backup_path}")
return backup_path
def close_active_backup_cycle(backup_root: Path, backup_path: Path | None = None) -> Path:
current_backup = active_backup_path(backup_root)
if current_backup is None:
raise RuntimeError("No active test backup cycle exists.")
if backup_path is not None and current_backup.resolve() != backup_path.resolve():
return current_backup
(backup_root / ACTIVE_BACKUP_FILENAME).unlink()
print(f"Closed test backup cycle: {current_backup}")
return current_backup
def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Run potentially destructive tests with backup/restore protection."
)
parser = argparse.ArgumentParser(description="Run potentially destructive tests with backup/restore protection.")
parser.add_argument("--auto-restore", action="store_true", help="Restore immediately after successful tests.")
parser.add_argument("--keep-backup", action="store_true", help="Keep the backup even after a successful restore.")
parser.add_argument(
@@ -130,14 +181,22 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
"--restore-from",
help="Restore from an existing backup name or absolute backup path instead of running tests.",
)
parser.add_argument(
"--accept-current-data",
action="store_true",
help="Close the active test cycle without restoring; preserve its backup.",
)
parser.add_argument("command", nargs=argparse.REMAINDER, help="Command to run after '--'.")
args = parser.parse_args(argv)
if args.restore_from and args.command:
parser.error("--restore-from cannot be combined with a test command.")
if args.restore_from and args.accept_current_data:
parser.error("--restore-from cannot be combined with --accept-current-data.")
if not args.restore_from and not args.command:
parser.error("A test command is required unless --restore-from is provided.")
if (args.restore_from or args.accept_current_data) and args.command:
parser.error("Restore/accept modes cannot be combined with a test command.")
if not args.restore_from and not args.accept_current_data and not args.command:
parser.error("A test command is required unless --restore-from or --accept-current-data is provided.")
return args
@@ -159,45 +218,50 @@ def run_command(command: list[str]) -> int:
return completed.returncode
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv or sys.argv[1:])
repo_root = Path(__file__).resolve().parent.parent
data_dir = repo_root / "data"
backup_root = repo_root / ".test-backups"
db_path = data_dir / "transcription.db"
def accept_current_data(backup_root: Path) -> int:
show_phase("Complete Test Cycle")
backup_path = close_active_backup_cycle(backup_root)
print(f"Current data accepted. Backup preserved at: {backup_path}")
return 0
if not data_dir.exists():
raise FileNotFoundError(f"Data directory not found: {data_dir}")
backup_root.mkdir(parents=True, exist_ok=True)
def restore_saved_backup(restore_from: str, data_path: Path, backup_root: Path) -> int:
restore_path = Path(restore_from)
if not restore_path.is_absolute():
restore_path = backup_root / restore_path
if args.restore_from:
restore_path = Path(args.restore_from)
if not restore_path.is_absolute():
restore_path = backup_root / restore_path
show_phase("Restore Phase")
if not wait_for_restore_preflight(data_path / "transcription.db"):
print(f"Restore cancelled. Backup preserved at: {restore_path}")
return 1
show_phase("Restore Phase")
if not wait_for_restore_preflight(db_path):
print(f"Restore cancelled. Backup preserved at: {restore_path}")
return 1
restore_backup(restore_path, data_path)
current_backup = active_backup_path(backup_root)
if current_backup is not None:
if current_backup.resolve() == restore_path.resolve():
close_active_backup_cycle(backup_root, restore_path)
else:
print(f"Active test cycle remains unchanged: {current_backup}")
return 0
restore_backup(restore_path, data_dir)
return 0
def restore_requested(args: argparse.Namespace) -> bool:
if args.auto_restore:
return True
if args.skip_restore_prompt:
return False
answer = input("Tests passed. Restore data backup now? [y/N] ").strip().lower()
return answer in {"y", "yes"}
def run_protected_tests(args: argparse.Namespace, data_path: Path, backup_root: Path) -> int:
command = normalize_command(list(args.command))
if not command:
raise ValueError("No test command provided.")
show_phase("Backup Phase")
if not test_file_unlocked(db_path):
print(f"WARNING: Backup preflight warning: database appears to be in use: {db_path}")
print("WARNING: Proceeding with backup, but hot backups can capture an in-flight state.")
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
backup_name = f"data-backup-{timestamp}"
backup_path = backup_root / backup_name
shutil.copytree(data_dir, backup_path)
print(f"Created backup: {backup_path}")
backup_path = create_or_reuse_backup(data_path, backup_root)
backup_name = backup_path.name
test_exit_code = run_command(command)
if test_exit_code != 0:
@@ -208,36 +272,30 @@ def main(argv: list[str] | None = None) -> int:
print(f" uv run python tools/run_destructive_tests.py --restore-from {backup_name}")
return test_exit_code
should_restore = False
if args.auto_restore:
should_restore = True
elif args.skip_restore_prompt:
should_restore = False
else:
answer = input("Tests passed. Restore data backup now? [y/N] ").strip().lower()
if answer in {"y", "yes"}:
should_restore = True
if not should_restore:
print(f"Restore skipped by user. Backup kept at: {backup_path}")
if not restore_requested(args):
print(f"Restore skipped by user. Test cycle remains active with backup: {backup_path}")
print("Further test runs will reuse this backup instead of creating another.")
print("Restore later with:")
print(f" uv run python tools/run_destructive_tests.py --restore-from {backup_name}")
print("Or accept the current data and close the cycle with:")
print(" uv run python tools/run_destructive_tests.py --accept-current-data")
return 0
show_phase("Restore Phase")
if not wait_for_restore_preflight(db_path):
if not wait_for_restore_preflight(data_path / "transcription.db"):
print(f"Restore cancelled. Backup preserved at: {backup_path}")
return 1
try:
restore_backup(backup_path, data_dir)
restore_backup(backup_path, data_path)
close_active_backup_cycle(backup_root, backup_path)
if args.keep_backup:
print(f"Kept backup: {backup_path}")
else:
shutil.rmtree(backup_path)
print(f"Deleted backup: {backup_path}")
except Exception as exc:
print("WARNING: Restore failed. Your current data remains unchanged.")
except (OSError, RuntimeError) as exc:
print("WARNING: Restore failed. The backup remains available.")
print(f"WARNING: {exc}")
print("Likely cause: another process has data/transcription.db open.")
print("Stop the process and retry restore with:")
@@ -247,5 +305,23 @@ def main(argv: list[str] | None = None) -> int:
return 0
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv or sys.argv[1:])
repo_root = Path(__file__).resolve().parent.parent
data_path = repo_root / "data"
backup_root = repo_root / ".test-backups"
if not data_path.exists():
raise FileNotFoundError(f"Data directory not found: {data_path}")
backup_root.mkdir(parents=True, exist_ok=True)
if args.accept_current_data:
return accept_current_data(backup_root)
if args.restore_from:
return restore_saved_backup(args.restore_from, data_path, backup_root)
return run_protected_tests(args, data_path, backup_root)
if __name__ == "__main__":
raise SystemExit(main())
raise SystemExit(main())