"""Day-wise run log for the automatic arXiv collection.

Run *state* (running / completed, when it started and finished) lives in
`arxiv_auto_runs`. Email *counts* are always aggregated live from
`collected_data_preprints_v2` rather than cached on the run, so the report can
never drift away from what was actually stored.
"""

from datetime import date, datetime, timedelta, timezone
from typing import Annotated, Any, Optional

from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
from pymongo import DESCENDING

from ..db import get_db
from bson import ObjectId
from bson.errors import InvalidId

from ..models import (
    AutoRunFinish,
    AutoRunOut,
    AutoRunStart,
    CampaignDataOut,
    CampaignDeleteResult,
    CampaignOut,
    CampaignPreview,
    CampaignResult,
    CampaignSendResult,
    CollectedLogOut,
    DisciplineCount,
    FilterResult,
    PendingDates,
    PrepareSendResult,
    ProcessRequest,
    ProcessResult,
    SendReportRequest,
    SendReportResult,
    SmtpTestRequest,
    SmtpTestResult,
)
from ..campaigns import (
    prepare_send_jobs,
    render_campaign_email,
    report_send,
    send_campaign,
    send_campaign_test,
)
from ..processing import create_campaigns, process_day

router = APIRouter(prefix="/arxiv-auto", tags=["auto arxiv"])

RUNS = "arxiv_auto_runs"
COLLECTED = "collected_data_preprints_v2"

# Exact-email block lists: a collected contact whose address appears in one of
# these is filtered out and attributed to that reason. The order is the
# attribution priority -- an address on more than one list counts once, for the
# first reason here.
EMAIL_FILTERS = [
    ("unsub", "collected_filter_unsub"),
    ("read", "collected_filter_read"),
    ("bounced", "collected_filter_bounced"),
    ("automatic", "collected_filter_automatic"),
    ("abc", "collected_filter_abc"),
]
# A preprint is `collected` when the SAME (title, email) already exists in the
# main collected-contacts collection -- a real duplicate, not just a shared
# address. Matched on the pair, not email alone. collected_data holds the title
# under `article_title` (older rows: `title`).
COLLECTED_REASON = "collected"
COLLECTED_SOURCE = "collected_data"

# Substring list (str_remove): a collected address that CONTAINS an active
# keyword is filtered out, e.g. postmaster@ / noreply@ / MAILER-DAEMON@.
KEYWORD_REASON = "extensions"
KEYWORD_LIST = "email_filter_keywords"

# Every reason, in attribution priority order.
REASONS = [reason for reason, _ in EMAIL_FILTERS] + [COLLECTED_REASON, KEYWORD_REASON]

BACKFILL_DAYS = 15


def runs():
    return get_db()[RUNS]


def collected():
    return get_db()[COLLECTED]


def as_day(value: Any) -> Optional[datetime]:
    """Midnight UTC, so a day compares equal however it arrived."""
    if value is None:
        return None
    if isinstance(value, datetime):
        return datetime(value.year, value.month, value.day, tzinfo=timezone.utc)
    if isinstance(value, date):
        return datetime(value.year, value.month, value.day, tzinfo=timezone.utc)
    try:
        parsed = date.fromisoformat(str(value)[:10])
    except ValueError:
        return None
    return datetime(parsed.year, parsed.month, parsed.day, tzinfo=timezone.utc)


def today_utc() -> datetime:
    now = datetime.now(timezone.utc)
    return datetime(now.year, now.month, now.day, tzinfo=timezone.utc)


def _reason_group() -> dict:
    """A $group stage fragment that counts rows per reason plus the day total."""
    group: dict = {"emails": {"$sum": 1}}
    for reason in REASONS:
        group[reason] = {"$sum": {"$cond": [{"$eq": ["$filter_reason", reason]}, 1, 0]}}
    return group


def _with_remaining(doc: dict) -> dict:
    """emails and every reason -> the same, plus remaining = emails - sum(reasons)."""
    emails = doc.get("emails", 0)
    reasons = {r: doc.get(r, 0) for r in REASONS}
    return {"emails": emails, **reasons, "remaining": emails - sum(reasons.values())}


async def _active_keywords() -> list[str]:
    keywords: list[str] = []
    async for doc in get_db()[KEYWORD_LIST].find({}):
        kw = (doc.get("keyword") or "").strip().lower()
        active = str(doc.get("is_active", "")).strip().lower() in ("true", "1", "yes")
        if kw and active:
            keywords.append(kw)
    return keywords


async def counts_by_day() -> dict[datetime, dict]:
    """Per-day {emails, <each reason>, remaining}."""
    group = {"_id": "$published_on", **_reason_group()}
    rows = await (
        await collected().aggregate([
            {"$match": {"published_on": {"$type": "date"}}},
            {"$group": group},
        ])
    ).to_list(length=None)
    return {as_day(r["_id"]): _with_remaining(r) for r in rows}


async def apply_filters(cp=None, match: Optional[dict] = None) -> dict:
    """Attribute every collected contact to at most one filter reason and write
    it to `filter_reason`. Idempotent: a re-run also clears rows that no longer
    match any list, so the counts always reflect the lists as they stand now.

    First match wins in EMAIL_FILTERS order, then the substring keyword list.

    `cp` is the collection to filter -- defaults to the arXiv collection; the
    preprints router passes its own so the same block lists apply there too.
    `match` limits which rows are (re)filtered -- e.g. {"discipline": slug} so
    only the selected discipline is touched.
    """
    cp = cp if cp is not None else collected()
    match = match or {}
    scanned = await cp.count_documents(match)
    addresses = [a for a in await cp.distinct("email", match) if a]
    lower = {a: a.strip().lower() for a in addresses}

    # 1) Email block lists -> a reason per address (unsub .. abc). Each list is
    #    queried by our addresses only, so a huge source is never loaded whole.
    email_reason: dict[str, str] = {}  # address -> reason (first match wins)
    if addresses:
        for reason, coll in EMAIL_FILTERS:
            present: set[str] = set()
            async for doc in get_db()[coll].find({"email": {"$in": addresses}}, {"email": 1}):
                email = (doc.get("email") or "").strip().lower()
                if email:
                    present.add(email)
            for addr in addresses:
                if addr not in email_reason and lower[addr] in present:
                    email_reason[addr] = reason

    # 2) Keyword substring -> tentative `extensions` (lowest priority).
    keywords = await _active_keywords()
    ext_email = {
        addr for addr in addresses
        if addr not in email_reason and any(kw in lower[addr] for kw in keywords)
    } if keywords else set()

    # 3) `collected`: the SAME (title, email) already exists in collected_data.
    collected_pairs: set[tuple[str, str]] = set()
    if addresses:
        async for doc in get_db()[COLLECTED_SOURCE].find(
                {"email": {"$in": addresses}}, {"article_title": 1, "title": 1, "email": 1}):
            e = (doc.get("email") or "").strip().lower()
            if not e:
                continue
            for raw_title in (doc.get("article_title"), doc.get("title")):
                t = (raw_title or "").strip().lower()
                if t:
                    collected_pairs.add((t, e))

    # 4) Walk our documents and pick each one's reason. An email-list reason
    #    applies to every row for that address; `collected` and `extensions` are
    #    decided per row (they depend on the title / the individual address).
    ids_by_reason: dict[str, list] = {r: [] for r in REASONS}
    async for doc in cp.find(match, {"title": 1, "email": 1}):
        email = doc.get("email") or ""
        reason = email_reason.get(email)
        if reason is None:
            pair = ((doc.get("title") or "").strip().lower(), email.strip().lower())
            if pair in collected_pairs:
                reason = COLLECTED_REASON
            elif email in ext_email:
                reason = KEYWORD_REASON
        if reason:
            ids_by_reason[reason].append(doc["_id"])

    # 5) Clear (within the match), then set each reason in bulk (drop legacy flag).
    await cp.update_many(match, {"$unset": {"filter_reason": "", "unsubscribed": ""}})
    for reason, ids in ids_by_reason.items():
        for start in range(0, len(ids), 10000):
            chunk = ids[start:start + 10000]
            if chunk:
                await cp.update_many({"_id": {"$in": chunk}}, {"$set": {"filter_reason": reason}})

    rows_by_reason = {r: len(ids_by_reason[r]) for r in REASONS}
    filtered = sum(rows_by_reason.values())
    return {"scanned": scanned, "filtered": filtered,
            "remaining": scanned - filtered, "by_reason": rows_by_reason}


@router.get("/runs", response_model=list[AutoRunOut], summary="Day-wise log report")
async def list_runs(limit: Annotated[int, Query(ge=1, le=365)] = 60):
    # NOTE: filtering is NOT run here -- it reads several large block lists and
    # would make every day-view load slow. The stored `filter_reason` is just
    # aggregated (indexed, fast). Filtering runs when a collection finishes and
    # on demand via POST /filters.
    per_day = await counts_by_day()
    # Days already saved into collected_data_de / with campaigns (indexed, cheap).
    # Scoped to arXiv (preprints.org keeps its own flags in preprints_auto).
    NOT_PP = {"$nin": ["preprints.org", "biorxiv", "medrxiv"]}  # arXiv = none of the others
    saved_days = {as_day(d) for d in await get_db()["collected_data_de"].distinct(
        "published_on", {"source": {"$exists": False}})}
    campaigned_days = {as_day(d) for d in await get_db()["cd_campaigns"].distinct(
        "published_on", {"source": NOT_PP})}
    # A day is "completed" once every one of its campaigns has been sent
    # (none still missing sent_on).
    pending_campaign_days = {as_day(d) for d in await get_db()["cd_campaigns"].distinct(
        "published_on", {"sent_on": {"$exists": False}, "source": NOT_PP})}
    completed_campaign_days = campaigned_days - pending_campaign_days

    recorded = await runs().find().sort("date", DESCENDING).to_list(length=None)
    by_date: dict[datetime, dict] = {}
    for run in recorded:
        day = as_day(run.get("date"))
        if day is not None:
            by_date[day] = run

    report: list[AutoRunOut] = []
    for day in sorted(set(by_date) | set(per_day), reverse=True)[:limit]:
        run = by_date.get(day, {})
        stats = per_day.get(day, {"emails": 0})
        report.append(AutoRunOut(
            date=day,
            # A day with rows but no run record was collected by hand.
            status=run.get("status") or ("completed" if stats.get("emails") else "not started"),
            saved=day in saved_days,
            campaigned=day in campaigned_days,
            campaigns_completed=day in completed_campaign_days,
            disciplines=run.get("disciplines") or [],
            started_on=run.get("started_on"),
            collected_on=run.get("completed_on"),
            user_id=run.get("user_id"),
            error=run.get("error"),
            **{k: stats.get(k, 0) for k in ("emails", "remaining", *REASONS)},
        ))
    return report


@router.get("/runs/{day}/disciplines", response_model=list[DisciplineCount],
            summary="Discipline-wise summary for one day")
async def disciplines_for(day: str):
    target = as_day(day)
    if target is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="date must be YYYY-MM-DD")

    group = {"_id": "$discipline", **_reason_group()}
    rows = await (
        await collected().aggregate([
            {"$match": {"published_on": target}},
            {"$group": group},
            {"$sort": {"emails": -1}},
        ])
    ).to_list(length=None)
    return [
        DisciplineCount(discipline=r["_id"] or "—", **_with_remaining(r))
        for r in rows
    ]


@router.post("/filters", response_model=FilterResult,
             summary="Run every block-list filter over the collected preprint contacts")
async def run_filters():
    """Run all filters on demand. The day-wise log applies them automatically
    too, so this is kept only for scripted / manual runs."""
    result = await apply_filters()
    return FilterResult(**result)


@router.post("/process", response_model=ProcessResult,
             summary="Save one day's clean rows into collected_data_de")
async def process(payload: ProcessRequest):
    """Runs automatically when a collection finishes; also exposed for the
    desktop's Process button."""
    day = as_day(payload.date)
    if day is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="date must be YYYY-MM-DD")
    result = await process_day(payload.user_id, day)
    return ProcessResult(**result)


@router.post("/campaigns", response_model=CampaignResult,
             summary="Create publisher-wise campaigns for one saved day")
async def campaigns(payload: ProcessRequest):
    """For each topic in the day's collected_data_de, create one cd_campaigns per
    matching journals publisher, each carrying that topic's cd_campaign_data."""
    day = as_day(payload.date)
    if day is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="date must be YYYY-MM-DD")
    try:
        result = await create_campaigns(payload.user_id, day)
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001 -- surface the real error, not a bare 500
        import traceback
        traceback.print_exc()
        raise HTTPException(status_code=500,
                            detail=f"{type(exc).__name__}: {exc}") from exc
    return CampaignResult(**result)


@router.get("/campaigns", response_model=list[CampaignOut],
            summary="Publisher-wise campaigns for one day")
async def campaigns_for(date: str):
    day = as_day(date)
    if day is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="date must be YYYY-MM-DD")
    # arXiv view shows arXiv campaigns only -- preprints campaigns (source=
    # 'preprints.org') live in the Preprints screen.
    rows = await get_db()["cd_campaigns"].find(
        {"published_on": day, "source": {"$nin": ["preprints.org", "biorxiv", "medrxiv"]}}
    ).sort("total_emails", DESCENDING).to_list(length=None)
    return [CampaignOut(**r) for r in rows]


@router.delete("/campaigns", response_model=CampaignDeleteResult,
               summary="Delete all campaigns for a day and their records")
async def campaigns_delete_day(date: str, source: Optional[str] = None):
    """Removes every cd_campaigns row for the day and its cd_campaign_data. The
    collected_data_de source rows are left in place. An explicit `source`
    (preprints.org / biorxiv) scopes to that source's campaigns; otherwise arXiv
    (everything that is neither preprints nor biorxiv)."""
    day = as_day(date)
    if day is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="date must be YYYY-MM-DD")
    db = get_db()
    if source in ("preprints.org", "biorxiv", "medrxiv"):
        src = {"source": source}
    else:  # arXiv = anything that isn't one of the other sources
        src = {"source": {"$nin": ["preprints.org", "biorxiv", "medrxiv"]}}
    query = {"published_on": day, **src}
    ids = await db["cd_campaigns"].distinct("_id", query)
    data_res = await db["cd_campaign_data"].delete_many({"campaign_id": {"$in": ids}})
    camp_res = await db["cd_campaigns"].delete_many(query)
    return CampaignDeleteResult(
        deleted=camp_res.deleted_count > 0,
        campaigns_deleted=camp_res.deleted_count,
        records_deleted=data_res.deleted_count,
        detail=f"Removed {camp_res.deleted_count} campaign(s) and "
               f"{data_res.deleted_count} record(s) for {date}.")


@router.delete("/campaigns/{campaign_id}", response_model=CampaignDeleteResult,
               summary="Delete one campaign and its records (leaves collected data untouched)")
async def campaign_delete(campaign_id: str):
    """Removes the cd_campaigns row and its cd_campaign_data rows only. The
    collected_data_de source rows are left in place."""
    try:
        cid = ObjectId(campaign_id)
    except (InvalidId, TypeError):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="invalid campaign id")
    db = get_db()
    data_res = await db["cd_campaign_data"].delete_many({"campaign_id": cid})
    camp_res = await db["cd_campaigns"].delete_one({"_id": cid})
    if camp_res.deleted_count == 0:
        return CampaignDeleteResult(deleted=False, records_deleted=data_res.deleted_count,
                                    detail="Campaign not found.")
    return CampaignDeleteResult(
        deleted=True, records_deleted=data_res.deleted_count,
        detail=f"Removed campaign and {data_res.deleted_count} record(s).")


@router.get("/campaigns/{campaign_id}/data", response_model=list[CampaignDataOut],
            summary="A campaign's email data")
async def campaign_data(campaign_id: str):
    try:
        cid = ObjectId(campaign_id)
    except (InvalidId, TypeError):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="invalid campaign id")
    rows = await get_db()["cd_campaign_data"].find(
        {"campaign_id": cid}).sort("name", 1).to_list(length=None)
    return [CampaignDataOut(**r) for r in rows]


@router.get("/collected-log", response_model=list[CollectedLogOut],
            summary="Date-wise count of records saved into collected_data (IST)")
async def collected_log(limit: Annotated[int, Query(ge=1, le=365)] = 90):
    """How many records the campaign sender mirrored into collected_data, grouped
    by the IST calendar day they were saved (`created_on`). Only mirror-saved
    rows are counted (they carry `created_on`)."""
    pipeline = [
        {"$match": {"created_on": {"$ne": None}}},
        {"$group": {"_id": {"$dateToString": {
            "format": "%Y-%m-%d", "date": "$created_on", "timezone": "+05:30"}},
            "count": {"$sum": 1}}},
        {"$sort": {"_id": DESCENDING}},
        {"$limit": limit},
    ]
    rows = await (await get_db()["collected_data"].aggregate(pipeline)).to_list(length=None)
    return [CollectedLogOut(date=r["_id"], count=r["count"]) for r in rows]


@router.get("/campaigns/{campaign_id}/preview", response_model=CampaignPreview,
            summary="Render one recipient's campaign email (subject + HTML, no send)")
async def campaign_preview(campaign_id: str):
    try:
        cid = ObjectId(campaign_id)
    except (InvalidId, TypeError):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="invalid campaign id")
    try:
        result = await render_campaign_email(cid, None)
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    return CampaignPreview(**result)


@router.post("/campaigns/{campaign_id}/test", response_model=SmtpTestResult,
             summary="Send the rendered campaign template to one entered address")
async def campaign_test(campaign_id: str, payload: SmtpTestRequest):
    try:
        cid = ObjectId(campaign_id)
    except (InvalidId, TypeError):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid campaign id")
    try:
        result = await send_campaign_test(cid, payload.email.strip())
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    return SmtpTestResult(**result)


@router.post("/campaigns/{campaign_id}/send", response_model=CampaignSendResult,
             summary="Send the campaign to all its recipients (once, in the background)")
async def campaign_send(campaign_id: str, background: BackgroundTasks):
    try:
        cid = ObjectId(campaign_id)
    except (InvalidId, TypeError):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid campaign id")
    camps = get_db()["cd_campaigns"]
    # Atomically claim the campaign: only proceeds if it hasn't been sent or
    # started before, so it never runs twice.
    claim = await camps.update_one(
        {"_id": cid, "sent_on": {"$exists": False}, "send_started_on": {"$exists": False}},
        {"$set": {"send_started_on": datetime.now(timezone.utc)}},
    )
    if claim.matched_count == 0:
        camp = await camps.find_one({"_id": cid}) or {}
        when = str(camp.get("sent_on") or camp.get("send_started_on") or "")[:16]
        return CampaignSendResult(started=False, total=camp.get("total_emails", 0),
                                  detail=f"Already sent on {when} — not sending again." if when
                                  else "Already sent — not sending again.")

    total = await get_db()["cd_campaign_data"].count_documents({"campaign_id": cid})
    if total == 0:
        await camps.update_one({"_id": cid}, {"$unset": {"send_started_on": ""}})
        return CampaignSendResult(started=False, total=0, detail="No recipients for this campaign.")
    background.add_task(send_campaign, cid)
    return CampaignSendResult(started=True, total=total,
                              detail=f"Sending to {total} recipient(s) in the background.")


@router.post("/campaigns/{campaign_id}/prepare-send", response_model=PrepareSendResult,
             summary="Claim a campaign and return rendered emails + SMTP for the desktop to send")
async def campaign_prepare_send(campaign_id: str):
    """The API prepares only -- it renders each email and hands back the SMTP to
    use. The desktop exe performs the actual send, then calls report-send."""
    try:
        cid = ObjectId(campaign_id)
    except (InvalidId, TypeError):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid campaign id")
    result = await prepare_send_jobs(cid)
    return PrepareSendResult(**result)


@router.post("/campaigns/{campaign_id}/report-send", response_model=SendReportResult,
             summary="Record what the desktop sent (mark rows, mirror, counters)")
async def campaign_report_send(campaign_id: str, payload: SendReportRequest):
    try:
        cid = ObjectId(campaign_id)
    except (InvalidId, TypeError):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid campaign id")
    result = await report_send(cid, payload.results, final=payload.final)
    return SendReportResult(**result)


@router.post("/campaigns/{campaign_id}/reset-send", response_model=CampaignSendResult,
             summary="Clear a stuck 'Sending' claim so the campaign can be sent again")
async def campaign_reset_send(campaign_id: str):
    """Un-sticks a campaign left mid-send: clears send_started_on (and partial-run
    markers) WITHOUT touching sent_on or the per-record status, so a resend only
    covers recipients that were never sent."""
    try:
        cid = ObjectId(campaign_id)
    except (InvalidId, TypeError):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid campaign id")
    db = get_db()
    await db["cd_campaigns"].update_one(
        {"_id": cid}, {"$unset": {"send_started_on": "", "pending_count": "", "last_run_on": ""}})
    remaining = await db["cd_campaign_data"].count_documents(
        {"campaign_id": cid, "mail_send_status": {"$ne": 1}})
    return CampaignSendResult(started=True, total=remaining,
                              detail=f"Ready to send — {remaining} recipient(s) remaining.")


@router.get("/pending", response_model=PendingDates,
            summary="Which days still need collecting (never today)")
async def pending(days: Annotated[int, Query(ge=1, le=90)] = BACKFILL_DAYS):
    today = today_utc()
    yesterday = today - timedelta(days=1)

    done = {
        as_day(r["date"])
        for r in await runs().find({"status": "completed"}, {"date": 1}).to_list(length=None)
    }
    last = max((d for d in done if d is not None), default=None)

    # Yesterday first, then back through the window. Today is never included.
    wanted = [yesterday] + [yesterday - timedelta(days=n) for n in range(1, days + 1)]
    todo = [d for d in wanted if d not in done]

    return PendingDates(last_completed=last, yesterday=yesterday, pending=todo)


@router.post("/runs/start", response_model=AutoRunOut, summary="Mark a day as running")
async def start_run(payload: AutoRunStart):
    day = as_day(payload.date)
    if day is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="date must be YYYY-MM-DD")
    if day >= today_utc():
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="The current day is never collected automatically; arXiv's listing "
                   "for today is still filling up.",
        )

    now = datetime.now(timezone.utc)
    await runs().update_one(
        {"date": day},
        {"$set": {"status": "running", "started_on": now,
                  "disciplines": payload.disciplines, "user_id": payload.user_id,
                  "error": None, "completed_on": None}},
        upsert=True,
    )
    return AutoRunOut(date=day, status="running", started_on=now,
                      disciplines=payload.disciplines, user_id=payload.user_id)


@router.post("/runs/finish", response_model=AutoRunOut, summary="Mark a day finished")
async def finish_run(payload: AutoRunFinish):
    day = as_day(payload.date)
    if day is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="date must be YYYY-MM-DD")

    now = datetime.now(timezone.utc)
    await runs().update_one(
        {"date": day},
        {"$set": {"status": payload.status, "completed_on": now, "error": payload.error}},
        upsert=True,
    )
    # Reconcile the freshly collected day against the block lists now (once per
    # collection), so the day-view stays fast and never filters on read.
    await apply_filters()
    doc = await runs().find_one({"date": day}) or {}
    # Then save the clean rows into collected_data_de. Best-effort: a processing
    # error must never fail the run's finish.
    if doc.get("user_id") and payload.status == "completed":
        try:
            await process_day(doc["user_id"], day)
        except Exception:
            pass
    per_day = await counts_by_day()
    stats = per_day.get(day, {"emails": 0})
    return AutoRunOut(
        date=day, status=payload.status,
        disciplines=doc.get("disciplines") or [], started_on=doc.get("started_on"),
        collected_on=now, user_id=doc.get("user_id"), error=payload.error,
        **{k: stats.get(k, 0) for k in ("emails", "remaining", *REASONS)},
    )


@router.delete("/runs", summary="Clear the run log (all, or one day)")
async def clear_runs(day: Annotated[Optional[str], Query()] = None):
    query = {}
    if day:
        target = as_day(day)
        if target is None:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                                detail="date must be YYYY-MM-DD")
        query["date"] = target
    result = await runs().delete_many(query)
    return {"deleted": result.deleted_count}