142 lines
4.5 KiB
Python
142 lines
4.5 KiB
Python
"""Alarm and clock management."""
|
|
import os
|
|
from backend.time_util import now as tz_now
|
|
from flask import current_app
|
|
from backend.models import get_db
|
|
|
|
# Per-user "already triggered today" state: {user_id: (date_str, set)}
|
|
_triggered_state: dict[int, tuple[str, set]] = {}
|
|
|
|
|
|
|
|
|
|
def _uid() -> int:
|
|
"""Resolve the current user (request session or scheduler thread)."""
|
|
try:
|
|
from backend.auth import current_user_id
|
|
return current_user_id() or 0
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
def get_alarms(user_id: int = None) -> list:
|
|
if user_id is None:
|
|
user_id = _uid()
|
|
db = get_db(current_app.config["DB_PATH"])
|
|
rows = [dict(r) for r in db.execute(
|
|
"SELECT * FROM alarms WHERE user_id=? ORDER BY time", (user_id,)
|
|
).fetchall()]
|
|
db.close()
|
|
return rows
|
|
|
|
|
|
def add_alarm(label: str, time: str, days: str = "1,2,3,4,5,6,7", enabled: bool = True,
|
|
sound: str = "default", once: bool = False, user_id: int = None) -> dict:
|
|
if user_id is None:
|
|
user_id = _uid()
|
|
db = get_db(current_app.config["DB_PATH"])
|
|
cur = db.execute(
|
|
"INSERT INTO alarms (user_id, label, time, days, enabled, sound, once) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
(user_id, label, time, days, 1 if enabled else 0, sound, 1 if once else 0)
|
|
)
|
|
db.commit()
|
|
row = dict(db.execute("SELECT * FROM alarms WHERE id=?", (cur.lastrowid,)).fetchone())
|
|
db.close()
|
|
return row
|
|
|
|
|
|
def update_alarm(alarm_id: int, label: str = None, time: str = None,
|
|
days: str = None, enabled: bool = None, sound: str = None, once: bool = None,
|
|
user_id: int = None):
|
|
if user_id is None:
|
|
user_id = _uid()
|
|
db = get_db(current_app.config["DB_PATH"])
|
|
fields = []
|
|
vals = []
|
|
for key, val in [("label", label), ("time", time), ("days", days),
|
|
("enabled", 1 if enabled else 0 if enabled is not None else None),
|
|
("sound", sound),
|
|
("once", 1 if once else 0 if once is not None else None)]:
|
|
if val is not None:
|
|
fields.append(f"{key}=?")
|
|
vals.append(val)
|
|
ok = False
|
|
if fields:
|
|
vals.append(user_id)
|
|
vals.append(alarm_id)
|
|
cur = db.execute(f"UPDATE alarms SET {', '.join(fields)} WHERE user_id=? AND id=?", vals)
|
|
db.commit()
|
|
ok = cur.rowcount > 0
|
|
db.close()
|
|
return ok
|
|
|
|
|
|
def delete_alarm(alarm_id: int, user_id: int = None) -> bool:
|
|
if user_id is None:
|
|
user_id = _uid()
|
|
db = get_db(current_app.config["DB_PATH"])
|
|
cur = db.execute("DELETE FROM alarms WHERE user_id=? AND id=?", (user_id, alarm_id))
|
|
db.commit()
|
|
db.close()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def check_alarms(user_id: int = None) -> list:
|
|
if user_id is None:
|
|
user_id = _uid()
|
|
|
|
db = get_db(current_app.config["DB_PATH"])
|
|
now = tz_now()
|
|
today_str = now.strftime("%Y%m%d")
|
|
|
|
triggered_date, triggered_today = _triggered_state.get(user_id, ("", set()))
|
|
if triggered_date != today_str:
|
|
triggered_today = set()
|
|
triggered_date = today_str
|
|
|
|
current_time = now.strftime("%H:%M")
|
|
weekday = str(now.isoweekday())
|
|
|
|
rows = db.execute(
|
|
"SELECT * FROM alarms WHERE user_id=? AND enabled=1 AND time=?",
|
|
(user_id, current_time)
|
|
).fetchall()
|
|
|
|
triggered = []
|
|
for r in rows:
|
|
days_list = r["days"].split(",")
|
|
if weekday in days_list:
|
|
alarm_id = r["id"]
|
|
if alarm_id in triggered_today:
|
|
continue
|
|
triggered_today.add(alarm_id)
|
|
triggered.append(dict(r))
|
|
# Once-only alarms: auto-disable after trigger
|
|
if r["once"]:
|
|
db.execute("UPDATE alarms SET enabled=0 WHERE user_id=? AND id=?", (user_id, alarm_id))
|
|
db.commit()
|
|
|
|
_triggered_state[user_id] = (triggered_date, triggered_today)
|
|
db.close()
|
|
|
|
if triggered:
|
|
current_app.logger.info(f"Triggering {len(triggered)} alarms: {[a['label'] for a in triggered]}")
|
|
current_app.event_bus.publish("alarm_triggered", {
|
|
"alarms": triggered,
|
|
"ts": now.isoformat(),
|
|
}, user_id=user_id)
|
|
return triggered
|
|
|
|
|
|
def get_ringtone_path(sound_name: str) -> str | None:
|
|
if not sound_name or sound_name == "default":
|
|
return None
|
|
ringtone_dir = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|
"frontend", "static", "ringtones"
|
|
)
|
|
path = os.path.join(ringtone_dir, sound_name)
|
|
if os.path.isfile(path):
|
|
return path
|
|
return None
|