backend/upgrade_scheduler.py
Rama master · réplica de lectura del repositorio original.
"""Subtitle upgrade scheduler.
Migrated from threading.Timer to APScheduler SublarrScheduler in Phase 5 / P4.
The actual scan body lives in ``upgrade_tick()`` which is invoked via a
JobSpec registered by ``services.scheduler._build_default_jobs``.
Legacy adapter API:
- ``start_upgrade_scheduler(app)`` updates the APScheduler interval based
on the current config value. Called by settings-save paths and from
app_schedulers._start_schedulers.
- ``stop_upgrade_scheduler()`` is a no-op.
- ``get_upgrade_scheduler()`` returns a small proxy used by
routes/system/tasks for UI state only.
"""
import logging
from datetime import UTC, datetime, timedelta
logger = logging.getLogger(__name__)
DEFAULT_INTERVAL_HOURS = 0 # Disabled by default; user must opt in
#: Subtitles with a score below this value are candidates regardless of format.
UPGRADE_SCORE_THRESHOLD = 500
_COMMIT_EVERY = 200
_state = {
"executing": False,
"last_run_at": None,
"interval_hours": DEFAULT_INTERVAL_HOURS,
}
# ---------------------------------------------------------------------------
# Public adapter API
# ---------------------------------------------------------------------------
def get_upgrade_scheduler():
"""Return a thin adapter for routes/system/tasks.py state rendering."""
return _UpgradeSchedulerProxy()
class _UpgradeSchedulerProxy:
"""Backwards-compatible shim exposing fields the tasks UI reads."""
@property
def _running(self) -> bool:
return _get_interval_hours() > 0
@property
def is_executing(self) -> bool:
return _state["executing"]
@property
def last_run_at(self):
return _state["last_run_at"]
@property
def next_run_at(self):
last = _state["last_run_at"]
interval = _state["interval_hours"]
if not last or not interval:
return None
try:
return (last + timedelta(hours=interval)).isoformat()
except Exception:
return None
def _execute_scan(self) -> dict:
return _execute_scan()
def start_upgrade_scheduler(app):
"""Adapter that updates the APScheduler upgrade job's interval.
Reads ``upgrade_scan_interval_hours`` from config. If the interval is
0 (disabled), the job is paused via ``scheduler.pause_job`` so that
the underlying JobSpec row stays in the store but never fires.
"""
from apscheduler.triggers.interval import IntervalTrigger
scheduler = app.extensions.get("scheduler") if app is not None else None
interval = _get_interval_hours()
_state["interval_hours"] = interval
if scheduler is None:
logger.debug("upgrade_scheduler: SublarrScheduler not available yet")
return
try:
if interval <= 0:
try:
scheduler.pause_job("upgrade_scan")
logger.info("upgrade: paused (interval=0)")
except Exception:
logger.debug("upgrade: pause_job failed (job may not exist yet)", exc_info=True)
return
scheduler.modify_trigger("upgrade_scan", IntervalTrigger(hours=max(1, interval)))
try:
scheduler.resume_job("upgrade_scan")
except Exception:
logger.debug(
"upgrade: could not resume the upgrade_scan job after changing its interval",
exc_info=True,
)
logger.info("upgrade: interval set to %dh via adapter", interval)
except Exception:
logger.error("upgrade: adapter failed", exc_info=True)
def stop_upgrade_scheduler():
"""No-op — APScheduler handles shutdown via SublarrScheduler.shutdown()."""
return None
# ---------------------------------------------------------------------------
# Tick + execution
# ---------------------------------------------------------------------------
def upgrade_tick() -> None:
"""Module-level tick function invoked by APScheduler."""
result = _execute_scan()
logger.info(
"Upgrade scan complete: %d queued, %d skipped",
result.get("queued", 0),
result.get("skipped", 0),
)
def _get_interval_hours() -> int:
try:
from config import get_settings
return get_settings().upgrade_scan_interval_hours
except Exception as e:
logger.debug("Could not read upgrade scan interval: %s", e)
return DEFAULT_INTERVAL_HOURS
def _execute_scan() -> dict:
"""Core scan logic: find eligible downloads and re-queue wanted items."""
from sqlalchemy import select
from sqlalchemy import update as sa_update
from config import get_settings
from db import get_db
from db.models.core import WantedItem
from db.models.providers import SubtitleDownload
from db.repositories.wanted import WantedRepository
_state["executing"] = True
queued = 0
skipped = 0
try:
settings = get_settings()
if not getattr(settings, "upgrade_enabled", True):
logger.info("upgrade_scan: upgrade_enabled=False — skipping scan")
return {"queued": 0, "skipped": 0}
window_days = settings.upgrade_window_days
interval_h = settings.upgrade_scan_interval_hours
# Only consider downloads old enough that the 2× delta protection has expired
cutoff_download = datetime.now(UTC) - timedelta(days=window_days)
# Don't re-queue items that were already searched recently
cutoff_search = datetime.now(UTC) - timedelta(hours=max(interval_h, 24))
# get_db() returns the Flask-SQLAlchemy scoped session — it is NOT a
# context manager ("with get_db() as db:" raises TypeError), and
# repositories bind to that session themselves (no constructor arg).
db = get_db()
wanted_repo = WantedRepository()
# Find subtitle_downloads older than window_days. Only the four columns
# the decision reads: whole ORM rows (decision logs and all) of a
# large download history cost a Pi hundreds of megabytes per scan.
stmt = select(
SubtitleDownload.source,
SubtitleDownload.score,
SubtitleDownload.format,
SubtitleDownload.file_path,
).where(SubtitleDownload.downloaded_at < cutoff_download)
downloads = db.execute(stmt).all()
for dl in downloads:
# Phase-1 provisional-MT guard (feature #8): defense-in-depth
# skip of the MT row itself. The PRIMARY/complete guard is the
# "provisional" status check below -- a translation also writes
# a second subtitle_downloads row for the site-provider subtitle
# (source="provider", file_path=the video path), which resolves
# back to the same wanted item and is NOT source==
# "machine_translation", so it would slip past this check alone.
# The original-only re-seek of these rows is feature #8b.
if (dl.source or "") == "machine_translation":
skipped += 1
continue
# Only consider low-score downloads or non-ASS formats
is_low_score = (dl.score or 0) < UPGRADE_SCORE_THRESHOLD
is_not_ass = (dl.format or "").lower() not in ("ass", "ssa")
if not (is_low_score or is_not_ass):
skipped += 1
continue
# Find corresponding wanted item by file_path
item = wanted_repo.get_wanted_by_file_path(dl.file_path)
if not item:
skipped += 1
continue
# Skip if already in "wanted" / "searching" / "provisional" state.
# "provisional" is the primary guard against re-reactivating an
# MT-kept-seeking item via its sibling site-provider download row
# (see the machine_translation early-skip comment above).
if item.get("status") in ("wanted", "searching", "provisional"):
skipped += 1
continue
# Skip if searched too recently
last_search = item.get("last_search_at")
if last_search:
try:
last_dt = datetime.fromisoformat(last_search)
if last_dt.tzinfo is None:
last_dt = last_dt.replace(tzinfo=UTC)
if last_dt > cutoff_search:
skipped += 1
continue
except Exception:
logger.debug(
"upgrade: could not parse the last search time, not skipping the item",
exc_info=True,
)
# Re-queue as upgrade candidate
db.execute(
sa_update(WantedItem)
.where(WantedItem.id == item["id"])
.values(
status="wanted",
upgrade_candidate=1,
current_score=dl.score or 0,
updated_at=datetime.now(UTC),
)
)
if queued % _COMMIT_EVERY == _COMMIT_EVERY - 1:
db.commit() # one commit per batch, not per row (SD cards)
logger.debug(
"Upgrade candidate queued: wanted_id=%d file=%s score=%d format=%s",
item["id"],
dl.file_path,
dl.score or 0,
dl.format or "?",
)
queued += 1
db.commit()
finally:
_state["executing"] = False
_state["last_run_at"] = datetime.now(UTC)
return {"queued": queued, "skipped": skipped}
# ---------------------------------------------------------------------------
# Legacy class retained only for tests that instantiate it directly.
# New code should use the adapter functions / ``upgrade_tick()`` above.
# ---------------------------------------------------------------------------
class UpgradeScheduler:
"""Backwards-compatible shim around the module-level tick functions.
Retained so existing unit tests (``tests/test_upgrade_scheduler.py``)
that exercise ``_execute_scan`` / ``_get_interval_hours`` semantics
continue to work without rewrites.
"""
def __init__(self, app):
self._app = app
self._timer = None
self._running = False
@property
def is_executing(self) -> bool:
return _state["executing"]
@property
def last_run_at(self):
return _state["last_run_at"]
def start(self):
interval = self._get_interval_hours()
if interval <= 0:
logger.info("Upgrade scheduler disabled (interval=0)")
self._running = False
self._timer = None
return
# Legacy tests assert ``_timer is not None`` after start().
self._running = True
self._timer = object()
def stop(self):
self._running = False
self._timer = None
def run_now(self) -> dict:
with self._app.app_context():
return _execute_scan()
def _get_interval_hours(self) -> int:
return _get_interval_hours()
def _execute_scan(self) -> dict:
return _execute_scan()