83 lines
2.1 KiB
Python
83 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Run plex-label-sync on a cron schedule from env."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from croniter import croniter
|
|
|
|
CRON_MACROS = {
|
|
'@yearly': '0 0 1 1 *',
|
|
'@annually': '0 0 1 1 *',
|
|
'@monthly': '0 0 1 * *',
|
|
'@weekly': '0 0 * * 0',
|
|
'@daily': '0 0 * * *',
|
|
'@midnight': '0 0 * * *',
|
|
'@hourly': '0 * * * *',
|
|
}
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
print(f"[{datetime.now().astimezone().isoformat(timespec='seconds')}] {msg}", flush=True)
|
|
|
|
|
|
def zone() -> ZoneInfo:
|
|
name = os.environ.get('TZ') or 'UTC'
|
|
try:
|
|
return ZoneInfo(name)
|
|
except Exception as exc:
|
|
raise SystemExit(f"Invalid TZ '{name}': {exc}") from exc
|
|
|
|
|
|
def resolve_cron_expr() -> str:
|
|
for key in ('CRON_SCHEDULE', 'CRON', 'SYNC_CRON'):
|
|
value = os.environ.get(key, '').strip()
|
|
if value:
|
|
expr = CRON_MACROS.get(value.lower(), value)
|
|
if not croniter.is_valid(expr):
|
|
raise SystemExit(
|
|
f"Invalid {key} '{value}' (expected a 5-field cron expression)"
|
|
)
|
|
return expr
|
|
return '0 */6 * * *'
|
|
|
|
|
|
def next_run(expr: str, after: datetime) -> datetime:
|
|
return croniter(expr, after).get_next(datetime)
|
|
|
|
|
|
def run_sync() -> int:
|
|
log('Starting sync…')
|
|
result = subprocess.run([sys.executable, '/app/sync.py'], check=False)
|
|
log(f'Sync finished with code {result.returncode}.')
|
|
return result.returncode
|
|
|
|
|
|
def main() -> int:
|
|
tz = zone()
|
|
expr = resolve_cron_expr()
|
|
run_on_start = os.environ.get('RUN_ON_START', 'true').lower() in {'1', 'true', 'yes'}
|
|
|
|
log(f"plex-label-sync scheduler (TZ={tz.key}, CRON_SCHEDULE='{expr}')")
|
|
|
|
if run_on_start:
|
|
run_sync()
|
|
|
|
while True:
|
|
now = datetime.now(tz=tz)
|
|
nxt = next_run(expr, now)
|
|
sleep_for = max(1.0, (nxt - now).total_seconds())
|
|
log(f"Next run at {nxt.isoformat(timespec='seconds')} (sleep {sleep_for:.0f}s)")
|
|
time.sleep(sleep_for)
|
|
run_sync()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|