#!/usr/bin/env python3 import sys import os import time sys.path.insert(0, '/website') os.chdir('/website') from apscheduler.schedulers.background import BackgroundScheduler from schedular import expire_files, downgrade_expired_subscriptions, send_manual_renewal_reminders import sqlite3 import json import time as _time import calendar as _cal from datetime import datetime as _dt def _expire_overdue_jobs(): _db_path = '/website/storage/sqlite.db' _FALLBACK_SECS = 7200 try: with sqlite3.connect(_db_path) as _conn: _conn.row_factory = sqlite3.Row _rows = _conn.execute(""" SELECT j.id, j.started_at, COALESCE(p.max_processing_seconds, 0) AS plan_max_secs FROM local_jobs j LEFT JOIN conversions c ON c.job_id = j.id LEFT JOIN users u ON u.id = c.user_id LEFT JOIN plans p ON p.id = COALESCE(u.plan_id, 1) WHERE j.status = 'started' AND j.started_at IS NOT NULL GROUP BY j.id """).fetchall() _now = _time.time() _overdue = [] for _row in _rows: _max_secs = int(_row['plan_max_secs'] or 0) if _max_secs <= 0: _max_secs = _FALLBACK_SECS try: _started = _dt.strptime(_row['started_at'], '%Y-%m-%d %H:%M:%S') _elapsed = _now - _cal.timegm(_started.timetuple()) if _elapsed > _max_secs: _overdue.append(_row['id']) except Exception: pass if _overdue: _msg = json.dumps({ 'error': True, 'message': 'Job expired: exceeded maximum processing time.', 'results': [] }) _ph = ','.join('?' * len(_overdue)) _conn.execute( f"UPDATE local_jobs SET status='failed', result_json=?, completed_at=datetime('now') WHERE id IN ({_ph})", [_msg] + _overdue ) _conn.commit() except Exception as e: print(f'Scheduler error: {e}') # --------------------------------------------------------------------------- # Backup sync jobs — run_local_sync() and run_ftp_sync() are called by # APScheduler. Settings are re-read from the DB on every execution so that # interval or path changes take effect without restarting the scheduler. # --------------------------------------------------------------------------- _DB_PATH = '/website/storage/sqlite.db' def _read_sync_settings() -> dict: """Read backup-sync related keys from site_settings table.""" defaults = { 'local_sync_enabled': '0', 'local_sync_path': '', 'local_sync_interval_secs': '60', 'ftp_sync_enabled': '0', 'ftp_host': '', 'ftp_port': '21', 'ftp_user': '', 'ftp_password': '', 'ftp_remote_path': '/', 'ftp_interval_mins': '60', } try: with sqlite3.connect(_DB_PATH) as conn: rows = conn.execute( "SELECT key, value FROM site_settings WHERE key LIKE 'local_sync%' OR key LIKE 'ftp_%'" ).fetchall() for k, v in rows: defaults[k] = v except Exception as e: print(f'[backup_sync] Could not read settings: {e}') return defaults def run_local_sync(): """APScheduler job: incremental local-path sync (runs every N seconds).""" settings = _read_sync_settings() if settings.get('local_sync_enabled') != '1': return path = (settings.get('local_sync_path') or '').strip() if not path: return try: from backup_sync import LocalSyncer result = LocalSyncer(path).run_sync() print(f'[local_sync] {result}') except Exception as exc: print(f'[local_sync] Error: {exc}') def run_ftp_sync(): """APScheduler job: incremental FTP sync (runs every N minutes).""" settings = _read_sync_settings() if settings.get('ftp_sync_enabled') != '1': return host = (settings.get('ftp_host') or '').strip() if not host: return try: from backup_sync import FTPSyncer syncer = FTPSyncer( host=host, port=int(settings.get('ftp_port') or 21), user=(settings.get('ftp_user') or '').strip(), password=(settings.get('ftp_password') or '').strip(), remote_path=(settings.get('ftp_remote_path') or '/').strip(), ) result = syncer.run_sync() print(f'[ftp_sync] {result}') except Exception as exc: print(f'[ftp_sync] Error: {exc}') # --------------------------------------------------------------------------- # Dynamic-interval job wrappers # --------------------------------------------------------------------------- # APScheduler does not natively support intervals that change at runtime. # We schedule both jobs at fixed short intervals and let each function decide # whether enough time has elapsed since the last actual run (tracked in the # state JSON file). This avoids restarting the scheduler when admins change # the interval. _LOCAL_POLL_SECS = 30 # check every 30 s; actual sync obeys local_sync_interval_secs _FTP_POLL_SECS = 60 # check every 60 s; actual sync obeys ftp_interval_mins def _run_local_sync_gated(): """Run local sync only if configured interval has elapsed.""" import backup_sync as _bs state = _bs._load_state() settings = _read_sync_settings() interval = int(settings.get('local_sync_interval_secs') or 60) last_ts = state.get('local_last_sync') if last_ts: try: last_dt = _dt.strptime(last_ts, '%Y-%m-%d %H:%M:%S UTC') elapsed = (_dt.utcnow() - last_dt).total_seconds() if elapsed < interval: return except Exception: pass run_local_sync() def _run_ftp_sync_gated(): """Run FTP sync only if configured interval has elapsed.""" import backup_sync as _bs state = _bs._load_state() settings = _read_sync_settings() interval_mins = int(settings.get('ftp_interval_mins') or 60) interval_secs = interval_mins * 60 last_ts = state.get('ftp_last_sync') if last_ts: try: last_dt = _dt.strptime(last_ts, '%Y-%m-%d %H:%M:%S UTC') elapsed = (_dt.utcnow() - last_dt).total_seconds() if elapsed < interval_secs: return except Exception: pass run_ftp_sync() # --------------------------------------------------------------------------- # Create and start scheduler # --------------------------------------------------------------------------- sched = BackgroundScheduler(daemon=True) sched.add_job(expire_files, 'interval', minutes=60) sched.add_job(_expire_overdue_jobs, 'interval', seconds=60) sched.add_job(downgrade_expired_subscriptions, 'interval', minutes=5) sched.add_job(send_manual_renewal_reminders, 'interval', hours=24) sched.add_job(_run_local_sync_gated, 'interval', seconds=_LOCAL_POLL_SECS) sched.add_job(_run_ftp_sync_gated, 'interval', seconds=_FTP_POLL_SECS) sched.start() # Run once immediately on startup to catch any expirations that occurred while offline try: downgrade_expired_subscriptions() except Exception as _e: print(f'[subscription] Startup downgrade check failed: {_e}') print('✓ Scheduler started successfully (with backup sync + subscription downgrade jobs)!') # Keep scheduler running while True: time.sleep(10)