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)