"""
BAN YARO — Hintergrund-Scheduler
Täglich: Gesundheits-Erinnerungen per Push versenden.
"""
import logging
from datetime import date, datetime, timedelta
from zoneinfo import ZoneInfo
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
_TZ = ZoneInfo("Europe/Berlin")
from database import db
from routes.push import send_push_to_user, send_push_to_all, send_push
import weather
logger = logging.getLogger(__name__)
_scheduler = AsyncIOScheduler(timezone="Europe/Berlin")
# In-Memory Job-Protokoll: {job_id: {"last_run": datetime, "result": str, "status": "ok"|"error"}}
_job_log: dict = {}
def start():
# ------------------------------------------------------------------
# Job-Staffelung in 5-Minuten-Intervallen — verhindert gleichzeitige
# Last-Spitzen (mehrere Jobs zur selben Sekunde 08:00 Uhr).
# coalesce=True: bei verpassten Läufen nur ein Lauf nachholen.
# misfire_grace_time: Mindestwert 300s, höher wo Job lange dauern kann.
# ------------------------------------------------------------------
_scheduler.add_job(
_job_health_reminders,
CronTrigger(hour=8, minute=0), # täglich 08:00 Uhr
id="health_reminders",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
_scheduler.add_job(
_job_poison_archive,
CronTrigger(hour=3, minute=0), # täglich 03:00 Uhr (ruhige Zeit)
id="poison_archive",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
_scheduler.add_job(
_job_purge_jwt_blacklist,
CronTrigger(hour=3, minute=30), # täglich 03:30 Uhr, nach poison_archive
id="purge_jwt_blacklist",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
_scheduler.add_job(
_job_weather_alert,
CronTrigger(hour=7, minute=30), # täglich 07:30 Uhr
id="weather_alert",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
_scheduler.add_job(
_job_milestone_check,
CronTrigger(hour=0, minute=5), # täglich 00:05 Uhr
id="milestone_check",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
_scheduler.add_job(
_job_import_events,
CronTrigger(month="1,4,7,10", day=2, hour=2), # quartalsweise Jan/Apr/Jul/Okt
id="import_events",
replace_existing=True,
misfire_grace_time=7200,
coalesce=True,
)
# Einmalig beim Start (nach 10s Verzögerung) für sofortige Befüllung
_scheduler.add_job(
_job_import_events,
'date',
run_date=datetime.now(tz=_TZ) + timedelta(seconds=10),
id="import_events_startup",
replace_existing=True,
)
# 1. des Monats 03:00 — Rassen aus TheDogAPI aktualisieren
_scheduler.add_job(
_job_seed_breeds,
CronTrigger(day=1, hour=3, minute=0), # 1. jedes Monats
id="seed_breeds",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# 1. des Monats 04:00 — fehlende Rassen aus Wikidata ergänzen
_scheduler.add_job(
_job_seed_wikidata_breeds,
CronTrigger(day=1, hour=4, minute=0), # 1. jedes Monats
id="seed_wikidata",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Jeden Montag 09:05 — Wöchentlicher Fortschritts-Lober (staggered)
_scheduler.add_job(
_job_weekly_praise,
CronTrigger(day_of_week='mon', hour=9, minute=5),
id="weekly_praise",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Täglich 06:00 Uhr Status-Report per Mail
_scheduler.add_job(
_job_status_report,
CronTrigger(hour=6, minute=0),
id="status_report",
replace_existing=True,
misfire_grace_time=1800,
coalesce=True,
)
# Täglich 12:00 — Moderation-Overdue-Check
_scheduler.add_job(
_job_moderation_overdue,
CronTrigger(hour=12, minute=0),
id="moderation_overdue",
replace_existing=True,
misfire_grace_time=1800,
coalesce=True,
)
# 1. Feb / Mai / Aug / Nov 07:10 — Quartalsbericht (staggered weg von 07:00)
_scheduler.add_job(
_job_quarterly_report,
CronTrigger(month="2,5,8,11", day=1, hour=7, minute=10),
id="quarterly_report",
replace_existing=True,
misfire_grace_time=7200,
coalesce=True,
)
# Jeden Montag 07:05 — KI-Gesundheitsberichte (staggered weg von 07:00)
_scheduler.add_job(
_job_ki_health_report,
CronTrigger(day_of_week='mon', hour=7, minute=5),
id="ki_health_report",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Täglich 06:30 — Wiederkehrende Ausgaben anlegen
_scheduler.add_job(
_job_recurring_expenses,
CronTrigger(hour=6, minute=30),
id="recurring_expenses",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# 1. des Monats 00:05 — Hund des Monats Sieger festlegen
_scheduler.add_job(
_job_hdm_winner,
CronTrigger(day=1, hour=0, minute=5),
id="hdm_winner",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Täglich 19:00 Uhr — Streak-Erinnerung
_scheduler.add_job(
_job_streak_reminder,
CronTrigger(hour=19, minute=0),
id="streak_reminder",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Täglich 08:05 Uhr — Tierfutter-Rückrufe prüfen (RASFF) (staggered weg von 08:00)
_scheduler.add_job(
_job_recall_check,
CronTrigger(hour=8, minute=5),
id="recall_check",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Täglich 03:40 Uhr — OSM-Beiträge: Pending-Retry + Revert-Überleben prüfen
# + Pro-Freischaltung (staggered, ruhige Zeit)
_scheduler.add_job(
_job_osm_confirm,
CronTrigger(hour=3, minute=40),
id="osm_confirm",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Jeden Montag 08:10 Uhr — Neue Foto-Challenge anlegen (staggered weg von 08:00)
_scheduler.add_job(
_job_new_foto_challenge,
CronTrigger(day_of_week='mon', hour=8, minute=10),
id="new_foto_challenge",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Täglich 07:00 Uhr — Goldene Gassi-Stunde
_scheduler.add_job(
_job_golden_gassi_hour,
CronTrigger(hour=7, minute=0),
id="golden_gassi_hour",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Täglich 09:00 Uhr — Jahrestags-Erinnerungen (Tagebuch-Einträge von heute vor X Jahren)
_scheduler.add_job(
_job_anniversary_reminders,
CronTrigger(hour=9, minute=0),
id="anniversary_reminders",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# 1. des Monats 10:00 — Monatlicher Rückblick per Push
_scheduler.add_job(
_job_monthly_recap,
CronTrigger(day=1, hour=10, minute=0),
id="monthly_recap",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Täglich 03:15 — Abo-Ablauf prüfen (staggered weg von 03:00 poison_archive)
_scheduler.add_job(
_job_subscription_check,
CronTrigger(hour=3, minute=15),
id="subscription_check",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
_scheduler.add_job(
_job_invoice_reminder,
CronTrigger(hour=8, minute=30), # täglich 08:30 Uhr
id="invoice_reminder",
replace_existing=True,
misfire_grace_time=3600,
coalesce=True,
)
# Täglich 06:30 — Error-Digest-Mail an ADMIN_EMAIL
_scheduler.add_job(
_job_error_digest,
CronTrigger(hour=6, minute=30),
id="error_digest",
replace_existing=True,
misfire_grace_time=1800,
coalesce=True,
)
_scheduler.start()
logger.info("Scheduler gestartet (gestaffelt) — Health-Reminder 08:00, Giftköder-Archiv 03:00, Wetter-Alert 07:30, Meilenstein-Check 00:05, Event-Import 1.+2./4./7./10. 02:00, Rassen-Seed 1. 03:00, Wikidata-Seed 1. 04:00, Status-Report 06:00, Moderation-Overdue 12:00, Quartalsbericht 1. Feb/Mai/Aug/Nov 07:10, KI-Gesundheitsbericht Mo 07:05, Streak-Reminder 19:00, Rückruf-Check 08:05, Goldene-Gassi-Stunde 07:00, Jahrestags-Erinnerungen 09:00, Monatlicher-Rückblick 1. 10:00, Foto-Challenge Mo 08:10, Weekly-Praise Mo 09:05, Abo-Check 03:15, Invoice-Reminder 08:30. OSM-Cache: on-demand (kein Prewarm).")
def stop():
_scheduler.shutdown(wait=False)
logger.info("Scheduler gestoppt.")
# ------------------------------------------------------------------
# JOB: Abo-Ablauf prüfen (täglich 03:00)
# ------------------------------------------------------------------
_TIER_PRICE = {"pro": 29.00, "breeder": 49.00}
async def _create_renewal_invoice_draft(user: dict, expires: date, tier_label: str):
"""Legt einen Rechnungs-Entwurf für die Abo-Verlängerung an, sofern noch keiner existiert."""
import os
from mailer import send_email, email_html
from routes.invoices import _next_invoice_number
# Gekündigte Abos bekommen keine Erneuerungsrechnung
if user.get("subscription_cancelled_at"):
logger.info(f"Kein Erneuerungsentwurf für {user['email']} — Abo ist gekündigt.")
return
tier = user["subscription_tier"]
price = _TIER_PRICE.get(tier, 29.00)
# Verlängerungszeitraum: Folgetag nach Ablauf bis +1 Jahr
start = expires + timedelta(days=1)
end = start.replace(year=start.year + 1) - timedelta(days=1)
period = f"{start.strftime('%d.%m.%Y')} - {end.strftime('%d.%m.%Y')}"
with db() as conn:
# Nur anlegen wenn noch kein Entwurf/offener Eintrag für diesen User + Zeitraum
existing = conn.execute(
"""SELECT id FROM invoices
WHERE user_id=? AND status IN ('draft','sent')
AND service_period=?""",
(user["id"], period)
).fetchone()
if existing:
logger.info(f"Erneuerungsrechnung bereits vorhanden für user {user['id']}")
return
# Billing-Adresse des Users laden
row = conn.execute(
"SELECT billing_address FROM users WHERE id=?", (user["id"],)
).fetchone()
billing_address = row["billing_address"] if row else None
# Rabatt berechnen (inline, da kein Admin-Import möglich)
disc_row = conn.execute(
"""SELECT u.is_founder, u.is_founder_pending, u.referred_by,
COALESCE((SELECT COUNT(*) FROM users WHERE referred_by=u.id), 0) AS referral_count
FROM users u WHERE u.id=?""",
(user["id"],)
).fetchone()
discount_pct = 0
discount_reason = None
referral_count = 0
if disc_row:
referral_count = disc_row["referral_count"]
if disc_row["is_founder"] or disc_row["is_founder_pending"]:
discount_pct = 100
discount_reason = "founder"
elif (disc_row["referred_by"] or 0) > 0:
ref = conn.execute(
"SELECT is_founder, is_founder_pending, founder_referral_tickets FROM users WHERE id=?",
(disc_row["referred_by"],)
).fetchone()
if ref and (ref["is_founder"] or ref["is_founder_pending"]):
# 50%-Weitergabe nur solange der Gründer Tickets hat: dieser Freund
# bekommt sie, wenn sein Rang unter den verifizierten Geworbenen
# (nach Anmeldedatum) das Ticket-Kontingent nicht übersteigt.
rank = conn.execute(
"""SELECT COUNT(*) FROM users
WHERE referred_by=? AND email_verified=1
AND created_at <= (SELECT created_at FROM users WHERE id=?)""",
(disc_row["referred_by"], user["id"])
).fetchone()[0]
if rank <= (ref["founder_referral_tickets"] or 0):
discount_pct = 50
discount_reason = "referred_by_founder"
if not discount_reason:
for thr, pct in [(50, 50), (20, 30), (10, 20)]:
if referral_count >= thr:
discount_pct = pct
discount_reason = "referral"
break
discount_amt = round(price * discount_pct / 100, 2)
after_disc = round(price - discount_amt, 2)
_AGB = "Jahresbeitrag gem. AGB. Bei vorzeitiger Kündigung keine anteilige Rückerstattung; Zugang bleibt bis Laufzeitende bestehen."
if discount_reason == "founder":
notes = f"Gründer-Sonderkonditionen: {tier_label} kostenfrei als Dankeschön für deine Unterstützung als Gründer! {_AGB} (Automatisch erstellt, Ablauf: {expires.strftime('%d.%m.%Y')})"
elif discount_reason == "referred_by_founder":
notes = f"Willkommen in der Gründer-Community! Als persönlich von einem Gründer eingeladenes Mitglied erhältst du dauerhaft {discount_pct}% Rabatt. {_AGB} (Automatisch erstellt, Ablauf: {expires.strftime('%d.%m.%Y')})"
elif discount_reason == "referral":
notes = f"Herzlichen Dank für deine Unterstützung! Für {referral_count} geworbene Freunde erhältst du {discount_pct}% Rabatt. {_AGB} (Automatisch erstellt, Ablauf: {expires.strftime('%d.%m.%Y')})"
else:
notes = f"{_AGB} (Automatisch erstellt, Ablauf: {expires.strftime('%d.%m.%Y')})"
invoice_number = _next_invoice_number(conn)
description = f"{tier_label} Jahresabo (Verlängerung)"
conn.execute("""
INSERT INTO invoices
(invoice_number, user_id, recipient_name, recipient_email, recipient_address,
description, service_period, amount_net, discount_pct, discount_amount,
amount_after_discount, tax_rate, tax_amount, amount_gross, notes)
VALUES (?,?,?,?,?,?,?,?,?,?,?,0,0,?,?)
""", (
invoice_number, user["id"], user["name"], user["email"], billing_address,
description, period,
price, discount_pct, discount_amt, after_disc, after_disc, notes,
))
conn.execute(
"INSERT INTO invoice_items (invoice_id, description, quantity, unit_price, total) VALUES (?,?,1,?,?)",
(conn.execute("SELECT last_insert_rowid()").fetchone()[0], description, price, price)
)
logger.info(f"Erneuerungsrechnung {invoice_number} als Entwurf angelegt für {user['email']}")
# Admin-Benachrichtigung
admin_email = os.getenv("ADMIN_EMAIL", "")
if admin_email:
app_url = os.getenv("APP_URL", "https://banyaro.app")
body = f"""
Für {user['name']} ({user['email']}) wurde automatisch ein
Rechnungsentwurf für die Abo-Verlängerung erstellt.
| Rechnung: | {invoice_number} |
| Tarif: | {tier_label} |
| Betrag: | {price:.2f} EUR |
| Zeitraum: | {period} |
| Abo läuft ab: | {expires.strftime('%d.%m.%Y')} (in 30 Tagen) |
Bitte prüfen, ggf. anpassen und rechtzeitig versenden.
"""
html = email_html(body, cta_url=f"{app_url}/#admin", cta_label="Zur Rechnung im Admin")
await send_email(
admin_email,
f"Erneuerungsrechnung {invoice_number} bereit — {user['name']}",
html,
f"Entwurf {invoice_number} für {user['name']} ({tier_label}, {price:.2f} EUR, {period}) bereit."
)
async def _remind_renewal_invoice(user: dict, expires: date, tier_label: str):
"""7-Tage-Erinnerung an René: Entwurf noch nicht versendet."""
import os
from mailer import send_email, email_html
with db() as conn:
draft = conn.execute(
"SELECT invoice_number FROM invoices WHERE user_id=? AND status='draft' LIMIT 1",
(user["id"],)
).fetchone()
if not draft:
return # kein offener Entwurf, nichts zu erinnern
admin_email = os.getenv("ADMIN_EMAIL", "")
if not admin_email:
return
app_url = os.getenv("APP_URL", "https://banyaro.app")
body = f"""
Achtung: Das Abo von {user['name']} ({user['email']})
läuft in 7 Tagen (am {expires.strftime('%d.%m.%Y')}) ab.
Rechnungsentwurf {draft['invoice_number']} wurde noch nicht versendet.
Bitte jetzt versenden damit der Kunde rechtzeitig bezahlen kann.
"""
html = email_html(body, cta_url=f"{app_url}/#admin", cta_label="Rechnung jetzt senden")
await send_email(
admin_email,
f"⚠ Noch 7 Tage — Erneuerungsrechnung {draft['invoice_number']} nicht versendet",
html,
f"Entwurf {draft['invoice_number']} für {user['name']} noch nicht versendet. Abo läuft in 7 Tagen ab."
)
logger.info(f"7-Tage-Erinnerung an Admin für {user['email']}: {draft['invoice_number']}")
async def _job_invoice_reminder():
"""
Unbezahlte Rechnungen (status='sent'):
- Nach 21 Tagen: Zahlungsmahnung mit 14-Tage-Frist (§286/314 BGB)
- Nach 35 Tagen (21+14): Fristlose Abo-Kündigung
"""
from database import db as _db
from mailer import send_email, email_html
from routes.invoices import _next_invoice_number
import html as _html
import os
APP_URL = os.getenv("APP_URL", "https://banyaro.app")
IBAN = os.getenv("RECHNUNG_IBAN", "")
ADMIN_MAIL = os.getenv("ADMIN_EMAIL", "")
today = datetime.now(_TZ).date()
with db() as conn:
open_invoices = conn.execute(
"""SELECT i.*, u.name AS user_name, u.subscription_tier, u.id AS uid
FROM invoices i
LEFT JOIN users u ON u.id = i.user_id
WHERE i.status = 'sent'
AND i.sent_at IS NOT NULL"""
).fetchall()
for inv in open_invoices:
try:
sent_date = datetime.fromisoformat(inv["sent_at"].replace("Z", "+00:00")).date()
days_open = (today - sent_date).days
rg = inv["invoice_number"]
name = inv["recipient_name"]
email = inv["recipient_email"]
amount = inv["amount_gross"]
frist = (today + timedelta(days=14)).strftime("%d.%m.%Y")
# ── 21 Tage: Zahlungsmahnung mit 14-Tage-Frist ───────────
if days_open == 21:
iban_line = f"IBAN: {IBAN} · Verwendungszweck: {rg}
" if IBAN else ""
body = f"""
Hallo {_html.escape(name)},
unsere Rechnung {rg} vom {datetime.fromisoformat(inv['created_at'][:10]).strftime('%d.%m.%Y')}
über {amount:.2f} EUR ist leider noch offen.
Bitte überweisen Sie den Betrag bis zum {frist}.
{iban_line}
Sollte die Zahlung bis zu diesem Datum nicht eingehen, sind wir leider gezwungen,
Ihr Abonnement fristlos zu kündigen (§ 314 BGB).
"""
html = email_html(body, cta_url=APP_URL, cta_label="Ban Yaro öffnen")
await send_email(
email,
f"Zahlungserinnerung: Rechnung {rg} — Ban Yaro",
html,
f"Hallo {name},\n\nRechnung {rg} über {amount:.2f} EUR ist noch offen.\n"
f"Bitte bis {frist} überweisen. Andernfalls kündigen wir fristlos.\n"
+ (f"IBAN: {IBAN}, Verwendungszweck: {rg}\n" if IBAN else "")
)
logger.info(f"Zahlungsmahnung gesendet: {rg} an {email} (21 Tage offen)")
# ── 35 Tage: Fristlose Kündigung ─────────────────────────
elif days_open == 35:
# Abo kündigen wenn Nutzer zugeordnet und aktives Abo
if inv["uid"] and inv["subscription_tier"] not in (None, "standard", "standard_test"):
with db() as conn2:
conn2.execute(
"""UPDATE users SET subscription_tier='standard',
subscription_expires_at=NULL,
subscription_cancelled_at=strftime('%Y-%m-%dT%H:%M:%SZ','now')
WHERE id=?""",
(inv["uid"],)
)
logger.info(f"Fristlose Kündigung: user {inv['uid']} wegen unbezahlter Rechnung {rg}")
body = f"""
Hallo {_html.escape(name)},
da die Zahlung für Rechnung {rg} ({amount:.2f} EUR)
trotz unserer Zahlungserinnerung nicht eingegangen ist,
haben wir Ihr Abonnement gemäß § 314 BGB fristlos gekündigt.
Ihre Daten bleiben vollständig erhalten. Sie können jederzeit ein neues Abonnement abschließen.
Bei Rückfragen antworten Sie einfach auf diese E-Mail.
"""
html = email_html(body, cta_url=APP_URL, cta_label="Ban Yaro öffnen")
await send_email(
email,
f"Ihr Ban Yaro Abonnement wurde gekündigt — Rechnung {rg}",
html,
f"Hallo {name},\n\nIhr Abo wurde wegen unbezahlter Rechnung {rg} fristlos gekündigt.\n"
f"Ihre Daten sind erhalten. Neue Buchung jederzeit möglich.\n"
)
if ADMIN_MAIL:
await send_email(
ADMIN_MAIL,
f"Fristlose Kündigung: {name} — {rg} ({amount:.2f} EUR unbezahlt)",
email_html(f"Abo von {_html.escape(name)} ({email}) wurde automatisch fristlos gekündigt (§314 BGB). Rechnung {rg} seit 35 Tagen offen.
"),
f"Abo {name} gekündigt wegen unbezahlter Rechnung {rg}."
)
except Exception as e:
logger.warning(f"Invoice-Reminder Fehler für {inv.get('invoice_number','?')}: {e}")
async def _job_subscription_check():
"""Abgelaufene Abos auf Standard setzen; Warnmails 30 und 7 Tage vorher."""
from database import db as _db
from mailer import send_email, email_html
import html as _html
now = datetime.now(_TZ)
today = now.date()
with _db() as conn:
users = conn.execute(
"""SELECT id, name, email, subscription_tier, subscription_expires_at,
subscription_cancelled_at
FROM users
WHERE subscription_tier IN ('pro','breeder')
AND subscription_expires_at IS NOT NULL"""
).fetchall()
for u in users:
try:
expires = datetime.fromisoformat(u["subscription_expires_at"].replace('Z', '+00:00')).date()
days_left = (expires - today).days
tier_label = {"pro": "Ban Yaro Pro", "breeder": "Züchter"}.get(u["subscription_tier"], u["subscription_tier"])
# Abgelaufen → auf Standard setzen
if days_left < 0:
with _db() as conn:
dog_count = conn.execute(
"SELECT COUNT(*) FROM dogs WHERE user_id=? AND is_active!=0", (u["id"],)
).fetchone()[0]
needs_sel = 1 if dog_count > 1 else 0
conn.execute(
"""UPDATE users SET subscription_tier='standard',
needs_dog_selection=? WHERE id=?""",
(needs_sel, u["id"])
)
logger.info(f"Abo abgelaufen: {u['email']} → standard (needs_dog_selection={needs_sel})")
body = f"""
Hallo {_html.escape(u['name'])},
dein {tier_label}-Abo ist heute abgelaufen.
Dein Account wurde auf den kostenlosen Tarif gesetzt.
Deine Daten sind vollständig erhalten. Du kannst jederzeit wieder upgraden.
"""
if needs_sel:
body += "Wichtig: Du hattest mehrere Hunde. Öffne die App und wähle deinen Haupthund aus — alle anderen Profile bleiben gespeichert.
"
html = email_html(body, cta_url="https://banyaro.app", cta_label="Ban Yaro öffnen")
await send_email(u["email"], f"Dein {tier_label}-Abo ist abgelaufen", html,
f"Hallo {u['name']},\ndein {tier_label}-Abo ist abgelaufen. Daten bleiben erhalten.")
# 30 Tage Warnung + Erneuerungsrechnung als Entwurf anlegen
elif days_left == 30:
body = f"""
Hallo {_html.escape(u['name'])},
dein {tier_label}-Abo läuft in 30 Tagen
(am {expires.strftime('%d.%m.%Y')}) ab.
Um weiterzumachen, überweise einfach den Jahresbetrag und schreib uns kurz —
wir verlängern deinen Zugang sofort.
"""
html = email_html(body, cta_url="https://banyaro.app", cta_label="Abo verlängern")
await send_email(u["email"], f"Dein {tier_label}-Abo läuft in 30 Tagen ab", html,
f"Hallo {u['name']},\ndein {tier_label}-Abo läuft in 30 Tagen ab ({expires}).")
# Erneuerungsrechnung als Entwurf anlegen (nur wenn noch keine existiert)
await _create_renewal_invoice_draft(u, expires, tier_label)
# 7 Tage — Warnung an User + Erinnerung an René falls Entwurf noch nicht versendet
elif days_left == 7:
body = f"""
Hallo {_html.escape(u['name'])},
dein {tier_label}-Abo läuft in 7 Tagen
(am {expires.strftime('%d.%m.%Y')}) ab.
Jetzt verlängern und nahtlos weitermachen!
"""
html = email_html(body, cta_url="https://banyaro.app", cta_label="Abo verlängern")
await send_email(u["email"], f"Nur noch 7 Tage — {tier_label}-Abo läuft ab", html,
f"Hallo {u['name']},\nnur noch 7 Tage für dein {tier_label}-Abo.")
await _remind_renewal_invoice(u, expires, tier_label)
except Exception as e:
logger.warning(f"subscription_check Fehler für {u['email']}: {e}")
# ------------------------------------------------------------------
# JOB: Gesundheits-Erinnerungen
# ------------------------------------------------------------------
async def _job_health_reminders():
"""
Findet alle Health-Einträge mit `naechstes`-Datum:
- genau heute → sofortige Erinnerung
- in 7 Tagen → Vorwarnung
- gestern → Überfällig-Erinnerung (nur einmal, 1 Tag nach Fälligkeit)
Schickt jeweils eine Push-Notification an den Hundebesitzer.
"""
today = date.today()
in7 = today + timedelta(days=7)
yesterday = today - timedelta(days=1)
logger.info(f"Health-Reminder Job läuft für {today}")
in3 = today + timedelta(days=3)
with db() as conn:
# Alle fälligen Einträge der nächsten 7 Tage + gestrige (überfällig)
# erinnerung=0 → User hat diese Erinnerung deaktiviert
rows = conn.execute("""
SELECT h.id, h.typ, h.bezeichnung, h.naechstes,
d.user_id, d.name AS hund_name
FROM health h
JOIN dogs d ON d.id = h.dog_id
WHERE h.naechstes IN (?, ?, ?, ?)
AND h.typ IN ('impfung', 'entwurmung', 'medikament',
'parasit', 'krallen', 'fellpflege')
AND (h.erinnerung IS NULL OR h.erinnerung = 1)
""", (str(today), str(in7), str(in3), str(yesterday))).fetchall()
sent_total = 0
for r in rows:
naechstes = date.fromisoformat(r["naechstes"])
delta = (naechstes - today).days
if delta == 7:
title = f"⏰ Erinnerung: {r['bezeichnung']}"
body = f"In 7 Tagen fällig für {r['hund_name']}."
elif delta == 3:
title = f"⏰ Erinnerung: {r['bezeichnung']}"
body = f"In 3 Tagen fällig für {r['hund_name']} — bald vorbereiten."
elif delta == 0:
title = f"📅 Heute fällig: {r['bezeichnung']}"
body = f"Bitte heute erledigen — {r['hund_name']} wartet."
else: # delta == -1 → gestern überfällig
title = f"⚠️ Überfällig: {r['bezeichnung']}"
body = f"War gestern fällig für {r['hund_name']} — bitte bald erledigen."
sent = send_push_to_user(r["user_id"], {
"type": "health_reminder",
"title": title,
"body": body,
"data": {"page": "health"},
})
sent_total += sent
if sent:
logger.info(f"Reminder Push: user={r['user_id']} entry={r['id']} delta={delta}d")
logger.info(f"Health-Reminder Job fertig — {len(rows)} Einträge, {sent_total} Push gesendet.")
_log_job("health_reminders", "ok", f"{len(rows)} Einträge, {sent_total} Push")
# ------------------------------------------------------------------
# JOB: Abgelaufene Giftköder-Meldungen archivieren
# Abgelaufene, aber noch nicht manuell aufgelöste Einträge werden
# sauber als geloest=1 markiert — für spätere KI-Musteranalyse.
# Die Zeilen selbst werden NIE gelöscht.
# ------------------------------------------------------------------
async def _job_poison_archive():
"""
Findet Giftköder-Meldungen deren expires_at verstrichen ist
und die noch nicht als geloest markiert wurden.
Setzt geloest=1, geloest_grund='automatisch_abgelaufen'.
"""
from datetime import datetime
now = datetime.utcnow().isoformat()
with db() as conn:
result = conn.execute("""
UPDATE poison
SET geloest = 1,
geloest_at = datetime('now'),
geloest_grund = 'automatisch_abgelaufen'
WHERE geloest = 0
AND expires_at < ?
""", (now,))
count = result.rowcount
_log_job("poison_archive", "ok", f"{count} Meldungen archiviert")
if count:
logger.info(f"Giftköder-Archiv: {count} abgelaufene Meldungen archiviert.")
# ------------------------------------------------------------------
# JOB: Wetter-Alarm (Hitzepfoten / Gewitter)
# ------------------------------------------------------------------
async def _job_weather_alert():
"""
STANDORTBEZOGENER Wetter-Alarm.
Gruppiert alle Push-Abonnenten nach ihrem letzten bekannten Standort
(gerundet auf 0.1° ≈ 11 km) und holt pro Cluster die lokale Tagesprognose.
Sendet nur an die Abonnenten DES JEWEILIGEN Clusters, wenn dort:
- Temperatur >= 28°C (Asphalt-Warnung für Pfoten), oder
- Gewitter wahrscheinlich (Hitze hat Vorrang).
Abonnenten ohne gespeicherten Standort erhalten keine Wetter-Warnung
(besser keine als eine für eine fremde Region).
"""
from collections import defaultdict
logger.info("Wetter-Alert Job läuft (standortbezogen)")
with db() as conn:
rows = conn.execute(
"SELECT * FROM push_subscriptions "
"WHERE last_lat IS NOT NULL AND last_lon IS NOT NULL"
).fetchall()
if not rows:
logger.info("Wetter-Alert: keine Abonnenten mit Standort.")
_log_job("weather_alert", "ok", "Keine Standorte")
return
# Nach gerundetem Standort clustern → ein Wetter-Abruf pro Cluster
clusters = defaultdict(list)
for row in rows:
key = (round(row["last_lat"], 1), round(row["last_lon"], 1))
clusters[key].append(row)
heat_clusters = thunder_clusters = total_sent = 0
for (lat, lon), subs in clusters.items():
try:
alert = await weather.get_day_alert(lat, lon)
except Exception as e:
logger.warning(f"Wetter-Alert: Abruf für {lat},{lon} fehlgeschlagen: {e}")
continue
max_temp = alert["max_temp_c"]
payload = None
if max_temp is not None and max_temp >= 28:
payload = {
"type": "weather_heat",
"title": "☀️ Heißer Asphalt heute",
"body": f"Bis {max_temp:.0f}°C heute — Asphalt kann über 50°C heiß werden. Frühmorgens oder abends gassi gehen!",
"data": {"tag": "weather-heat"},
}
heat_clusters += 1
elif alert["thunderstorm"]:
payload = {
"type": "weather_thunder",
"title": "⛈️ Gewitter möglich",
"body": "Heute Gewitter wahrscheinlich. Gassi-Tour früh einplanen und Hund beruhigen.",
"data": {"tag": "weather-thunder"},
}
thunder_clusters += 1
if payload:
for row in subs:
if send_push(row, payload):
total_sent += 1
msg = f"{heat_clusters} Hitze- / {thunder_clusters} Gewitter-Cluster, {total_sent} Push"
logger.info(f"Wetter-Alert: {msg}")
_log_job("weather_alert", "ok", msg)
# ------------------------------------------------------------------
# JOB: Geburtstags- und Monats-Meilensteine
# Läuft täglich um 00:05 Uhr (Europe/Berlin).
# Prüft alle Hunde mit gesetztem Geburtstag und erstellt bei Treffern
# einen Tagebucheintrag (is_milestone=1) + Push-Notification.
# ------------------------------------------------------------------
async def _job_milestone_check():
"""
Prüft für jeden Hund mit bekanntem Geburtstag ob heute ein
Meilenstein-Tag ist:
- Jahrestag (1. Geburtstag, 2. Geburtstag, …)
- Monatsjubiläum in den ersten 12 Monaten (1 Monat, 2 Monate, …, 11 Monate)
Doppelt-Schutz: Wenn bereits ein Meilenstein-Eintrag mit demselben
Titel für heute existiert, wird kein zweiter erstellt.
"""
today = date.today()
logger.info(f"Meilenstein-Check läuft für {today}")
with db() as conn:
dogs = conn.execute("""
SELECT d.id, d.name, d.user_id, d.geburtstag
FROM dogs d
WHERE d.geburtstag IS NOT NULL
AND d.geburtstag != ''
""").fetchall()
created_total = 0
for dog in dogs:
try:
bday = date.fromisoformat(dog["geburtstag"])
except ValueError:
logger.warning(f"Meilenstein: ungültiges Geburtstag für Hund {dog['id']}: {dog['geburtstag']!r}")
continue
milestone = _compute_milestone(today, bday, dog["name"])
if milestone is None:
continue
titel, text = milestone
with db() as conn:
# Doppelt-Schutz: kein zweiter Eintrag am selben Tag mit gleichem Titel
exists = conn.execute("""
SELECT id FROM diary
WHERE dog_id = ? AND datum = ? AND titel = ? AND is_milestone = 1
""", (dog["id"], str(today), titel)).fetchone()
if exists:
logger.info(f"Meilenstein bereits vorhanden: Hund {dog['id']} '{titel}'")
continue
# Tagebucheintrag anlegen
cur = conn.execute("""
INSERT INTO diary (dog_id, datum, typ, titel, text, is_milestone)
VALUES (?, ?, 'milestone', ?, ?, 1)
""", (dog["id"], str(today), titel, text))
entry_id = cur.lastrowid
# Junction-Tabelle befüllen
conn.execute("""
INSERT OR IGNORE INTO diary_dogs (diary_id, dog_id) VALUES (?, ?)
""", (entry_id, dog["id"]))
# Push an Besitzer
send_push_to_user(dog["user_id"], {
"type": "milestone",
"title": titel,
"body": text,
"data": {"page": "diary"},
"tag": f"milestone-{dog['id']}-{today}",
})
logger.info(f"Meilenstein erstellt: Hund {dog['id']} '{titel}' → diary_id={entry_id}")
created_total += 1
logger.info(f"Meilenstein-Check fertig — {created_total} Einträge erstellt.")
_log_job("milestone_check", "ok", f"{created_total} Meilensteine erstellt")
# ------------------------------------------------------------------
# JOB: VDH-Events importieren
# ------------------------------------------------------------------
async def _job_import_events():
"""
Scrapt Veranstaltungen von vdh.de und importiert neue Events in die DB.
Bereits vorhandene external_ids werden übersprungen (Upsert-Logik).
"""
try:
from scraper.events_vdh import fetch_vdh_events
except ImportError as e:
logger.error(f"Event-Import: Scraper konnte nicht geladen werden: {e}")
return
try:
events = await fetch_vdh_events()
except Exception as e:
logger.error(f"Event-Import: Fehler beim Scrapen: {e}")
return
imported = 0
with db() as conn:
for ev in events:
try:
exists = conn.execute(
"SELECT id FROM events WHERE external_id = ?",
(ev['external_id'],)
).fetchone()
if not exists:
conn.execute("""
INSERT INTO events (user_id, titel, datum, ort_name, typ, link, quelle, external_id, status)
VALUES (NULL, ?, ?, ?, ?, ?, 'vdh', ?, 'aktiv')
""", (
ev['titel'],
ev['datum'],
ev.get('ort_name'),
ev['typ'],
ev.get('link'),
ev['external_id'],
))
imported += 1
except Exception as e:
logger.warning(f"Event-Import: Fehler beim Speichern von '{ev.get('titel')}': {e}")
logger.info(f"Event-Import: {imported} neue Events importiert (von {len(events)} geparsten).")
_log_job("import_events", "ok", f"{imported} neue von {len(events)} Events")
# ------------------------------------------------------------------
# JOB: Rassen aus TheDogAPI seeden
# ------------------------------------------------------------------
async def _job_seed_breeds():
"""Lädt alle Hunderassen von TheDogAPI und speichert sie in wiki_rassen."""
try:
from scraper.breeds import fetch_and_seed_breeds, mirror_breed_photos
except ImportError as e:
logger.error(f"Breed-Seed: Scraper konnte nicht geladen werden: {e}")
return
try:
count = await fetch_and_seed_breeds()
logger.info(f"Breed seed job done: {count} breeds")
mirrored = await mirror_breed_photos()
logger.info(f"Breed photo mirror done: {mirrored} photos")
except Exception as e:
logger.error(f"Breed-Seed: Fehler: {e}")
# ------------------------------------------------------------------
# JOB: Fehlende Rassen aus Wikidata ergänzen
# ------------------------------------------------------------------
async def _job_seed_wikidata_breeds():
"""Lädt fehlende Hunderassen von Wikidata und spiegelt Fotos lokal."""
try:
from scraper.wikidata_breeds import fetch_and_seed_wikidata_breeds, mirror_wikidata_photos
except ImportError as e:
logger.error(f"Wikidata-Seed: Scraper konnte nicht geladen werden: {e}")
return
try:
count = await fetch_and_seed_wikidata_breeds()
logger.info(f"Wikidata breed seed done: {count} neue Rassen")
mirrored = await mirror_wikidata_photos()
logger.info(f"Wikidata photo mirror done: {mirrored} Fotos")
# Wikipedia-Fotos für Rassen die noch kein Bild haben
from scraper.wikipedia_photos import fetch_wikipedia_photos
wp_count = await fetch_wikipedia_photos()
logger.info(f"Wikipedia photo fetch done: {wp_count} Fotos")
_log_job("seed_wikidata", "ok", f"{count} Rassen, {mirrored}+{wp_count} Fotos")
except Exception as e:
logger.error(f"Wikidata-Seed: Fehler: {e}")
_log_job("seed_wikidata", "error", str(e))
# ------------------------------------------------------------------
# Hilfsfunktion: Job-Protokoll aktualisieren
# ------------------------------------------------------------------
def _log_job(job_id: str, status: str, result: str):
_job_log[job_id] = {
"last_run": datetime.now(tz=_TZ),
"status": status,
"result": result,
}
# ------------------------------------------------------------------
# Hilfsfunktion: Lob-Text für einen Hund generieren
# ------------------------------------------------------------------
async def _generate_praise_for_dog(dog: dict, user_id: int) -> str:
"""Generiert einen Lob-Text für einen Hund basierend auf der letzten Woche."""
from ki import complete, KIUnavailableError
import json as _json
from datetime import date, timedelta
since = (date.today() - timedelta(days=7)).isoformat()
week_num = date.today().isocalendar()[1]
name = dog["name"]
rasse = dog.get("rasse") or "Hund"
stats = {}
try:
with db() as conn:
stats["diary"] = conn.execute("SELECT COUNT(*) FROM diary WHERE dog_id=? AND datum>=?", (dog["id"], since)).fetchone()[0]
stats["training"] = conn.execute("SELECT COUNT(*) FROM training_sessions WHERE dog_id=? AND datum>=?", (dog["id"], since)).fetchone()[0]
stats["top_training"] = conn.execute("SELECT COUNT(*) FROM training_sessions WHERE dog_id=? AND datum>=? AND ist_top=1", (dog["id"], since)).fetchone()[0]
stats["health"] = conn.execute("SELECT COUNT(*) FROM health WHERE dog_id=? AND datum>=?", (dog["id"], since)).fetchone()[0]
stats["days_active"] = conn.execute(
"SELECT COUNT(DISTINCT datum) FROM diary WHERE dog_id=? AND datum>=?", (dog["id"], since)
).fetchone()[0]
# Wie viele Wochen ist der User dabei?
first = conn.execute("SELECT MIN(datum) FROM diary WHERE dog_id=?", (dog["id"],)).fetchone()[0]
if first:
weeks_total = max(1, (date.today() - date.fromisoformat(first)).days // 7)
else:
weeks_total = 1
stats["weeks_total"] = weeks_total
except Exception:
pass
# Prompt aufbauen
aktivitaet_parts = []
if stats.get("diary", 0):
aktivitaet_parts.append(f"{stats['diary']} Tagebuch-Eintr\u00e4ge")
if stats.get("training", 0):
t = f"{stats['training']} Trainingseinheiten"
if stats.get("top_training", 0):
t += f" (davon {stats['top_training']} Top-Training)"
aktivitaet_parts.append(t)
if stats.get("health", 0):
aktivitaet_parts.append(f"{stats['health']} Gesundheitseintr\u00e4ge")
if not aktivitaet_parts:
aktivitaet_text = "Diese Woche war ruhig \u2014 keine erfassten Aktivit\u00e4ten."
else:
aktivitaet_text = ", ".join(aktivitaet_parts)
# W\u00f6chentlich rotierender Fokus \u2192 die KI klingt nicht jede Woche gleich.
_toene = [
"Betone die enge Verbundenheit zwischen {name} und dir.",
"Hebe die kleinen Abenteuer und besonderen Momente hervor.",
"W\u00fcrdige die sch\u00f6ne gemeinsame Routine und Verl\u00e4sslichkeit.",
"Feiere das gemeinsame Wachsen und die Fortschritte.",
"Betone Ruhe, Geborgenheit und Vertrauen.",
"Schreibe verspielt und mit einem Augenzwinkern.",
]
ton = _toene[week_num % len(_toene)].replace("{name}", name)
prompt = f"""Du bist ein warmer, wohlwollender Begleiter f\u00fcr Hundebesitzer. Schreibe eine kurze pers\u00f6nliche Lob-Nachricht (2-3 S\u00e4tze) f\u00fcr die vergangene Woche.
Hund: {name} ({rasse})
Letzte 7 Tage: {aktivitaet_text}
Fokus dieser Woche: {ton}
Regeln (unbedingt einhalten):
- Nur loben, NIEMALS Ratschl\u00e4ge geben oder auf Fehlendes hinweisen
- Sprich \u00fcber den Hund: "{name} hatte eine tolle Woche" \u2014 nicht \u00fcber den Besitzer
- Auch bei 0 Aktivit\u00e4ten: positive Formulierung (\u201eAuch ruhige Wochen geh\u00f6ren dazu\u201c)
- Variiere Einstieg und Wortwahl \u2014 klinge NICHT wie letzte Woche
- Erw\u00e4hne KEINE Wochenzahl und keine nackten Statistik-Zahlen
- Maximal 3 kurze S\u00e4tze
- Warm, pers\u00f6nlich, keine Floskeln
- Kein "Du solltest...", kein "Vergiss nicht...", keine Empfehlungen"""
try:
text = await complete(
prompt,
system="Du schreibst kurze, warme Lob-Nachrichten f\u00fcr Hundebesitzer. Nur Lob, keine Ratschl\u00e4ge.",
max_tokens=150,
)
return text.strip()
except Exception:
# Fallback wenn KI nicht verfügbar — Varianten-Pool, deterministisch pro
# Woche+Hund gewählt, damit der Text nicht jede Woche identisch klingt.
aktiv_varianten = [
f"{name} hatte eine richtig aktive Woche — {aktivitaet_text}. Stark! 🐾",
f"Was für eine Woche, {name}! {aktivitaet_text} — das kann sich sehen lassen. 🌟",
f"{name} war diese Woche voll dabei: {aktivitaet_text}. Weiter so! 🐶",
f"Tolle Woche mit {name} — {aktivitaet_text}. Ihr seid ein super Team! 🐾",
f"{aktivitaet_text} — dafür hat sich {name} eine extra Streicheleinheit verdient. ✨",
]
ruhig_varianten = [
f"Auch ruhige Wochen gehören dazu. {name} weiß, dass du für ihn da bist. 🐾",
f"Diese Woche war's gemütlich — und das ist völlig okay. {name} genießt die Zeit mit dir. 🌿",
f"Nicht jede Woche muss voll sein. {name} fühlt sich bei dir einfach wohl. ☀️",
f"Eine entspannte Woche mit {name} — manchmal ist genau das das Schönste. 🐾",
f"{name} und du — auch ohne großes Programm seid ihr ein eingespieltes Team. 🐶",
]
pool = aktiv_varianten if aktivitaet_parts else ruhig_varianten
return pool[(week_num + dog["id"]) % len(pool)]
# ------------------------------------------------------------------
# JOB: Wöchentlicher Fortschritts-Lober
# ------------------------------------------------------------------
async def _job_weekly_praise():
"""Jeden Montag: Lob-Text f\u00fcr alle aktiven Hunde generieren + Push senden."""
from datetime import date
import json as _json
today = date.today()
d = today.isocalendar()
week_key = f"{d[0]}-W{d[1]:02d}"
logger.info(f"Weekly Praise Job startet f\u00fcr Woche {week_key}")
# Alle Hunde laden, für die noch kein Lob diese Woche existiert
with db() as conn:
dogs = conn.execute("""
SELECT d.id, d.name, d.rasse, d.user_id, d.foto_url
FROM dogs d
WHERE NOT EXISTS (
SELECT 1 FROM weekly_praise wp
WHERE wp.dog_id=d.id AND wp.week_key=?
)
ORDER BY d.id
""", (week_key,)).fetchall()
dogs = [dict(d) for d in dogs]
logger.info(f"Weekly Praise: {len(dogs)} Hunde ohne Lob diese Woche.")
import asyncio
generated = 0
for dog in dogs:
try:
praise = await _generate_praise_for_dog(dog, dog["user_id"])
with db() as conn:
conn.execute("""
INSERT OR IGNORE INTO weekly_praise (user_id, dog_id, week_key, praise_text)
VALUES (?,?,?,?)
""", (dog["user_id"], dog["id"], week_key, praise))
# Push-Notification — erste 100 Zeichen als Preview
preview = praise[:100] + "\u2026" if len(praise) > 100 else praise
send_push_to_user(dog["user_id"], {
"type": "weekly_praise",
"title": f"\U0001f43e R\u00fcckblick f\u00fcr {dog['name']}",
"body": preview,
"data": {"page": "diary"},
"tag": f"weekly-praise-{dog['id']}-{week_key}",
})
generated += 1
await asyncio.sleep(2) # Rate limiting für KI
except Exception as e:
logger.error(f"Weekly Praise: Fehler f\u00fcr Hund {dog['id']}: {e}")
logger.info(f"Weekly Praise Job fertig \u2014 {generated}/{len(dogs)} Lob-Texte generiert.")
_log_job("weekly_praise", "ok", f"{generated} Lob-Texte f\u00fcr KW {d[1]}")
# ------------------------------------------------------------------
# JOB: KI-Gesundheitsberichte (alle 2 Wochen, jeden Montag 07:00)
# ------------------------------------------------------------------
async def _job_ki_health_report():
"""
Erstellt für jeden Hund, der seit mehr als 13 Tagen keinen KI-Gesundheitsbericht
hat (oder noch keinen hatte), einen neuen Bericht via ki.health_summary() und
schickt eine Push-Notification an den Besitzer. Maximal 20 Hunde pro Lauf.
"""
import ki as KI
with db() as conn:
dogs = conn.execute("""
SELECT d.id AS dog_id, d.name, d.rasse, d.geburtstag, d.gewicht_kg, d.user_id
FROM dogs d
WHERE d.id NOT IN (
SELECT dog_id FROM ki_health_reports
WHERE erstellt_at >= datetime('now', '-13 days')
)
ORDER BY d.id
LIMIT 20
""").fetchall()
dogs = [dict(d) for d in dogs]
if not dogs:
logger.info("KI-Gesundheitsbericht: Keine fälligen Hunde.")
_log_job("ki_health_report", "ok", "0 Berichte erstellt")
return
count = 0
for dog in dogs:
try:
with db() as conn:
health_rows = conn.execute(
"SELECT * FROM health WHERE dog_id=? ORDER BY datum DESC",
(dog["dog_id"],)
).fetchall()
health_data = [dict(r) for r in health_rows]
dog_info = {
"name": dog["name"],
"rasse": dog.get("rasse"),
"geburtstag": dog.get("geburtstag"),
"gewicht_kg": dog.get("gewicht_kg"),
}
bericht = await KI.health_summary(health_data=health_data, dog_info=dog_info)
with db() as conn:
conn.execute(
"INSERT INTO ki_health_reports (dog_id, user_id, bericht) VALUES (?, ?, ?)",
(dog["dog_id"], dog["user_id"], bericht)
)
send_push_to_user(dog["user_id"], {
"type": "ki_health_report",
"title": f"Gesundheitsbericht für {dog['name']}",
"body": "Dein KI-Assistent hat einen neuen Bericht erstellt.",
"data": {"page": "health"},
})
count += 1
logger.info(f"KI-Gesundheitsbericht: Bericht für Hund {dog['dog_id']} ({dog['name']}) erstellt.")
except Exception as e:
logger.error(f"KI-Gesundheitsbericht: Fehler für Hund {dog['dog_id']} ({dog['name']}): {e}")
logger.info(f"KI-Gesundheitsbericht Job fertig — {count}/{len(dogs)} Berichte erstellt.")
_log_job("ki_health_report", "ok", f"{count} Berichte erstellt")
# ------------------------------------------------------------------
async def _job_moderation_overdue():
"""Sendet Alarm-Mail wenn Moderations-Einträge seit >24h offen sind."""
import os
from mailer import send_email
admin = os.getenv("ADMIN_EMAIL", "")
if not admin:
return
SLA_H = 24
threshold = f"datetime('now', '-{SLA_H} hours')"
overdue = {}
try:
with db() as conn:
n = conn.execute(f"SELECT COUNT(*) FROM job_applications WHERE status IN ('pending','reviewing') AND created_at < {threshold}").fetchone()[0]
if n: overdue["Bewerbungen"] = n
n = conn.execute(f"SELECT COUNT(*) FROM users WHERE breeder_status='pending' AND created_at < {threshold}").fetchone()[0]
if n: overdue["Züchter-Anträge"] = n
n = conn.execute(f"SELECT COUNT(*) FROM forum_reports WHERE resolved=0 AND created_at < {threshold}").fetchone()[0]
if n: overdue["Forum-Meldungen"] = n
n = conn.execute(f"SELECT COUNT(*) FROM wiki_foto_submissions WHERE status='pending' AND created_at < {threshold}").fetchone()[0]
if n: overdue["Foto-Einreichungen"] = n
n = conn.execute(f"SELECT COUNT(*) FROM osm_poi_edits WHERE status='pending' AND created_at < {threshold}").fetchone()[0]
if n: overdue["POI-Korrekturen"] = n
n = conn.execute(f"SELECT COUNT(*) FROM wiki_zuchter WHERE verified=0 AND created_at < {threshold}").fetchone()[0]
if n: overdue["Züchter-Einreichungen (Wiki)"] = n
except Exception as e:
logger.error(f"Moderation-Overdue-Check: DB-Fehler: {e}")
return
if not overdue:
logger.info("Moderation-Overdue-Check: Alles im SLA.")
return
now_str = datetime.now(tz=_TZ).strftime("%d.%m.%Y %H:%M")
rows_html = "".join(
f'| {label} | '
f'{count} |
'
for label, count in overdue.items()
)
html = f"""\
⚠️ Moderation überfällig
{now_str} · SLA: {SLA_H}h
Folgende Einträge warten seit mehr als {SLA_H} Stunden auf Bearbeitung:
| Bereich |
Anzahl |
{rows_html}
Ban Yaro · banyaro.app
"""
plain = f"Ban Yaro — Moderation überfällig ({now_str})\n\nSeit >{SLA_H}h offen:\n" + \
"\n".join(f" • {l}: {c}" for l, c in overdue.items()) + \
"\n\nhttps://banyaro.app/app/admin"
try:
await send_email(admin, f"⚠️ Ban Yaro — Moderation überfällig ({', '.join(overdue)})", html, plain)
logger.info(f"Moderation-Overdue-Mail gesendet: {overdue}")
except Exception as e:
logger.error(f"Moderation-Overdue-Mail fehlgeschlagen: {e}")
def _action_items_html(metrics: dict) -> str:
items = [
("jobs_pending", "Bewerbungen offen"),
("breeder_pending", "Züchter-Anträge"),
("reports_open", "Forum-Meldungen"),
("fotos_pending", "Foto-Einreichungen"),
("poi_edits_pending", "POI-Korrekturen"),
]
open_items = [(label, metrics.get(key, 0)) for key, label in items if metrics.get(key, 0) > 0]
if not open_items:
body = '✅ Alles erledigt — nichts offen'
else:
pills = "".join(
f''
f'{label} {count}'
for label, count in open_items
)
body = f'⚠️ {len(open_items)} Punkt{"e" if len(open_items)!=1 else ""} brauchen deine Aufmerksamkeit
{pills}'
link = ''
return f'' \
f'
Heute zu erledigen
' \
f'{body}{link}
| {label} | {ts} | {status} {result} |
'
job_rows_txt += f" {status} {label}: {ts} — {result}\n"
else:
job_rows_html += f'| {label} | — noch nicht gelaufen |
'
job_rows_txt += f" — {label}: noch nicht gelaufen\n"
html = f"""\
🐾 Ban Yaro — Status-Report
{now_str} Uhr
{_action_items_html(metrics)}
Community
{"".join(f'
' for k,v in [
("Nutzer gesamt",metrics["users"]),
("Neue Nutzer heute",metrics["users_today"]),
("Hunde",metrics["dogs"]),
("Tagebuch-Einträge",metrics["diary_entries"]),
("Aktive Giftköder",metrics["poison_active"]),
("Vermisste Hunde",metrics["lost_active"]),
("'So einen hab ich'",metrics["interesse_hat"]),
("'Interessiert mich'",metrics["interesse_will"]),
])}
Ban Yaro · banyaro.app · Nächster Report in ~6h
"""
action_open = [l for k,l in [
("jobs_pending","Bewerbungen"),("breeder_pending","Züchter-Anträge"),
("reports_open","Meldungen"),("fotos_pending","Fotos"),("poi_edits_pending","POI-Korrekturen"),
] if metrics.get(k,0) > 0]
plain = f"""Ban Yaro Status-Report — {now_str}
=== HEUTE ZU ERLEDIGEN ===
{"✅ Alles erledigt" if not action_open else "⚠️ " + ", ".join(f"{l} ({metrics[k]})" for k,l in [
("jobs_pending","Bewerbungen"),("breeder_pending","Züchter-Anträge"),
("reports_open","Meldungen"),("fotos_pending","Fotos"),("poi_edits_pending","POI-Korrekturen"),
] if metrics.get(k,0) > 0)}
=== Scheduler-Jobs ===
{job_rows_txt}
=== Community ===
Nutzer gesamt: {metrics['users']} (+{metrics['users_today']} heute)
Hunde: {metrics['dogs']}
Tagebuch-Einträge: {metrics['diary_entries']}
Aktive Giftköder: {metrics['poison_active']}
Vermisste Hunde: {metrics['lost_active']}
'So einen hab ich': {metrics['interesse_hat']}
'Interessiert mich': {metrics['interesse_will']}
"""
try:
await send_email(admin, f"Ban Yaro Status {now_str}", html, plain)
logger.info(f"Status-Report gesendet an {admin}.")
except Exception as e:
logger.error(f"Status-Report: Mail-Fehler: {e}")
async def _job_quarterly_report():
"""Quartalsbericht: alle Report-Sections per Mail an ADMIN_EMAIL."""
import os, sys
from mailer import send_email, email_html
admin = os.getenv("ADMIN_EMAIL", "")
if not admin:
logger.info("Quartalsbericht: ADMIN_EMAIL nicht gesetzt, übersprungen.")
_log_job("quarterly_report", "ok", "ADMIN_EMAIL nicht gesetzt")
return
now_str = datetime.now(tz=_TZ).strftime("%d.%m.%Y")
quarter = (datetime.now(tz=_TZ).month - 1) // 3 + 1
try:
# Report-Script importieren und alle Sections aufrufen
sys.path.insert(0, "/app/scripts")
import importlib, generate_reports as gr
importlib.reload(gr) # sicherstellen dass aktuelle Version
sections = [
("Sicherheit", gr.report_sicherheit),
("Funktionsumfang", gr.report_funktionsumfang),
("Dateien", gr.report_dateien),
("Nutzerübersicht", gr.report_nutzer),
("Partnerliste", gr.report_partner),
("Server & Speicher", gr.report_server),
]
def md_to_html_simple(text: str) -> str:
"""Minimale Markdown→HTML-Konvertierung für E-Mail."""
import html as _h
lines_out = []
in_code = False
in_table = False
for line in text.split("\n"):
if line.startswith("```"):
if in_code:
lines_out.append("")
in_code = False
else:
lines_out.append('')
in_code = True
continue
if in_code:
lines_out.append(_h.escape(line))
continue
if line.startswith("#### "):
lines_out.append(f'{line[5:]}
')
elif line.startswith("### "):
lines_out.append(f'{line[4:]}
')
elif line.startswith("## "):
lines_out.append(f'{line[3:]}
')
elif line.startswith("# "):
pass # Haupttitel kommt vom äußeren Template
elif line.startswith("---"):
pass # Trennlinie überspringen
elif line.startswith("| "):
if not in_table:
lines_out.append('')
in_table = True
if set(line.replace("|","").replace("-","").replace(" ","")) == set():
continue # Trenn-Zeile
cells = [c.strip() for c in line.split("|")[1:-1]]
row_html = "".join(f'| {_h.escape(c)} | ' for c in cells)
lines_out.append(f"{row_html}
")
continue
elif line.startswith("- ") or line.startswith("* "):
if in_table:
lines_out.append("
")
in_table = False
lines_out.append(f'{line[2:]}')
elif line.startswith("> "):
if in_table:
lines_out.append("")
in_table = False
lines_out.append(f'{line[2:]}
')
elif line.strip() == "":
if in_table:
lines_out.append("")
in_table = False
lines_out.append("")
else:
if in_table:
lines_out.append("")
in_table = False
styled = line.replace("**", "", 1).replace("**", "", 1)
lines_out.append(f'{styled}
')
if in_table:
lines_out.append("")
if in_code:
lines_out.append("
")
return "\n".join(lines_out)
# Body aus allen Sections zusammensetzen
body_parts = []
plain_parts = [f"Ban Yaro Quartalsbericht Q{quarter} {now_str}\n", "=" * 50]
for title, fn in sections:
try:
md = fn()
body_parts.append(
f''
f'
{title}
'
f'{md_to_html_simple(md)}'
f''
)
plain_parts.append(f"\n=== {title.upper()} ===\n{md}\n")
except Exception as e:
body_parts.append(f'Fehler in Section {title}: {e}
')
plain_parts.append(f"\n=== {title.upper()} ===\nFehler: {e}\n")
full_body = "\n".join(body_parts)
full_plain = "\n".join(plain_parts)
subject = f"Ban Yaro Quartalsbericht Q{quarter}/{datetime.now(tz=_TZ).year} — {now_str}"
html = email_html(full_body, footer_text=f"Ban Yaro · banyaro.app · Quartalsbericht Q{quarter}")
await send_email(admin, subject, html, full_plain)
logger.info(f"Quartalsbericht Q{quarter} gesendet an {admin}.")
_log_job("quarterly_report", "ok", f"Q{quarter} → {admin}")
except Exception as e:
logger.error(f"Quartalsbericht: Fehler: {e}")
_log_job("quarterly_report", "error", str(e))
def _compute_milestone(today: date, bday: date, dog_name: str):
"""
Gibt (titel, text) zurück wenn heute ein Meilenstein-Tag ist,
sonst None.
Regeln:
- Jahrestag (Monat + Tag stimmen überein, Jahrgang ≥ 1):
"🎂 ist X Jahr(e) alt!"
- Monatsjubiläum in den ersten 11 Monaten (Geburtsmonats-Tag):
"🐾 ist heute X Monat(e) alt!"
"""
# Jahrestag?
if today.month == bday.month and today.day == bday.day:
years = today.year - bday.year
if years <= 0:
return None # Geburtstag selbst (Tag 0) → kein Eintrag
years_label = f"{years} Jahr" if years == 1 else f"{years} Jahre"
titel = f"🎂 {dog_name} ist {years_label} alt!"
text = (
f"Heute feiern wir {dog_name}s {years}. Geburtstag! 🐾🎉 "
f"Herzlichen Glückwunsch zum {years_label}!"
)
return titel, text
# Monatsjubiläum (nur innerhalb des ersten Lebensjahres)?
# today liegt im selben Monatstag wie der Geburtstag aber in einem anderen Monat.
if today.day == bday.day:
# Vollständige Monate seit Geburt berechnen
months = (today.year - bday.year) * 12 + (today.month - bday.month)
if 1 <= months <= 11:
months_label = f"{months} Monat" if months == 1 else f"{months} Monate"
titel = f"🐾 {dog_name} ist heute {months_label} alt!"
text = (
f"{dog_name} wird heute {months_label} alt — "
f"was für ein tolles kleines Hundeleben! 🥳"
)
return titel, text
return None
# ------------------------------------------------------------------
# JOB: Hund des Monats — Sieger des Vormonats festlegen
# ------------------------------------------------------------------
async def _job_hdm_winner():
"""Läuft am 1. des Monats 00:05 und schreibt den Sieger des Vormonats."""
today = datetime.now(tz=_TZ)
# Vormonat berechnen
first_this = today.replace(day=1)
last_month = (first_this - timedelta(days=1)).replace(day=1)
monat = last_month.strftime("%Y-%m")
with db() as conn:
# Schon eingetragen?
existing = conn.execute(
"SELECT id FROM hund_des_monats_wins WHERE monat=?", (monat,)
).fetchone()
if existing:
logger.info(f"HdM-Winner {monat}: bereits eingetragen, übersprungen.")
_log_job("hdm_winner", "ok", f"bereits vorhanden für {monat}")
return
winner = conn.execute("""
SELECT v.dog_id, d.name, d.user_id, COUNT(v.id) AS stimmen
FROM hund_des_monats_votes v
JOIN dogs d ON d.id = v.dog_id
WHERE v.monat = ?
GROUP BY v.dog_id
ORDER BY stimmen DESC
LIMIT 1
""", (monat,)).fetchone()
if not winner:
logger.info(f"HdM-Winner {monat}: keine Stimmen, kein Sieger.")
_log_job("hdm_winner", "ok", f"keine Stimmen für {monat}")
return
conn.execute(
"INSERT OR IGNORE INTO hund_des_monats_wins (dog_id, monat, stimmen) VALUES (?, ?, ?)",
(winner["dog_id"], monat, winner["stimmen"]),
)
month_label = last_month.strftime("%B %Y")
send_push_to_user(winner["user_id"], {
"type": "hdm_winner",
"title": f"🏆 {winner['name']} ist Hund des Monats!",
"body": f"{winner['name']} hat den {month_label} gewonnen — herzlichen Glückwunsch!",
"data": {"page": "forum"},
"tag": f"hdm-{monat}",
})
logger.info(f"HdM-Winner {monat}: Hund {winner['dog_id']} ('{winner['name']}', {winner['stimmen']} Stimmen) eingetragen.")
_log_job("hdm_winner", "ok", f"{monat}: {winner['name']} ({winner['stimmen']} Stimmen)")
# ------------------------------------------------------------------
# JOB: Streak-Erinnerung (täglich 19:00)
# ------------------------------------------------------------------
async def _job_streak_reminder():
"""
Findet alle User die heute noch nicht trainiert haben (last_training_date < heute)
und deren current_streak > 0. Sendet einen motivierenden Push pro Hund.
"""
today = str(date.today())
logger.info(f"Streak-Reminder Job läuft für {today}")
with db() as conn:
rows = conn.execute("""
SELECT ts.user_id, ts.dog_id, ts.current_streak, d.name AS dog_name
FROM training_streaks ts
JOIN dogs d ON d.id = ts.dog_id
WHERE ts.current_streak > 0
AND (ts.last_training_date IS NULL OR ts.last_training_date < ?)
""", (today,)).fetchall()
sent_total = 0
for r in rows:
n = r["current_streak"]
sent = send_push_to_user(r["user_id"], {
"type": "streak_reminder",
"title": f"🔥 {r['dog_name']} wartet auf sein Training!",
"body": f"Streak: {n} {'Tag' if n == 1 else 'Tage'} — nicht jetzt aufhören.",
"data": {"page": "uebungen"},
"tag": f"streak-{r['dog_id']}-{today}",
})
sent_total += sent
logger.info(f"Streak-Reminder Job fertig — {len(rows)} Hunde geprüft, {sent_total} Push gesendet.")
_log_job("streak_reminder", "ok", f"{sent_total} Push an {len(rows)} Hunde")
# ------------------------------------------------------------------
# JOB: Tierfutter-Rückrufe prüfen (RASFF, täglich 08:00)
# ------------------------------------------------------------------
async def _job_osm_confirm():
"""OSM-Beiträge: Pending-Retry + Revert-Überleben prüfen + Pro-Freischaltung.
Import innen → kein Zirkel-Import beim Modul-Load."""
try:
from routes.osm_contrib import run_confirmation_round
await run_confirmation_round()
except Exception as e:
logger.warning("OSM-Confirm-Job Fehler: %s", e)
async def _job_recall_check():
"""
Fragt täglich die RASFF EU-API nach neuen Tierfutter-Rückrufen ab.
Neue Einträge werden in DB gespeichert, für jeden wird ein Push
an alle abonnierten User gesendet.
"""
logger.info("Rückruf-Check Job läuft")
try:
from routes.recalls import fetch_rasff_recalls, save_new_recalls
entries = await fetch_rasff_recalls()
if not entries:
logger.info("Rückruf-Check: Keine Einträge von RASFF erhalten (API-Fehler oder leer).")
_log_job("recall_check", "ok", "0 neue Rückrufe (API leer)")
return
new_entries = save_new_recalls(entries)
logger.info(f"Rückruf-Check: {len(new_entries)} neue von {len(entries)} geprüften Einträgen.")
for entry in new_entries:
produkt = entry.get("produkt") or entry.get("titel") or "Unbekanntes Produkt"
gefahr = entry.get("gefahr") or "Bitte Produktdetails prüfen"
ext_id = entry["external_id"]
body = f"{produkt} — {gefahr[:80]}"
send_push_to_all({
"title": "⚠️ Tierfutter-Rückruf",
"body": body,
"data": {"page": "recalls"},
"tag": f"recall-{ext_id}",
})
logger.info(f"Rückruf-Push gesendet: {ext_id} — {produkt}")
_log_job("recall_check", "ok", f"{len(new_entries)} neue Rückrufe")
except Exception as e:
logger.error(f"Rückruf-Check: unerwarteter Fehler: {e}")
_log_job("recall_check", "error", str(e))
# ------------------------------------------------------------------
# JOB: Wiederkehrende Ausgaben anlegen
# ------------------------------------------------------------------
async def _job_recurring_expenses():
try:
from routes.expenses import process_due_recurring
count = process_due_recurring()
logger.info(f"Daueraufträge: {count} Einträge angelegt.")
_log_job("recurring_expenses", "ok", f"{count} Einträge")
except Exception as e:
logger.error(f"Daueraufträge-Job Fehler: {e}")
_log_job("recurring_expenses", "error", str(e))
# ------------------------------------------------------------------
# JOB: Goldene Gassi-Stunde (täglich 07:00 Uhr)
# ------------------------------------------------------------------
async def _job_golden_gassi_hour():
"""
Berechnet für jeden User mit aktivierter Einstellung (gassi_stunde_push=1)
das beste 2h-Wetterfenster des Tages und schickt eine Push-Notification.
Score-Logik pro Stunde (max. 10 Punkte):
- Temperatur 10–20°C → +3
- Temperatur 5–10°C → +1
- Niederschlagswahrsch. <20% → +3, <40% → +1
- Windgeschwindigkeit <20 km/h → +2, <30 km/h → +1
- Stunden 07–19 Uhr (Tageslicht) → +2
Bestes fortlaufendes 2h-Fenster (Summe zweier aufeinanderfolgender Stunden).
"""
import httpx
from datetime import date as _date
logger.info("Goldene-Gassi-Stunde Job läuft")
# Alle User mit aktivierter Einstellung + mindestens einer Push-Subscription
with db() as conn:
users = conn.execute("""
SELECT DISTINCT u.id AS user_id,
ps.last_lat, ps.last_lon
FROM users u
JOIN push_subscriptions ps ON ps.user_id = u.id
WHERE u.gassi_stunde_push = 1
""").fetchall()
users = [dict(u) for u in users]
logger.info(f"Goldene-Gassi-Stunde: {len(users)} User mit aktivierter Einstellung.")
if not users:
_log_job("golden_gassi_hour", "ok", "0 User mit Einstellung aktiv")
return
sent_total = 0
for u in users:
lat = u["last_lat"] or 48.1351 # Fallback: München
lon = u["last_lon"] or 11.5820
try:
hourly = await _fetch_hourly_weather(lat, lon)
except Exception as e:
logger.warning(f"Goldene-Gassi-Stunde: Wetter-Fehler für user {u['user_id']}: {e}")
continue
if not hourly:
continue
best_start, best_score, best_temp, best_wind = _find_best_gassi_window(hourly)
if best_score < 3:
# Heute kein gutes Wetterfenster → kein Push
logger.info(f"Goldene-Gassi-Stunde: user {u['user_id']} — kein gutes Fenster (score={best_score})")
continue
hour_end = (best_start + 2) % 24
temp_str = f"{best_temp:.0f}°C" if best_temp is not None else "–"
wind_str = "Kaum Wind" if (best_wind is not None and best_wind < 20) else (
f"{best_wind:.0f} km/h Wind" if best_wind is not None else "")
body_parts = [f"Bestes Wetter zwischen {best_start:02d}:00–{hour_end:02d}:00 Uhr",
f"· {temp_str}"]
if wind_str:
body_parts.append(f"· {wind_str}")
sent = send_push_to_user(u["user_id"], {
"type": "golden_gassi_hour",
"title": "☀️ Goldene Gassi-Stunde heute!",
"body": " ".join(body_parts),
"data": {"page": "wetter"},
"tag": f"gassi-{_date.today()}",
})
sent_total += sent
logger.info(f"Goldene-Gassi-Stunde: user {u['user_id']} → {best_start:02d}:00 (score={best_score}, {temp_str}) — Push: {sent}")
logger.info(f"Goldene-Gassi-Stunde Job fertig — {len(users)} User, {sent_total} Push gesendet.")
_log_job("golden_gassi_hour", "ok", f"{sent_total} Push an {len(users)} User")
# ------------------------------------------------------------------
# JOB: Jahrestags-Erinnerungen (täglich 09:00)
# ------------------------------------------------------------------
async def _job_anniversary_reminders():
"""Prüft ob heute ein Jahrestag für diary-Einträge vorliegt und sendet Push."""
today = datetime.now(tz=_TZ)
today_md = today.strftime('%m-%d') # Monat-Tag ohne Jahr
logger.info(f"Jahrestags-Erinnerungen Job läuft für {today_md}")
with db() as conn:
# diary hat keinen user_id — User kommt über dogs.user_id
entries = conn.execute("""
SELECT d.id, d.titel, d.datum, dogs.user_id, d.dog_id,
(SELECT dm.url FROM diary_media dm
WHERE dm.diary_id=d.id LIMIT 1) AS foto_url
FROM diary d
JOIN dogs ON dogs.id = d.dog_id
WHERE strftime('%m-%d', d.datum) = ?
AND d.datum < date('now')
AND d.titel IS NOT NULL
AND d.is_milestone = 0
""", (today_md,)).fetchall()
sent_total = 0
for e in entries:
try:
jahre = today.year - int(e['datum'][:4])
if jahre < 1:
continue
jahre_label = f"{jahre} Jahr" if jahre == 1 else f"{jahre} Jahren"
send_push_to_user(e['user_id'], {
'type': 'anniversary_reminder',
'title': f'📅 Vor {jahre_label}: {(e["titel"] or "")[:40]}',
'body': 'Erinnerung an diesen besonderen Tag mit deinem Hund',
'data': {'page': 'diary'},
'tag': f'anniversary-{e["id"]}-{today.year}',
})
sent_total += 1
except Exception as ex:
logger.warning(f"Jahrestag-Reminder: Fehler für Eintrag {e['id']}: {ex}")
logger.info(f"Jahrestags-Erinnerungen Job fertig — {len(entries)} Einträge geprüft, {sent_total} Push gesendet.")
_log_job("anniversary_reminders", "ok", f"{sent_total} Push von {len(entries)} Einträgen")
# ------------------------------------------------------------------
# JOB: Monatlicher Rückblick (1. des Monats 10:00)
# ------------------------------------------------------------------
async def _job_monthly_recap():
"""Sendet jedem User am 1. des Monats einen Rückblick des Vormonats."""
today = datetime.now(tz=_TZ)
first_this = today.replace(day=1)
last_month_end = first_this - timedelta(days=1)
last_month_start = last_month_end.replace(day=1)
year_str = last_month_start.strftime('%Y')
month_str = last_month_start.strftime('%m')
month_label = last_month_start.strftime('%B %Y')
logger.info(f"Monatlicher Rückblick Job läuft für {month_label}")
with db() as conn:
# Alle User mit mindestens einem Hund
users = conn.execute(
"SELECT DISTINCT user_id FROM dogs"
).fetchall()
sent_total = 0
for u in users:
user_id = u["user_id"]
try:
with db() as conn:
# Hunde des Users
dog_rows = conn.execute(
"SELECT id, name FROM dogs WHERE user_id=?", (user_id,)
).fetchall()
if not dog_rows:
continue
dog_ids = [d["id"] for d in dog_rows]
placeholders = ','.join('?' * len(dog_ids))
# km (Routen des Users im Vormonat)
km_row = conn.execute(
"SELECT ROUND(COALESCE(SUM(distanz_km),0),1) AS km FROM routes "
"WHERE user_id=? AND strftime('%Y',created_at)=? AND strftime('%m',created_at)=?",
(user_id, year_str, month_str)
).fetchone()
gesamt_km = km_row["km"] or 0.0
# Tagebucheinträge
eintraege = conn.execute(
f"SELECT COUNT(*) AS n FROM diary "
f"WHERE dog_id IN ({placeholders}) AND strftime('%Y',datum)=? AND strftime('%m',datum)=?",
(*dog_ids, year_str, month_str)
).fetchone()["n"]
# Training-Sessions
training = conn.execute(
f"SELECT COUNT(*) AS n FROM training_sessions "
f"WHERE dog_id IN ({placeholders}) AND strftime('%Y',created_at)=? AND strftime('%m',created_at)=?",
(*dog_ids, year_str, month_str)
).fetchone()["n"]
# Lieblingsfoto (erstes Foto im Vormonat)
foto_row = conn.execute(
f"SELECT dm.url FROM diary_media dm "
f"JOIN diary d ON d.id=dm.diary_id "
f"WHERE d.dog_id IN ({placeholders}) AND dm.media_type='image' "
f"AND strftime('%Y',d.datum)=? AND strftime('%m',d.datum)=? "
f"ORDER BY d.datum ASC LIMIT 1",
(*dog_ids, year_str, month_str)
).fetchone()
foto_url = foto_row["url"] if foto_row else None
# Nur senden wenn mindestens eine Aktivität vorhanden
if eintraege == 0 and training == 0 and gesamt_km == 0:
continue
dog_name = dog_rows[0]["name"]
parts = []
if gesamt_km > 0:
parts.append(f"{gesamt_km} km gelaufen")
if eintraege > 0:
parts.append(f"{eintraege} Tagebucheintr{'ä' if True else 'a'}ge")
if training > 0:
parts.append(f"{training} Training-Sessions")
body_text = " · ".join(parts)
send_push_to_user(user_id, {
'type': 'monthly_recap',
'title': f'📅 {month_label}: Rückblick für {dog_name}',
'body': body_text,
'data': {'page': 'diary'},
'tag': f'monthly-recap-{year_str}-{month_str}',
})
sent_total += 1
except Exception as ex:
logger.error(f"Monatlicher Rückblick: Fehler für user {user_id}: {ex}")
logger.info(f"Monatlicher Rückblick Job fertig — {len(users)} User geprüft, {sent_total} Push gesendet.")
_log_job("monthly_recap", "ok", f"{sent_total} Push für {month_label}")
async def _job_new_foto_challenge():
"""Jeden Montag 08:00 — neue Foto-Challenge für die aktuelle Woche anlegen."""
from datetime import date, timedelta
from routes.challenges import _CHALLENGE_THEMEN, _current_week_monday, _current_week_sunday
monday = _current_week_monday()
sunday = _current_week_sunday()
week_num = date.today().isocalendar()[1]
thema = _CHALLENGE_THEMEN[week_num % len(_CHALLENGE_THEMEN)]
with db() as conn:
existing = conn.execute(
"SELECT id FROM foto_challenge WHERE start_date = ?", (monday,)
).fetchone()
if existing:
logger.info(f"Foto-Challenge: Woche {monday} bereits vorhanden (id={existing['id']}).")
_log_job("new_foto_challenge", "ok", f"Bereits vorhanden für {monday}")
return
cur = conn.execute(
"INSERT INTO foto_challenge (thema, beschreibung, start_date, end_date, created_by) "
"VALUES (?, ?, ?, ?, NULL)",
(thema, f"Diese Woche: {thema}", monday, sunday)
)
challenge_id = cur.lastrowid
# Push an alle User
send_push_to_all({
"type": "foto_challenge",
"title": "📸 Neue Foto-Challenge!",
"body": f"Diese Woche: {thema} — mach mit!",
"data": {"page": "walks", "tab": "challenge"},
"tag": f"challenge-{monday}",
})
logger.info(f"Foto-Challenge angelegt: '{thema}' für {monday}–{sunday} (id={challenge_id}).")
_log_job("new_foto_challenge", "ok", f"'{thema}' für {monday}")
async def _fetch_hourly_weather(lat: float, lon: float) -> list[dict]:
"""Holt stündliche Wetterdaten für heute von Open-Meteo."""
import httpx
from datetime import date as _date
today = _date.today().isoformat()
url = (
"https://api.open-meteo.com/v1/forecast"
f"?latitude={lat}&longitude={lon}"
"&hourly=temperature_2m,precipitation_probability,windspeed_10m"
"&timezone=Europe%2FBerlin&forecast_days=1"
)
async with httpx.AsyncClient(timeout=8.0) as client:
resp = await client.get(url)
resp.raise_for_status()
raw = resp.json()
hourly = raw.get("hourly", {})
times = hourly.get("time", [])
temps = hourly.get("temperature_2m", [])
precips = hourly.get("precipitation_probability", [])
winds = hourly.get("windspeed_10m", [])
result = []
for i, ts in enumerate(times):
if not ts.startswith(today):
continue
hour = int(ts[11:13])
result.append({
"hour": hour,
"temp": temps[i] if i < len(temps) else None,
"precip": precips[i] if i < len(precips) else None,
"wind": winds[i] if i < len(winds) else None,
})
return result
def _score_hour(h: dict) -> int:
"""Berechnet Gassi-Score für eine einzelne Stunde (0–10 Punkte)."""
score = 0
temp = h.get("temp")
precip = h.get("precip")
wind = h.get("wind")
hour = h.get("hour", 12)
# Temperatur
if temp is not None:
if 10 <= temp <= 20:
score += 3
elif 5 <= temp < 10 or 20 < temp <= 25:
score += 1
# Niederschlagswahrscheinlichkeit
if precip is not None:
if precip < 20:
score += 3
elif precip < 40:
score += 1
# Wind
if wind is not None:
if wind < 20:
score += 2
elif wind < 30:
score += 1
# Tageslicht (07–19 Uhr)
if 7 <= hour <= 19:
score += 2
return score
def _find_best_gassi_window(hourly: list[dict]) -> tuple[int, int, float | None, float | None]:
"""
Findet das beste aufeinanderfolgende 2h-Fenster.
Gibt (start_hour, total_score, avg_temp, avg_wind) zurück.
"""
best_start = 8
best_score = -1
best_temp = None
best_wind = None
for i in range(len(hourly) - 1):
h1 = hourly[i]
h2 = hourly[i + 1]
combined = _score_hour(h1) + _score_hour(h2)
if combined > best_score:
best_score = combined
best_start = h1["hour"]
# Durchschnittswerte für Anzeige
temps = [x for x in [h1.get("temp"), h2.get("temp")] if x is not None]
winds = [x for x in [h1.get("wind"), h2.get("wind")] if x is not None]
best_temp = sum(temps) / len(temps) if temps else None
best_wind = sum(winds) / len(winds) if winds else None
return best_start, best_score, best_temp, best_wind
# ------------------------------------------------------------------
# JOB: Error-Digest (täglich 06:30 Uhr)
# ------------------------------------------------------------------
# Quellen fuer Fehler:
# 1. _job_log dieser Datei — Status der einzelnen Scheduler-Jobs
# 2. main.log_buffer — In-Memory-Buffer der letzten 500 Log-Zeilen
# Limitierung: log_buffer ist nicht persistent. Ein vollstaendiger 24h-Digest
# wuerde eine externe Log-Senke (Datei oder DB) brauchen. Bis dahin liefert
# der Job das was im Memory verfuegbar ist plus alle Scheduler-Errors.
# ------------------------------------------------------------------
async def _job_error_digest():
"""Schickt eine Zusammenfassung aller bekannten ERROR/EXCEPTION-Eintraege
der letzten 24h an ADMIN_EMAIL."""
import os
import html as _html
from collections import Counter
from mailer import send_email, email_html
admin = os.getenv("ADMIN_EMAIL", "")
if not admin:
logger.info("Error-Digest: ADMIN_EMAIL nicht gesetzt, uebersprungen.")
_log_job("error_digest", "ok", "ADMIN_EMAIL nicht gesetzt")
return
now = datetime.now(tz=_TZ)
cutoff = now - timedelta(hours=24)
# ── 1. Scheduler-Job-Errors einsammeln ───────────────────────
scheduler_errors = []
for jid, log in _job_log.items():
last_run = log.get("last_run")
if last_run and last_run >= cutoff and log.get("status") == "error":
scheduler_errors.append({
"job": jid,
"ts": last_run.strftime("%d.%m. %H:%M"),
"result": log.get("result", ""),
})
# ── 2. In-Memory-Log-Buffer einsammeln (best-effort) ─────────
log_errors = []
try:
from main import log_buffer # type: ignore
for entry in list(log_buffer):
lvl = entry.get("l", "")
if lvl in ("ERROR", "CRITICAL", "EXCEPTION"):
log_errors.append({
"ts": entry.get("t", ""),
"lvl": lvl,
"name": entry.get("n", ""),
"msg": entry.get("m", ""),
})
except Exception as e:
logger.debug(f"Error-Digest: log_buffer nicht verfuegbar: {e}")
# Gruppieren: Fehler-Meldungen mit gleichem msg-Prefix zusammenfassen
grouped: Counter = Counter()
for e in log_errors:
# Erste 80 Zeichen als Schluessel — schluckt Variablenwerte am Ende
key = (e["name"], e["msg"][:80])
grouped[key] += 1
# ── Wenn nichts zu melden ist: leise raus ────────────────────
if not scheduler_errors and not grouped:
logger.info("Error-Digest: Keine Errors in den letzten 24h.")
_log_job("error_digest", "ok", "0 Errors")
return
# ── HTML-Body bauen ──────────────────────────────────────────
parts = []
parts.append(f'Fehler-Zusammenfassung der letzten 24h ({now.strftime("%d.%m.%Y %H:%M")}).
')
if scheduler_errors:
rows_html = "".join(
f''
f'| {_html.escape(e["job"])} | '
f'{e["ts"]} | '
f'{_html.escape(e["result"])} | '
f'
'
for e in scheduler_errors
)
parts.append(
'Scheduler-Job-Fehler ({0})
'.format(len(scheduler_errors))
+ f''
)
if grouped:
sorted_groups = sorted(grouped.items(), key=lambda kv: -kv[1])
rows_html = "".join(
f''
f'| {count}x | '
f'{_html.escape(name)} | '
f'{_html.escape(msg)} | '
f'
'
for (name, msg), count in sorted_groups[:30]
)
parts.append(
'In-Memory-Log-Errors ({0} Typen, {1} Eintraege)
'.format(len(grouped), sum(grouped.values()))
+ f''
)
parts.append('Quellen: scheduler._job_log (24h) + main.log_buffer (Memory, max. 500 Eintraege). Fuer vollstaendige 24h-Historie waere eine persistente Log-Senke noetig.
')
body = "\n".join(parts)
html = email_html(body, footer_text="Ban Yaro · Error-Digest")
plain_lines = [f"Ban Yaro Error-Digest — {now.strftime('%d.%m.%Y %H:%M')}", ""]
if scheduler_errors:
plain_lines.append("Scheduler-Job-Fehler:")
for e in scheduler_errors:
plain_lines.append(f" - {e['ts']} {e['job']}: {e['result']}")
plain_lines.append("")
if grouped:
plain_lines.append("Log-Errors (Top 30 Typen, gruppiert):")
for (name, msg), count in sorted(grouped.items(), key=lambda kv: -kv[1])[:30]:
plain_lines.append(f" {count}x [{name}] {msg}")
plain = "\n".join(plain_lines)
try:
await send_email(
admin,
f"Ban Yaro Error-Digest ({len(scheduler_errors)} Scheduler + {len(grouped)} Log-Errors)",
html,
plain,
)
logger.info(f"Error-Digest gesendet an {admin} — {len(scheduler_errors)} scheduler, {len(grouped)} log-types.")
_log_job("error_digest", "ok", f"{len(scheduler_errors)} scheduler / {len(grouped)} log-types")
except Exception as e:
logger.error(f"Error-Digest: Mail-Fehler: {e}")
_log_job("error_digest", "error", str(e))
def _job_purge_jwt_blacklist():
"""Räumt abgelaufene Einträge aus jwt_blacklist auf — sonst wächst die
Tabelle monoton mit jedem Logout. Läuft täglich 03:30."""
try:
from auth import _purge_expired_jwt
deleted = _purge_expired_jwt()
logger.info(f"jwt_blacklist: {deleted} abgelaufene Einträge gelöscht.")
_log_job("purge_jwt_blacklist", "ok", f"{deleted} entries deleted")
except Exception as e:
logger.exception(f"jwt_blacklist purge fehlgeschlagen: {e}")
_log_job("purge_jwt_blacklist", "error", str(e))