database_verifier

Database verification functionality for Music Genre Updater.

This module handles verifying the track database against Music.app and managing incremental run timestamps.

DryRunAction

Bases: TypedDict

Describes a single dry-run action recorded during verification.

DatabaseVerifier

DatabaseVerifier(
    ap_client,
    console_logger,
    error_logger,
    db_verify_logger,
    *,
    analytics,
    config,
    dry_run=False
)

Manages database verification and incremental run tracking.

Initialize the DatabaseVerifier.

Parameters:

Name	Type	Description	Default
ap_client	AppleScriptClientProtocol	AppleScript client for Music.app communication	required
console_logger	Logger	Logger for console output	required
error_logger	Logger	Logger for error messages	required
db_verify_logger	Logger	Logger for verification log file	required
analytics	Analytics	Analytics instance for tracking	required
config	AppConfig	Typed application configuration	required
dry_run	bool	Whether to run in dry-run mode	False

Source code in src/app/features/verify/database_verifier.py

def __init__(
    self,
    ap_client: AppleScriptClientProtocol,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
    db_verify_logger: logging.Logger,
    *,
    analytics: Analytics,
    config: AppConfig,
    dry_run: bool = False,
) -> None:
    """Initialize the DatabaseVerifier.

    Args:
        ap_client: AppleScript client for Music.app communication
        console_logger: Logger for console output
        error_logger: Logger for error messages
        db_verify_logger: Logger for verification log file
        analytics: Analytics instance for tracking
        config: Typed application configuration
        dry_run: Whether to run in dry-run mode

    """
    self.ap_client = ap_client
    self.console_logger = console_logger
    self.error_logger = error_logger
    self.db_verify_logger = db_verify_logger
    self.analytics = analytics
    self.config = config
    self.dry_run = dry_run
    self._dry_run_actions: list[DryRunAction] = []
    self._verify_start_time: float = 0.0

should_auto_verify async

should_auto_verify()

Check if automatic database verification should run.

Returns True if: - auto_verify_days has passed since last verification - No previous verification exists

Returns:

Type	Description
bool	True if auto-verify should run, False otherwise

Source code in src/app/features/verify/database_verifier.py

async def should_auto_verify(self) -> bool:
    """Check if automatic database verification should run.

    Returns True if:
    - auto_verify_days has passed since last verification
    - No previous verification exists

    Returns:
        True if auto-verify should run, False otherwise

    """
    auto_verify_days = self.config.database_verification.auto_verify_days

    if auto_verify_days <= 0:
        return False

    csv_path = get_full_log_path(
        self.config,
        "csv_output_file",
        "csv/track_list.csv",
    )
    last_verify_file = csv_path.replace(".csv", LAST_VERIFY_SUFFIX)
    last_verify_path = Path(last_verify_file)

    if not last_verify_path.exists():
        self.console_logger.debug("No previous verification found, auto-verify needed")
        return True

    try:
        loop = asyncio.get_running_loop()

        def _read_last_verify() -> str:
            with last_verify_path.open(encoding="utf-8") as f:
                return f.read().strip()

        last_verify_str = await loop.run_in_executor(None, _read_last_verify)
        last_verify = datetime.fromisoformat(last_verify_str)

        if last_verify.tzinfo is None:
            last_verify = last_verify.replace(tzinfo=UTC)

        days_since = (datetime.now(tz=UTC) - last_verify).days

        if days_since >= auto_verify_days:
            self.console_logger.info(
                "%s needed: %s days since last check %s",
                LogFormat.label("AUTO-VERIFY"),
                LogFormat.number(days_since),
                LogFormat.dim(f"(threshold: {auto_verify_days})"),
            )
            return True

        self.console_logger.debug(
            "Auto-verify not needed: %d days since last check (threshold: %d)",
            days_since,
            auto_verify_days,
        )
        return False

    except (OSError, ValueError, RuntimeError) as e:
        self.error_logger.warning("Error checking auto-verify status: %s", e)
        return True  # Run verification if we can't determine last run

can_run_incremental async

can_run_incremental(force_run=False)

Check if enough time has passed since the last incremental run.

Parameters:

Name	Type	Description	Default
force_run	bool	If True, skip the time check	False

Returns:

Type	Description
bool	True if incremental run should proceed, False otherwise

Source code in src/app/features/verify/database_verifier.py

async def can_run_incremental(self, force_run: bool = False) -> bool:
    """Check if enough time has passed since the last incremental run.

    Args:
        force_run: If True, skip the time check

    Returns:
        True if incremental run should proceed, False otherwise

    """
    if force_run:
        self.console_logger.info("Force run requested, skipping interval check")
        return True

    # Get configuration values
    interval_minutes = self.config.incremental_interval_minutes
    last_run_file = get_full_log_path(
        self.config,
        "last_incremental_run_file",
        "last_incremental_run.log",
    )

    # Check if the last run file exists
    last_run_path = Path(last_run_file)
    if not last_run_path.exists():
        self.console_logger.info(
            "No previous incremental run found, proceeding with run",
        )
        return True

    try:
        # Read last run time using async file operation
        loop = asyncio.get_running_loop()

        def _read_file() -> str:
            with last_run_path.open(encoding="utf-8") as f:
                return f.read().strip()

        last_run_str = await loop.run_in_executor(None, _read_file)

        # Try multiple datetime formats for compatibility
        try:
            last_run_time = datetime.fromisoformat(last_run_str)
            # Ensure timezone awareness - if naive, assume UTC
            if last_run_time.tzinfo is None:
                last_run_time = last_run_time.replace(tzinfo=UTC)
        except ValueError:
            # Handle legacy format: YYYY-MM-DD HH:MM:SS
            try:
                last_run_time = datetime.strptime(last_run_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=UTC)
            except ValueError:
                # Handle date-only format: YYYY-MM-DD
                last_run_time = datetime.strptime(last_run_str, "%Y-%m-%d").replace(tzinfo=UTC)

        # Handle future timestamps (corrupted/invalid files)
        now = datetime.now(tz=UTC)
        if last_run_time > now:
            self.console_logger.warning(
                "Last run timestamp is in the future (%s). Treating as if no previous run exists.",
                last_run_time.strftime("%Y-%m-%d %H:%M"),
            )
            return True

        # Check if enough time has passed
        time_since_last = now - last_run_time
        required_interval = timedelta(minutes=interval_minutes)

        if time_since_last >= required_interval:
            self.console_logger.info(
                "Last run: %s. Sufficient time has passed, proceeding.",
                last_run_time.strftime("%Y-%m-%d %H:%M"),
            )
            return True

        # Not enough time has passed
        remaining = required_interval - time_since_last
        remaining_minutes = int(remaining.total_seconds() / 60)
        self.console_logger.info(
            "Last run: %s. Next run in %d minutes. Skipping.",
            last_run_time.strftime("%Y-%m-%d %H:%M"),
            remaining_minutes,
        )

    except (ValueError, OSError):
        self.error_logger.exception("Error reading last incremental run time")
        # On error, allow the run to proceed
        return True

    # Execution completed successfully - not enough time has passed
    return False

update_last_incremental_run async

update_last_incremental_run()

Update the timestamp of the last incremental run.

Source code in src/app/features/verify/database_verifier.py

async def update_last_incremental_run(self) -> None:
    """Update the timestamp of the last incremental run."""
    tracker = IncrementalRunTracker(self.config)
    await tracker.update_last_run_timestamp()

    self.console_logger.info(
        "Updated last incremental run timestamp in %s",
        tracker.get_last_run_file_path(),
    )

verify_and_clean_track_database async

verify_and_clean_track_database(
    force=False, apply_test_filter=False
)

Verify the track database against Music.app and remove invalid entries.

Parameters:

Name	Type	Description	Default
force	bool	Force verification even if recently done	False
apply_test_filter	bool	Apply test artist filter in dry-run mode	False

Returns:

Type	Description
int	Number of invalid tracks removed

Source code in src/app/features/verify/database_verifier.py

async def verify_and_clean_track_database(
    self,
    force: bool = False,
    apply_test_filter: bool = False,
) -> int:
    """Verify the track database against Music.app and remove invalid entries.

    Args:
        force: Force verification even if recently done
        apply_test_filter: Apply test artist filter in dry-run mode

    Returns:
        Number of invalid tracks removed

    """
    # Load configuration and database
    auto_verify_days = self.config.database_verification.auto_verify_days

    csv_path = get_full_log_path(
        self.config,
        "csv_output_file",
        "csv/track_list.csv",
    )
    track_dict = load_track_list(csv_path)
    existing_tracks = list(track_dict.values())

    if not existing_tracks:
        self.console_logger.info("No existing track database to verify")
        return 0

    # Check if verification should be skipped
    if await self._should_skip_verification(force, csv_path, auto_verify_days):
        return 0

    # Get tracks to verify (with optional test filter)
    tracks_to_verify = self._get_tracks_to_verify(existing_tracks, apply_test_filter)

    # Log verification start
    self._log_verify_start(len(tracks_to_verify))

    # Verify tracks using bulk ID comparison (single AppleScript call)
    invalid_tracks = await self._verify_tracks_bulk(tracks_to_verify)

    # Handle invalid tracks (removal or dry-run logging)
    self._handle_invalid_tracks(invalid_tracks, existing_tracks, csv_path)

    # Update last verification timestamp
    await self._update_verification_timestamp(csv_path)

    # Log verification complete
    removed_count = 0 if self.dry_run else len(invalid_tracks)
    self._log_verify_complete(len(tracks_to_verify), len(invalid_tracks), removed_count)

    return len(invalid_tracks)

get_dry_run_actions

get_dry_run_actions()

Get the list of dry-run actions recorded.

Returns:

Type	Description
list[DryRunAction]	List of dry-run action dictionaries

Source code in src/app/features/verify/database_verifier.py

def get_dry_run_actions(self) -> list[DryRunAction]:
    """Get the list of dry-run actions recorded.

    Returns:
        List of dry-run action dictionaries

    """
    return self._dry_run_actions