"""Day-wise collection log for preprints.org (Selenium-collected). Mirrors arxiv_auto but is scoped to its OWN collection, `collected_data_preprints_org` (never touches `collected_data_preprints_v2`). Run state lives in `preprints_auto_runs`; email counts are aggregated live from the collection. """ from datetime import datetime, timedelta, timezone from typing import Annotated, Any, Optional from fastapi import APIRouter, HTTPException, Query, status from pydantic import BaseModel from pymongo import ASCENDING, DESCENDING, UpdateOne from ..db import get_db from ..models import ( AutoRunFinish, AutoRunOut, AutoRunStart, CampaignOut, CampaignResult, CollectedBatch, CollectedPreprintOut, CollectedResult, DisciplineCount, Page, PendingDates, ProcessRequest, ProcessResult, ) from ..processing import ( DE_DATA_TYPE, _as_object_id, _is_others_topic, _others_journals, create_campaigns, process_day, ) from .arxiv_auto import ( BACKFILL_DAYS, REASONS, _reason_group, _with_remaining, apply_filters, as_day, today_utc, ) from .collected import as_bson_date, valid_email router = APIRouter(prefix="/preprints-auto", tags=["preprints auto"]) COLLECTED = "collected_data_preprints_org" RUNS = "preprints_auto_runs" class UserRequest(BaseModel): user_id: str class DisciplineRequest(BaseModel): user_id: str discipline: str def collected(): return get_db()[COLLECTED] def runs(): return get_db()[RUNS] async def counts_by_day() -> dict[datetime, dict]: """published_on -> {emails, , remaining}. Reasons come from the stored `filter_reason` (set by apply_filters); remaining = emails - filtered.""" group = {"_id": "$published_on", **_reason_group()} rows = await (await collected().aggregate([ {"$match": {"published_on": {"$type": "date"}}}, {"$group": group}, ])).to_list(length=None) # Normalise the aggregation _id (naive BSON date) through as_day so it matches # the tz-aware days from run records -- otherwise sorted() mixes naive+aware. return {as_day(r["_id"]): _with_remaining(r) for r in rows} # ------------------------------------------------------------ save harvested @router.post("/collected", response_model=CollectedResult, summary="Save a batch of preprints.org contacts") async def save_batch(payload: CollectedBatch): if not payload.rows: return CollectedResult(received=0, inserted=0, skipped=0) now = datetime.now(timezone.utc) ops = [] seen: set[tuple[str, str]] = set() for row in payload.rows: doc = row.model_dump(exclude_none=True) email = (doc.get("email") or "").strip().lower() title = (doc.get("title") or "").strip() if not title or not valid_email(email): continue doc["email"], doc["title"] = email, title published = as_bson_date(doc.get("published_on")) if published is not None: doc["published_on"] = published else: doc.pop("published_on", None) if (title, email) in seen: continue seen.add((title, email)) doc["collected_on"] = now ops.append(UpdateOne({"title": title, "email": email}, {"$setOnInsert": doc}, upsert=True)) if not ops: return CollectedResult(received=len(payload.rows), inserted=0, skipped=len(payload.rows)) result = await collected().bulk_write(ops, ordered=False) inserted = result.upserted_count return CollectedResult(received=len(payload.rows), inserted=inserted, skipped=len(payload.rows) - inserted, updated=0) # ------------------------------------------------------------ day-wise log @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): per_day = await counts_by_day() saved_days = {as_day(d) for d in await get_db()["collected_data_de"].distinct( "published_on", {"source": "preprints.org"})} campaigned_days = {as_day(d) for d in await get_db()["cd_campaigns"].distinct( "published_on", {"source": "preprints.org"})} 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, "remaining": 0}) report.append(AutoRunOut( date=day, status=run.get("status") or ("completed" if stats.get("emails") else "not started"), saved=day in saved_days, campaigned=day in campaigned_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("/collected-ids", summary="Manuscript ids already saved for a discipline") async def collected_ids(discipline: str): """The set of preprints.org manuscript ids already stored for this discipline, so a re-run (Play) can skip them and resume where it left off.""" ids = await collected().distinct("arxiv_id", {"discipline": discipline}) return {"ids": [i for i in ids if i]} @router.get("/disciplines-summary", response_model=list[DisciplineCount], summary="Per-discipline counts of the CURRENT (non-archived) batch") async def disciplines_summary(): # Working set only: rows not yet archived into a campaign batch. group = {"_id": "$discipline", **_reason_group()} rows = await (await collected().aggregate([ {"$match": {"campaigned_on": {"$exists": False}}}, {"$group": group}, {"$sort": {"emails": -1}}, ])).to_list(length=None) by_disc = {r["_id"]: r for r in rows} # Disciplines that have campaigns -> so 'Created' still shows at 0 working rows. camp_counts: dict = {} for c in await (await get_db()["cd_campaigns"].aggregate([ {"$match": {"source": SOURCE_TAG}}, {"$group": {"_id": "$discipline", "n": {"$sum": 1}}}, ])).to_list(length=None): camp_counts[c["_id"]] = c["n"] out: list[DisciplineCount] = [] for disc in set(by_disc) | set(camp_counts): r = by_disc.get(disc, {}) # New = distinct surviving (unfiltered), non-archived emails for the batch. clean = {"discipline": disc, "campaigned_on": {"$exists": False}, "$or": [{"filter_reason": {"$exists": False}}, {"filter_reason": None}]} new = len([e for e in await collected().distinct("email", clean) if e]) out.append(DisciplineCount(discipline=disc or "—", new=new, campaigned=camp_counts.get(disc, 0), **_with_remaining(r))) return out @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") pipeline = [ {"$match": {"published_on": target}}, {"$group": {"_id": "$discipline", "emails": {"$sum": 1}}}, {"$sort": {"emails": -1}}, ] rows = await (await collected().aggregate(pipeline)).to_list(length=None) return [DisciplineCount(discipline=r["_id"] or "—", emails=r["emails"], remaining=r["emails"]) for r in rows] @router.get("/collected", response_model=Page[CollectedPreprintOut], summary="Records for a day (optionally one discipline)") async def list_collected( published_on: Annotated[Optional[str], Query()] = None, discipline: Annotated[Optional[str], Query()] = None, limit: Annotated[int, Query(ge=1, le=500)] = 200, sort_dir: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc", ): query: dict[str, Any] = {} if published_on: day = as_bson_date(published_on) if day is None: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="published_on must be YYYY-MM-DD") query["published_on"] = day if discipline: query["discipline"] = discipline col = collected() total = await col.count_documents(query) cursor = col.find(query).sort("collected_on", ASCENDING if sort_dir == "asc" else DESCENDING).limit(limit) return Page[CollectedPreprintOut](total=total, skip=0, limit=limit, items=await cursor.to_list(length=limit)) # ------------------------------------------------------------ run tracking @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) 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") 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) stats = (await counts_by_day()).get(day, {"emails": 0, "remaining": 0}) doc = await runs().find_one({"date": day}) or {} return AutoRunOut(date=day, status=payload.status, started_on=doc.get("started_on"), collected_on=now, disciplines=doc.get("disciplines") or [], user_id=doc.get("user_id"), error=payload.error, emails=stats.get("emails", 0), remaining=stats.get("remaining", 0)) # ------------------------------------------------- process + campaigns SOURCE_COLL = "collected_data_preprints_org" SOURCE_TAG = "preprints.org" @router.post("/process", response_model=ProcessResult, summary="Save one day's rows into collected_data_de (preprints source)") async def process(payload: ProcessRequest): 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, SOURCE_COLL, SOURCE_TAG) return ProcessResult(**result) @router.post("/campaigns", response_model=CampaignResult, summary="Create publisher-wise campaigns for a preprints day") async def create_campaigns_route(payload: ProcessRequest): 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 create_campaigns(payload.user_id, day, SOURCE_TAG) return CampaignResult(**result) async def _preprint_days(match: Optional[dict] = None) -> list[datetime]: """Distinct published_on dates in the preprints collection, optionally limited to a match (e.g. one discipline).""" query = {"published_on": {"$type": "date"}, **(match or {})} raw = await collected().distinct("published_on", query) days = sorted({as_day(d) for d in raw if d is not None}, reverse=True) return [d for d in days if d is not None] @router.post("/filters", summary="Run every block-list filter over the preprints contacts") async def run_filters(discipline: Optional[str] = None): """Mark each preprints.org contact with its filter_reason (unsub/read/… ), so the reason columns populate and Process only saves the survivors. Pass a discipline to filter just that subject.""" match = {"discipline": discipline} if discipline else None return await apply_filters(collected(), match) @router.post("/process-discipline", response_model=ProcessResult, summary="Filter + process ONE discipline's rows into collected_data_de") async def process_discipline(payload: DisciplineRequest): # Filter just this discipline's CURRENT (non-archived) rows, then process them. batch = {"discipline": payload.discipline, "campaigned_on": {"$exists": False}} await apply_filters(collected(), batch) scanned = inserted = unmatched = 0 for day in await _preprint_days(batch): res = await process_day(payload.user_id, day, SOURCE_COLL, SOURCE_TAG, discipline=payload.discipline) scanned += res.get("scanned", 0) inserted += res.get("inserted", 0) unmatched += res.get("unmatched_topic", 0) return ProcessResult(scanned=scanned, inserted=inserted, unmatched_topic=unmatched, already_saved=False) @router.post("/campaigns-discipline", response_model=CampaignResult, summary="Filter+process the discipline's current batch into a campaign dated today, " "then archive it so the discipline resets to 0") async def create_campaigns_discipline(payload: DisciplineRequest): try: return await _build_discipline_campaigns(payload) 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 async def _build_discipline_campaigns(payload: DisciplineRequest) -> CampaignResult: """Batch model: take the discipline's CURRENT (non-archived) collected emails, filter + process them, build one campaign per publisher dated TODAY, then mark that batch archived (`campaigned_on`). The discipline row then reports 0 until Play collects a fresh batch. """ db = get_db() cp = collected() de = db["collected_data_de"] journals = db["journals"] camps = db["cd_campaigns"] camp_data = db["cd_campaign_data"] collector = _as_object_id(payload.user_id) now = datetime.now(timezone.utc) # Full timestamp (not midnight): the cd_campaigns unique index is # {user, publisher, primary_topic_id, published_on} and does NOT include the # discipline -- so disciplines that share a topic (e.g. several -> "Others") # would collide on a shared day-stamp. A per-batch timestamp keeps each # Create Campaign its own campaign. It still displays as today's date. today = now discipline = payload.discipline batch = {"discipline": discipline, "campaigned_on": {"$exists": False}} # The current batch's emails (before archiving). batch_emails = [e for e in await cp.distinct("email", batch) if e] if not batch_emails: return CampaignResult(topics=0, campaigns=0, records=0, detail="No new emails to campaign for this discipline.") # Filter + process just this batch into collected_data_de. await apply_filters(cp, batch) scanned = inserted = unmatched = 0 for day in await _preprint_days(batch): res = await process_day(payload.user_id, day, SOURCE_COLL, SOURCE_TAG, discipline=discipline) scanned += res.get("scanned", 0) inserted += res.get("inserted", 0) unmatched += res.get("unmatched_topic", 0) # This batch's de rows (scoped by discipline AND the batch's emails). de_rows = await de.find( {"discipline": discipline, "source": SOURCE_TAG, "email": {"$in": batch_emails}}).to_list(length=None) # Only NEW emails: drop any already SENT in a prior preprints campaign # (mail_send_status == 1) so nobody is emailed twice. prior_ids = await camps.distinct("_id", {"source": SOURCE_TAG}) sent_emails: set = set() if prior_ids: sent_emails = {e.strip().lower() for e in await camp_data.distinct( "email", {"campaign_id": {"$in": prior_ids}, "mail_send_status": 1}) if e} n_before = len(de_rows) if sent_emails: de_rows = [r for r in de_rows if (r.get("email") or "").strip().lower() not in sent_emails] already_sent = n_before - len(de_rows) by_topic: dict = {} for rec in de_rows: by_topic.setdefault(rec.get("primary_topic_id"), []).append(rec) topics = campaigns = records_copied = 0 no_publisher_topics: list[str] = [] for topic_id, records in by_topic.items(): if topic_id is None or not records: continue topic_name = records[0].get("primary_topic") if _is_others_topic(topic_name): publishers = await _others_journals(db, topic_id) else: publishers = await journals.find({"topic_id": topic_id}).to_list(length=None) if not publishers: no_publisher_topics.append(str(topic_name or topic_id)) continue topics += 1 for journal in publishers: publisher_id = journal.get("publisher_id") if publisher_id is None: continue # One campaign per (user, publisher, topic, discipline, source, TODAY) -- # today's date keys each batch as its own campaign. key = {"user_id": collector, "publisher_id": publisher_id, "primary_topic_id": topic_id, "discipline": discipline, "source": SOURCE_TAG, "published_on": today} # Key fields (source/discipline/published_on/topic id/publisher) come # from `key` on insert -- keep them OUT of $set to avoid an upsert # path-conflict. res = await camps.update_one( key, {"$set": {"publisher_name": journal.get("publisher_name"), "journal_id": journal.get("_id"), "primary_topic": topic_name, "topic": topic_name, "data_type": DE_DATA_TYPE, "total_emails": len(records), "status": 0, "mail_send_status": 0}, "$setOnInsert": {"created_on": now}}, upsert=True) if res.upserted_id is not None: campaign_id = res.upserted_id campaigns += 1 else: existing = await camps.find_one(key, {"_id": 1}) campaign_id = existing["_id"] ops = [] for rec in records: data_key = {"campaign_id": campaign_id, "de_id": rec.get("_id")} ops.append(UpdateOne( data_key, {"$setOnInsert": { **data_key, "user_id": collector, "publisher_id": publisher_id, "primary_topic_id": topic_id, "primary_topic": topic_name, "topic": rec.get("topic"), "published_on": today, "data_type": DE_DATA_TYPE, "title": rec.get("title"), "name": rec.get("name"), "email": rec.get("email"), "country_name": rec.get("country_name"), "country_id": rec.get("country_id"), "country_level": rec.get("country_level"), "affiliation": rec.get("affiliation"), "section_id": rec.get("section_id"), "discipline": discipline, "mail_send_status": 0, "subscribe_send_status": 0, "created_on": now}}, upsert=True)) for start in range(0, len(ops), 1000): out = await camp_data.bulk_write(ops[start:start + 1000], ordered=False) records_copied += out.upserted_count # total_emails = the campaign's ACTUAL data count (correct when a # second same-day batch merges into today's campaign). total = await camp_data.count_documents({"campaign_id": campaign_id}) await camps.update_one({"_id": campaign_id}, {"$set": {"total_emails": total}}) # Archive (reset the row to 0) when campaigns were created, OR when every # email in the batch was already sent (nothing new to do). A mapping/publisher # failure leaves the batch so you can fix it and retry. if campaigns > 0: await cp.update_many(batch, {"$set": {"campaigned_on": now}}) detail = (f"Skipped {already_sent} already-sent email(s)." if already_sent else None) elif not de_rows and already_sent: await cp.update_many(batch, {"$set": {"campaigned_on": now}}) detail = (f"No new campaigns: all {already_sent} email(s) in this batch were " f"already sent in a prior campaign.") elif not de_rows: detail = (f"No campaigns: 0 rows reached collected_data_de " f"(scanned {scanned}, {unmatched} had no discipline→topic mapping). " f"Check the 'disciplines' mapping for {discipline}.") elif no_publisher_topics: detail = (f"No campaigns: {len(de_rows)} processed row(s), but topic(s) " f"[{', '.join(sorted(set(no_publisher_topics)))}] have no publisher " f"in 'journals' (and aren't 'Others'). Add a journals row linking " f"the topic to a publisher, or rename/route it to 'Others'.") else: detail = "No campaigns created for this discipline." return CampaignResult(topics=topics, campaigns=campaigns, records=records_copied, detail=detail) @router.get("/campaigns", response_model=list[CampaignOut], summary="Preprints 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") rows = await get_db()["cd_campaigns"].find( {"published_on": day, "source": SOURCE_TAG} ).sort("total_emails", DESCENDING).to_list(length=None) return [CampaignOut(**r) for r in rows] @router.post("/campaigns-purge", summary="Delete ALL preprints campaigns and their data") async def campaigns_purge(): db = get_db() ids = await db["cd_campaigns"].distinct("_id", {"source": SOURCE_TAG}) records = 0 if ids: records = (await db["cd_campaign_data"].delete_many( {"campaign_id": {"$in": ids}})).deleted_count campaigns = (await db["cd_campaigns"].delete_many({"source": SOURCE_TAG})).deleted_count return {"campaigns": campaigns, "records": records} @router.get("/campaigns-by-discipline", response_model=list[CampaignOut], summary="This discipline's preprints campaigns") async def campaigns_by_discipline(discipline: str): rows = await get_db()["cd_campaigns"].find( {"discipline": discipline, "source": SOURCE_TAG} ).sort("total_emails", DESCENDING).to_list(length=None) return [CampaignOut(**r) for r in rows]