281 lines
8.0 KiB
Python
281 lines
8.0 KiB
Python
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
|
||
from app.db.database import get_connection
|
||
from app.db.migrations import run_migrations
|
||
|
||
# Plánovač (monitor) spouští pouze záznamy s `next_run_at IS NOT NULL`
|
||
# (a zároveň `next_run_at <= CURRENT_TIMESTAMP`).
|
||
# Portál proto MUSÍ `next_run_at` naplnit při create/update/enable, jinak se
|
||
# zapnutý skript nikdy nespustí. Monitor si po každém běhu sám dopočítá další
|
||
# `next_run_at` – Portál pouze připravuje data, scheduler zůstává v monitoru.
|
||
#
|
||
# Časy v DB jsou ukládané přes SQLite CURRENT_TIMESTAMP, tj. v UTC ve formátu
|
||
# "YYYY-MM-DD HH:MM:SS". `next_run_at` musíme počítat ve stejném formátu a
|
||
# časové zóně, aby porovnání v monitoru sedělo.
|
||
_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"
|
||
|
||
|
||
def _parse_schedule_time(schedule_time: str | None) -> tuple[int, int]:
|
||
"""Rozparsuje "HH:MM" na (hodina, minuta) nebo vyhodí ValueError."""
|
||
value = (schedule_time or "").strip()
|
||
if not value:
|
||
raise ValueError("Čas spuštění (schedule_time) je povinný pro tuto frekvenci")
|
||
parts = value.split(":")
|
||
if len(parts) != 2:
|
||
raise ValueError(f"Neplatný formát času: {value!r} (očekává se HH:MM)")
|
||
try:
|
||
hour = int(parts[0])
|
||
minute = int(parts[1])
|
||
except ValueError:
|
||
raise ValueError(f"Neplatný formát času: {value!r} (očekává se HH:MM)")
|
||
if not (0 <= hour <= 23 and 0 <= minute <= 59):
|
||
raise ValueError(f"Neplatný čas: {value!r} (hodina 0-23, minuta 0-59)")
|
||
return hour, minute
|
||
|
||
|
||
def _add_one_month(dt: datetime) -> datetime:
|
||
"""Posune datum o jeden měsíc dopředu, ošetří kratší měsíce."""
|
||
month = dt.month + 1
|
||
year = dt.year + (month - 1) // 12
|
||
month = (month - 1) % 12 + 1
|
||
# Ošetření např. 31. -> kratší měsíc; den ořízneme na poslední platný.
|
||
day = dt.day
|
||
while True:
|
||
try:
|
||
return dt.replace(year=year, month=month, day=day)
|
||
except ValueError:
|
||
day -= 1
|
||
if day < 1:
|
||
raise
|
||
|
||
|
||
def compute_next_run_at(
|
||
schedule_type: str | None,
|
||
schedule_time: str | None,
|
||
*,
|
||
now: datetime | None = None,
|
||
) -> str:
|
||
"""Spočítá `next_run_at` stejně jako monitor.
|
||
|
||
Vrací řetězec ve formátu "YYYY-MM-DD HH:MM:SS" v UTC.
|
||
Vyhodí ValueError, pokud čas nelze spočítat (neplatná frekvence/čas).
|
||
"""
|
||
now = (now or datetime.utcnow()).replace(microsecond=0)
|
||
|
||
if schedule_type == "hourly":
|
||
# Další celá hodina.
|
||
nxt = now.replace(minute=0, second=0) + timedelta(hours=1)
|
||
return nxt.strftime(_TIMESTAMP_FORMAT)
|
||
|
||
hour, minute = _parse_schedule_time(schedule_time)
|
||
candidate = now.replace(hour=hour, minute=minute, second=0)
|
||
|
||
if schedule_type == "daily":
|
||
if candidate <= now:
|
||
candidate += timedelta(days=1)
|
||
elif schedule_type == "weekly":
|
||
if candidate <= now:
|
||
candidate += timedelta(days=7)
|
||
elif schedule_type == "monthly":
|
||
if candidate <= now:
|
||
candidate = _add_one_month(candidate)
|
||
else:
|
||
raise ValueError(f"Neznámá frekvence: {schedule_type!r}")
|
||
|
||
return candidate.strftime(_TIMESTAMP_FORMAT)
|
||
|
||
|
||
def resolve_next_run_at(metadata: dict[str, Any]) -> str | None:
|
||
"""Vrátí `next_run_at` pro zadaná metadata.
|
||
|
||
- vypnutý skript (`is_enabled = 0`) => None
|
||
- zapnutý skript => spočítaný čas (může vyhodit ValueError)
|
||
"""
|
||
if not metadata.get("is_enabled"):
|
||
return None
|
||
return compute_next_run_at(metadata.get("schedule_type"), metadata.get("schedule_time"))
|
||
|
||
|
||
def get_scheduled_scripts():
|
||
run_migrations()
|
||
con = get_connection()
|
||
|
||
rows = con.execute(
|
||
"""
|
||
SELECT
|
||
id,
|
||
name,
|
||
description,
|
||
script_name,
|
||
schedule_type,
|
||
schedule_time,
|
||
timeout_seconds,
|
||
COALESCE(is_enabled, 1) AS is_enabled,
|
||
COALESCE(is_running, 0) AS is_running,
|
||
last_run_at,
|
||
last_status,
|
||
last_job_id,
|
||
last_error_text,
|
||
next_run_at,
|
||
created_at,
|
||
updated_at
|
||
FROM scheduled_scripts
|
||
ORDER BY name
|
||
"""
|
||
).fetchall()
|
||
|
||
con.close()
|
||
return [dict(row) for row in rows]
|
||
|
||
|
||
def get_scheduled_script(script_id: int):
|
||
run_migrations()
|
||
con = get_connection()
|
||
|
||
row = con.execute(
|
||
"""
|
||
SELECT
|
||
id,
|
||
name,
|
||
description,
|
||
script_name,
|
||
schedule_type,
|
||
schedule_time,
|
||
timeout_seconds,
|
||
COALESCE(is_enabled, 1) AS is_enabled,
|
||
COALESCE(is_running, 0) AS is_running,
|
||
last_run_at,
|
||
last_status,
|
||
last_job_id,
|
||
last_error_text,
|
||
next_run_at,
|
||
created_at,
|
||
updated_at
|
||
FROM scheduled_scripts
|
||
WHERE id = ?
|
||
""",
|
||
(script_id,),
|
||
).fetchone()
|
||
|
||
con.close()
|
||
return dict(row) if row else None
|
||
|
||
|
||
def create_scheduled_script(metadata: dict[str, Any]) -> int:
|
||
run_migrations()
|
||
con = get_connection()
|
||
|
||
# Zapnutý skript musí dostat `next_run_at`, jinak ho monitor nikdy nespustí.
|
||
next_run_at = resolve_next_run_at(metadata)
|
||
|
||
cur = con.execute(
|
||
"""
|
||
INSERT INTO scheduled_scripts (
|
||
name,
|
||
description,
|
||
script_name,
|
||
schedule_type,
|
||
schedule_time,
|
||
timeout_seconds,
|
||
is_enabled,
|
||
is_running,
|
||
next_run_at,
|
||
created_at,
|
||
updated_at
|
||
)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||
""",
|
||
(
|
||
metadata.get("name"),
|
||
metadata.get("description") or None,
|
||
metadata.get("script_name"),
|
||
metadata.get("schedule_type"),
|
||
metadata.get("schedule_time") or None,
|
||
metadata.get("timeout_seconds"),
|
||
1 if metadata.get("is_enabled") else 0,
|
||
next_run_at,
|
||
),
|
||
)
|
||
script_id = cur.lastrowid
|
||
con.commit()
|
||
con.close()
|
||
return script_id
|
||
|
||
|
||
def update_scheduled_script(script_id: int, metadata: dict[str, Any]) -> None:
|
||
run_migrations()
|
||
con = get_connection()
|
||
|
||
# Při editaci vždy přepočítáme `next_run_at` podle (případně změněné)
|
||
# frekvence/času. Vypnutý skript dostane NULL, zapnutý nový čas.
|
||
next_run_at = resolve_next_run_at(metadata)
|
||
|
||
con.execute(
|
||
"""
|
||
UPDATE scheduled_scripts
|
||
SET name = ?,
|
||
description = ?,
|
||
schedule_type = ?,
|
||
schedule_time = ?,
|
||
timeout_seconds = ?,
|
||
is_enabled = ?,
|
||
next_run_at = ?,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = ?
|
||
""",
|
||
(
|
||
metadata.get("name"),
|
||
metadata.get("description") or None,
|
||
metadata.get("schedule_type"),
|
||
metadata.get("schedule_time") or None,
|
||
metadata.get("timeout_seconds"),
|
||
1 if metadata.get("is_enabled") else 0,
|
||
next_run_at,
|
||
script_id,
|
||
),
|
||
)
|
||
|
||
con.commit()
|
||
con.close()
|
||
|
||
|
||
def set_scheduled_script_enabled(
|
||
script_id: int, is_enabled: bool, next_run_at: str | None
|
||
) -> None:
|
||
run_migrations()
|
||
con = get_connection()
|
||
|
||
# Zapnutí: dopočítaný `next_run_at`. Vypnutí: NULL.
|
||
con.execute(
|
||
"""
|
||
UPDATE scheduled_scripts
|
||
SET is_enabled = ?,
|
||
next_run_at = ?,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = ?
|
||
""",
|
||
(1 if is_enabled else 0, next_run_at if is_enabled else None, script_id),
|
||
)
|
||
|
||
con.commit()
|
||
con.close()
|
||
|
||
|
||
def delete_scheduled_script(script_id: int) -> bool:
|
||
run_migrations()
|
||
con = get_connection()
|
||
|
||
cur = con.execute(
|
||
"""
|
||
DELETE FROM scheduled_scripts
|
||
WHERE id = ?
|
||
AND COALESCE(is_running, 0) = 0
|
||
""",
|
||
(script_id,),
|
||
)
|
||
deleted = cur.rowcount == 1
|
||
con.commit()
|
||
con.close()
|
||
return deleted
|