"""Turn a day's clean preprint contacts into `collected_data_de` rows. "Clean" = survived every block-list filter (`filter_reason` absent). Each row is mapped through `disciplines` -> `section_topics` (the PHIDESK topic) and `countries`, then upserted into `collected_data_de` with the fixed field set. Idempotent (upsert on a natural key), so re-processing a day -- the automatic trigger and then the Process button, say -- never duplicates. """ import re from datetime import datetime, timezone from typing import Any, Optional from bson import ObjectId from bson.errors import InvalidId from pymongo import UpdateOne from .db import get_db # Fixed data_type stamped on every collected_data_de row (preprint id). DE_DATA_TYPE = ObjectId("686f42d8333418fdc7368ca3") PREPRINTS = "collected_data_preprints_v2" DE = "collected_data_de" JOURNALS = "journals" CAMPAIGNS = "cd_campaigns" CAMPAIGN_DATA = "cd_campaign_data" # When a preprints record's topic is the catch-all "Others", its campaign is # released to these three publishers (matched by publisher_name, case-insensitive) # instead of whatever journals map to the topic: # igmin -> "IgMin Publications Inc" # peertechz -> "Peertechz Publications Pvt Ltd" # heighten -> "Heighten Science Publications Inc." OTHERS_TOPIC = "others" OTHERS_PUBLISHERS = ["igmin", "peertechz", "heighten"] def _is_others_topic(topic_name: Any) -> bool: return (str(topic_name or "")).strip().lower() == OTHERS_TOPIC async def _others_journals(db, topic_id) -> list: """One journal each for the Others-topic publishers (Igmin / Peertechz / Heightpubs), attached to `topic_id`. Missing publishers are simply skipped.""" out: list = [] seen: set = set() for name in OTHERS_PUBLISHERS: journal = await db[JOURNALS].find_one( {"publisher_name": {"$regex": re.escape(name), "$options": "i"}}) if journal and journal.get("publisher_id") not in seen: seen.add(journal.get("publisher_id")) journal = dict(journal) journal["topic_id"] = topic_id # attach to the Others topic out.append(journal) return out async def _all_active_publisher_journals(db, topic_id) -> list: """One journal per ACTIVE publisher, attached to `topic_id` -- used for the bioRxiv 'Others' topic (a discipline that maps to Others releases to every active publisher). A publisher is active when its users row has status == 1; if no publisher is marked active, fall back to every publisher with a journal.""" active: set = set() async for u in db["users"].find( {"publisher_id": {"$exists": True}, "status": 1}, {"publisher_id": 1}): if u.get("publisher_id") is not None: active.add(u["publisher_id"]) out: list = [] seen: set = set() async for journal in db[JOURNALS].find({}): pid = journal.get("publisher_id") if pid is None or pid in seen: continue if active and pid not in active: # keep only active publishers (if known) continue seen.add(pid) journal = dict(journal) journal["topic_id"] = topic_id # attach to the Others topic out.append(journal) return out def _de_src(source_tag: Optional[str]) -> dict: """Scope collected_data_de / cd_campaigns by source. arXiv rows (the default, source_tag=None) carry no `source`; preprints rows carry source='preprints.org' so the two pipelines never mix in the shared de / campaign collections.""" return {"source": source_tag} if source_tag else {"source": {"$exists": False}} def _as_object_id(value: Any) -> Any: """Coerce to ObjectId when possible; otherwise return the value unchanged.""" if isinstance(value, ObjectId): return value try: return ObjectId(str(value)) except (InvalidId, TypeError): return value def _to_int(value: Any) -> Optional[int]: try: return int(value) except (ValueError, TypeError): return None def _norm_source(value: Any) -> str: """Normalise the source so the preprint value 'arxiv.org' matches the disciplines value 'arxiv' -- both become 'arxiv'.""" return str(value or "").strip().lower().split(".")[0] async def _discipline_topics() -> dict: """(source, shortname) -> phidesktopic (its section_topics `topic_name`).""" out: dict = {} async for d in get_db()["disciplines"].find( {}, {"source": 1, "shortname": 1, "phidesktopic": 1}): topic = (d.get("phidesktopic") or "").strip() if topic: out[(_norm_source(d.get("source")), d.get("shortname"))] = topic return out async def _country_index() -> dict: """Lower-cased country name / alias -> country doc (name, _id, income).""" idx: dict = {} async for c in get_db()["countries"].find( {}, {"name": 1, "similar_country_names": 1, "income": 1}): name = (c.get("name") or "").strip().lower() if name: idx.setdefault(name, c) for alias in (c.get("similar_country_names") or []): a = (str(alias) or "").strip().lower() if a: idx.setdefault(a, c) return idx async def process_day(user_id: str, day: datetime, source_coll: str = PREPRINTS, source_tag: Optional[str] = None, discipline: Optional[str] = None, force: bool = False) -> dict: """Build collected_data_de rows for one day from `source_coll`. arXiv is the default; preprints passes collected_data_preprints_org + 'preprints.org'. `discipline` limits processing to one discipline's rows (used by the discipline-wise preprints view). The per-day already-saved short-circuit is skipped in that mode -- other disciplines may share the day -- and the upsert key keeps it idempotent. `force` also skips the short-circuit, so re-running Process picks up newly-mapped disciplines (the upsert prevents duplicates).""" db = get_db() pp = db[source_coll] de = db[DE] section_topics = db["section_topics"] # Already saved for this day (for this source)? Nothing to do. Skipped for a # single discipline or a forced re-run (a day can gain new mappable rows). if discipline is None and not force and await de.count_documents( {"published_on": day, **_de_src(source_tag)}) > 0: return {"scanned": 0, "inserted": 0, "unmatched_topic": 0, "already_saved": True} collector = _as_object_id(user_id) disc_topics = await _discipline_topics() countries = await _country_index() topic_cache: dict[str, Optional[dict]] = {} now = datetime.now(timezone.utc) # Final filtered rows: this day, survived every filter (no `filter_reason`), # and not yet archived into a campaign batch (`campaigned_on` absent). clean = { "published_on": day, "campaigned_on": {"$exists": False}, "$or": [{"filter_reason": {"$exists": False}}, {"filter_reason": None}], } if discipline: clean["discipline"] = discipline ops: list = [] scanned = unmatched_topic = 0 async for rec in pp.find(clean): scanned += 1 phidesktopic = disc_topics.get((_norm_source(rec.get("source")), rec.get("discipline"))) if not phidesktopic: unmatched_topic += 1 continue if phidesktopic not in topic_cache: topic_cache[phidesktopic] = await section_topics.find_one( {"topic_name": phidesktopic}) topic = topic_cache[phidesktopic] if not topic: unmatched_topic += 1 continue topic_id = topic.get("_id") topic_name = topic.get("topic_name") section_id = topic.get("section_id") country = countries.get((rec.get("country") or "").strip().lower()) name = rec.get("name") if not name or not str(name).strip(): name = "Doctor" doc = { "user_id": collector, "data_type": DE_DATA_TYPE, "topic": topic_name, "country_name": country.get("name") if country else rec.get("country"), "country_id": country.get("_id") if country else None, "country_level": _to_int(country.get("income")) if country else None, "title": rec.get("title"), "name": name, "email": rec.get("email"), "mail_send_status": 0, "subscribe_send_status": 0, "posted_on": now, "primary_topic_id": topic_id, "primary_topic": topic_name, "section_id": section_id, "affiliation": rec.get("affiliation"), "published_on": rec.get("published_on"), } # source / discipline go in $set (not $setOnInsert) so re-processing # BACKFILLS them onto de rows saved before discipline-tagging existed -- # per-discipline campaigns depend on this tag. backfill: dict = {} if source_tag: backfill["source"] = source_tag if rec.get("discipline"): backfill["discipline"] = rec.get("discipline") # Identity: one row per (collector, email, title, topic, day). key = {"user_id": collector, "email": doc["email"], "title": doc["title"], "primary_topic_id": topic_id, "published_on": doc["published_on"]} update: dict = {"$setOnInsert": doc} if backfill: update["$set"] = backfill ops.append(UpdateOne(key, update, upsert=True)) inserted = 0 for start in range(0, len(ops), 1000): result = await de.bulk_write(ops[start:start + 1000], ordered=False) inserted += result.upserted_count return {"scanned": scanned, "inserted": inserted, "unmatched_topic": unmatched_topic, "already_saved": False} async def create_campaigns(user_id: str, day: datetime, source_tag: Optional[str] = None, discipline: Optional[str] = None) -> dict: """Publisher-wise campaigns from a saved day's collected_data_de (scoped to the given source, and optionally one discipline). Idempotent: re-running upserts, never duplicates.""" db = get_db() de = db[DE] journals = db[JOURNALS] camps = db[CAMPAIGNS] camp_data = db[CAMPAIGN_DATA] collector = _as_object_id(user_id) now = datetime.now(timezone.utc) src = _de_src(source_tag) if discipline: src = {**src, "discipline": discipline} campaign_source = source_tag or "arxiv" topics = 0 campaigns = 0 records_copied = 0 for topic_id in await de.distinct("primary_topic_id", {"published_on": day, **src}): if topic_id is None: continue records = await de.find( {"published_on": day, "primary_topic_id": topic_id, **src}).to_list(length=None) if not records: continue topic_name = records[0].get("primary_topic") # 'Others' catch-all topic (biorxiv / medrxiv / preprints) -> the 3 active # publishers (Igmin / Peertechz / Heighten), each with the same data. # Any other topic (or source) uses the topic's own mapped journals. if _is_others_topic(topic_name) and source_tag in ("biorxiv", "medrxiv", "preprints.org"): publishers = await _others_journals(db, topic_id) else: publishers = await journals.find({"topic_id": topic_id}).to_list(length=None) if not publishers: continue # no publisher for this topic -> no campaign topics += 1 for journal in publishers: publisher_id = journal.get("publisher_id") if publisher_id is None: continue # Key on the cd_campaigns UNIQUE INDEX (user, publisher, topic, # published_on) -- NOT source -- so re-running always UPDATES the # existing campaign at that slot instead of trying to insert a # duplicate (which fails when the existing row has a different or # missing source). `source` is applied in $set. key = {"user_id": collector, "publisher_id": publisher_id, "primary_topic_id": topic_id, "published_on": day} 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, "source": campaign_source, "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"] # Copy the topic's records into this campaign's data set. 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": day, "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"), "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 return {"topics": topics, "campaigns": campaigns, "records": records_copied}