462 lines
17 KiB
Python
462 lines
17 KiB
Python
import datetime
|
|
import logging
|
|
import os
|
|
import re
|
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from sqlalchemy import or_
|
|
from models import Job, Recording, get_session, prune_logs, utcnow
|
|
|
|
logger = logging.getLogger("scheduler")
|
|
|
|
DAY_MAP = {
|
|
"mo": 1, "mon": 1, "montag": 1,
|
|
"di": 2, "tue": 2, "dienstag": 2,
|
|
"mi": 3, "wed": 3, "mittwoch": 3,
|
|
"do": 4, "thu": 4, "donnerstag": 4,
|
|
"fr": 5, "fri": 5, "freitag": 5,
|
|
"sa": 6, "sat": 6, "samstag": 6,
|
|
"so": 7, "sun": 7, "sonntag": 7,
|
|
}
|
|
|
|
_TIME_RE = re.compile(r"^([01]\d|2[0-3]):([0-5]\d)$")
|
|
|
|
|
|
def _day_to_num(day: str) -> int:
|
|
return DAY_MAP.get(day.lower().strip(), 0)
|
|
|
|
|
|
def _matches_day(today_num: int, schedule: str) -> bool:
|
|
if schedule.strip() == "*":
|
|
return True
|
|
for entry in schedule.split(","):
|
|
entry = entry.strip()
|
|
if "-" in entry:
|
|
parts = entry.split("-", 1)
|
|
start = _day_to_num(parts[0])
|
|
end = _day_to_num(parts[1])
|
|
if not start or not end:
|
|
continue
|
|
if start <= end:
|
|
if start <= today_num <= end:
|
|
return True
|
|
else:
|
|
if today_num >= start or today_num <= end:
|
|
return True
|
|
else:
|
|
if _day_to_num(entry) == today_num:
|
|
return True
|
|
return False
|
|
|
|
|
|
def is_valid_time(t) -> bool:
|
|
return isinstance(t, str) and bool(_TIME_RE.match(t))
|
|
|
|
|
|
def is_valid_date(d) -> bool:
|
|
if not isinstance(d, str) or not re.match(r"^\d{4}-\d{2}-\d{2}$", d):
|
|
return False
|
|
try:
|
|
datetime.date.fromisoformat(d)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def is_valid_days(days) -> bool:
|
|
if not isinstance(days, str):
|
|
return False
|
|
days = days.strip()
|
|
if days == "*":
|
|
return True
|
|
if not days:
|
|
return False
|
|
for entry in days.split(","):
|
|
entry = entry.strip()
|
|
parts = entry.split("-", 1) if "-" in entry else [entry]
|
|
if not all(_day_to_num(p) for p in parts):
|
|
return False
|
|
return True
|
|
|
|
|
|
def validate_schedule(job: dict):
|
|
"""Prüft Zeitplanfelder. Wirft ValueError mit deutscher Meldung."""
|
|
start = job.get("schedule_start") or ""
|
|
stop = job.get("schedule_stop") or ""
|
|
date = job.get("schedule_date") or ""
|
|
end_date = job.get("schedule_end_date") or ""
|
|
days = job.get("schedule_days")
|
|
days = "*" if days is None else days
|
|
|
|
if start and not is_valid_time(start):
|
|
raise ValueError("Startzeit muss im Format HH:MM angegeben werden")
|
|
if stop and not is_valid_time(stop):
|
|
raise ValueError("Stoppzeit muss im Format HH:MM angegeben werden")
|
|
if date and not is_valid_date(date):
|
|
raise ValueError("Datum muss im Format JJJJ-MM-TT angegeben werden")
|
|
if end_date and not is_valid_date(end_date):
|
|
raise ValueError("Enddatum muss im Format JJJJ-MM-TT angegeben werden")
|
|
if not is_valid_days(days):
|
|
raise ValueError("Ungültige Wochentage (erlaubt: * oder z.B. mo-fr,sa)")
|
|
if job.get("schedule_enabled") and not start:
|
|
raise ValueError("Für einen aktiven Zeitplan ist eine Startzeit erforderlich")
|
|
if job.get("schedule_once") and not date:
|
|
raise ValueError("Für eine einmalige Aufnahme ist ein Datum erforderlich")
|
|
if end_date:
|
|
if not date:
|
|
raise ValueError("Enddatum erfordert ein Startdatum")
|
|
if end_date < date:
|
|
raise ValueError("Enddatum darf nicht vor dem Startdatum liegen")
|
|
|
|
|
|
def _time_to_minutes(t: str) -> int:
|
|
if not is_valid_time(t):
|
|
raise ValueError(f"Ungültige Uhrzeit: {t!r}")
|
|
parts = t.split(":")
|
|
return int(parts[0]) * 60 + int(parts[1])
|
|
|
|
|
|
def _in_time_window(now_time: str, start: str, stop: str) -> bool:
|
|
now_min = _time_to_minutes(now_time)
|
|
start_min = _time_to_minutes(start)
|
|
stop_min = _time_to_minutes(stop)
|
|
if start_min <= stop_min:
|
|
return start_min <= now_min < stop_min
|
|
else:
|
|
return now_min >= start_min or now_min < stop_min
|
|
|
|
|
|
def _apply_buffers(job: dict) -> dict:
|
|
eff = dict(job)
|
|
pre = job.get("pre_buffer", 0) or 0
|
|
post = job.get("post_buffer", 0) or 0
|
|
if pre > 0 and is_valid_time(eff.get("schedule_start")):
|
|
m = _time_to_minutes(eff["schedule_start"]) - pre
|
|
m %= 1440
|
|
eff["schedule_start"] = f"{m // 60:02d}:{m % 60:02d}"
|
|
if post > 0 and is_valid_time(eff.get("schedule_stop")):
|
|
m = _time_to_minutes(eff["schedule_stop"]) + post
|
|
m %= 1440
|
|
eff["schedule_stop"] = f"{m // 60:02d}:{m % 60:02d}"
|
|
return eff
|
|
|
|
|
|
def _is_multi_day(job: dict) -> bool:
|
|
return bool(job.get("schedule_once") and job.get("schedule_date") and job.get("schedule_end_date"))
|
|
|
|
|
|
def compute_duration(job: dict, apply_buffers: bool) -> int | None:
|
|
"""Aufnahmedauer in Sekunden aus max_duration bzw. Zeitplan. Wirft ValueError bei ungültigen Daten."""
|
|
duration = job.get("max_duration")
|
|
if duration:
|
|
return int(duration)
|
|
start = job.get("schedule_start") or ""
|
|
stop = job.get("schedule_stop") or ""
|
|
if not start or not stop:
|
|
return None
|
|
|
|
pre = (job.get("pre_buffer") or 0) if apply_buffers else 0
|
|
post = (job.get("post_buffer") or 0) if apply_buffers else 0
|
|
|
|
if _is_multi_day(job):
|
|
try:
|
|
start_dt = datetime.datetime.strptime(f"{job['schedule_date']} {start}", "%Y-%m-%d %H:%M")
|
|
end_dt = datetime.datetime.strptime(f"{job['schedule_end_date']} {stop}", "%Y-%m-%d %H:%M")
|
|
except (TypeError, ValueError):
|
|
raise ValueError("Ungültiges Datum oder Uhrzeit im Zeitplan")
|
|
start_dt -= datetime.timedelta(minutes=pre)
|
|
end_dt += datetime.timedelta(minutes=post)
|
|
return max(0, int((end_dt - start_dt).total_seconds()))
|
|
|
|
eff = _apply_buffers({**job, "pre_buffer": pre, "post_buffer": post})
|
|
start_min = _time_to_minutes(eff["schedule_start"])
|
|
stop_min = _time_to_minutes(eff["schedule_stop"])
|
|
if stop_min > start_min:
|
|
return (stop_min - start_min) * 60
|
|
return (1440 - start_min + stop_min) * 60
|
|
|
|
|
|
class StreamScheduler:
|
|
def __init__(self, recorder):
|
|
self.recorder = recorder
|
|
self._scheduler = BackgroundScheduler()
|
|
self._started_keys: set[str] = set()
|
|
self._last_errors: dict[int, str] = {}
|
|
|
|
def start(self):
|
|
self._scheduler.add_job(self._tick, "interval", seconds=30, id="scheduler_tick",
|
|
replace_existing=True, max_instances=1)
|
|
self._scheduler.add_job(self.recorder.check_scheduled_stops, "interval", seconds=10,
|
|
id="check_stops", replace_existing=True, max_instances=1)
|
|
self._scheduler.add_job(self.prune_logs, "interval", hours=24,
|
|
id="prune_logs", replace_existing=True, max_instances=1)
|
|
self._scheduler.start()
|
|
|
|
def stop(self):
|
|
self._scheduler.shutdown(wait=False)
|
|
|
|
@staticmethod
|
|
def prune_logs():
|
|
try:
|
|
days = int(os.environ.get("LOG_RETENTION_DAYS", "90"))
|
|
except ValueError:
|
|
days = 90
|
|
return prune_logs(days)
|
|
|
|
def _original_occ_key(self, job: dict, now: datetime.datetime | None = None) -> str:
|
|
now = now or datetime.datetime.now()
|
|
today_date = now.strftime("%Y-%m-%d")
|
|
date = today_date if not job["schedule_once"] else job["schedule_date"]
|
|
return f"{job['id']}|{job['schedule_once']}|{date}|{job['schedule_start']}"
|
|
|
|
def _schedule_state(self, job: dict, now: datetime.datetime | None = None) -> dict:
|
|
now = now or datetime.datetime.now()
|
|
today_num = now.isoweekday()
|
|
today_date = now.strftime("%Y-%m-%d")
|
|
now_min = now.hour * 60 + now.minute
|
|
|
|
occ_key = (
|
|
f"{job['id']}|{job['schedule_once']}|"
|
|
f"{today_date if not job['schedule_once'] else job['schedule_date']}|"
|
|
f"{job['schedule_start']}"
|
|
)
|
|
|
|
multi_day = _is_multi_day(job)
|
|
state = {
|
|
"occ_key": occ_key,
|
|
"in_window": False,
|
|
"multi_day": multi_day,
|
|
"is_cross_midnight": False,
|
|
}
|
|
|
|
start = job.get("schedule_start") or ""
|
|
stop = job.get("schedule_stop") or ""
|
|
if not is_valid_time(start):
|
|
return state
|
|
if not is_valid_time(stop):
|
|
stop = ""
|
|
days = job.get("schedule_days") or "*"
|
|
|
|
is_today = False
|
|
if job["schedule_once"]:
|
|
if job["schedule_date"] == today_date:
|
|
is_today = True
|
|
else:
|
|
if _matches_day(today_num, days):
|
|
is_today = True
|
|
|
|
start_min = _time_to_minutes(start)
|
|
in_window = False
|
|
is_cross_midnight = False
|
|
if is_today:
|
|
if multi_day or not stop:
|
|
in_window = now_min >= start_min
|
|
else:
|
|
stop_min = _time_to_minutes(stop)
|
|
if stop_min <= start_min:
|
|
is_cross_midnight = True
|
|
in_window = now_min >= start_min
|
|
else:
|
|
in_window = start_min <= now_min < stop_min
|
|
elif stop:
|
|
stop_min = _time_to_minutes(stop)
|
|
if stop_min <= start_min and stop_min > 0 and now_min < stop_min:
|
|
yesterday = now - datetime.timedelta(days=1)
|
|
if job["schedule_once"]:
|
|
was_yesterday = job["schedule_date"] == yesterday.strftime("%Y-%m-%d")
|
|
else:
|
|
was_yesterday = _matches_day(yesterday.isoweekday(), days)
|
|
if was_yesterday:
|
|
is_cross_midnight = True
|
|
in_window = True
|
|
|
|
state["in_window"] = in_window
|
|
state["is_cross_midnight"] = is_cross_midnight
|
|
return state
|
|
|
|
def suppress_current_occurrence(self, job_id: int) -> bool:
|
|
session = get_session()
|
|
try:
|
|
j = session.query(Job).filter_by(id=job_id).first()
|
|
if not j or not j.schedule_enabled or not j.schedule_start:
|
|
return False
|
|
job = {
|
|
"id": j.id,
|
|
"schedule_once": j.schedule_once,
|
|
"schedule_date": j.schedule_date,
|
|
"schedule_days": j.schedule_days,
|
|
"schedule_start": j.schedule_start,
|
|
"schedule_stop": j.schedule_stop,
|
|
"schedule_end_date": j.schedule_end_date,
|
|
"pre_buffer": j.pre_buffer or 0,
|
|
"post_buffer": j.post_buffer or 0,
|
|
}
|
|
finally:
|
|
session.close()
|
|
|
|
try:
|
|
occ_key = self._original_occ_key(job)
|
|
state = self._schedule_state(_apply_buffers(job))
|
|
except Exception as e:
|
|
logger.warning("Zeitplan von Job %s ungültig: %s", job_id, e)
|
|
return False
|
|
if not state["in_window"]:
|
|
return False
|
|
|
|
self._started_keys.add(occ_key)
|
|
return True
|
|
|
|
def _occurrence_start_utc(self, job: dict, now: datetime.datetime) -> datetime.datetime | None:
|
|
date_text = job["schedule_date"] if job["schedule_once"] else now.strftime("%Y-%m-%d")
|
|
try:
|
|
local_start = datetime.datetime.strptime(
|
|
f"{date_text} {job['schedule_start']}",
|
|
"%Y-%m-%d %H:%M",
|
|
)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return local_start.astimezone(datetime.timezone.utc).replace(tzinfo=None)
|
|
|
|
def _has_recording_for_occurrence(self, job: dict, now: datetime.datetime) -> bool:
|
|
occurrence_start = self._occurrence_start_utc(job, now)
|
|
if occurrence_start is None:
|
|
return False
|
|
|
|
session = get_session()
|
|
try:
|
|
return session.query(Recording).filter(
|
|
Recording.job_id == job["id"],
|
|
or_(
|
|
Recording.started_at >= occurrence_start,
|
|
Recording.stopped_at >= occurrence_start,
|
|
),
|
|
).first() is not None
|
|
finally:
|
|
session.close()
|
|
|
|
def _is_job_recording(self, job_id: int) -> bool:
|
|
session = get_session()
|
|
try:
|
|
return session.query(Recording).filter(
|
|
Recording.job_id == job_id,
|
|
Recording.status.in_(["running", "starting"]),
|
|
).first() is not None
|
|
finally:
|
|
session.close()
|
|
|
|
def _get_active_scheduled_recording_for_job(self, job_id: int, *, respect_stop: bool = False):
|
|
session = get_session()
|
|
try:
|
|
rec = session.query(Recording).filter(
|
|
Recording.job_id == job_id,
|
|
Recording.status.in_(["running", "starting"]),
|
|
Recording.is_scheduled == True,
|
|
).first()
|
|
if not rec:
|
|
return None
|
|
if respect_stop and rec.scheduled_stop and rec.scheduled_stop > utcnow():
|
|
return None
|
|
return rec.id
|
|
finally:
|
|
session.close()
|
|
|
|
def _prune_started_keys(self, now: datetime.datetime):
|
|
cutoff = (now - datetime.timedelta(days=2)).strftime("%Y-%m-%d")
|
|
for key in list(self._started_keys):
|
|
parts = key.split("|")
|
|
date = parts[2] if len(parts) >= 4 else ""
|
|
if not is_valid_date(date) or date < cutoff:
|
|
self._started_keys.discard(key)
|
|
|
|
def _tick(self):
|
|
now = datetime.datetime.now()
|
|
self._prune_started_keys(now)
|
|
|
|
session = get_session()
|
|
try:
|
|
jobs = session.query(Job).filter_by(schedule_enabled=True).all()
|
|
job_list = []
|
|
for j in jobs:
|
|
job_list.append({
|
|
"id": j.id, "name": j.name, "stream_url": j.stream_url,
|
|
"stream_type": j.stream_type, "output_format": j.output_format,
|
|
"max_duration": j.max_duration, "extra_ffmpeg_args": j.extra_ffmpeg_args,
|
|
"segment_duration": j.segment_duration,
|
|
"schedule_once": j.schedule_once, "schedule_date": j.schedule_date,
|
|
"schedule_days": j.schedule_days, "schedule_start": j.schedule_start,
|
|
"schedule_stop": j.schedule_stop, "schedule_end_date": j.schedule_end_date,
|
|
"pre_buffer": j.pre_buffer or 0, "post_buffer": j.post_buffer or 0,
|
|
"description": j.description,
|
|
"ntfy_enabled": j.ntfy_enabled,
|
|
"plik_enabled": j.plik_enabled,
|
|
"delete_after_upload": j.delete_after_upload,
|
|
})
|
|
finally:
|
|
session.close()
|
|
|
|
active_ids = set()
|
|
for job in job_list:
|
|
active_ids.add(job["id"])
|
|
try:
|
|
self._process_job(job, now)
|
|
self._last_errors.pop(job["id"], None)
|
|
except Exception as e:
|
|
err = f"{type(e).__name__}: {e}"
|
|
if self._last_errors.get(job["id"]) != err:
|
|
self._last_errors[job["id"]] = err
|
|
logger.error("Zeitplan-Fehler bei Job %s (%s): %s", job["id"], job.get("name"), err)
|
|
try:
|
|
self.recorder._log("ERROR", f"Zeitplan-Fehler: {e}", job.get("name"))
|
|
except Exception:
|
|
pass
|
|
|
|
for job_id in list(self._last_errors):
|
|
if job_id not in active_ids:
|
|
self._last_errors.pop(job_id, None)
|
|
|
|
def _process_job(self, job: dict, now: datetime.datetime):
|
|
if not job["schedule_start"]:
|
|
return
|
|
|
|
occ_key = self._original_occ_key(job, now)
|
|
eff_job = _apply_buffers(job)
|
|
state = self._schedule_state(eff_job, now)
|
|
|
|
if state["in_window"]:
|
|
if (
|
|
occ_key not in self._started_keys
|
|
and not self._is_job_recording(job["id"])
|
|
and not self._has_recording_for_occurrence(job, now)
|
|
):
|
|
duration = compute_duration(job, apply_buffers=True)
|
|
# Erst merken, damit ein Startfehler nicht alle 30s erneut versucht wird
|
|
self._started_keys.add(occ_key)
|
|
self.recorder.start_recording(
|
|
job_id=job["id"],
|
|
name=job["name"],
|
|
stream_url=job["stream_url"],
|
|
stream_type=job["stream_type"],
|
|
output_format=job["output_format"] or "",
|
|
max_duration=duration,
|
|
extra_ffmpeg_args=job["extra_ffmpeg_args"] or "",
|
|
segment_duration=job["segment_duration"],
|
|
is_scheduled=True,
|
|
description=job.get("description", ""),
|
|
ntfy_enabled=job.get("ntfy_enabled", False),
|
|
plik_enabled=job.get("plik_enabled", False),
|
|
delete_after_upload=job.get("delete_after_upload", False),
|
|
)
|
|
else:
|
|
if (
|
|
not state["multi_day"]
|
|
and not state["is_cross_midnight"]
|
|
and eff_job["schedule_stop"]
|
|
and self._is_job_recording(job["id"])
|
|
):
|
|
rec_id = self._get_active_scheduled_recording_for_job(
|
|
job["id"], respect_stop=True,
|
|
)
|
|
if rec_id:
|
|
self.recorder.stop_recording(rec_id)
|
|
self._started_keys.discard(occ_key)
|