banyaro/backend/scheduler.py

"""
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"""
        <p>Für <strong>{user['name']}</strong> ({user['email']}) wurde automatisch ein
           Rechnungsentwurf für die Abo-Verlängerung erstellt.</p>
        <table style="border-collapse:collapse;font-size:14px;margin:12px 0">
          <tr><td style="padding:4px 12px 4px 0;color:#888">Rechnung:</td><td><strong>{invoice_number}</strong></td></tr>
          <tr><td style="padding:4px 12px 4px 0;color:#888">Tarif:</td><td>{tier_label}</td></tr>
          <tr><td style="padding:4px 12px 4px 0;color:#888">Betrag:</td><td>{price:.2f} EUR</td></tr>
          <tr><td style="padding:4px 12px 4px 0;color:#888">Zeitraum:</td><td>{period}</td></tr>
          <tr><td style="padding:4px 12px 4px 0;color:#888">Abo läuft ab:</td><td>{expires.strftime('%d.%m.%Y')} (in 30 Tagen)</td></tr>
        </table>
        <p>Bitte prüfen, ggf. anpassen und rechtzeitig versenden.</p>"""
        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"""
    <p><strong>Achtung:</strong> Das Abo von <strong>{user['name']}</strong> ({user['email']})
       läuft in <strong>7 Tagen</strong> (am {expires.strftime('%d.%m.%Y')}) ab.</p>
    <p>Rechnungsentwurf <strong>{draft['invoice_number']}</strong> wurde noch nicht versendet.
       Bitte jetzt versenden damit der Kunde rechtzeitig bezahlen kann.</p>"""
    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"<p style='margin:0 0 8px;font-size:13px'>IBAN: <strong>{IBAN}</strong> · Verwendungszweck: {rg}</p>" if IBAN else ""
                body = f"""
                <p style='margin:0 0 12px'>Hallo <b>{_html.escape(name)}</b>,</p>
                <p style='margin:0 0 12px'>
                  unsere Rechnung <b>{rg}</b> vom {datetime.fromisoformat(inv['created_at'][:10]).strftime('%d.%m.%Y')}
                  über <b>{amount:.2f} EUR</b> ist leider noch offen.
                </p>
                <p style='margin:0 0 12px'>
                  Bitte überweisen Sie den Betrag bis zum <b>{frist}</b>.
                  {iban_line}
                </p>
                <p style='margin:0;font-size:13px;color:#888'>
                  Sollte die Zahlung bis zu diesem Datum nicht eingehen, sind wir leider gezwungen,
                  Ihr Abonnement fristlos zu kündigen (§&nbsp;314 BGB).
                </p>"""
                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"""
                    <p style='margin:0 0 12px'>Hallo <b>{_html.escape(name)}</b>,</p>
                    <p style='margin:0 0 12px'>
                      da die Zahlung für Rechnung <b>{rg}</b> ({amount:.2f} EUR)
                      trotz unserer Zahlungserinnerung nicht eingegangen ist,
                      haben wir Ihr Abonnement gemäß §&nbsp;314 BGB fristlos gekündigt.
                    </p>
                    <p style='margin:0 0 12px'>
                      Ihre Daten bleiben vollständig erhalten. Sie können jederzeit ein neues Abonnement abschließen.
                    </p>
                    <p style='margin:0;font-size:13px;color:#888'>
                      Bei Rückfragen antworten Sie einfach auf diese E-Mail.
                    </p>"""
                    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"<p>Abo von <b>{_html.escape(name)}</b> ({email}) wurde automatisch fristlos gekündigt (§314 BGB). Rechnung {rg} seit 35 Tagen offen.</p>"),
                            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"""
                <p>Hallo {_html.escape(u['name'])},</p>
                <p>dein <strong>{tier_label}</strong>-Abo ist heute abgelaufen.
                   Dein Account wurde auf den kostenlosen Tarif gesetzt.</p>
                <p>Deine Daten sind vollständig erhalten. Du kannst jederzeit wieder upgraden.</p>"""
                if needs_sel:
                    body += "<p><strong>Wichtig:</strong> Du hattest mehrere Hunde. Öffne die App und wähle deinen Haupthund aus — alle anderen Profile bleiben gespeichert.</p>"
                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"""
                <p>Hallo {_html.escape(u['name'])},</p>
                <p>dein <strong>{tier_label}</strong>-Abo läuft in <strong>30 Tagen</strong>
                   (am {expires.strftime('%d.%m.%Y')}) ab.</p>
                <p>Um weiterzumachen, überweise einfach den Jahresbetrag und schreib uns kurz —
                   wir verlängern deinen Zugang sofort.</p>"""
                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"""
                <p>Hallo {_html.escape(u['name'])},</p>
                <p>dein <strong>{tier_label}</strong>-Abo läuft in <strong>7 Tagen</strong>
                   (am {expires.strftime('%d.%m.%Y')}) ab.</p>
                <p>Jetzt verlängern und nahtlos weitermachen!</p>"""
                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()
    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)

    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}
Dabei seit: {stats.get('weeks_total', 1)} Wochen

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)
- 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
        if aktivitaet_parts:
            return f"{name} hatte eine aktive Woche \u2014 {aktivitaet_text}. Das ist toll! \U0001f43e"
        else:
            return f"Auch ruhige Wochen geh\u00f6ren dazu. {name} wei\u00df, dass du f\u00fcr ihn da bist. \U0001f43e"


# ------------------------------------------------------------------
# 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'<tr><td style="padding:6px 12px;font-weight:600;color:#c45000">{label}</td>'
        f'<td style="padding:6px 12px;font-size:18px;font-weight:800;color:#c45000">{count}</td></tr>'
        for label, count in overdue.items()
    )
    html = f"""\
<!DOCTYPE html><html lang="de"><head><meta charset="utf-8"></head>
<body style="font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;background:#f5f0ea;margin:0;padding:0">
<div style="max-width:560px;margin:24px auto;background:#fff;border-radius:14px;overflow:hidden;box-shadow:0 2px 16px rgba(0,0,0,.08)">
  <div style="background:linear-gradient(135deg,#c45000,#e8733a);padding:22px 28px;color:#fff">
    <div style="font-size:20px;font-weight:800;margin-bottom:2px">⚠️ Moderation überfällig</div>
    <div style="opacity:.88;font-size:13px">{now_str} · SLA: {SLA_H}h</div>
  </div>
  <div style="padding:22px 28px">
    <p style="color:#444;margin:0 0 16px">Folgende Einträge warten seit mehr als {SLA_H} Stunden auf Bearbeitung:</p>
    <table style="width:100%;border-collapse:collapse;font-size:14px">
      <thead><tr style="border-bottom:2px solid #f0e8dc">
        <th style="text-align:left;padding:6px 12px;color:#888;font-size:11px;text-transform:uppercase;letter-spacing:.06em">Bereich</th>
        <th style="text-align:left;padding:6px 12px;color:#888;font-size:11px;text-transform:uppercase;letter-spacing:.06em">Anzahl</th>
      </tr></thead>
      <tbody>{rows_html}</tbody>
    </table>
    <div style="margin-top:20px">
      <a href="https://banyaro.app/app/admin" style="display:inline-block;background:#c45000;color:#fff;
         text-decoration:none;padding:10px 22px;border-radius:8px;font-weight:700;font-size:14px">
        → Admin-Panel öffnen
      </a>
    </div>
  </div>
  <div style="padding:12px 28px;background:#fdf6ef;font-size:11px;color:#aaa;text-align:center">
    Ban Yaro · banyaro.app
  </div>
</div></body></html>"""

    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 = '<span style="color:#16a34a;font-weight:700">✅ Alles erledigt — nichts offen</span>'
    else:
        pills = "".join(
            f'<span style="display:inline-block;background:#fff3e0;color:#c45000;border:1px solid #e8a857;'
            f'border-radius:999px;padding:3px 12px;font-size:12px;font-weight:700;margin:2px 4px 2px 0">'
            f'{label} <strong style="background:#c45000;color:#fff;border-radius:999px;'
            f'padding:0 7px;margin-left:4px">{count}</strong></span>'
            for label, count in open_items
        )
        body = f'<div style="font-size:13px;font-weight:600;color:#c45000;margin-bottom:8px">⚠️ {len(open_items)} Punkt{"e" if len(open_items)!=1 else ""} brauchen deine Aufmerksamkeit</div>{pills}'

    link = '<div style="margin-top:10px"><a href="https://banyaro.app/app/admin" style="font-size:12px;color:#C4843A">→ Admin-Panel öffnen</a></div>'
    return f'<div style="padding:20px 28px;border-bottom:2px solid #e8a857;background:#fffbf5">' \
           f'<div style="font-size:11px;font-weight:700;text-transform:uppercase;letter-spacing:.06em;color:#C4843A;margin-bottom:10px">Heute zu erledigen</div>' \
           f'{body}{link}</div>'


# JOB: Status-Report per Mail (täglich 06:00 Uhr)
# ------------------------------------------------------------------
async def _job_status_report():
    """Sendet einen HTML-Status-Report an ADMIN_EMAIL."""
    import os
    from mailer import send_email

    admin = os.getenv("ADMIN_EMAIL", "")
    if not admin:
        logger.info("Status-Report: ADMIN_EMAIL nicht gesetzt, übersprungen.")
        return

    now_str = datetime.now(tz=_TZ).strftime("%d.%m.%Y %H:%M")

    # --- DB-Metriken abrufen ---
    metrics = {}
    try:
        with db() as conn:
            # Züchter
            try:
                metrics["zuchter_pending"]  = conn.execute("SELECT COUNT(*) FROM wiki_zuchter WHERE verified=0").fetchone()[0]
                metrics["zuchter_verified"] = conn.execute("SELECT COUNT(*) FROM wiki_zuchter WHERE verified=1").fetchone()[0]
            except Exception:
                metrics["zuchter_pending"] = metrics["zuchter_verified"] = 0

            # Community
            metrics["users"]          = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
            metrics["users_today"]    = conn.execute("SELECT COUNT(*) FROM users WHERE DATE(created_at)=DATE('now')").fetchone()[0]
            metrics["dogs"]           = conn.execute("SELECT COUNT(*) FROM dogs").fetchone()[0]
            metrics["diary_entries"]  = conn.execute("SELECT COUNT(*) FROM diary").fetchone()[0]
            metrics["poison_active"]  = conn.execute("SELECT COUNT(*) FROM poison WHERE geloest=0").fetchone()[0]
            try:
                metrics["lost_active"] = conn.execute("SELECT COUNT(*) FROM lost WHERE gefunden=0").fetchone()[0]
            except Exception:
                metrics["lost_active"] = 0

            # Action Items
            try:
                metrics["jobs_pending"]      = conn.execute("SELECT COUNT(*) FROM job_applications WHERE status IN ('pending','reviewing')").fetchone()[0]
            except Exception:
                metrics["jobs_pending"] = 0
            try:
                metrics["breeder_pending"]   = conn.execute("SELECT COUNT(*) FROM users WHERE breeder_status='pending'").fetchone()[0]
            except Exception:
                metrics["breeder_pending"] = 0
            try:
                metrics["reports_open"]      = conn.execute("SELECT COUNT(*) FROM forum_reports WHERE resolved=0").fetchone()[0]
            except Exception:
                metrics["reports_open"] = 0
            try:
                metrics["fotos_pending"]     = conn.execute("SELECT COUNT(*) FROM wiki_foto_submissions WHERE status='pending'").fetchone()[0]
            except Exception:
                metrics["fotos_pending"] = 0
            try:
                metrics["poi_edits_pending"] = conn.execute("SELECT COUNT(*) FROM osm_poi_edits WHERE status='pending'").fetchone()[0]
            except Exception:
                metrics["poi_edits_pending"] = 0

            # Wiki-Interesse
            try:
                metrics["interesse_hat"]  = conn.execute("SELECT COUNT(*) FROM wiki_breed_interest WHERE typ='hat'").fetchone()[0]
                metrics["interesse_will"] = conn.execute("SELECT COUNT(*) FROM wiki_breed_interest WHERE typ='will'").fetchone()[0]
            except Exception:
                metrics["interesse_hat"] = metrics["interesse_will"] = 0
    except Exception as e:
        logger.error(f"Status-Report: DB-Fehler: {e}")
        return

    # --- Job-Log-Tabelle ---
    job_labels = {
        "health_reminders":     "Gesundheits-Erinnerungen",
        "poison_archive":       "Giftköder-Archiv",
        "weather_alert":        "Wetter-Alert",
        "milestone_check":      "Meilenstein-Check",
        "import_events":        "Event-Import (VDH)",
        "seed_breeds":         "Rassen-Seed (TheDogAPI, monatlich)",
        "seed_wikidata":       "Rassen-Seed (Wikidata, monatlich)",
        "weekly_praise":        "Wöchentlicher Lober (Mo 09:00)",
        "ki_health_report":     "KI-Gesundheitsberichte",
        "quarterly_report":     "Quartalsbericht (1. Feb/Mai/Aug/Nov)",
        "streak_reminder":      "Streak-Erinnerung (täglich 19:00)",
        "recall_check":         "Tierfutter-Rückrufe (RASFF, täglich 08:00)",
        "golden_gassi_hour":    "Goldene Gassi-Stunde (täglich 07:00)",
        "anniversary_reminders": "Jahrestags-Erinnerungen (täglich 09:00)",
        "monthly_recap":        "Monatlicher Rückblick (1. des Monats 10:00)",
    }
    job_rows_html = ""
    job_rows_txt  = ""
    for jid, label in job_labels.items():
        log = _job_log.get(jid)
        if log:
            ts      = log["last_run"].strftime("%d.%m. %H:%M")
            status  = "✅" if log["status"] == "ok" else "❌"
            result  = log["result"]
            color   = "#16a34a" if log["status"] == "ok" else "#dc2626"
            job_rows_html += f'<tr><td style="padding:5px 10px;color:#555">{label}</td><td style="padding:5px 10px;font-family:monospace;font-size:12px">{ts}</td><td style="padding:5px 10px;color:{color}">{status} {result}</td></tr>'
            job_rows_txt  += f"  {status} {label}: {ts} — {result}\n"
        else:
            job_rows_html += f'<tr><td style="padding:5px 10px;color:#555">{label}</td><td style="padding:5px 10px;color:#aaa" colspan="2">— noch nicht gelaufen</td></tr>'
            job_rows_txt  += f"  — {label}: noch nicht gelaufen\n"

    html = f"""\
<!DOCTYPE html>
<html lang="de">
<head><meta charset="utf-8"></head>
<body style="font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;background:#f5f0ea;margin:0;padding:0">
<div style="max-width:600px;margin:24px auto;background:#fff;border-radius:14px;overflow:hidden;box-shadow:0 2px 16px rgba(0,0,0,.08)">

  <!-- Header -->
  <div style="background:linear-gradient(135deg,#C4843A,#e8a857);padding:24px 28px;color:#fff">
    <div style="font-size:22px;font-weight:800;margin-bottom:2px">🐾 Ban Yaro — Status-Report</div>
    <div style="opacity:.88;font-size:13px">{now_str} Uhr</div>
  </div>

  <!-- Action Items -->
  {_action_items_html(metrics)}

  <!-- Scheduler-Status -->
  <div style="padding:20px 28px;border-bottom:1px solid #f0e8dc">
    <div style="font-size:11px;font-weight:700;text-transform:uppercase;letter-spacing:.06em;color:#C4843A;margin-bottom:10px">Scheduler-Jobs</div>
    <table style="width:100%;border-collapse:collapse;font-size:13px">
      {job_rows_html}
    </table>
  </div>

  <!-- Community-Metriken -->
  <div style="padding:20px 28px;border-bottom:1px solid #f0e8dc">
    <div style="font-size:11px;font-weight:700;text-transform:uppercase;letter-spacing:.06em;color:#C4843A;margin-bottom:10px">Community</div>
    <div style="display:grid;grid-template-columns:1fr 1fr;gap:8px">
      {"".join(f'<div style="background:#fdf6ef;border-radius:8px;padding:10px 14px"><div style="font-size:20px;font-weight:800;color:#C4843A">{v}</div><div style="font-size:11px;color:#888">{k}</div></div>' 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"]),
      ])}
    </div>
  </div>

  <!-- Footer -->
  <div style="padding:14px 28px;background:#fdf6ef;font-size:11px;color:#aaa;text-align:center">
    Ban Yaro · banyaro.app · Nächster Report in ~6h
  </div>

</div>
</body>
</html>"""

    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("</code></pre>")
                        in_code = False
                    else:
                        lines_out.append('<pre style="background:#f5f0ea;padding:10px;border-radius:6px;font-size:12px;overflow-x:auto"><code>')
                        in_code = True
                    continue
                if in_code:
                    lines_out.append(_h.escape(line))
                    continue
                if line.startswith("#### "):
                    lines_out.append(f'<h4 style="margin:12px 0 4px;color:#333">{line[5:]}</h4>')
                elif line.startswith("### "):
                    lines_out.append(f'<h3 style="margin:16px 0 6px;color:#555;font-size:14px;text-transform:uppercase;letter-spacing:.04em">{line[4:]}</h3>')
                elif line.startswith("## "):
                    lines_out.append(f'<h2 style="margin:20px 0 8px;color:#C4843A;font-size:16px;border-bottom:1px solid #f0e8dc;padding-bottom:4px">{line[3:]}</h2>')
                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('<table style="width:100%;border-collapse:collapse;font-size:13px;margin:8px 0">')
                        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'<td style="padding:4px 8px;border-bottom:1px solid #f0e8dc">{_h.escape(c)}</td>' for c in cells)
                    lines_out.append(f"<tr>{row_html}</tr>")
                    continue
                elif line.startswith("- ") or line.startswith("* "):
                    if in_table:
                        lines_out.append("</table>")
                        in_table = False
                    lines_out.append(f'<li style="margin:2px 0;color:#444">{line[2:]}</li>')
                elif line.startswith("> "):
                    if in_table:
                        lines_out.append("</table>")
                        in_table = False
                    lines_out.append(f'<blockquote style="border-left:3px solid #C4843A;margin:8px 0;padding:6px 12px;background:#fdf6ef;color:#555;font-size:13px">{line[2:]}</blockquote>')
                elif line.strip() == "":
                    if in_table:
                        lines_out.append("</table>")
                        in_table = False
                    lines_out.append("")
                else:
                    if in_table:
                        lines_out.append("</table>")
                        in_table = False
                    styled = line.replace("**", "<b>", 1).replace("**", "</b>", 1)
                    lines_out.append(f'<p style="margin:4px 0;color:#444;font-size:14px">{styled}</p>')
            if in_table:
                lines_out.append("</table>")
            if in_code:
                lines_out.append("</code></pre>")
            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'<div style="margin-bottom:32px">'
                    f'<h1 style="font-size:18px;font-weight:800;color:#C4843A;margin:0 0 12px;'
                    f'border-bottom:2px solid #f0e8dc;padding-bottom:6px">{title}</h1>'
                    f'{md_to_html_simple(md)}'
                    f'</div>'
                )
                plain_parts.append(f"\n=== {title.upper()} ===\n{md}\n")
            except Exception as e:
                body_parts.append(f'<p style="color:#dc2626">Fehler in Section {title}: {e}</p>')
                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):
      "🎂 <Name> ist X Jahr(e) alt!"
    - Monatsjubiläum in den ersten 11 Monaten (Geburtsmonats-Tag):
      "🐾 <Name> 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'<p style="margin:0 0 16px;color:#444">Fehler-Zusammenfassung der letzten 24h ({now.strftime("%d.%m.%Y %H:%M")}).</p>')

    if scheduler_errors:
        rows_html = "".join(
            f'<tr>'
            f'<td style="padding:6px 12px;color:#c45000;font-weight:600">{_html.escape(e["job"])}</td>'
            f'<td style="padding:6px 12px;font-family:monospace;font-size:12px">{e["ts"]}</td>'
            f'<td style="padding:6px 12px;color:#555">{_html.escape(e["result"])}</td>'
            f'</tr>'
            for e in scheduler_errors
        )
        parts.append(
            '<h2 style="font-size:14px;color:#c45000;margin:20px 0 6px">Scheduler-Job-Fehler ({0})</h2>'.format(len(scheduler_errors))
            + f'<table style="width:100%;border-collapse:collapse;font-size:13px"><tbody>{rows_html}</tbody></table>'
        )

    if grouped:
        sorted_groups = sorted(grouped.items(), key=lambda kv: -kv[1])
        rows_html = "".join(
            f'<tr>'
            f'<td style="padding:6px 12px;text-align:right;font-weight:700;color:#dc2626">{count}x</td>'
            f'<td style="padding:6px 12px;font-family:monospace;font-size:11px;color:#888">{_html.escape(name)}</td>'
            f'<td style="padding:6px 12px;color:#444">{_html.escape(msg)}</td>'
            f'</tr>'
            for (name, msg), count in sorted_groups[:30]
        )
        parts.append(
            '<h2 style="font-size:14px;color:#c45000;margin:20px 0 6px">In-Memory-Log-Errors ({0} Typen, {1} Eintraege)</h2>'.format(len(grouped), sum(grouped.values()))
            + f'<table style="width:100%;border-collapse:collapse;font-size:12px"><tbody>{rows_html}</tbody></table>'
        )

    parts.append('<p style="margin:24px 0 0;font-size:11px;color:#aaa">Quellen: scheduler._job_log (24h) + main.log_buffer (Memory, max. 500 Eintraege). Fuer vollstaendige 24h-Historie waere eine persistente Log-Senke noetig.</p>')

    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))