dee/core/ip_radar.py

"""Live Collision IP Radar — v1 (literature, Europe PMC).

Cross-references a freshly generated CRISPR guide / evolved variant against
recent literature (preprints + published papers) so the user is warned when
their design overlaps existing work. Patents are deferred to v2 (need a
Lens.org token).

Design constraints (decided 2026-06-02):
  * OPT-IN, METADATA-FIRST. We never receive the user's full novel sequence —
    only the gene symbol, the specific mutation set, and (for CRISPR) the
    20-mer guide spacer. Those are what we send to Europe PMC.
  * MATCH ON SPECIFIC IDENTIFIERS, not gene+keyword noise. "TP53 AND CRISPR"
    returns ~12k papers — meaningless as a collision. The exact 20-mer spacer
    (full-text phrase) or the exact mutation set is what makes a hit real:
    a novel guide returns 0 hits; a published one returns the papers that
    printed it.
  * NO definitive infringement / "patented by X" claims. We surface "possible
    prior art / overlap — verify" with the citation; the framing is the
    frontend's job, this module just returns matches.
  * Non-blocking + cached. A 24 h in-process TTL cache (the HF Space is a
    single instance — no Redis) means 50 users checking the same standard
    TP53 guide hit Europe PMC once, then serve cached. Empty results are
    cached too (a clean guide shouldn't re-ping on every view).

Pure stdlib (urllib + json). No new dependencies.
"""
from __future__ import annotations

import datetime as _dt
import hashlib
import json
import os
import re
import threading
import time
import urllib.parse
import urllib.request
from typing import Dict, List, Optional

_PMC = "https://www.ebi.ac.uk/europepmc/webservices/rest/search"
_TIMEOUT = 8          # seconds per Europe PMC call
_CACHE_TTL = 24 * 3600
_CACHE_MAX = 512
_MAX_RESULTS = 4

# A valid CRISPR spacer for searching: pure ACGT, plausible guide length.
_SPACER_RE = re.compile(r"^[ACGT]{17,25}$")
# HGVS-ish protein point mutation, e.g. T47A, S101G, R248* (stop).
_MUT_RE = re.compile(r"^[A-Za-z](\d{1,5})[A-Za-z*]$")

# ── tiny thread-safe TTL cache ─────────────────────────────────────────────
_cache: "dict[str, tuple[float, dict]]" = {}
_cache_lock = threading.Lock()


def _cache_get(key: str) -> Optional[dict]:
    with _cache_lock:
        hit = _cache.get(key)
        if not hit:
            return None
        expires, val = hit
        if expires < time.time():
            _cache.pop(key, None)
            return None
        return val


def _cache_put(key: str, val: dict) -> None:
    with _cache_lock:
        if len(_cache) >= _CACHE_MAX:
            # drop the oldest-expiring entry — cheap bound, no LRU bookkeeping
            oldest = min(_cache, key=lambda k: _cache[k][0])
            _cache.pop(oldest, None)
        _cache[key] = (time.time() + _CACHE_TTL, val)


def _cache_clear() -> None:  # test hook
    with _cache_lock:
        _cache.clear()


# ── Europe PMC ─────────────────────────────────────────────────────────────
def _pmc_search(query: str, page_size: int = 5) -> List[dict]:
    """Run one Europe PMC query, newest first. Returns the raw result dicts.
    Network/parse errors raise — the caller decides how to degrade."""
    qs = urllib.parse.urlencode({
        "query": query,
        "format": "json",
        "resultType": "lite",
        "pageSize": str(page_size),
        "sort": "P_PDATE_D desc",
    })
    req = urllib.request.Request(
        f"{_PMC}?{qs}",
        headers={"User-Agent": "TuringDNA-IPRadar/1.0 (research; contact via turingdna.com)"},
    )
    with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
        data = json.loads(resp.read().decode("utf-8", "replace"))
    return (data.get("resultList") or {}).get("result") or []


def _days_ago(date_str: str) -> Optional[int]:
    try:
        d = _dt.date.fromisoformat(date_str[:10])
        return max(0, (_dt.date.today() - d).days)
    except (ValueError, TypeError):
        return None


def _source_label(source: str, doi: str) -> str:
    if source == "PPR":
        # 10.1101 is the Cold Spring Harbor prefix (bioRxiv / medRxiv).
        return "bioRxiv / medRxiv" if (doi or "").startswith("10.1101") else "a preprint"
    if source == "PAT":
        return "a patent"
    return "a journal"


def _authors(author_string: str) -> str:
    if not author_string:
        return ""
    first = author_string.split(",", 1)[0].strip().rstrip(".")
    return f"{first} et al." if "," in author_string else first


def _url_for(r: dict) -> str:
    doi = r.get("doi")
    if doi:
        return f"https://doi.org/{doi}"
    pmid = r.get("pmid")
    if pmid:
        return f"https://europepmc.org/abstract/MED/{pmid}"
    rid, src = r.get("id"), r.get("source")
    if rid and src:
        return f"https://europepmc.org/article/{src}/{rid}"
    return ""


def _shape(r: dict, match_type: str, match_label: str) -> dict:
    date = r.get("firstPublicationDate") or ""
    return {
        "title": (r.get("title") or "").rstrip(". "),
        "authors": _authors(r.get("authorString") or ""),
        "date": date,
        "days_ago": _days_ago(date),
        "year": r.get("pubYear") or "",
        "source": _source_label(r.get("source") or "", r.get("doi") or ""),
        "is_preprint": (r.get("source") == "PPR"),
        "url": _url_for(r),
        "match_type": match_type,     # 'guide_exact' | 'mutation_set'
        "match_label": match_label,   # human label of WHAT matched
    }


def _build_queries(gene_symbol: str, mutations: List[str], spacer: str) -> List[tuple]:
    """→ list of (query, match_type, match_label). Only SPECIFIC identifiers —
    never a bare gene+keyword (that's noise, not a collision)."""
    out = []
    if spacer and _SPACER_RE.match(spacer):
        out.append((f'"{spacer}"', "guide_exact", "this exact guide sequence"))
    if gene_symbol and mutations:
        muts = " AND ".join(f'"{m}"' for m in mutations[:6])
        out.append((f"({gene_symbol}) AND ({muts})", "mutation_set",
                    f"{gene_symbol} + {', '.join(mutations[:6])}"))
    return out


_PATENTSVIEW = "https://search.patentsview.org/api/v1/patent/"


def check_patents(gene_symbol: str = "", max_results: int = 3) -> Dict:
    """Recent patents that mention the target gene + genome-editing terms, via
    the FREE PatentsView Search API. Needs a (free, registered) env key
    PATENTSVIEW_API_KEY; without it we return available=False and the radar
    stays literature-only.

    This is GENE-LEVEL "related patents in this space" — NOT a precise
    sequence-to-claim match (that needs Lens PatSeq, which is paid). The UI
    frames it as context to review, never as an infringement finding.
    """
    gene_symbol = (gene_symbol or "").strip()
    key = os.environ.get("PATENTSVIEW_API_KEY", "").strip()
    if not key or not gene_symbol:
        return {"available": bool(key), "patents": []}

    cache_key = "pat:" + gene_symbol.lower()
    cached = _cache_get(cache_key)
    if cached is not None:
        return cached

    q = json.dumps({"_and": [
        {"_text_phrase": {"patent_abstract": gene_symbol}},
        {"_text_any": {"patent_abstract": "CRISPR Cas9 gene editing guide RNA"}},
    ]})
    params = urllib.parse.urlencode({
        "q": q,
        "f": json.dumps(["patent_id", "patent_title", "patent_date",
                         "assignees.assignee_organization"]),
        "o": json.dumps({"size": max_results}),
        "s": json.dumps([{"patent_date": "desc"}]),
    })
    out = {"available": True, "patents": []}
    try:
        req = urllib.request.Request(
            f"{_PATENTSVIEW}?{params}",
            headers={"X-Api-Key": key, "User-Agent": "TuringDNA-IPRadar/1.0"},
        )
        with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
            data = json.loads(resp.read().decode("utf-8", "replace"))
        for p in (data.get("patents") or [])[:max_results]:
            assignees = p.get("assignees") or []
            assignee = (assignees[0] or {}).get("assignee_organization", "") if assignees else ""
            pid = p.get("patent_id") or ""
            out["patents"].append({
                "title": (p.get("patent_title") or "").rstrip(". "),
                "assignee": assignee,
                "date": p.get("patent_date") or "",
                "year": (p.get("patent_date") or "")[:4],
                "url": f"https://patents.google.com/patent/US{pid}" if pid else "",
            })
    except Exception:
        pass  # patents are best-effort; literature still returns
    _cache_put(cache_key, out)
    return out


def check_literature(
    *,
    gene_symbol: str = "",
    mutations: Optional[List[str]] = None,
    spacer: str = "",
    mode: str = "knockout",
    max_results: int = _MAX_RESULTS,
) -> Dict:
    """Metadata-first literature collision check. Returns
    {ok, matches:[...], checked:[labels]} — matches sorted newest-first,
    de-duplicated across queries. Cached (incl. empty results)."""
    gene_symbol = (gene_symbol or "").strip()
    spacer = (spacer or "").strip().upper()
    mutations = [m for m in (mutations or []) if _MUT_RE.match(str(m))][:8]

    queries = _build_queries(gene_symbol, mutations, spacer)
    if not queries:
        # Nothing specific enough to be a real collision signal.
        return {"ok": True, "matches": [], "checked": [],
                "note": "Need an exact guide spacer or a gene + mutation set to scan."}

    key = "ipr:" + hashlib.sha1(
        json.dumps([q for q, _, _ in queries], sort_keys=True).encode()
    ).hexdigest()
    cached = _cache_get(key)
    if cached is not None:
        return cached

    matches: List[dict] = []
    seen = set()
    checked: List[str] = []
    for query, mtype, mlabel in queries:
        checked.append(mlabel)
        try:
            results = _pmc_search(query)
        except Exception:
            continue  # one source failing shouldn't sink the whole check
        for r in results:
            dedup = r.get("doi") or r.get("pmid") or r.get("id")
            if dedup in seen:
                continue
            seen.add(dedup)
            matches.append(_shape(r, mtype, mlabel))
            if len(matches) >= max_results:
                break
        if len(matches) >= max_results:
            break

    matches.sort(key=lambda m: m.get("date") or "", reverse=True)
    out = {"ok": True, "matches": matches, "checked": checked}
    _cache_put(key, out)
    return out