Blame for src/identifiers.py

"""What may leave the Space: identifiers only (ADR-0001 §6).

Every endpoint argument passes through here BEFORE any outbound call. Allowed are HGNC symbols,
Ensembl gene ids, UniProt accession numbers, disease ids from EFO, MONDO or Orphanet, NCT ids, and
a disease name as a bounded string. Free text, sample ids, numbers (expression values) and file contents are
refused as data — ``invalid_identifier`` — and more than 100 identifiers as
``too_many_identifiers``. A refusal never echoes more than a short, truncated token back.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field

MAX_IDENTIFIERS = 100
MAX_DISEASE_NAME = 80
_ECHO = 24  # longest token a refusal message repeats back

# HGNC approved symbols start with a letter and use letters, digits and hyphens (C1orf112,
# HLA-DRB1, NKX2-1; immunoglobulin segments such as IGHV1-69-2 carry two hyphens).
SYMBOL = re.compile(r"^[A-Za-z][A-Za-z0-9]*(?:-[A-Za-z0-9]+){0,2}$")
ENSEMBL_GENE = re.compile(r"^ENSG\d{11}(?:\.\d+)?$")
# UniProt's published accession pattern.
UNIPROT = re.compile(
    r"^(?:[OPQ][0-9][A-Z0-9]{3}[0-9]|[A-NR-Z][0-9](?:[A-Z][A-Z0-9]{2}[0-9]){1,2})$"
)
DISEASE_ID = re.compile(r"^(?:EFO|MONDO|Orphanet)_\d{1,10}$")
NCT = re.compile(r"^NCT\d{8}$")
DISEASE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ,'\-]*$")

# Shapes that would pass as a symbol but are sample or run identifiers (TCGA barcodes, GEO and
# SRA accessions, CPTAC case ids, ICGC donor/sample ids). They never leave the Space: a sample id
# is study data, not a public identifier. Every prefix here is checked against real symbols that
# share its letters (CPT1A, ERRFI1, DOK1, SAA1 stay valid; tests/test_identifiers.py).
_SAMPLE_ID = re.compile(
    r"^(?:TCGA-|GSM\d|GSE\d|SRR\d|ERR\d|DRR\d|SRS\d|SAMN\d|C3[LN]-|DO\d|SA\d)",
    re.IGNORECASE,
)
_SPLIT = re.compile(r"[\s,;]+")

GENE_KINDS = frozenset({"symbol", "ensembl"})
PROTEIN_OR_GENE_KINDS = frozenset({"symbol", "ensembl", "uniprot"})


@dataclass(frozen=True)
class Identifier:
    value: str
    kind: str  # symbol | ensembl | uniprot | disease_id | disease_name | nct


@dataclass(frozen=True)
class Parsed:
    """Either a list of identifiers or a refusal (``reason`` + ``message``), never both."""

    identifiers: tuple[Identifier, ...] = ()
    reason: str | None = None
    message: str | None = None
    duplicates_dropped: int = field(default=0)

    @property
    def ok(self) -> bool:
        return self.reason is None


def _echo(token: str) -> str:
    return token if len(token) <= _ECHO else token[:_ECHO] + "…"


def _refuse(reason: str, message: str) -> Parsed:
    return Parsed(reason=reason, message=message)


def _dedupe_key(token: str) -> str:
    """Symbols compare case-insensitively (kras == KRAS); Ensembl ids exactly."""
    return token if token.startswith("ENSG") else token.upper()


def classify_gene_token(
    token: str, *, allow_uniprot: bool = False, allow_ensembl: bool = True
) -> str | None:
    """The identifier kind for one gene/protein token, or None if it is not one."""
    if _SAMPLE_ID.match(token):
        return None
    if ENSEMBL_GENE.match(token):
        return "ensembl" if allow_ensembl else None
    if allow_uniprot and UNIPROT.match(token):
        return "uniprot"
    if len(token) <= 25 and SYMBOL.match(token):
        return "symbol"
    return None


def parse_gene_list(
    raw: str | list[str] | None, *, allow_uniprot: bool = False, allow_ensembl: bool = True
) -> Parsed:
    """Split, de-duplicate (order kept) and validate a gene list.

    ``allow_ensembl=False`` is for the endpoints that resolve through HGNC symbols
    (annotate_genes, pathways_for_genes); drugs_for_targets accepts Ensembl ids directly.
    """
    if isinstance(raw, list):
        tokens = [str(t).strip() for t in raw]
    else:
        tokens = _SPLIT.split((raw or "").strip())
    tokens = [t for t in tokens if t]
    if not tokens:
        return _refuse("invalid_identifier", "No gene identifiers were given.")
    unique = {_dedupe_key(t) for t in tokens}
    if len(unique) > MAX_IDENTIFIERS:
        return _refuse(
            "too_many_identifiers",
            f"{len(unique)} identifiers given; at most {MAX_IDENTIFIERS} per call.",
        )
    kinds = {"allow_uniprot": allow_uniprot, "allow_ensembl": allow_ensembl}
    bad = [t for t in tokens if classify_gene_token(t, **kinds) is None]
    if bad:
        shown = ", ".join(_echo(t) for t in bad[:5])
        more = f" and {len(bad) - 5} more" if len(bad) > 5 else ""
        accepted = (
            "HGNC symbols"
            + (" or Ensembl gene ids" if allow_ensembl else "")
            + (" or UniProt accessions" if allow_uniprot else "")
        )
        return _refuse(
            "invalid_identifier",
            f"Not a gene identifier: {shown}{more}. Only {accepted} are sent to external sources; "
            "free text, sample ids and numbers are not.",
        )
    out: list[Identifier] = []
    emitted: set[str] = set()
    for t in tokens:
        key = _dedupe_key(t)
        if key in emitted:
            continue
        emitted.add(key)
        out.append(Identifier(t, classify_gene_token(t, **kinds) or "symbol"))
    return Parsed(identifiers=tuple(out), duplicates_dropped=len(tokens) - len(out))


def parse_disease(raw: str | None) -> Parsed:
    """One disease: an EFO/MONDO/Orphanet id, or a bounded name."""
    text = (raw or "").strip()
    if not text:
        return _refuse("invalid_identifier", "No disease was given.")
    if DISEASE_ID.match(text):
        return Parsed(identifiers=(Identifier(text, "disease_id"),))
    if (
        len(text) <= MAX_DISEASE_NAME
        and DISEASE_NAME.match(text)
        and any(c.isalpha() for c in text)
        and not _SAMPLE_ID.match(text)
    ):
        return Parsed(identifiers=(Identifier(" ".join(text.split()), "disease_name"),))
    return _refuse(
        "invalid_identifier",
        f"Not a disease name or id: {_echo(text)}. Give an EFO/MONDO id or a name of at most "
        f"{MAX_DISEASE_NAME} characters (letters, digits, spaces, hyphens, commas, apostrophes).",
    )


def parse_nct(raw: str | None) -> Parsed:
    text = (raw or "").strip().upper()
    if NCT.match(text):
        return Parsed(identifiers=(Identifier(text, "nct"),))
    return _refuse("invalid_identifier", f"Not an NCT id: {_echo(text)}.")