mirror of
https://github.com/wizarrrr/wizarr.git
synced 2026-07-30 23:07:19 -04:00
73c29458fe
- Fix 55 test failures caused by missing request contexts and incorrect session_transaction() usage across 8 test files - Fix ruff import sorting errors and unused imports - Fix 122 type errors: rename method override parameters to match base classes, add None guards for fetchone()/datetime, widen dict type annotations, add type: ignore for SQLAlchemy stub limitations - Add [tool.ty.rules] config to suppress unsupported-base warnings - Fix _ variable shadowing gettext in wizard routes - Add noqa: ARG002 for unused method arguments required by base class
256 lines
9.3 KiB
Python
256 lines
9.3 KiB
Python
"""
|
|
Maintenance tasks for Wizarr activity data.
|
|
|
|
These helpers support scheduled jobs that clean up or reconcile sessions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import structlog
|
|
|
|
try:
|
|
from app.extensions import db # type: ignore
|
|
except ImportError: # pragma: no cover - during tests
|
|
db = None # type: ignore
|
|
|
|
from app.models import ActivitySession
|
|
|
|
|
|
class ActivityMaintenanceService:
|
|
"""Lifecycle management for session data."""
|
|
|
|
def __init__(self):
|
|
self.logger = structlog.get_logger(__name__)
|
|
|
|
def cleanup_old_activity(self, retention_days: int = 90) -> int:
|
|
"""Delete activity sessions older than the retention window."""
|
|
if db is None:
|
|
return 0
|
|
|
|
try:
|
|
cutoff_date = datetime.now(UTC) - timedelta(days=retention_days)
|
|
deleted_count = (
|
|
db.session.query(ActivitySession) # type: ignore
|
|
.filter(ActivitySession.started_at < cutoff_date)
|
|
.delete()
|
|
)
|
|
|
|
db.session.commit() # type: ignore
|
|
self.logger.info("Cleaned up %s old activity sessions", deleted_count)
|
|
return deleted_count
|
|
|
|
except Exception as exc: # pragma: no cover - log and rollback
|
|
self.logger.error("Failed to cleanup old activity: %s", exc, exc_info=True)
|
|
db.session.rollback() # type: ignore
|
|
return 0
|
|
|
|
def end_stale_sessions(self, timeout_hours: int = 24) -> int:
|
|
"""Mark sessions as ended when they have not updated within timeout."""
|
|
if db is None:
|
|
return 0
|
|
|
|
try:
|
|
cutoff_time = datetime.now(UTC) - timedelta(hours=timeout_hours)
|
|
|
|
stale_sessions = (
|
|
db.session.query(ActivitySession) # type: ignore
|
|
.filter(
|
|
ActivitySession.active.is_(True),
|
|
ActivitySession.updated_at < cutoff_time,
|
|
)
|
|
.all()
|
|
)
|
|
|
|
ended_count = 0
|
|
for session in stale_sessions:
|
|
session.active = False
|
|
metadata = session.get_metadata()
|
|
metadata["status"] = "ended"
|
|
metadata["stale_closed_at"] = datetime.now(UTC).isoformat()
|
|
session.set_metadata(metadata)
|
|
self._end_session_gracefully(session)
|
|
ended_count += 1
|
|
|
|
if ended_count:
|
|
db.session.commit() # type: ignore
|
|
|
|
self.logger.info("Ended %s stale activity sessions", ended_count)
|
|
return ended_count
|
|
|
|
except Exception as exc: # pragma: no cover - log and rollback
|
|
self.logger.error("Failed to end stale sessions: %s", exc, exc_info=True)
|
|
db.session.rollback() # type: ignore
|
|
return 0
|
|
|
|
def recover_sessions_on_startup(self) -> int:
|
|
"""Validate sessions when the application boots and fix stale entries."""
|
|
if db is None:
|
|
return 0
|
|
|
|
try:
|
|
active_sessions = (
|
|
db.session.query(ActivitySession) # type: ignore
|
|
.filter(ActivitySession.active.is_(True))
|
|
.all()
|
|
)
|
|
|
|
if not active_sessions:
|
|
return 0
|
|
|
|
self.logger.info(
|
|
"Validating %s active sessions during startup", len(active_sessions)
|
|
)
|
|
|
|
sessions_by_server: dict[int, list[ActivitySession]] = {}
|
|
for session in active_sessions:
|
|
sessions_by_server.setdefault(session.server_id, []).append(session)
|
|
|
|
ended_count = 0
|
|
recovered_count = 0
|
|
|
|
for server_id, sessions in sessions_by_server.items():
|
|
try:
|
|
result = self._validate_server_sessions(server_id, sessions)
|
|
ended_count += result["ended"]
|
|
recovered_count += result["recovered"]
|
|
except Exception as exc:
|
|
self.logger.warning(
|
|
"Failed to validate sessions for server %s: %s",
|
|
server_id,
|
|
exc,
|
|
)
|
|
for session in sessions:
|
|
self._end_session_gracefully(session)
|
|
ended_count += 1
|
|
|
|
if ended_count or recovered_count:
|
|
db.session.commit() # type: ignore
|
|
self.logger.info(
|
|
"Session recovery completed: %s recovered, %s ended",
|
|
recovered_count,
|
|
ended_count,
|
|
)
|
|
|
|
return ended_count
|
|
|
|
except Exception as exc: # pragma: no cover
|
|
self.logger.error("Failed to recover sessions: %s", exc, exc_info=True)
|
|
db.session.rollback() # type: ignore
|
|
return 0
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
def _validate_server_sessions(
|
|
self,
|
|
server_id: int,
|
|
sessions: list[ActivitySession],
|
|
) -> dict[str, int]:
|
|
"""Poll the media server to verify which sessions are still active."""
|
|
from app.models import MediaServer
|
|
from app.services.media.service import get_client_for_media_server
|
|
|
|
ended_count = 0
|
|
recovered_count = 0
|
|
|
|
try:
|
|
server = db.session.query(MediaServer).filter_by(id=server_id).first() # type: ignore
|
|
if not server:
|
|
self.logger.warning(
|
|
"Server %s not found during validation. Ending sessions.", server_id
|
|
)
|
|
for session in sessions:
|
|
self._end_session_gracefully(session)
|
|
ended_count += 1
|
|
return {"ended": ended_count, "recovered": recovered_count}
|
|
|
|
client = get_client_for_media_server(server)
|
|
if not client:
|
|
self.logger.warning(
|
|
"No client for server %s. Ending active sessions.", server_id
|
|
)
|
|
for session in sessions:
|
|
self._end_session_gracefully(session)
|
|
ended_count += 1
|
|
return {"ended": ended_count, "recovered": recovered_count}
|
|
|
|
try:
|
|
current_sessions = client.now_playing()
|
|
except Exception as exc:
|
|
self.logger.warning(
|
|
"Failed to poll server %s sessions: %s", server_id, exc
|
|
)
|
|
cutoff_time = datetime.now(UTC) - timedelta(hours=1)
|
|
for session in sessions:
|
|
updated_at = session.updated_at
|
|
if updated_at.tzinfo is None: # type: ignore
|
|
updated_at = updated_at.replace(tzinfo=UTC) # type: ignore
|
|
if updated_at < cutoff_time:
|
|
self._end_session_gracefully(session)
|
|
ended_count += 1
|
|
else:
|
|
recovered_count += 1
|
|
return {"ended": ended_count, "recovered": recovered_count}
|
|
|
|
active_session_ids = set()
|
|
if current_sessions:
|
|
for current_session in current_sessions:
|
|
session_id = self._extract_session_id(current_session)
|
|
if session_id:
|
|
active_session_ids.add(session_id)
|
|
|
|
for session in sessions:
|
|
if session.session_id in active_session_ids:
|
|
recovered_count += 1
|
|
else:
|
|
self._end_session_gracefully(session)
|
|
ended_count += 1
|
|
|
|
except Exception as exc:
|
|
self.logger.error(
|
|
"Error validating sessions for server %s: %s", server_id, exc
|
|
)
|
|
cutoff_time = datetime.now(UTC) - timedelta(hours=1)
|
|
for session in sessions:
|
|
updated_at = session.updated_at
|
|
if updated_at.tzinfo is None: # type: ignore
|
|
updated_at = updated_at.replace(tzinfo=UTC) # type: ignore
|
|
if updated_at < cutoff_time:
|
|
self._end_session_gracefully(session)
|
|
ended_count += 1
|
|
else:
|
|
recovered_count += 1
|
|
|
|
return {"ended": ended_count, "recovered": recovered_count}
|
|
|
|
def _extract_session_id(self, current_session: dict) -> str | None:
|
|
"""Return the session identifier from a now_playing payload."""
|
|
possible_fields = ["sessionKey", "session_id", "id", "key", "playSessionId"]
|
|
for field in possible_fields:
|
|
value = current_session.get(field)
|
|
if value:
|
|
return str(value)
|
|
return None
|
|
|
|
def _end_session_gracefully(self, session: ActivitySession) -> None:
|
|
"""Mark a session inactive while estimating its final metadata."""
|
|
estimated_end_time = session.updated_at or session.started_at
|
|
|
|
session.active = False
|
|
metadata = session.get_metadata()
|
|
metadata["graceful_closed_at"] = (
|
|
estimated_end_time.isoformat() if estimated_end_time else None
|
|
)
|
|
metadata["status"] = "ended"
|
|
session.set_metadata(metadata)
|
|
|
|
self.logger.debug(
|
|
"Gracefully ended session %s",
|
|
session.session_id,
|
|
)
|
|
|
|
|
|
__all__ = ["ActivityMaintenanceService"]
|