Patrick AsmusandClaude Opus 4.6
b446ce9b3e
- Gitea CI-Workflows fuer Multi-Arch Docker Builds (amd64 + arm64) - Beschreibungsfeld und mehrtaegige Zeitplanung fuer Jobs - Benachrichtigungen auf Deutsch - docker-compose nutzt Registry-Image Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
212 lines
9.0 KiB
Python
212 lines
9.0 KiB
Python
import datetime
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from models import Job, Recording, get_session
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
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 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 _time_to_minutes(t: str) -> int:
|
|
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
|
|
|
|
|
|
class StreamScheduler:
|
|
def __init__(self, recorder):
|
|
self.recorder = recorder
|
|
self._scheduler = BackgroundScheduler()
|
|
self._started_keys: set[str] = set()
|
|
|
|
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.start()
|
|
|
|
def stop(self):
|
|
self._scheduler.shutdown(wait=False)
|
|
|
|
def _occurrence_key(self, job: Job) -> str:
|
|
today = datetime.date.today().isoformat()
|
|
if job.schedule_once:
|
|
today = job.schedule_date or today
|
|
return f"{job.id}|{job.schedule_once}|{today}|{job.schedule_start}"
|
|
|
|
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):
|
|
session = get_session()
|
|
try:
|
|
rec = session.query(Recording).filter(
|
|
Recording.job_id == job_id,
|
|
Recording.status.in_(["running", "starting"]),
|
|
Recording.is_scheduled == True,
|
|
).first()
|
|
return rec.id if rec else None
|
|
finally:
|
|
session.close()
|
|
|
|
def _tick(self):
|
|
now = datetime.datetime.now()
|
|
today_num = now.isoweekday()
|
|
today_date = now.strftime("%Y-%m-%d")
|
|
now_time = now.strftime("%H:%M")
|
|
|
|
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,
|
|
"description": j.description,
|
|
"metadata_monitor_enabled": j.metadata_monitor_enabled,
|
|
"metadata_pattern": j.metadata_pattern,
|
|
"metadata_grace_period": j.metadata_grace_period,
|
|
"metadata_poll_interval": j.metadata_poll_interval,
|
|
"ntfy_enabled": j.ntfy_enabled,
|
|
"plik_enabled": j.plik_enabled,
|
|
"delete_after_upload": j.delete_after_upload,
|
|
})
|
|
finally:
|
|
session.close()
|
|
|
|
for job in job_list:
|
|
if not job["schedule_start"]:
|
|
continue
|
|
|
|
is_today = False
|
|
multi_day = bool(job.get("schedule_end_date") and job["schedule_once"])
|
|
|
|
if job["schedule_once"]:
|
|
if job["schedule_date"] == today_date:
|
|
is_today = True
|
|
else:
|
|
if _matches_day(today_num, job["schedule_days"] or "*"):
|
|
is_today = True
|
|
|
|
in_window = False
|
|
is_cross_midnight = False
|
|
if is_today:
|
|
if multi_day:
|
|
now_min = _time_to_minutes(now_time)
|
|
start_min = _time_to_minutes(job["schedule_start"])
|
|
in_window = now_min >= start_min
|
|
elif job["schedule_stop"]:
|
|
start_min = _time_to_minutes(job["schedule_start"])
|
|
stop_min = _time_to_minutes(job["schedule_stop"])
|
|
if stop_min <= start_min:
|
|
is_cross_midnight = True
|
|
now_min = _time_to_minutes(now_time)
|
|
in_window = now_min >= start_min
|
|
else:
|
|
in_window = _in_time_window(now_time, job["schedule_start"], job["schedule_stop"])
|
|
else:
|
|
now_min = _time_to_minutes(now_time)
|
|
start_min = _time_to_minutes(job["schedule_start"])
|
|
in_window = now_min >= start_min
|
|
|
|
occ_key = f"{job['id']}|{job['schedule_once']}|{today_date if not job['schedule_once'] else job['schedule_date']}|{job['schedule_start']}"
|
|
|
|
if in_window:
|
|
if occ_key not in self._started_keys and not self._is_job_recording(job["id"]):
|
|
duration = job["max_duration"]
|
|
if not duration and job["schedule_stop"]:
|
|
if multi_day:
|
|
start_dt = datetime.datetime.strptime(
|
|
f"{job['schedule_date']} {job['schedule_start']}", "%Y-%m-%d %H:%M")
|
|
end_dt = datetime.datetime.strptime(
|
|
f"{job['schedule_end_date']} {job['schedule_stop']}", "%Y-%m-%d %H:%M")
|
|
duration = max(0, int((end_dt - start_dt).total_seconds()))
|
|
else:
|
|
start_min = _time_to_minutes(job["schedule_start"])
|
|
stop_min = _time_to_minutes(job["schedule_stop"])
|
|
if stop_min > start_min:
|
|
duration = (stop_min - start_min) * 60
|
|
else:
|
|
duration = (1440 - start_min + stop_min) * 60
|
|
|
|
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", ""),
|
|
metadata_monitor_enabled=job.get("metadata_monitor_enabled", False),
|
|
metadata_pattern=job.get("metadata_pattern", ""),
|
|
metadata_grace_period=job.get("metadata_grace_period") or 300,
|
|
metadata_poll_interval=job.get("metadata_poll_interval") or 30,
|
|
ntfy_enabled=job.get("ntfy_enabled", False),
|
|
plik_enabled=job.get("plik_enabled", False),
|
|
delete_after_upload=job.get("delete_after_upload", False),
|
|
)
|
|
self._started_keys.add(occ_key)
|
|
else:
|
|
if not multi_day and not is_cross_midnight and job["schedule_stop"] and self._is_job_recording(job["id"]):
|
|
rec_id = self._get_active_scheduled_recording_for_job(job["id"])
|
|
if rec_id:
|
|
self.recorder.stop_recording(rec_id)
|
|
self._started_keys.discard(occ_key)
|