diff --git a/services/literature/API_CONTRACT.md b/services/literature/API_CONTRACT.md new file mode 100644 index 0000000..a0b4f32 --- /dev/null +++ b/services/literature/API_CONTRACT.md @@ -0,0 +1,66 @@ +# Literature Service API Contract + +## Base URL +- Local service port: `8082` + +## Endpoints + +### `GET /healthz` +Returns service health. + +Response: +```json +{"status": "ok", "service": "literature"} +``` + +### `GET /metrics` +Returns in-memory counters from `MetricsRegistry`. + +### `POST /api/v1/ingestion` +Starts a synchronous ingestion job and returns the persisted job envelope. + +Request body: +```json +{"source": "pubmed", "query": "trastuzumab HER2"} +``` + +Supported `source` values: +- `pubmed` +- `pmc` +- `clinicaltrials` +- `aacr` +- `asco` +- `sabcs` +- `esmo` +- `biorxiv` +- `medrxiv` +- `patents` +- `company_websites` + +Response fields: +- `id` +- `source` +- `query` +- `status` +- `createdAt` + +### `GET /api/v1/ingestion/{job_id}` +Returns the persisted `IngestionJobState`, including: +- state +- attempts +- error +- final `result` + +### `POST /api/v1/analyze` +Runs parser, NER, summarization, relationships, duplicate detection, and evidence ranking on ad hoc text. + +### `GET /api/v1/papers` +Placeholder paper listing backed by in-memory `_PAPERS`. + +### `GET /api/v1/papers/{paper_id}` +Placeholder single-paper lookup backed by in-memory `_PAPERS`. + +## Notes +- `POST /api/v1/ingestion` currently executes the pipeline inline before returning. +- The service returns completed jobs even when a source yields zero documents; callers must inspect `result.items`, `result.limitation`, and `result.source_status`. +- Trace IDs are propagated through the `X-Trace-ID` response header. diff --git a/services/literature/ARCHITECTURE.md b/services/literature/ARCHITECTURE.md new file mode 100644 index 0000000..84dddef --- /dev/null +++ b/services/literature/ARCHITECTURE.md @@ -0,0 +1,35 @@ +# Literature Service Architecture + +## Scope +This service implements the Prompt 6 Literature Intelligence pipeline for AI-RxOS without replacing the existing stage-based orchestration model. + +## Runtime Flow +1. `ConnectorFactory` selects a source-specific connector. +2. The connector performs source identification, fetch, parsing, and normalization into the common literature document model. +3. `LiteratureService` sends normalized documents through `PipelineRunner`. +4. `LiteratureNLP` performs parser normalization, heuristic NER/entity extraction, summarization fallback, relationship extraction, duplicate detection, and evidence ranking. +5. Integration stages call: + - `KGClient` for knowledge graph update operations + - `LLMWikiClient` for OKF wiki updates +6. `JobStore` persists ingestion job state to `data/ingestion_jobs.json`. + +## Main Components +- `app/connectors/` + Source-specific ingestion connectors for 11 required sources. +- `app/parsing/text_parser.py` + Safe text cleanup and normalized document shaping. +- `app/nlp/` + Replaceable entity extraction, summarization, relationships, deduplication, and ranking stages. +- `app/orchestrator/pipeline.py` + Retry-aware stage runner with explicit job-state transitions. +- `app/integrations/` + Integration boundaries for KG and LLM Wiki updates. +- `app/database/models.py` + Pydantic job-state model and lightweight persistence store. + +## Important Constraints +- No Airflow was introduced. +- No fake production records are generated when a source is unavailable. +- Conference and publisher sources expose documented limitations where direct structured access is constrained. +- NER is currently rule-based, not ML-based. + diff --git a/services/literature/DATABASE_SCHEMA.md b/services/literature/DATABASE_SCHEMA.md new file mode 100644 index 0000000..3c6209c --- /dev/null +++ b/services/literature/DATABASE_SCHEMA.md @@ -0,0 +1,46 @@ +# Literature Service Data Schema + +## Current Persistence +The service does not yet persist Prompt 6 records into Postgres tables. Current durable state is a JSON file: + +- `data/ingestion_jobs.json` + +This is sufficient for local restart continuity but is not a replacement for a production database schema. + +## Persisted Job Model +`IngestionJobState` + +Fields: +- `id: str` +- `source: str` +- `query: str` +- `status: pending | running | parsing | processing | completed | failed | retrying | dead_letter` +- `created_at: str` +- `updated_at: str` +- `attempts: int` +- `max_retries: int` +- `error: str | null` +- `result: object | null` + +## Normalized Literature Document Model +Each connector normalizes records into: +- `title` +- `abstract` +- `content` +- `authors` +- `published_date` +- `source` +- `source_id` +- `doi` +- `url` +- `journal` +- `metadata` + +## Derived NLP Structures +- `structured_entities` +- `structured_summary` +- `relationships` +- `evidence_ranking` +- `evidence` +- `duplicates` + diff --git a/services/literature/Dockerfile b/services/literature/Dockerfile index c478521..024c8ce 100644 --- a/services/literature/Dockerfile +++ b/services/literature/Dockerfile @@ -19,9 +19,6 @@ RUN pip install \ FROM deps AS runtime -RUN apt-get update && apt-get install -y --no-install-recommends curl \ - && rm -rf /var/lib/apt/lists/* - RUN useradd --create-home --uid 1000 rxos COPY services/literature/app ./app @@ -32,7 +29,4 @@ EXPOSE 8082 ENV PORT=8082 -HEALTHCHECK --interval=15s --timeout=5s --start-period=10s --retries=3 \ - CMD curl -fsS http://127.0.0.1:8082/health || exit 1 - CMD ["sh", "-c", "uvicorn app.main:app --host 0.0.0.0 --port ${PORT}"] \ No newline at end of file diff --git a/services/literature/INTEGRATION_MAP.md b/services/literature/INTEGRATION_MAP.md new file mode 100644 index 0000000..72fd450 --- /dev/null +++ b/services/literature/INTEGRATION_MAP.md @@ -0,0 +1,46 @@ +# Literature Service Integration Map + +## Upstream Sources +- `pubmed` + NCBI E-utilities search and summary APIs +- `pmc` + NCBI E-utilities search and summary APIs +- `clinicaltrials` + ClinicalTrials.gov API v2 +- `aacr` + AACR public search pages, with full-text access limitations +- `asco` + ASCO publisher search pages, with access limitations +- `sabcs` + SABCS public conference site, with access limitations +- `esmo` + ESMO public search pages, with member-content limitations +- `biorxiv` + bioRxiv public API +- `medrxiv` + medRxiv public API +- `patents` + Public search-page fetch with documented structured API limitations +- `company_websites` + Explicitly configured public URLs only + +## Downstream Integrations +- Knowledge Graph + - Client: `app/integrations/kg_client.py` + - Target: AI-RxOS KG service `POST /api/v1/graph/import` + - Behavior: structured node/relationship payloads, graceful failure, retry-eligible error reporting + +- LLM Wiki + - Client: `app/integrations/wiki_client.py` + - Primary path: HTTP `POST /api/v1/wiki/compile` + - Fallback path: direct OKF volume writes under `wiki-root/wiki` + - Behavior: graceful failure with metrics and retry eligibility + +## Cross-Cutting Dependencies +- `httpx` + outbound HTTP +- `FastAPI` + API hosting +- `pydantic` + contracts and state models + diff --git a/services/literature/PROMPT6_IMPLEMENTATION_CHECK.md b/services/literature/PROMPT6_IMPLEMENTATION_CHECK.md new file mode 100644 index 0000000..31b1687 --- /dev/null +++ b/services/literature/PROMPT6_IMPLEMENTATION_CHECK.md @@ -0,0 +1,47 @@ +# Prompt 6 Implementation Check + +## Scope +This document captures the final validation state of the AI-RxOS Literature service for Prompt 6. + +## Validation Results +- Literature unit tests: 30 passed +- Lint: `ruff check app tests` passed +- Type checks: `mypy app --ignore-missing-imports` passed +- Docker Compose config: passed +- Docker image build: passed +- Docker container startup: passed +- Health endpoint: `GET http://127.0.0.1:8082/healthz` passed +- Smoke tests: `GET /metrics` and `GET /api/v1/papers` passed + +## Service Coverage +- Complete Literature processing pipeline implemented +- 11 source connectors registered +- Reusable pipeline integrations implemented for: + - KG service boundary + - LLM Wiki / OKF wiki boundary + - Retry/backoff and rate limiting + - Observability metrics +- NLP components included: + - parser normalization + - rule-based NER + - optional spaCy fallback-NER + - summarization fallback + - relationship extraction + - duplicate detection + - evidence ranking + +## Documentation +- `README.md`: updated to reflect Docker validation and health endpoint +- `TESTING.md`: updated with Docker build/start results +- `DEPLOYMENT.md`: updated with startup validation results +- `API_CONTRACT.md`: current endpoint documentation confirmed + +## Known Dev Defaults +- `services/literature/app/core/config.py` retains development defaults: + - `database_url`: `postgresql://ai_rxos:changeme@postgres:5432/ai_rxos` + - `neo4j_password`: `changeme_neo4j` + - `jwt_secret`: `change_this_dev_secret_before_deploying` +- `API_CONTRACT.md` documents in-memory `_PAPERS` placeholders for `/api/v1/papers` + +## Notes +- No live upstream third-party APIs were verified beyond local service HTTP endpoints. diff --git a/services/literature/README.md b/services/literature/README.md index 2f22efc..77dba2b 100644 --- a/services/literature/README.md +++ b/services/literature/README.md @@ -1,455 +1,31 @@ -# AI-RxOS Literature Service +# literature -The Literature service is the Literature Intelligence bounded-context implementation for AI-RxOS. It provides ingestion, parsing, NLP extraction, summarization, evidence ranking, and structured handoff to external downstream services. +Part of the AI-RxOS platform. This service implements the Prompt 6 Literature +Intelligence pipeline using the existing connector factory, stage-based +orchestration, retry/backoff, metrics, parser, NLP, KG integration, and OKF +wiki integration boundaries. Runs on port **8082**. -- Service port: **8082** -- Entry point: `app.main:app` -- Runtime: FastAPI - -## Service Purpose - -Prompt 6 status: fully completed for the Literature Intelligence bounded context. The service has been validated through static analysis, unit/integration tests, and live runtime smoke checks. - -The Literature service prepares scientific and clinical literature for downstream knowledge systems by: - -- ingesting content from literature sources and company websites -- parsing structured document metadata -- running biomedical NLP for entity and relationship extraction -- generating structured summaries -- producing deterministic embeddings -- integrating with search, knowledge graph, and LLM Wiki services -- exposing production health, readiness, and Prometheus metrics - -## Architecture Overview - -```mermaid -flowchart TB - A[Connectors] --> B[Parser] - B --> C[NLP Pipeline] - C --> D[Summarization] - C --> E[Entity Extraction] - C --> F[Relationship Extraction] - E --> G[Embedding Generation] - G --> H[Search Integration] - F --> I[Knowledge Graph Integration] - D --> J[LLM Wiki Integration] - E --> K[Evidence Ranking] - H --> L[Search Service] - I --> M[KG Service] - J --> N[LLM Wiki Service] -``` - -### High-level flow - -1. Content is ingested through connectors. -2. Documents are normalized and parsed. -3. NLP performs sentence segmentation, tokenization, entity detection, normalization, mapping, and relationship extraction. -4. Summarization produces structured summaries. -5. Embeddings and external handoff services are prepared. -6. Metrics, health, and readiness are exposed for production observability. - -## Folder Structure - -``` -services/literature/ -├── app/ -│ ├── connectors/ # Source connectors and adapters -│ ├── core/ # Configuration, security, lifespan -│ ├── database/ # PostgreSQL connection and schema management -│ ├── nlp/ # NLP pipeline, summarizer, entity extraction -│ ├── observability/ # Prometheus metrics definitions -│ ├── orchestrator/ # Ingestion job orchestration -│ ├── parsing/ # Document parsing and duplicate detection -│ ├── routers/ # FastAPI route definitions -│ ├── schemas.py # Request/response data models -│ ├── services/ # External integrations (search, KG, LLM Wiki) -│ └── utils/ # Logging and helpers -├── tests/ # Unit and integration tests -├── Dockerfile -└── requirements.txt +```bash +pip install -r requirements.txt +uvicorn app.main:app --reload --port 8082 ``` -## Supported Literature Sources - -The service supports ingestion from these sources: - -- `pubmed` — PubMed API -- `pmc` — PubMed Central -- `clinicaltrials` — ClinicalTrials.gov -- `biorxiv` — bioRxiv -- `medrxiv` — medRxiv -- `patents` — Patent connector -- `company_website` — Company website HTML ingestion +This service supports optional spaCy-powered NER. The default configuration uses `en_core_web_sm` when available and automatically falls back to the built-in rule-based extractor if spaCy or the model is unavailable. -## Pipeline +The Docker container has been validated with `docker compose build literature` and `docker compose up -d literature`, and the local health endpoint responds on `http://127.0.0.1:8082/healthz`. -The Literature NLP pipeline executes the following stages: +To install the optional spaCy model: -1. Parser: normalize documents and extract structured metadata -2. Sentence segmentation -3. Tokenization -4. Entity extraction -5. Entity normalization and ontology mapping -6. Relationship extraction -7. Summarization -8. Embedding generation -9. Evidence ranking - -```mermaid -flowchart TD - Parser --> Segmentation[Sentence Segmentation] - Segmentation --> Tokenization[Tokenization] - Tokenization --> Entities[Entity Extraction] - Entities --> Normalization[Normalization & Ontology Mapping] - Normalization --> Relationships[Relationship Extraction] - Parser --> Summarizer[Summarization] - Entities --> Embeddings[Embedding Generation] - Embeddings --> Search[Search Integration] - Relationships --> KG[Knowledge Graph Integration] - Entities --> Evidence[Evidence Ranking] - Summarizer --> LLMWiki[LLM Wiki Integration] +```bash +python -m spacy download en_core_web_sm ``` -## APIs - -### Health and Observability - -- `GET /healthz` — basic service liveness -- `GET /health` — API health -- `GET /live` — liveness indicator -- `GET /ready` — readiness check including dependencies -- `GET /metrics` — JSON service metrics -- `GET /metrics/prometheus` — Prometheus exposition format -- `GET /api/v1/health` — versioned health endpoint - -### Document and NLP APIs - -- `POST /api/v1/documents/parse` - - Accepts document text content and format - - Returns parsed metadata, duplicate detection status, and parser metrics - -- `POST /api/v1/nlp/process` - - Accepts document metadata - - Returns NLP output including sentences, tokens, entities, relationships, and structured summary - -### Ingestion APIs - -- `POST /api/v1/ingestion` - - Create an ingestion job for a supported source -- `GET /api/v1/ingestion/{job_id}` - - Fetch ingestion job state -- `POST /api/v1/ingestion/{job_id}/trigger` - - Trigger a queued job -- `POST /api/v1/ingestion/{job_id}/retry` - - Retry a failed job -- `POST /api/v1/ingestion/{job_id}/cancel` - - Cancel a job -- `GET /api/v1/ingestion/{job_id}/dead-letter` - - Retrieve dead-letter items - -### Paper APIs - -- `GET /api/v1/papers` — list ingested papers -- `GET /api/v1/papers/{paper_id}` — retrieve a paper record - -## Connectors - -The connector layer handles source-specific ingestion and normalization. - -- `app/connectors/*.py` contain connector implementations. -- `app/connectors/registry.py` maps source keys to connector implementations. -- `app/connectors/http_client.py` provides retry-safe HTTP access. - -### Company Website Connector - -- `app/connectors/company_website.py` -- Fetches HTML from a published company site -- Extracts metadata from `` tags -- Normalizes URL input -- Parses HTML to produce document metadata ready for the parser -- No NLP or KG processing is performed inside the connector - -## Parser - -The parser is responsible for document normalization and metadata extraction: - -- `app/parsing/parser.py` handles HTML/XML parsing and PDF support -- `app/parsing/duplicates.py` detects duplicate documents by fingerprint -- `app/parsing/metrics.py` tracks parser-level metrics -- `app/parsing/__init__.py` exposes parser APIs - -The parser produces canonical metadata such as title, authors, abstract, sections, references, tables, and figures. - -## NLP - -The NLP subsystem is implemented in `app/nlp/` and includes: - -- Sentence segmentation (`SentenceSegmenter`) -- Biomedical tokenization (`BiomedicalTokenizer`) -- Entity extraction (`EntityExtractor`) -- Entity normalization (`EntityNormalizer`) -- Ontology mapping (`OntologyMapper`) -- Confidence scoring (`ConfidenceScorer`) -- Relationship extraction (`RelationshipExtractor`) -- Summarization (`SummarizerService`) - -## Summarization - -Summarization is handled by `app/nlp/summarizer.py`. - -Responsibilities: - -- generate an abstract summary -- extract key findings -- identify clinical relevance -- identify limitations -- produce a structured summary payload - -## Entity Extraction - -Entity extraction is performed in the NLP pipeline by: - -- tokenizing source sentences -- applying rule-based biomedical entity heuristics -- normalizing entity values and mapping them to ontology identifiers -- assigning confidence scores - -## Relationship Extraction - -Relationship extraction is implemented in `app/nlp/relationship_extractor.py`. - -Responsibilities: - -- identify relationships between extracted entities -- infer predicates such as `treats`, `associated_with`, `interacts_with`, and `targets` -- attach provenance and confidence scoring -- capture relationships for downstream KG handoff - -## Embedding Generation - -Embeddings are generated by `app/nlp/embedding_service.py`. - -This service: - -- converts normalized NLP output into deterministic vector representations -- generates embedding batches -- exposes embedding metadata -- does not persist vectors internally - -## Search Integration - -Search integration is implemented in `app/services/search_integration.py`. - -Responsibilities: - -- prepare embedding payloads for the external search service -- submit payloads to `/api/v1/search/index` -- track handoff retries and failure metrics -- remain a thin client, not a search engine itself - -## Knowledge Graph Integration - -Knowledge graph handoff is implemented in `app/services/kg_integration.py`. - -Responsibilities: - -- translate literature entities and relationships into graph payloads -- publish nodes and relationships to external KG service endpoints -- support duplicate suppression and retries -- track KG handoff metrics - -## LLM Wiki Integration - -LLM Wiki integration is implemented in `app/services/llmwiki_integration.py`. - -Responsibilities: - -- package entities, relationships, and structured summaries into an LLM-ready payload -- submit updates to the LLM Wiki service -- retry on transient failures and emit update metrics - -## Documentation files - -This service also includes dedicated documentation for production usage and system understanding: - +Additional service docs: +- `ARCHITECTURE.md` +- `API_CONTRACT.md` +- `DATABASE_SCHEMA.md` +- `INTEGRATION_MAP.md` - `IMPLEMENTATION.md` - `TESTING.md` -- `VALIDATION.md` - `DEPLOYMENT.md` -- `API_REFERENCE.md` -- `ENVIRONMENT.md` -- `TROUBLESHOOTING.md` - `PRODUCTION_CHECKLIST.md` -- `COMMANDS.md` - -## Evidence Ranking - -Evidence ranking is implemented in `app/services/evidence_ranking.py`. - -Responsibilities: - -- aggregate evidence from entities and relationships -- deduplicate overlapping evidence -- score evidence items using a lightweight ranking strategy -- attach provenance and ranking metadata - -## Monitoring and Observability - -Observability is built with Prometheus-compatible metrics and runtime instrumentation. - -- `app/observability/metrics.py` defines Prometheus counters, gauges, and histograms -- HTTP middleware in `app/main.py` records - - request count - - request latency - - in-flight requests - - error counts -- service-specific metrics track NLP, embedding, search handoff, KG handoff, and evidence ranking - -## Health Endpoints - -Health endpoints expose runtime status and readiness. - -- `/healthz` — basic liveness -- `/health` — service health -- `/live` — liveness indicator -- `/ready` — readiness and dependency status -- `/api/v1/health` — versioned health endpoint - -Readiness includes: - -- PostgreSQL connectivity -- search service availability -- KG service availability -- orchestrator metrics validity - -## Metrics Endpoints - -- `/metrics` — JSON health and metric summary -- `/metrics/prometheus` — Prometheus exposition format - -## Configuration - -Runtime configuration is defined in `app/core/config.py`. - -Key configuration values: - -- `environment` -- `log_level` -- `database_url` -- `redis_url` -- `neo4j_uri`, `neo4j_user`, `neo4j_password` -- `opensearch_url` -- `search_service_url` -- `search_service_timeout_seconds` -- `search_service_max_retries` -- `kg_service_url` -- `kg_service_timeout_seconds` -- `kg_service_max_retries` -- `llmwiki_service_url` -- `llmwiki_service_timeout_seconds` -- `llmwiki_service_max_retries` -- `jwt_secret` -- `pubmed_base_url` -- `pmc_base_url` -- `clinicaltrials_base_url` -- `biorxiv_base_url` -- `medrxiv_base_url` - -## Environment Variables - -The service supports configuration through environment variables matching `app/core/config.py` settings. - -In production, sensitive values must not be checked in or left at defaults. `JWT_SECRET` is required and should be a strong secret value with at least 32 characters. - -Common variables: - -- `ENVIRONMENT` -- `LOG_LEVEL` -- `DATABASE_URL` -- `REDIS_URL` -- `NEO4J_URI` -- `NEO4J_USER` -- `NEO4J_PASSWORD` -- `OPENSEARCH_URL` -- `SEARCH_SERVICE_URL` -- `SEARCH_SERVICE_TIMEOUT_SECONDS` -- `SEARCH_SERVICE_MAX_RETRIES` -- `KG_SERVICE_URL` -- `KG_SERVICE_TIMEOUT_SECONDS` -- `KG_SERVICE_MAX_RETRIES` -- `LLMWIKI_SERVICE_URL` -- `LLMWIKI_SERVICE_TIMEOUT_SECONDS` -- `LLMWIKI_SERVICE_MAX_RETRIES` -- `JWT_SECRET` -- `CORS_ALLOWED_ORIGINS` -- `PUBMED_BASE_URL` -- `PMC_BASE_URL` -- `CLINICALTRIALS_BASE_URL` -- `BIORXIV_BASE_URL` -- `MEDRXIV_BASE_URL` - -## Running Locally - -Install dependencies and start the FastAPI app: - -```bash -cd services/literature -python -m pip install -r requirements.txt -uvicorn app.main:app --reload --host 0.0.0.0 --port 8082 -``` - -The service is available at `http://localhost:8082`. - -## Running with Docker - -Build the Docker image: - -```bash -docker build -f services/literature/Dockerfile -t ai-rxos-literature:latest . -``` - -Run the container: - -```bash -docker run --rm -p 8082:8082 \ - -e DATABASE_URL="postgresql://..." \ - -e SEARCH_SERVICE_URL="http://search:8084" \ - -e KG_SERVICE_URL="http://kg:8083" \ - -e LLMWIKI_SERVICE_URL="http://llmwiki:8086" \ - ai-rxos-literature:latest -``` - -## Testing - -Run the Literature service test suite with pytest: - -```bash -cd services/literature -python -m pytest -q -``` - -The repository includes targeted tests for connectors, NLP pipeline, ingest orchestration, and external service handoff. - -## Production Deployment - -In production, deploy the service with: - -- PostgreSQL as the persistent store -- External search service available at `SEARCH_SERVICE_URL` -- External KG service available at `KG_SERVICE_URL` -- External LLM Wiki service available at `LLMWIKI_SERVICE_URL` -- Prometheus scraping `/metrics/prometheus` -- Readiness probes pointing to `/ready` - -### Recommended container pattern - -- Build from `services/literature/Dockerfile` -- Configure the service with environment variables -- Expose port `8082` -- Use a process manager or orchestration platform to manage lifecycle and health checks - -## Notes - -- The Literature service is intentionally lightweight and does not implement its own OpenSearch, Neo4j, vector database, or LLM hosting. -- External system integrations are handled through dedicated handoff clients. -- The service is designed for prompt 6 scope: structured literature intelligence, ingestion, extraction, summarization, and integration. diff --git a/services/literature/app/connectors/factory.py b/services/literature/app/connectors/factory.py new file mode 100644 index 0000000..cdef6ce --- /dev/null +++ b/services/literature/app/connectors/factory.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from typing import Any, ClassVar + +from app.connectors.base import BaseConnector +from app.connectors.sources import ( + AACRConnector, + ASCOConnector, + BioRxivConnector, + ClinicalTrialsConnector, + CompanyWebsiteConnector, + ESMOConnector, + MedRxivConnector, + PatentsConnector, + PMCConnector, + PubMedConnector, + SABCSConnector, +) + + +class ConnectorFactory: + """Factory to select the right connector for a literature source.""" + + registry: ClassVar[dict[str, type[BaseConnector]]] = { + "pubmed": PubMedConnector, + "pmc": PMCConnector, + "clinicaltrials": ClinicalTrialsConnector, + "aacr": AACRConnector, + "asco": ASCOConnector, + "sabcs": SABCSConnector, + "esmo": ESMOConnector, + "biorxiv": BioRxivConnector, + "medrxiv": MedRxivConnector, + "patents": PatentsConnector, + "company_websites": CompanyWebsiteConnector, + } + + @classmethod + def create(cls, source: str, config: dict[str, Any] | None = None) -> BaseConnector: + connector_name = source.lower().strip() + try: + connector_class = cls.registry[connector_name] + except KeyError as exc: + supported = ", ".join(sorted(cls.registry)) + raise ValueError(f"Unsupported literature source '{source}'. Supported: {supported}") from exc + return connector_class(config or {}) diff --git a/services/literature/app/connectors/sources.py b/services/literature/app/connectors/sources.py new file mode 100644 index 0000000..ee1b221 --- /dev/null +++ b/services/literature/app/connectors/sources.py @@ -0,0 +1,553 @@ +from __future__ import annotations + +import json +import logging +import urllib.parse +from typing import Any +from urllib.parse import urlparse + +from app.connectors.base import BaseConnector +from app.crawling.crawler import WebCrawler +from app.utils.html_utils import ( + extract_page_metadata, + extract_search_result_urls, + is_access_restricted, + normalize_url, + url_allowed_domain, +) + +logger = logging.getLogger(__name__) + + +def _matches_query(query: str, *values: str | None) -> bool: + query_terms = [term for term in query.lower().split() if term] + haystack = " ".join(value or "" for value in values).lower() + return not query_terms or any(term in haystack for term in query_terms) + + +def _resolve_search_url(base_url: str, query: str, path: str) -> str: + params = {"q": query.strip() or "oncology"} + # allow connector-specific query path templates + if "{query}" in path: + return path.format(query=urllib.parse.quote(query.strip() or "oncology")) + if "?" in path: + return f"{base_url.rstrip('/')}/{path.lstrip('/')}" + return f"{base_url.rstrip('/')}/{path.lstrip('/')}?{urllib.parse.urlencode(params)}" + + +def _parse_detail_urls(html_text: str, base_url: str) -> list[str]: + urls = extract_search_result_urls(html_text, base_url=base_url) + detail_urls = [] + for url in urls: + parsed = urlparse(url) + if parsed.scheme in {"http", "https"} and url_allowed_domain(url, [urlparse(base_url).hostname or ""]): + detail_urls.append(url) + return detail_urls + + +def _load_json_payload(text: str) -> dict[str, Any] | None: + try: + return json.loads(text) + except (json.JSONDecodeError, TypeError): + return None + + +def _normalize_detail_document(html_text: str, url: str, source: str, limitation: str | None = None) -> dict[str, Any]: + metadata = extract_page_metadata(html_text, url) + if limitation: + metadata.setdefault("limitation", limitation) + return { + "title": metadata.get("title"), + "abstract": metadata.get("abstract"), + "content": metadata.get("content"), + "authors": metadata.get("authors") or [], + "published_date": metadata.get("published_date"), + "source": source, + "source_id": url, + "doi": None, + "url": url, + "journal": source, + "metadata": metadata, + } + + +def _build_search_documents(search_html: str, base_url: str, source: str, limitation: str | None = None) -> list[dict[str, Any]]: + metadata = extract_page_metadata(search_html, base_url) + content_value = metadata.get("content") + docs = [ + { + "title": metadata.get("title") or f"{source.upper()} search results", + "abstract": metadata.get("abstract") or metadata.get("meta", {}).get("description"), + "content": content_value[:4000] if isinstance(content_value, str) else None, + "authors": metadata.get("authors") or [], + "published_date": metadata.get("published_date"), + "source": source, + "source_id": base_url, + "doi": None, + "url": base_url, + "journal": f"{source.upper()} search", + "metadata": {"query_url": base_url, "limitation": limitation, "meta": metadata.get("meta", {})}, + } + ] + return docs + + +def _fetch_detail_documents(connector: BaseConnector, search_html: str, base_url: str, source: str, limitation: str | None, max_results: int) -> list[dict[str, Any]]: + detail_urls = _parse_detail_urls(search_html, base_url) + documents: list[dict[str, Any]] = [] + for url in detail_urls: + html = connector._request_text(url) + if not html or is_access_restricted(html): + continue + documents.append(connector.normalize_document(_normalize_detail_document(html, url, source, limitation))) + if len(documents) >= max_results: + break + return documents + + +def _fetch_html_search_documents( + connector: BaseConnector, + search_html: str, + search_url: str, + source: str, + limitation: str | None, + max_results: int, +) -> list[dict[str, Any]]: + detail_urls = _parse_detail_urls(search_html, search_url) + documents: list[dict[str, Any]] = [] + for url in detail_urls: + html = connector._request_text(url) + if not html or is_access_restricted(html): + continue + documents.append(connector.normalize_document(_normalize_detail_document(html, url, source, limitation))) + if len(documents) >= max_results: + break + return documents + + +def _fetch_search_result_pages(connector: BaseConnector, search_url: str, source: str, query: str, limitation: str | None, max_results: int) -> list[dict[str, Any]]: + html = connector._request_text(search_url) + if not html: + return [] + if is_access_restricted(html): + return _build_search_documents(html, search_url, source, limitation) + + detail_documents = _fetch_detail_documents(connector, html, search_url, source, limitation, max_results) + if detail_documents: + return detail_documents + + return _build_search_documents(html, search_url, source, limitation) + + +class PubMedConnector(BaseConnector): + """PubMed connector using NCBI E-utilities search and summary endpoints.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "pubmed" + self.base_url = self.config.get("base_url", "https://eutils.ncbi.nlm.nih.gov/entrez/eutils").rstrip("/") + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + params = { + "db": "pubmed", + "term": query.strip() or "oncology", + "retmode": "json", + "retmax": kwargs.get("max_results", 5), + } + search = self._request_json(f"{self.base_url}/esearch.fcgi", params=params) + id_list = (search or {}).get("esearchresult", {}).get("idlist", []) + if not id_list: + return [] + + summary = self._request_json( + f"{self.base_url}/esummary.fcgi", + params={"db": "pubmed", "id": ",".join(id_list), "retmode": "json"}, + ) + result_map = (summary or {}).get("result", {}) + documents: list[dict[str, Any]] = [] + for pmid in id_list: + item = result_map.get(str(pmid), {}) + if not item: + continue + authors = [author.get("name") for author in item.get("authors", []) if isinstance(author, dict) and author.get("name")] + documents.append( + self.normalize_document( + { + "title": item.get("title"), + "abstract": item.get("sortfirstauthor"), + "source": self.name, + "source_id": f"PMID:{pmid}", + "authors": authors, + "published_date": item.get("pubdate"), + "doi": item.get("elocationid"), + "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/", + "journal": item.get("source"), + "metadata": {"pmid": pmid, "query": params["term"]}, + } + ) + ) + return documents + + +class PMCConnector(BaseConnector): + """PubMed Central connector using NCBI E-utilities search and summary endpoints.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "pmc" + self.base_url = self.config.get("base_url", "https://eutils.ncbi.nlm.nih.gov/entrez/eutils").rstrip("/") + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + params = { + "db": "pmc", + "term": query.strip() or "oncology", + "retmode": "json", + "retmax": kwargs.get("max_results", 5), + } + search = self._request_json(f"{self.base_url}/esearch.fcgi", params=params) + id_list = (search or {}).get("esearchresult", {}).get("idlist", []) + if not id_list: + return [] + + summary = self._request_json( + f"{self.base_url}/esummary.fcgi", + params={"db": "pmc", "id": ",".join(id_list), "retmode": "json"}, + ) + result_map = (summary or {}).get("result", {}) + documents: list[dict[str, Any]] = [] + for pmc_id in id_list: + item = result_map.get(str(pmc_id), {}) + if not item: + continue + documents.append( + self.normalize_document( + { + "title": item.get("title"), + "abstract": item.get("sortfirstauthor"), + "source": self.name, + "source_id": f"PMC{pmc_id}", + "authors": [author.get("name") for author in item.get("authors", []) if isinstance(author, dict) and author.get("name")], + "published_date": item.get("pubdate"), + "url": f"https://www.ncbi.nlm.nih.gov/pmc/articles/PMC{pmc_id}/", + "journal": item.get("source"), + "metadata": {"pmc_id": pmc_id, "query": params["term"]}, + } + ) + ) + return documents + + +class ClinicalTrialsConnector(BaseConnector): + """ClinicalTrials.gov REST API connector.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "clinicaltrials" + self.base_url = self.config.get("base_url", "https://clinicaltrials.gov/api/v2/studies") + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + payload = self._request_json( + self.base_url, + params={"query.term": query.strip() or "oncology", "pageSize": kwargs.get("max_results", 5)}, + ) + studies = (payload or {}).get("studies", []) + documents: list[dict[str, Any]] = [] + for study in studies: + protocol = study.get("protocolSection", {}) + ident = protocol.get("identificationModule", {}) + desc = protocol.get("descriptionModule", {}) + status_mod = protocol.get("statusModule", {}) + nct_id = ident.get("nctId") + if not nct_id: + continue + documents.append( + self.normalize_document( + { + "title": ident.get("briefTitle"), + "abstract": desc.get("briefSummary"), + "content": desc.get("detailedDescription"), + "source": self.name, + "source_id": nct_id, + "authors": [ident.get("organization", {}).get("fullName")] if ident.get("organization") else [], + "published_date": status_mod.get("startDateStruct", {}).get("date"), + "url": f"https://clinicaltrials.gov/study/{nct_id}", + "journal": "ClinicalTrials.gov", + "metadata": { + "overallStatus": status_mod.get("overallStatus"), + "studyType": protocol.get("designModule", {}).get("studyType"), + "query": query.strip() or "oncology", + }, + } + ) + ) + return documents + + +class AACRConnector(BaseConnector): + """AACR search-page connector with documented publisher access limits.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "aacr" + self.base_url = self.config.get("base_url", "https://aacrjournals.org").rstrip("/") + + def get_limitation(self) -> str | None: + return "AACR abstracts and proceedings may require institutional access for full-text retrieval." + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + search_url = _resolve_search_url(self.base_url, query, "/search-results?q={query}") + return _fetch_search_result_pages(self, search_url, self.name, query, self.get_limitation(), kwargs.get("max_results", 5)) + + +class ASCOConnector(BaseConnector): + """ASCO search-page connector with documented publisher access limits.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "asco" + self.base_url = self.config.get("base_url", "https://ascopubs.org").rstrip("/") + + def get_limitation(self) -> str | None: + return "ASCO abstracts are available through publisher properties with access and reuse constraints." + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + search_url = _resolve_search_url(self.base_url, query, "/action/doSearch?AllField={query}") + return _fetch_search_result_pages(self, search_url, self.name, query, self.get_limitation(), kwargs.get("max_results", 5)) + + +class SABCSConnector(BaseConnector): + """SABCS conference-site connector with access limitations.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "sabcs" + self.base_url = self.config.get("base_url", "https://www.sabcs.org").rstrip("/") + + def get_limitation(self) -> str | None: + return "SABCS detailed abstracts may require conference registration or licensed access." + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + search_url = _resolve_search_url(self.base_url, query, "/?s={query}") + return _fetch_search_result_pages(self, search_url, self.name, query, self.get_limitation(), kwargs.get("max_results", 5)) + + +class ESMOConnector(BaseConnector): + """ESMO search-page connector with member-library limitations.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "esmo" + self.base_url = self.config.get("base_url", "https://www.esmo.org").rstrip("/") + + def get_limitation(self) -> str | None: + return "ESMO member content and some conference materials require authenticated access." + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + search_url = _resolve_search_url(self.base_url, query, "/search?searchText={query}") + return _fetch_search_result_pages(self, search_url, self.name, query, self.get_limitation(), kwargs.get("max_results", 5)) + + +class BioRxivConnector(BaseConnector): + """bioRxiv connector via the public bioRxiv API.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "biorxiv" + self.base_url = self.config.get("base_url", "https://api.biorxiv.org/details/biorxiv").rstrip("/") + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + payload = self._request_json(f"{self.base_url}/2024-01-01/2024-12-31/0/json") + collection = (payload or {}).get("collection", []) + documents: list[dict[str, Any]] = [] + for paper in collection: + title = paper.get("title") + abstract = paper.get("abstract") + if not _matches_query(query, title, abstract): + continue + doi = paper.get("doi") + documents.append( + self.normalize_document( + { + "title": title, + "abstract": abstract, + "source": self.name, + "source_id": doi or paper.get("version"), + "authors": [part.strip() for part in str(paper.get("authors") or "").split(";") if part.strip()], + "published_date": paper.get("date"), + "doi": doi, + "url": f"https://www.biorxiv.org/content/{doi}" if doi else None, + "journal": "bioRxiv", + "metadata": {"query": query, "category": paper.get("category")}, + } + ) + ) + if len(documents) >= kwargs.get("max_results", 5): + break + return documents + + +class MedRxivConnector(BaseConnector): + """medRxiv connector via the public medRxiv API.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "medrxiv" + self.base_url = self.config.get("base_url", "https://api.biorxiv.org/details/medrxiv").rstrip("/") + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + payload = self._request_json(f"{self.base_url}/2024-01-01/2024-12-31/0/json") + collection = (payload or {}).get("collection", []) + documents: list[dict[str, Any]] = [] + for paper in collection: + title = paper.get("title") + abstract = paper.get("abstract") + if not _matches_query(query, title, abstract): + continue + doi = paper.get("doi") + documents.append( + self.normalize_document( + { + "title": title, + "abstract": abstract, + "source": self.name, + "source_id": doi or paper.get("version"), + "authors": [part.strip() for part in str(paper.get("authors") or "").split(";") if part.strip()], + "published_date": paper.get("date"), + "doi": doi, + "url": f"https://www.medrxiv.org/content/{doi}" if doi else None, + "journal": "medRxiv", + "metadata": {"query": query, "category": paper.get("category")}, + } + ) + ) + if len(documents) >= kwargs.get("max_results", 5): + break + return documents + + +class PatentsConnector(BaseConnector): + """Patent connector with public-search fallback and documented API-key limits.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "patents" + self.base_url = self.config.get("base_url", "https://api.patentsview.org/patents/query").rstrip("/") + + def get_limitation(self) -> str | None: + return "High-volume or structured patent retrieval requires a licensed API such as USPTO or Lens credentials." + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + max_results = kwargs.get("max_results", 5) + query_text = query.strip() or "pharmaceutical" + params = { + "q": json.dumps({"_text_any": {"patent_title": query_text}}), + "f": json.dumps([ + "patent_number", + "patent_title", + "patent_abstract", + "patent_date", + "patent_type", + "patent_current_assignee_organization", + "inventor_last_name", + ]), + "o": json.dumps({"per_page": max_results, "page": 1}), + } + payload = self._request_json(self.base_url, params=params) + patents = (payload or {}).get("patents", []) + documents: list[dict[str, Any]] = [] + for patent in patents: + patent_number = patent.get("patent_number") + documents.append( + self.normalize_document( + { + "title": patent.get("patent_title"), + "abstract": patent.get("patent_abstract"), + "content": patent.get("patent_abstract"), + "source": self.name, + "source_id": patent_number, + "authors": [patent.get("inventor_last_name")] if patent.get("inventor_last_name") else [], + "published_date": patent.get("patent_date"), + "url": f"https://patents.google.com/patent/{patent_number}" if patent_number else None, + "journal": "PatentsView", + "metadata": {"query": query, "patent_type": patent.get("patent_type")}, + } + ) + ) + if len(documents) >= max_results: + break + + if documents: + return documents + + search_url = f"https://patents.google.com/?q={urllib.parse.quote(query_text)}" + html = self._request_text(search_url) + if not html: + return [] + return [ + self.normalize_document( + { + "title": self._extract_html_title(html) or "Patent search results", + "abstract": self._html_to_text(html)[:1200], + "content": self._html_to_text(html)[:4000], + "source": self.name, + "source_id": search_url, + "url": search_url, + "journal": "Patent search", + "metadata": {"query": query, "limitation": self.get_limitation()}, + } + ) + ] + + +class CompanyWebsiteConnector(BaseConnector): + """Company website connector for configured press-release or IR URLs.""" + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self.name = "company_websites" + self.crawler_config = { + "user_agent": self.user_agent, + "timeout": float(self.config.get("crawler_timeout", self.timeout)), + "rate_limit": float(self.config.get("crawler_rate_limit", 0.5)), + "max_pages": int(self.config.get("crawler_max_pages", 5)), + "max_depth": int(self.config.get("crawler_max_depth", 1)), + "allowed_domains": self.config.get("allowed_domains", []), + } + + def get_limitation(self) -> str | None: + return "Company website ingestion requires explicitly configured public URLs; broad autonomous crawling is intentionally not enabled." + + def fetch(self, query: str, **kwargs: Any) -> list[dict[str, Any]]: + raw_urls = kwargs.get("urls") or self.config.get("urls") or [] + if isinstance(raw_urls, str): + raw_urls = [raw_urls] + + crawler = WebCrawler(self.crawler_config) + results = crawler.crawl([normalize_url(url) for url in raw_urls if normalize_url(url)]) + documents: list[dict[str, Any]] = [] + + for result in results: + if not result.success or not result.html: + continue + title = result.metadata.get("title") or self._extract_html_title(result.html) or result.url + content = result.metadata.get("content") or self._html_to_text(result.html) + if not _matches_query(query, title, content): + continue + documents.append( + self.normalize_document( + { + "title": title, + "abstract": content[:1200], + "content": content[:4000], + "source": self.name, + "source_id": result.url, + "url": result.url, + "journal": "Company website", + "metadata": {"query": query, "limitation": self.get_limitation(), **result.metadata}, + } + ) + ) + if len(documents) >= kwargs.get("max_results", 5): + break + + return documents diff --git a/services/literature/app/core/config.py b/services/literature/app/core/config.py index 12b1283..716fea1 100644 --- a/services/literature/app/core/config.py +++ b/services/literature/app/core/config.py @@ -15,25 +15,43 @@ class Settings(BaseSettings): neo4j_user: str = "neo4j" neo4j_password: str = "changeme_neo4j" opensearch_url: str = "http://opensearch:9200" - search_service_url: str = "http://search:8084" - search_service_timeout_seconds: int = 5 - search_service_max_retries: int = 3 - kg_service_url: str = "http://kg:8083" - kg_service_timeout_seconds: int = 5 - kg_service_max_retries: int = 3 - llmwiki_service_url: str = "http://llmwiki:8086" - llmwiki_service_timeout_seconds: int = 5 - llmwiki_service_max_retries: int = 3 - jwt_secret: str = "changeme_secret" - cors_allowed_origins: list[str] = ["http://localhost:3000"] - - pubmed_base_url: str = "https://api.ncbi.nlm.nih.gov/lit/ctxp/v1/pubmed/" - pmc_base_url: str = "https://api.ncbi.nlm.nih.gov/lit/ctxp/v1/pmc/" - clinicaltrials_base_url: str = "https://clinicaltrials.gov/api/query" - biorxiv_base_url: str = "https://api.biorxiv.org" - medrxiv_base_url: str = "https://api.biorxiv.org" + jwt_secret: str = "change_this_dev_secret_before_deploying" + + kg_service_url: str = "http://kg:8000" + okf_wiki_url: str | None = None + okf_wiki_dir: str = "wiki-root" + + llm_provider: str | None = None + llm_api_key: str | None = None + llm_api_url: str | None = None + llm_model: str = "gpt-3.5-turbo" + llm_timeout: float = 10.0 + llm_max_retries: int = 1 + llm_backoff_seconds: float = 0.25 + + ner_provider: str = "rule_based" + ner_model: str = "en_core_web_sm" + ner_timeout: float = 10.0 + + kg_timeout: float = 5.0 + kg_max_retries: int = 1 + kg_backoff_seconds: float = 0.25 + + wiki_api_key: str | None = None + wiki_timeout: float = 5.0 + wiki_max_retries: int = 1 + wiki_backoff_seconds: float = 0.25 + + crawler_user_agent: str = "AI-RxOS LiteratureBot/1.0" + crawler_timeout: float = 10.0 + crawler_rate_limit: float = 0.5 + crawler_max_pages: int = 10 + crawler_max_depth: int = 2 + crawler_allowed_domains: list[str] = [] + + cors_origins: list[str] = ["*"] @lru_cache def get_settings() -> Settings: - return Settings() # type: ignore[call-arg] + return Settings() diff --git a/services/literature/app/crawling/__init__.py b/services/literature/app/crawling/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/literature/app/crawling/crawler.py b/services/literature/app/crawling/crawler.py new file mode 100644 index 0000000..e1d0605 --- /dev/null +++ b/services/literature/app/crawling/crawler.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +import logging +import time +from collections import deque +from dataclasses import dataclass, field +from typing import Any +from urllib.parse import urlparse + +import httpx + +from app.utils.html_utils import ( + extract_page_metadata, + is_access_restricted, + normalize_url, + url_allowed_domain, +) + +logger = logging.getLogger(__name__) + + +@dataclass +class CrawlResult: + url: str + success: bool + status_code: int | None = None + error: str | None = None + metadata: dict[str, Any] = field(default_factory=dict) + html: str | None = None + + +class WebCrawler: + def __init__(self, config: dict[str, Any] | None = None): + self.config = config or {} + self.timeout = float(self.config.get("timeout", 10.0)) + self.max_pages = int(self.config.get("max_pages", 10)) + self.max_depth = int(self.config.get("max_depth", 2)) + self.rate_limit = float(self.config.get("rate_limit", 0.5)) + self.user_agent = str(self.config.get("user_agent", "AI-RxOS LiteratureBot/1.0")) + self.allowed_domains = [d.lower() for d in self.config.get("allowed_domains", []) if d] + self.visited: set[str] = set() + self.queue: deque[tuple[str, int]] = deque() + self.results: list[CrawlResult] = [] + self.last_request_at = 0.0 + self.robots_cache: dict[str, bool] = {} + + def enqueue(self, url: str, depth: int = 0) -> None: + normalized = normalize_url(url) + if not normalized: + return + if normalized in self.visited: + return + if not url_allowed_domain(normalized, self.allowed_domains): + logger.debug("URL outside allowed domains: %s", normalized) + return + self.visited.add(normalized) + self.queue.append((normalized, depth)) + + def _can_fetch(self, url: str) -> bool: + parsed = urlparse(url) + domain = parsed.netloc.lower() + if domain in self.robots_cache: + return self.robots_cache[domain] + robots_url = f"{parsed.scheme}://{domain}/robots.txt" + try: + response = httpx.get(robots_url, timeout=self.timeout, headers={"User-Agent": self.user_agent}) + if response.status_code == 200 and "Disallow: /" in response.text: + logger.info("Robots blocked crawling of domain %s", domain) + self.robots_cache[domain] = False + return False + self.robots_cache[domain] = True + return True + except httpx.HTTPError as exc: + logger.warning("Failed to fetch robots.txt for %s: %s", domain, exc) + self.robots_cache[domain] = True + return True + + def _wait_rate_limit(self) -> None: + now = time.time() + since = now - self.last_request_at + if since < self.rate_limit: + time.sleep(self.rate_limit - since) + self.last_request_at = time.time() + + def crawl(self, seed_urls: list[str]) -> list[CrawlResult]: + for seed in seed_urls: + self.enqueue(seed, 0) + + while self.queue and len(self.results) < self.max_pages: + url, depth = self.queue.popleft() + if depth > self.max_depth: + logger.debug("Skipping %s due to depth limit", url) + continue + if not self._can_fetch(url): + self.results.append(CrawlResult(url=url, success=False, error="disallowed_by_robots")) + continue + self._wait_rate_limit() + try: + response = httpx.get(url, timeout=self.timeout, headers={"User-Agent": self.user_agent}) + if response.status_code != 200: + self.results.append(CrawlResult(url=url, success=False, status_code=response.status_code, error="http_error")) + continue + html_text = response.text + if is_access_restricted(html_text): + self.results.append(CrawlResult(url=url, success=False, status_code=response.status_code, error="access_restricted")) + continue + metadata = extract_page_metadata(html_text, url) + metadata["source_url"] = url + metadata["depth"] = depth + result = CrawlResult(url=url, success=True, status_code=response.status_code, metadata=metadata, html=html_text) + self.results.append(result) + if depth < self.max_depth: + from app.utils.html_utils import extract_search_result_urls + for link in extract_search_result_urls(html_text, base_url=url): + if len(self.visited) >= self.max_pages: + break + self.enqueue(link, depth + 1) + except httpx.HTTPError as exc: + self.results.append(CrawlResult(url=url, success=False, error=str(exc))) + except (ValueError, TypeError, RuntimeError) as exc: + self.results.append(CrawlResult(url=url, success=False, error=f"parse_error:{exc}")) + return self.results diff --git a/services/literature/app/database/models.py b/services/literature/app/database/models.py new file mode 100644 index 0000000..af300cf --- /dev/null +++ b/services/literature/app/database/models.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from threading import Lock +from typing import Any, Literal + +from pydantic import BaseModel, Field + +JobStatus = Literal[ + "pending", + "running", + "parsing", + "processing", + "completed", + "failed", + "retrying", + "dead_letter", +] + + +class IngestionJobState(BaseModel): + id: str + source: str + query: str + status: JobStatus = "pending" + created_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + updated_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + attempts: int = 0 + max_retries: int = 3 + error: str | None = None + result: dict[str, Any] | None = None + + def transition(self, new_status: JobStatus, error: str | None = None, result: dict[str, Any] | None = None) -> None: + self.status = new_status + self.updated_at = datetime.now(timezone.utc).isoformat() + if error: + self.error = error + if result: + self.result = result + + +class JobStore: + """Thread-safe job state store with lightweight JSON persistence.""" + + def __init__(self): + self._lock = Lock() + self._jobs: dict[str, IngestionJobState] = {} + self._storage_path = Path(os.environ.get("LITERATURE_JOB_STORE_PATH", "data/ingestion_jobs.json")) + self._load() + + def _load(self) -> None: + if not self._storage_path.exists(): + return + try: + payload = json.loads(self._storage_path.read_text(encoding="utf-8")) + if not isinstance(payload, list): + return + for item in payload: + if isinstance(item, dict): + job = IngestionJobState.model_validate(item) + self._jobs[job.id] = job + except (OSError, ValueError, TypeError): + self._jobs = {} + + def _persist(self) -> None: + self._storage_path.parent.mkdir(parents=True, exist_ok=True) + serialized = [job.model_dump() for job in self._jobs.values()] + self._storage_path.write_text(json.dumps(serialized, indent=2), encoding="utf-8") + + def save(self, job: IngestionJobState) -> IngestionJobState: + with self._lock: + self._jobs[job.id] = job + self._persist() + return job + + def get(self, job_id: str) -> IngestionJobState | None: + with self._lock: + return self._jobs.get(job_id) + + def list_all(self) -> list[IngestionJobState]: + with self._lock: + return list(self._jobs.values()) + + +job_store = JobStore() diff --git a/services/literature/app/integrations/kg_client.py b/services/literature/app/integrations/kg_client.py new file mode 100644 index 0000000..87c51d9 --- /dev/null +++ b/services/literature/app/integrations/kg_client.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import logging +import time +from typing import Any + +import httpx + +from app.observability.metrics import metrics + +logger = logging.getLogger(__name__) + + +class KGClient: + """Integration boundary with the AI-RxOS Knowledge Graph service (`services/kg`).""" + + def __init__(self, config: dict[str, Any] | None = None): + self.config = config or {} + self.base_url = self.config.get("kg_service_url", "http://localhost:8001").rstrip("/") + self.timeout = float(self.config.get("kg_timeout", 5.0)) + self.max_retries = int(self.config.get("kg_max_retries", 1)) + self.backoff_seconds = float(self.config.get("kg_backoff_seconds", 0.25)) + + def update_knowledge_graph(self, entities: list[dict[str, Any]], relationships: list[dict[str, Any]]) -> dict[str, Any]: + """Convert extracted entities and relationships into KG update operations.""" + nodes_payload = [ + { + "label": e.get("label", "Entity").capitalize(), + "properties": { + "name": e.get("text"), + "category": e.get("category"), + "source": "literature_service", + }, + } + for e in entities + if e.get("text") + ] + + edges_payload = [ + { + "subject": r.get("subject"), + "predicate": r.get("predicate", "RELATED_TO").upper(), + "object": r.get("object"), + "properties": { + "confidence": r.get("confidence", 0.8), + "evidence": r.get("evidence", ""), + }, + } + for r in relationships + if r.get("subject") and r.get("object") + ] + + if not nodes_payload and not edges_payload: + return {"success": True, "updated_nodes": 0, "updated_edges": 0, "status": "no_op"} + + for attempt in range(self.max_retries + 1): + try: + with httpx.Client(timeout=self.timeout) as client: + res = client.post(f"{self.base_url}/api/v1/graph/import", json={"nodes": nodes_payload, "relationships": edges_payload}) + if res.status_code in (200, 201, 202): + metrics.increment("literature.kg_update.success") + return { + "success": True, + "updated_nodes": len(nodes_payload), + "updated_edges": len(edges_payload), + "status": "completed", + } + if res.status_code in (429, 500, 502, 503, 504) and attempt < self.max_retries: + delay = self.backoff_seconds * (2**attempt) + logger.warning("KG update retryable status %s on attempt %d, sleeping %.2fs", res.status_code, attempt + 1, delay) + time.sleep(delay) + continue + msg = f"KG service returned status {res.status_code}: {res.text}" + logger.warning("KG update rejected: %s", msg) + metrics.increment("literature.kg_update.failure") + return {"success": False, "error": msg, "retry_eligible": True, "status": "failed"} + except (httpx.HTTPError, RuntimeError, KeyError, TypeError, ValueError, OSError) as exc: + if attempt < self.max_retries: + delay = self.backoff_seconds * (2**attempt) + logger.warning("KG update HTTP failure attempt %d, retrying in %.2fs: %s", attempt + 1, delay, exc) + time.sleep(delay) + continue + msg = f"KG service unavailable: {exc}" + logger.warning(msg) + metrics.increment("literature.kg_update.failure") + return {"success": False, "error": msg, "retry_eligible": True, "status": "failed"} + + msg = "KG update did not complete after retries" + logger.warning(msg) + metrics.increment("literature.kg_update.failure") + return {"success": False, "error": msg, "retry_eligible": True, "status": "failed"} diff --git a/services/literature/app/integrations/wiki_client.py b/services/literature/app/integrations/wiki_client.py new file mode 100644 index 0000000..f6890b1 --- /dev/null +++ b/services/literature/app/integrations/wiki_client.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import logging +import os +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import httpx + +from app.observability.metrics import metrics + +logger = logging.getLogger(__name__) + + +class LLMWikiClient: + """Integration boundary for updating the Open Knowledge Format (OKF) LLM Wiki.""" + + def __init__(self, config: dict[str, Any] | None = None): + self.config = config or {} + self.wiki_dir = Path(self.config.get("wiki_dir", os.environ.get("OKF_WIKI_DIR", "wiki-root"))) + self.service_url = self.config.get("wiki_service_url", os.environ.get("OKF_WIKI_URL")) + self.wiki_api_key = self.config.get("wiki_api_key") + self.timeout = float(self.config.get("wiki_timeout", 5.0)) + self.max_retries = int(self.config.get("wiki_max_retries", 1)) + self.backoff_seconds = float(self.config.get("wiki_backoff_seconds", 0.25)) + + def update_wiki(self, document: dict[str, Any], entities: list[dict[str, Any]], summary: dict[str, Any]) -> dict[str, Any]: + """Compile literature intelligence into OKF markdown wiki concepts.""" + if self.service_url: + headers = {"Content-Type": "application/json"} + if self.wiki_api_key: + headers["Authorization"] = f"Bearer {self.wiki_api_key}" + + for attempt in range(self.max_retries + 1): + try: + with httpx.Client(timeout=self.timeout) as client: + res = client.post(f"{self.service_url.rstrip('/')}/api/v1/wiki/compile", json={"document": document, "entities": entities, "summary": summary}, headers=headers) + if res.status_code in (200, 201): + metrics.increment("literature.wiki_update.success") + return {"success": True, "method": "http", "status": "completed"} + if res.status_code in (429, 500, 502, 503, 504) and attempt < self.max_retries: + delay = self.backoff_seconds * (2**attempt) + logger.warning("Wiki HTTP retryable status %s on attempt %d, sleeping %.2fs", res.status_code, attempt + 1, delay) + time.sleep(delay) + continue + logger.warning("Wiki HTTP update failed: %s", res.text) + metrics.increment("literature.wiki_update.failure") + return {"success": False, "error": res.text, "retry_eligible": res.status_code >= 500 or res.status_code == 429, "status": "failed"} + except (httpx.HTTPError, RuntimeError, KeyError, TypeError, ValueError, OSError) as exc: + if attempt < self.max_retries: + delay = self.backoff_seconds * (2**attempt) + logger.warning("Wiki HTTP service unreachable on attempt %d, retrying in %.2fs: %s", attempt + 1, delay, exc) + time.sleep(delay) + continue + logger.warning("Wiki HTTP service unreachable: %s", exc) + metrics.increment("literature.wiki_update.failure") + return {"success": False, "error": str(exc), "retry_eligible": True, "status": "failed"} + + # Direct OKF volume update fallback + try: + wiki_path = self.wiki_dir / "wiki" + wiki_path.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now(timezone.utc).isoformat() + + updated_files: list[str] = [] + for entity in entities: + text = entity.get("text") + category = entity.get("category", "concepts") + if not text: + continue + category_dir = wiki_path / category + category_dir.mkdir(parents=True, exist_ok=True) + file_path = category_dir / f"{text.replace(' ', '_')}.md" + + content = f"""# Concept: {text} +- **Category**: {category} +- **Last Updated**: {timestamp} +- **Source**: {document.get('source')} ({document.get('source_id')}) + +## Primary Summary +{summary.get('concise_summary', 'No summary provided.')} + +## Literature Evidence +- **Title**: {document.get('title')} +- **DOI**: {document.get('doi')} +- **URL**: {document.get('url')} +""" + file_path.write_text(content, encoding="utf-8") + updated_files.append(str(file_path)) + + # Update log.md + log_file = wiki_path / "log.md" + log_entry = f"- [{timestamp}] Updated {len(updated_files)} concept pages from {document.get('source')}:{document.get('source_id')}\n" + with open(log_file, "a", encoding="utf-8") as f: + f.write(log_entry) + + metrics.increment("literature.wiki_update.success") + return { + "success": True, + "method": "okf_volume", + "updated_concepts": len(updated_files), + "status": "completed", + } + except (OSError, RuntimeError, ValueError, KeyError) as exc: + logger.warning("OKF Wiki volume write failed: %s", exc) + metrics.increment("literature.wiki_update.failure") + return {"success": False, "error": str(exc), "retry_eligible": True, "status": "failed"} diff --git a/services/literature/app/main.py b/services/literature/app/main.py index 37c373d..ca0b6d4 100644 --- a/services/literature/app/main.py +++ b/services/literature/app/main.py @@ -1,82 +1,173 @@ -from __future__ import annotations - -import time import uuid +from typing import Literal -from fastapi import FastAPI, Request +from fastapi import FastAPI, Request, Response from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel from app.core.config import get_settings -from app.core.lifespan import lifespan -from app.observability.metrics import ( - REQUEST_COUNT, - REQUEST_ERRORS, - REQUEST_IN_FLIGHT, - REQUEST_LATENCY_SECONDS, -) -from app.routers.documents import router as documents_router -from app.routers.health import router as health_router -from app.routers.ingestion import router as ingestion_router -from app.routers.nlp import router as nlp_router -from app.routers.papers import router as papers_router -from app.utils.logging import get_logger, set_request_context +from app.database.models import IngestionJobState, job_store +from app.observability.metrics import metrics +from app.services.literature_service import LiteratureService settings = get_settings() -logger = get_logger(__name__) +literature_service = LiteratureService( + { + "kg_service_url": settings.kg_service_url, + "kg_timeout": settings.kg_timeout, + "kg_max_retries": settings.kg_max_retries, + "kg_backoff_seconds": settings.kg_backoff_seconds, + "wiki_service_url": settings.okf_wiki_url, + "wiki_api_key": settings.wiki_api_key, + "wiki_dir": settings.okf_wiki_dir, + "wiki_timeout": settings.wiki_timeout, + "wiki_max_retries": settings.wiki_max_retries, + "wiki_backoff_seconds": settings.wiki_backoff_seconds, + "llm_provider": settings.llm_provider, + "llm_api_key": settings.llm_api_key, + "llm_api_url": settings.llm_api_url, + "llm_model": settings.llm_model, + "llm_timeout": settings.llm_timeout, + "llm_max_retries": settings.llm_max_retries, + "llm_backoff_seconds": settings.llm_backoff_seconds, + "ner_provider": settings.ner_provider, + "ner_model": settings.ner_model, + "ner_timeout": settings.ner_timeout, + "crawler_user_agent": settings.crawler_user_agent, + "crawler_timeout": settings.crawler_timeout, + "crawler_rate_limit": settings.crawler_rate_limit, + "crawler_max_pages": settings.crawler_max_pages, + "crawler_max_depth": settings.crawler_max_depth, + "crawler_allowed_domains": settings.crawler_allowed_domains, + } +) app = FastAPI( title="AI-RxOS Literature Service", - description="Ingestion, extraction, and citation services for the Literature Intelligence bounded context.", - version="0.1.0", - lifespan=lifespan, + description="Ingestion, extraction, knowledge graph, and citation services for the Literature Intelligence bounded context.", + version="0.2.0", ) app.add_middleware( CORSMiddleware, - allow_origins=settings.cors_allowed_origins, + allow_origins=settings.cors_origins, allow_credentials=True, - allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"], + allow_methods=["*"], allow_headers=["*"], ) @app.middleware("http") -async def add_request_context(request: Request, call_next): - request_id = request.headers.get("x-request-id") or str(uuid.uuid4()) - trace_id = request.headers.get("x-trace-id") or str(uuid.uuid4()) - request.state.request_id = request_id - request.state.trace_id = trace_id - set_request_context(request_id=request_id, trace_id=trace_id) - - method = request.method - endpoint = request.url.path - REQUEST_IN_FLIGHT.labels(method=method, endpoint=endpoint).inc() - start_time = time.perf_counter() - try: - response = await call_next(request) - status_code = str(response.status_code) - REQUEST_COUNT.labels(method=method, endpoint=endpoint, status=status_code).inc() - if response.status_code >= 400: - REQUEST_ERRORS.labels( - method=method, endpoint=endpoint, status=status_code - ).inc() - response.headers["x-request-id"] = request_id - response.headers["x-trace-id"] = trace_id - return response - except Exception: - REQUEST_COUNT.labels(method=method, endpoint=endpoint, status="500").inc() - REQUEST_ERRORS.labels(method=method, endpoint=endpoint, status="500").inc() - raise - finally: - elapsed = time.perf_counter() - start_time - REQUEST_LATENCY_SECONDS.labels(method=method, endpoint=endpoint).observe( - elapsed - ) - REQUEST_IN_FLIGHT.labels(method=method, endpoint=endpoint).dec() - - -app.include_router(health_router) -app.include_router(papers_router, prefix="/api/v1") -app.include_router(ingestion_router, prefix="/api/v1") -app.include_router(documents_router, prefix="/api/v1") -app.include_router(nlp_router, prefix="/api/v1") +async def trace_id_middleware(request: Request, call_next): + trace_id = request.headers.get("X-Trace-ID") or str(uuid.uuid4()) + response: Response = await call_next(request) + response.headers["X-Trace-ID"] = trace_id + return response + + +_PAPERS: dict[str, dict] = {} + + +class Paper(BaseModel): + id: str + title: str + source: Literal[ + "pubmed", + "pmc", + "clinicaltrials", + "aacr", + "asco", + "sabcs", + "esmo", + "biorxiv", + "medrxiv", + "patents", + "company_websites", + ] + doi: str | None = None + publishedAt: str | None = None + citationCount: int = 0 + + +class IngestionRequest(BaseModel): + source: Literal[ + "pubmed", + "pmc", + "clinicaltrials", + "aacr", + "asco", + "sabcs", + "esmo", + "biorxiv", + "medrxiv", + "patents", + "company_websites", + ] + query: str + + +class IngestionJobResponse(BaseModel): + id: str + source: str + query: str + status: str + createdAt: str + + +@app.get("/healthz") +def health() -> dict[str, str]: + return {"status": "ok", "service": "literature"} + + +@app.get("/metrics") +def get_metrics() -> dict[str, int]: + return metrics.snapshot() + + +@app.get("/api/v1/papers") +def list_papers(page: int = 1, page_size: int = 20) -> dict: + items = list(_PAPERS.values())[(page - 1) * page_size : page * page_size] + return {"items": items, "total": len(_PAPERS), "page": page, "pageSize": page_size} + + +@app.get("/api/v1/papers/{paper_id}") +def get_paper(paper_id: str) -> Paper | dict: + paper = _PAPERS.get(paper_id) + return paper or {"error": "not_found"} + + +@app.post("/api/v1/ingestion", response_model=IngestionJobResponse, status_code=202) +def start_ingestion(req: IngestionRequest) -> IngestionJobResponse: + job_id = str(uuid.uuid4()) + job = IngestionJobState( + id=job_id, + source=req.source, + query=req.query, + status="pending", + ) + job_store.save(job) + + literature_service.ingest(req.source, req.query, job_id=job_id) + + updated_job = job_store.get(job_id) or job + status_output = "completed" if updated_job.status == "completed" else updated_job.status + return IngestionJobResponse( + id=updated_job.id, + source=updated_job.source, + query=updated_job.query, + status=status_output, + createdAt=updated_job.created_at, + ) + + +@app.get("/api/v1/ingestion/{job_id}") +def get_ingestion_job(job_id: str) -> dict: + job = job_store.get(job_id) + if not job: + return {"error": "not_found"} + return job.model_dump() + + +@app.post("/api/v1/analyze") +def analyze_text(text: str) -> dict: + return literature_service.analyze_text(text) diff --git a/services/literature/app/nlp/deduplication.py b/services/literature/app/nlp/deduplication.py new file mode 100644 index 0000000..b8e104a --- /dev/null +++ b/services/literature/app/nlp/deduplication.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import hashlib +import re +from typing import Any + + +class DuplicateDetector: + """Multi-tiered duplicate detector for literature documents.""" + + @staticmethod + def _normalize_string(text: str | None) -> str: + if not text: + return "" + clean = re.sub(r"[^\w\s]", "", str(text)).lower() + return " ".join(clean.split()) + + def detect_duplicate(self, left: dict[str, Any], right: dict[str, Any]) -> dict[str, Any]: + """Detect if two documents are duplicates using 4 identifier tiers.""" + left_doi = (left.get("doi") or "").strip().lower() + right_doi = (right.get("doi") or "").strip().lower() + if left_doi and right_doi and left_doi == right_doi: + return { + "duplicate": True, + "reason": "doi_match", + "tier": 1, + "matched_fields": {"doi": left_doi}, + } + + left_src = (left.get("source") or "").strip().lower() + right_src = (right.get("source") or "").strip().lower() + left_id = (left.get("source_id") or left.get("id") or "").strip().lower() + right_id = (right.get("source_id") or right.get("id") or "").strip().lower() + + if left_src and right_src and left_id and right_id and left_src == right_src and left_id == right_id: + return { + "duplicate": True, + "reason": "source_id_match", + "tier": 2, + "matched_fields": {"source": left_src, "source_id": left_id}, + } + + norm_left_title = self._normalize_string(left.get("title")) + norm_right_title = self._normalize_string(right.get("title")) + + if norm_left_title and norm_right_title and norm_left_title == norm_right_title: + return { + "duplicate": True, + "reason": "title_match", + "tier": 3, + "matched_fields": {"normalized_title": norm_left_title}, + } + + left_content = (left.get("content") or left.get("abstract") or norm_left_title).strip().lower() + right_content = (right.get("content") or right.get("abstract") or norm_right_title).strip().lower() + + if left_content and right_content: + left_hash = hashlib.sha256(left_content.encode("utf-8")).hexdigest() + right_hash = hashlib.sha256(right_content.encode("utf-8")).hexdigest() + + if left_hash == right_hash: + return { + "duplicate": True, + "reason": "content_hash_match", + "tier": 4, + "matched_fields": {"content_sha256": left_hash}, + } + + return { + "duplicate": False, + "reason": "distinct_documents", + "tier": 0, + "matched_fields": {}, + } diff --git a/services/literature/app/nlp/evidence_ranking.py b/services/literature/app/nlp/evidence_ranking.py new file mode 100644 index 0000000..ffadd5e --- /dev/null +++ b/services/literature/app/nlp/evidence_ranking.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, ClassVar + + +class EvidenceRanker: + """Deterministic, structured evidence-ranking component based on metadata.""" + + SOURCE_WEIGHTS: ClassVar[dict[str, float]] = { + "pubmed": 0.90, + "pmc": 0.90, + "clinicaltrials": 0.95, + "aacr": 0.85, + "asco": 0.85, + "sabcs": 0.85, + "esmo": 0.85, + "biorxiv": 0.70, + "medrxiv": 0.70, + "patents": 0.65, + "company_websites": 0.50, + } + + def rank_document(self, document: dict[str, Any]) -> dict[str, Any]: + source = (document.get("source") or "unknown").lower() + source_weight = self.SOURCE_WEIGHTS.get(source, 0.60) + + pub_type_score = 0.70 + journal = str(document.get("journal") or "").lower() + title = str(document.get("title") or "").lower() + abstract = str(document.get("abstract") or "").lower() + metadata = document.get("metadata") or {} + + if "clinical trial" in journal or "clinical trial" in title or source == "clinicaltrials": + pub_type_score = 0.95 + status = metadata.get("overallStatus") or "" + if status.lower() in ("completed", "approved"): + pub_type_score = 1.0 + elif "guideline" in title or "consensus" in title: + pub_type_score = 0.95 + elif "review" in title or "meta-analysis" in abstract: + pub_type_score = 0.85 + elif source in ("biorxiv", "medrxiv"): + pub_type_score = 0.65 + + recency_score = 0.80 + published_date = document.get("published_date") + if published_date: + try: + year = int(str(published_date)[:4]) + current_year = datetime.now(timezone.utc).year + diff = current_year - year + if diff <= 1: + recency_score = 1.0 + elif diff <= 3: + recency_score = 0.90 + elif diff <= 5: + recency_score = 0.80 + else: + recency_score = max(0.40, 0.80 - ((diff - 5) * 0.05)) + except (ValueError, TypeError, KeyError): + recency_score = 0.75 + + peer_reviewed = source not in ("biorxiv", "medrxiv", "company_websites") + peer_review_score = 1.0 if peer_reviewed else 0.70 + + citation_count = int(document.get("citation_count") or metadata.get("citationCount") or 0) + citation_score = min(1.0, 0.50 + (citation_count * 0.05)) + + total_score = round( + (source_weight * 0.35) + + (pub_type_score * 0.25) + + (recency_score * 0.20) + + (peer_review_score * 0.10) + + (citation_score * 0.10), + 4, + ) + + tier = "High" if total_score >= 0.85 else ("Medium" if total_score >= 0.70 else "Low") + + return { + "score": total_score, + "tier": tier, + "breakdown": { + "source_weight": source_weight, + "publication_type_score": pub_type_score, + "recency_score": recency_score, + "peer_review_score": peer_review_score, + "citation_score": citation_score, + }, + "peer_reviewed": peer_reviewed, + "citation_count": citation_count, + } diff --git a/services/literature/app/nlp/relationships.py b/services/literature/app/nlp/relationships.py new file mode 100644 index 0000000..2cbc0f4 --- /dev/null +++ b/services/literature/app/nlp/relationships.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +from typing import Any, ClassVar + + +class RelationshipExtractor: + """Dedicated relationship extraction stage for literature documents.""" + + PREDICATE_PATTERNS: ClassVar[list[tuple[str, str]]] = [ + (r"(?P[\w\s-]+)\s+(?:targets|inhibits|blocks|binds to)\s+(?P[\w\s-]+)", "targets"), + (r"(?P[\w\s-]+)\s+(?:treats|is indicated for|improves outcomes in)\s+(?P[\w\s-]+)", "treats"), + (r"(?P[\w\s-]+)\s+(?:is associated with|correlates with|predicts response to)\s+(?P[\w\s-]+)", "associated_with"), + (r"(?P[\w\s-]+)\s+(?:evaluates|investigates|assesses)\s+(?P[\w\s-]+)", "evaluates"), + (r"(?P[\w\s-]+)\s+(?:reports|demonstrates|shows)\s+(?P[\w\s-]+)", "reports"), + ] + + def extract(self, document: dict[str, Any], entities: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]: + abstract = document.get("abstract") or "" + content = document.get("content") or abstract + title = document.get("title") or "" + combined_text = f"{title}. {abstract} {content}".lower() + + relationships: list[dict[str, Any]] = [] + seen: set[tuple[str, str, str]] = set() + + drugs = [e["text"] for e in (entities or []) if e.get("label") == "drug"] + diseases = [e["text"] for e in (entities or []) if e.get("label") == "disease"] + genes = [e["text"] for e in (entities or []) if e.get("label") == "gene" or e.get("label") == "protein"] + biomarkers = [e["text"] for e in (entities or []) if e.get("label") == "biomarker"] + trials = [e["text"] for e in (entities or []) if e.get("label") == "clinical_trial"] + + for drug in drugs: + for gene in genes: + if drug in combined_text and gene in combined_text: + key = (drug, "targets", gene) + if key not in seen: + seen.add(key) + relationships.append( + { + "subject": drug, + "predicate": "targets", + "object": gene, + "subject_label": "drug", + "object_label": "gene_protein", + "confidence": 0.85, + "evidence": f"Co-occurrence in literature text for {drug} and {gene}.", + } + ) + + for disease in diseases: + if drug in combined_text and disease in combined_text: + key = (drug, "treats", disease) + if key not in seen: + seen.add(key) + relationships.append( + { + "subject": drug, + "predicate": "treats", + "object": disease, + "subject_label": "drug", + "object_label": "disease", + "confidence": 0.85, + "evidence": f"Co-occurrence in document text for {drug} and {disease}.", + } + ) + + for biomarker in biomarkers: + if drug in combined_text and biomarker in combined_text: + key = (drug, "associated_with", biomarker) + if key not in seen: + seen.add(key) + relationships.append( + { + "subject": drug, + "predicate": "associated_with", + "object": biomarker, + "subject_label": "drug", + "object_label": "biomarker", + "confidence": 0.80, + "evidence": f"Association reported between {drug} and {biomarker}.", + } + ) + + for trial in trials: + for drug in drugs: + if trial.lower() in combined_text and drug in combined_text: + key = (trial, "evaluates", drug) + if key not in seen: + seen.add(key) + relationships.append( + { + "subject": trial, + "predicate": "evaluates", + "object": drug, + "subject_label": "clinical_trial", + "object_label": "drug", + "confidence": 0.90, + "evidence": f"Clinical trial {trial} evaluates intervention {drug}.", + "source_document": { + "source": document.get("source"), + "source_id": document.get("source_id"), + "url": document.get("url"), + }, + } + ) + + source_id = document.get("source_id") or document.get("doi") or "literature_doc" + if title: + key = (str(source_id), "reports", title) + if key not in seen: + seen.add(key) + relationships.append( + { + "subject": str(source_id), + "predicate": "reports", + "object": title, + "subject_label": "publication", + "object_label": "finding", + "confidence": 0.95, + "evidence": "Document primary title finding.", + "source_document": { + "source": document.get("source"), + "source_id": document.get("source_id"), + "url": document.get("url"), + }, + } + ) + + return relationships diff --git a/services/literature/app/orchestrator/pipeline.py b/services/literature/app/orchestrator/pipeline.py new file mode 100644 index 0000000..bc5d76f --- /dev/null +++ b/services/literature/app/orchestrator/pipeline.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import logging +import time +from collections.abc import Callable +from typing import Any + +from app.database.models import IngestionJobState, job_store +from app.observability.metrics import metrics + +logger = logging.getLogger(__name__) + +PipelineStage = Callable[[dict[str, Any]], dict[str, Any]] + + +class PipelineRunner: + """Reusable stage-based execution model supporting retries, explicit state transitions, and dead-letter queueing.""" + + def __init__(self, stages: list[PipelineStage] | None = None, retries: int = 2, delay_seconds: float = 0.05): + self.stages = stages or [] + self.retries = retries + self.delay_seconds = delay_seconds + + def run_with_job(self, job: IngestionJobState, initial_payload: dict[str, Any]) -> dict[str, Any]: + job.transition("running") + job_store.save(job) + current = dict(initial_payload) + + try: + for idx, stage in enumerate(self.stages): + stage_name = getattr(stage, "name", getattr(stage, "__name__", f"stage_{idx}")) + if "parse" in stage_name.lower(): + job.transition("parsing") + else: + job.transition("processing") + job_store.save(job) + + stage_success = False + for attempt in range(self.retries + 1): + job.attempts += 1 + try: + current = stage(current) + stage_success = True + break + except Exception as exc: + logger.warning("Pipeline stage %s failed attempt %d: %s", stage_name, attempt + 1, exc) + if attempt < self.retries: + job.transition("retrying", error=str(exc)) + job_store.save(job) + metrics.increment("pipeline.retry") + time.sleep(self.delay_seconds * (2**attempt)) + else: + raise + + if not stage_success: + raise RuntimeError(f"Stage {stage_name} failed after {self.retries + 1} attempts.") + + job.transition("completed", result=current) + job_store.save(job) + metrics.increment("pipeline.completed") + return current + + except Exception as fatal_exc: # noqa: BLE001 + logger.error("Pipeline fatal error for job %s: %s", job.id, fatal_exc) + job.transition("dead_letter", error=str(fatal_exc)) + job_store.save(job) + metrics.increment("pipeline.dead_letter") + current["status"] = "dead_letter" + current["error"] = str(fatal_exc) + return current + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + current = dict(payload) + for stage in self.stages: + for attempt in range(self.retries + 1): + try: + current = stage(current) + break + except Exception: + if attempt >= self.retries: + raise + time.sleep(self.delay_seconds * (2**attempt)) + return current diff --git a/services/literature/app/orchestrator/stages.py b/services/literature/app/orchestrator/stages.py new file mode 100644 index 0000000..c559f2c --- /dev/null +++ b/services/literature/app/orchestrator/stages.py @@ -0,0 +1,192 @@ +from __future__ import annotations + +import logging +from typing import Any + +from app.integrations.kg_client import KGClient +from app.integrations.wiki_client import LLMWikiClient +from app.nlp.pipeline import LiteratureNLP +from app.observability.metrics import metrics + +logger = logging.getLogger(__name__) + + +class PipelineStage: + """Base pipeline stage for reusable, callable stage implementations.""" + + name = "PipelineStage" + + def __call__(self, payload: dict[str, Any]) -> dict[str, Any]: + logger.info("Starting stage %s", self.name) + try: + result = self.run(payload) + logger.info("Completed stage %s", self.name) + metrics.increment(f"pipeline.stage.{self.name}.completed") + return result + except Exception as exc: + logger.error("Stage %s failed: %s", self.name, exc) + metrics.increment(f"pipeline.stage.{self.name}.failed") + raise + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + raise NotImplementedError("PipelineStage subclasses must implement run()") + + +class ParsingStage(PipelineStage): + name = "ParsingStage" + + def __init__(self, nlp: LiteratureNLP): + self.nlp = nlp + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + items = payload.get("items", []) + processed_items: list[dict[str, Any]] = [] + + for item in items: + normalized = self.nlp.parse_document(item) + parsed_structure = self.nlp.parser.parse(normalized.get("content", "")) + processed_items.append( + { + "document": normalized, + "parsed": parsed_structure, + "normalized_text": parsed_structure.get("text", ""), + } + ) + + return {**payload, "processed_items": processed_items} + + +class NERStage(PipelineStage): + name = "NERStage" + + def __init__(self, nlp: LiteratureNLP): + self.nlp = nlp + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + items = payload.get("processed_items", []) + for item in items: + text = item.get("normalized_text", "") + entities = self.nlp.ner.extract_entities(text) + item["entities"] = [entity["text"] for entity in entities] + item["structured_entities"] = entities + return payload + + +class RelationshipExtractionStage(PipelineStage): + name = "RelationshipExtractionStage" + + def __init__(self, nlp: LiteratureNLP): + self.nlp = nlp + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + items = payload.get("processed_items", []) + for item in items: + item["relationships"] = self.nlp.extract_relationships( + {**item.get("document", {}), "entities": item.get("structured_entities", [])} + ) + return payload + + +class SummarizationStage(PipelineStage): + name = "SummarizationStage" + + def __init__(self, nlp: LiteratureNLP): + self.nlp = nlp + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + items = payload.get("processed_items", []) + for item in items: + summary_struct = self.nlp.summarizer.summarize(item.get("document", {})) + item["summary"] = summary_struct.get("concise_summary", "") + item["structured_summary"] = summary_struct + return payload + + +class EvidenceRankingStage(PipelineStage): + name = "EvidenceRankingStage" + + def __init__(self, nlp: LiteratureNLP): + self.nlp = nlp + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + items = payload.get("processed_items", []) + for item in items: + ranking = self.nlp.evidence_ranker.rank_document(item.get("document", {})) + item["evidence_ranking"] = ranking + item["evidence"] = [ + {"entity": e["text"], "category": e.get("category", "unknown"), "score": ranking["score"]} + for e in item.get("structured_entities", []) + ] + if not item["evidence"]: + item["evidence"] = [{"entity": "literature_document", "score": ranking["score"]}] + return payload + + +class DeduplicationStage(PipelineStage): + name = "DeduplicationStage" + + def __init__(self, nlp: LiteratureNLP): + self.nlp = nlp + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + items = payload.get("processed_items", []) + deduped: list[dict[str, Any]] = [] + duplicates_found: list[dict[str, Any]] = [] + + for item in items: + doc = item.get("document", {}) + is_duplicate = False + for existing in deduped: + existing_doc = existing.get("document", {}) + dup_result = self.nlp.detect_duplicate(doc, existing_doc) + if dup_result.get("duplicate"): + is_duplicate = True + duplicates_found.append( + { + "item": doc, + "duplicate_of": existing_doc, + "reason": dup_result.get("reason"), + } + ) + metrics.increment("literature.duplicate_detected") + break + if not is_duplicate: + deduped.append(item) + + return {**payload, "items": deduped, "duplicates": duplicates_found} + + +class KGUpdateStage(PipelineStage): + name = "KGUpdateStage" + + def __init__(self, kg_client: KGClient): + self.kg_client = kg_client + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + kg_results: list[dict[str, Any]] = [] + for item in payload.get("items", []): + kg_results.append( + self.kg_client.update_knowledge_graph( + item.get("structured_entities", []), item.get("relationships", []) + ) + ) + return {**payload, "kg_updates": kg_results} + + +class WikiUpdateStage(PipelineStage): + name = "WikiUpdateStage" + + def __init__(self, wiki_client: LLMWikiClient): + self.wiki_client = wiki_client + + def run(self, payload: dict[str, Any]) -> dict[str, Any]: + wiki_results: list[dict[str, Any]] = [] + for item in payload.get("items", []): + wiki_results.append( + self.wiki_client.update_wiki( + item.get("document", {}), + item.get("structured_entities", []), + item.get("structured_summary", {}), + ) + ) + return {**payload, "wiki_updates": wiki_results, "status": "completed"} diff --git a/services/literature/app/parsing/text_parser.py b/services/literature/app/parsing/text_parser.py new file mode 100644 index 0000000..9803e5d --- /dev/null +++ b/services/literature/app/parsing/text_parser.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import html +import re +from typing import Any + + +class TextParser: + """Prompt 6 document parser with deterministic normalization, HTML stripping, and safe optional field handling.""" + + def strip_tags(self, text: str) -> str: + """Strip HTML/XML markup safely.""" + if not text: + return "" + clean = re.sub(r"<[^>]+>", "", text) + clean = html.unescape(clean) + clean = re.sub(r"\s+", " ", clean) + return re.sub(r"\s+([.,!?;:])", r"\1", clean).strip() + + def parse(self, text: str | None) -> dict[str, Any]: + raw = text or "" + cleaned = self.strip_tags(raw) + sentences = [segment.strip() for segment in re.split(r"(?<=[.!?])\s+", cleaned) if segment.strip()] + paragraphs = [paragraph.strip() for paragraph in re.split(r"\n+", cleaned) if paragraph.strip()] + tokens = re.findall(r"\b[\w-]+\b", cleaned) + return { + "text": cleaned, + "sentences": sentences, + "paragraphs": paragraphs, + "token_count": len(tokens), + } + + def normalize_document(self, document: dict[str, Any]) -> dict[str, Any]: + raw_title = str(document.get("title") or document.get("headline") or "").strip() + raw_abstract = str(document.get("abstract") or document.get("summary") or "").strip() + raw_content = str(document.get("content") or document.get("full_text") or raw_abstract or raw_title) + + title = self.strip_tags(raw_title) + abstract = self.strip_tags(raw_abstract) + content = self.strip_tags(raw_content) + + source = str(document.get("source") or "unknown").strip().lower() + source_id_val = document.get("source_id") or document.get("id") + source_id = str(source_id_val).strip() if source_id_val is not None else None + + doi_val = document.get("doi") + doi = str(doi_val).strip() if doi_val is not None else None + + url_val = document.get("url") or document.get("link") + url = str(url_val).strip() if url_val is not None else None + + pub_date_val = document.get("published_date") or document.get("date") or document.get("publishedAt") + published_date = str(pub_date_val).strip() if pub_date_val is not None else None + + authors_raw = document.get("authors") or [] + if isinstance(authors_raw, str): + authors = [a.strip() for a in authors_raw.split(",") if a.strip()] + elif isinstance(authors_raw, list): + authors = [str(a).strip() for a in authors_raw if a] + else: + authors = [] + + journal_val = document.get("journal") or document.get("conference") + journal = str(journal_val).strip() if journal_val is not None else None + + metadata = document.get("metadata") + if not isinstance(metadata, dict): + metadata = {} + + document_type = document.get("document_type") or document.get("type") or source + return { + "title": title, + "abstract": abstract, + "content": content, + "authors": authors, + "published_date": published_date, + "source": source, + "source_id": source_id, + "doi": doi, + "url": url, + "journal": journal, + "document_type": str(document_type).strip() if document_type is not None else None, + "metadata": metadata, + } diff --git a/services/literature/app/services/literature_service.py b/services/literature/app/services/literature_service.py new file mode 100644 index 0000000..54bb92a --- /dev/null +++ b/services/literature/app/services/literature_service.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import logging +from typing import Any + +from app.connectors.factory import ConnectorFactory +from app.database.models import IngestionJobState, job_store +from app.integrations.kg_client import KGClient +from app.integrations.wiki_client import LLMWikiClient +from app.nlp.pipeline import LiteratureNLP +from app.observability.metrics import metrics +from app.orchestrator.pipeline import PipelineRunner +from app.orchestrator.stages import ( + DeduplicationStage, + EvidenceRankingStage, + KGUpdateStage, + NERStage, + ParsingStage, + RelationshipExtractionStage, + SummarizationStage, + WikiUpdateStage, +) + +logger = logging.getLogger(__name__) + + +class LiteratureService: + """Service layer coordinating connectors, NLP pipeline, KG & LLM Wiki updates, and state persistence.""" + + def __init__(self, config: dict[str, Any] | None = None): + self.config = config or {} + self.nlp = LiteratureNLP(self.config) + self.kg_client = KGClient(self.config) + self.wiki_client = LLMWikiClient(self.config) + + self.pipeline = PipelineRunner( + stages=[ + ParsingStage(self.nlp), + NERStage(self.nlp), + RelationshipExtractionStage(self.nlp), + SummarizationStage(self.nlp), + EvidenceRankingStage(self.nlp), + DeduplicationStage(self.nlp), + KGUpdateStage(self.kg_client), + WikiUpdateStage(self.wiki_client), + ], + retries=2, + delay_seconds=0.05, + ) + + + def ingest(self, source: str, query: str, job_id: str | None = None, **kwargs: Any) -> dict[str, Any]: + connector_config = {**self.config, **kwargs} + connector = ConnectorFactory.create(source, connector_config) + source_status = connector.connect() + results = connector.fetch(query, **kwargs) + metrics.increment(f"literature.{source}.ingest") + if not results: + metrics.increment(f"literature.{source}.no_results") + + initial_payload = { + "source": source, + "query": query, + "items": results, + "limitation": connector.get_limitation(), + "source_status": source_status, + } + + if job_id: + job = job_store.get(job_id) or IngestionJobState(id=job_id, source=source, query=query) + return self.pipeline.run_with_job(job, initial_payload) + + return self.pipeline.run(initial_payload) + + def analyze_text(self, text: str) -> dict[str, Any]: + result = self.nlp.run(text) + metrics.increment("literature.nlp.analyze") + return result diff --git a/services/literature/app/utils/html_utils.py b/services/literature/app/utils/html_utils.py new file mode 100644 index 0000000..55f64c5 --- /dev/null +++ b/services/literature/app/utils/html_utils.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import html +import re +from collections.abc import Iterable +from html.parser import HTMLParser +from typing import Any +from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse + +TRACKING_QUERY_PARAMS = { + "utm_source", + "utm_medium", + "utm_campaign", + "utm_term", + "utm_content", + "fbclid", + "gclid", + "mc_cid", + "mc_eid", + "ref", + "ref_src", +} + +RESTRICTED_KEYWORDS = [ + "sign in", + "sign_in", + "log in", + "login", + "access denied", + "403", + "404", + "paywall", + "subscription", + "restricted", +] + + +def normalize_url(url: str, base_url: str | None = None) -> str: + if base_url: + url = urljoin(base_url, url) + url = html.unescape(url.strip()) + parsed = urlparse(url) + cleaned_query = urlencode( + [(k, v) for k, v in parse_qsl(parsed.query, keep_blank_values=True) if k.lower() not in TRACKING_QUERY_PARAMS] + ) + normalized = urlunparse((parsed.scheme, parsed.netloc.lower(), parsed.path or "/", parsed.params, cleaned_query, "")) + return normalized + + +def is_access_restricted(html_text: str) -> bool: + if not html_text: + return False + lower = html_text.lower() + return any(token in lower for token in RESTRICTED_KEYWORDS) + + +class LinkAndMetaParser(HTMLParser): + def __init__(self, base_url: str | None = None): + super().__init__() + self.base_url = base_url + self.title_fragments: list[str] = [] + self.meta: dict[str, str] = {} + self.links: list[str] = [] + self._in_title = False + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]): + tag = tag.lower() + attrs_dict = {name.lower(): (value or "") for name, value in attrs} + if tag == "title": + self._in_title = True + elif tag == "meta": + name = attrs_dict.get("name") or attrs_dict.get("property") + content = attrs_dict.get("content") or attrs_dict.get("value") + if name and content: + self.meta[name.lower()] = content.strip() + elif tag == "a": + href = attrs_dict.get("href") + if href: + normalized = normalize_url(href, self.base_url) + if normalized: + self.links.append(normalized) + + def handle_endtag(self, tag: str): + if tag.lower() == "title": + self._in_title = False + + def handle_data(self, data: str): + if self._in_title: + self.title_fragments.append(data) + + def get_title(self) -> str | None: + title = " ".join(self.title_fragments).strip() + return title or None + + def get_meta(self) -> dict[str, str]: + return self.meta + + def get_links(self) -> list[str]: + unique_links: list[str] = [] + seen: set[str] = set() + for url in self.links: + if url not in seen: + seen.add(url) + unique_links.append(url) + return unique_links + + +def extract_page_metadata(html_text: str, url: str) -> dict[str, Any]: + parser = LinkAndMetaParser(base_url=url) + parser.feed(html_text) + title = parser.get_title() or parser.get_meta().get("og:title") or parser.get_meta().get("twitter:title") + description = ( + parser.get_meta().get("description") + or parser.get_meta().get("og:description") + or parser.get_meta().get("twitter:description") + ) + authors = [] + if "citation_author" in parser.get_meta(): + authors = [author.strip() for author in parser.get_meta()["citation_author"].split(";") if author.strip()] + elif "author" in parser.get_meta(): + authors = [author.strip() for author in parser.get_meta()["author"].split(",") if author.strip()] + published_date = ( + parser.get_meta().get("citation_publication_date") + or parser.get_meta().get("article:published_time") + or parser.get_meta().get("date") + or parser.get_meta().get("pubdate") + ) + content = strip_tags(html_text) + return { + "title": title, + "abstract": description, + "content": content, + "authors": authors, + "published_date": published_date, + "url": url, + "metadata": {"meta": parser.get_meta()}, + } + + +def extract_search_result_urls(html_text: str, base_url: str | None = None) -> list[str]: + parser = LinkAndMetaParser(base_url=base_url) + parser.feed(html_text) + return parser.get_links() + + +def strip_tags(html_text: str) -> str: + if not html_text: + return "" + text = re.sub(r"", " ", html_text, flags=re.IGNORECASE | re.DOTALL) + text = re.sub(r"", " ", text, flags=re.IGNORECASE | re.DOTALL) + text = re.sub(r"", " ", text, flags=re.DOTALL) + text = re.sub(r"<[^>]+>", " ", text) + text = html.unescape(text) + text = re.sub(r"\s+", " ", text) + return text.strip() + + +def url_allowed_domain(url: str, allowed_domains: Iterable[str]) -> bool: + parsed = urlparse(url) + hostname = parsed.hostname or "" + hostname = hostname.lower() + if not hostname: + return False + if not allowed_domains: + return True + for allowed in allowed_domains: + if hostname == allowed.lower() or hostname.endswith("." + allowed.lower()): + return True + return False diff --git a/services/literature/data/ingestion_jobs.json b/services/literature/data/ingestion_jobs.json new file mode 100644 index 0000000..d23a203 --- /dev/null +++ b/services/literature/data/ingestion_jobs.json @@ -0,0 +1,644 @@ +[ + { + "id": "job-test-1", + "source": "pubmed", + "query": "oncology", + "status": "dead_letter", + "created_at": "2026-08-09T18:18:41.860017+00:00", + "updated_at": "2026-08-09T18:18:41.881321+00:00", + "attempts": 2, + "max_retries": 3, + "error": "Stage failure simulation", + "result": null + }, + { + "id": "88fd4d71-ad14-45a3-96a1-f5d23c64c5d3", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T15:36:34.655036+00:00", + "updated_at": "2026-08-09T15:36:35.776023+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "2867cc60-a9a2-4319-8ba2-cf069df3a4eb", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T15:39:00.534692+00:00", + "updated_at": "2026-08-09T15:39:01.601027+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "d885a5ce-6eae-4a54-81d7-88cd94ac2732", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T15:55:11.178201+00:00", + "updated_at": "2026-08-09T15:55:12.135817+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "7be4b43d-b905-4c4a-a3c1-b033ce81ccb9", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T16:36:16.840544+00:00", + "updated_at": "2026-08-09T16:36:17.913566+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "0e279eb7-9a08-4530-a76e-ccb234a64030", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T16:44:58.981446+00:00", + "updated_at": "2026-08-09T16:44:59.594785+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "b2716595-ad05-400b-8719-dbe729544c84", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T16:45:52.686726+00:00", + "updated_at": "2026-08-09T16:45:53.594496+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "b04952c9-c8d2-49ab-a296-ed1bae76cdf4", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T16:51:21.094238+00:00", + "updated_at": "2026-08-09T16:51:21.859473+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "1eb90b9c-f53b-4179-a093-66ce0d28e367", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T16:51:52.359921+00:00", + "updated_at": "2026-08-09T16:51:52.887987+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "b472b5b9-5002-4b35-bd5a-59ac84026114", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T17:01:50.096537+00:00", + "updated_at": "2026-08-09T17:01:50.457348+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "3e928a11-33a8-4e85-bd1c-b6251f544236", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T17:09:25.730838+00:00", + "updated_at": "2026-08-09T17:09:26.098394+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "5584ec5b-1036-4a27-92bb-42949d019531", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T17:12:38.917534+00:00", + "updated_at": "2026-08-09T17:12:39.495457+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "37ce44c3-cf73-4bae-8696-13c97dcbf6e1", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T17:14:27.258929+00:00", + "updated_at": "2026-08-09T17:14:27.736948+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "c0c3581b-e472-4c24-9da5-52b020ca848a", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T17:22:39.974790+00:00", + "updated_at": "2026-08-09T17:22:40.343901+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "9d9e46c6-a49b-42f5-9b9e-bd82ec7b3123", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T17:30:38.787581+00:00", + "updated_at": "2026-08-09T17:30:39.147444+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "c2502caf-9c8a-41ed-8995-63574a9f1294", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T17:37:05.924430+00:00", + "updated_at": "2026-08-09T17:37:06.835643+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "f5b70b64-d139-48b6-b41d-de2c8e71e11c", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T17:41:58.607354+00:00", + "updated_at": "2026-08-09T17:41:59.036607+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "7a946b83-90fb-43b6-92ab-0ad65fdaf991", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T17:59:23.039521+00:00", + "updated_at": "2026-08-09T17:59:23.470983+00:00", + "attempts": 3, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "status": "completed", + "kg_updates": [], + "wiki_updates": [] + } + }, + { + "id": "f93e622d-1c8f-40dc-b13d-b86c48118a50", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T18:14:32.553191+00:00", + "updated_at": "2026-08-09T18:14:33.090576+00:00", + "attempts": 8, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "kg_updates": [], + "wiki_updates": [], + "status": "completed" + } + }, + { + "id": "70397351-08a3-4124-8949-7ad691fdb0b2", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T18:16:48.991167+00:00", + "updated_at": "2026-08-09T18:16:49.478011+00:00", + "attempts": 8, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "kg_updates": [], + "wiki_updates": [], + "status": "completed" + } + }, + { + "id": "118426b3-eb35-4331-b105-f4eff540ba4e", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T18:17:26.917287+00:00", + "updated_at": "2026-08-09T18:17:27.430458+00:00", + "attempts": 8, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "kg_updates": [], + "wiki_updates": [], + "status": "completed" + } + }, + { + "id": "6c6feea4-b06b-4c09-9c41-3fa6b736bfb8", + "source": "pubmed", + "query": "trastuzumab HER2", + "status": "completed", + "created_at": "2026-08-09T18:18:41.891042+00:00", + "updated_at": "2026-08-09T18:18:42.287977+00:00", + "attempts": 8, + "max_retries": 3, + "error": null, + "result": { + "source": "pubmed", + "query": "trastuzumab HER2", + "items": [], + "limitation": null, + "source_status": { + "source": "pubmed", + "connected": true, + "timeout": 10.0, + "user_agent": "AI-RxOS LiteratureBot/1.0", + "max_retries": 2, + "limitation": null + }, + "processed_items": [], + "duplicates": [], + "kg_updates": [], + "wiki_updates": [], + "status": "completed" + } + } +] \ No newline at end of file diff --git a/services/literature/prompt6_endpoint_check.py b/services/literature/prompt6_endpoint_check.py new file mode 100644 index 0000000..43bbd25 --- /dev/null +++ b/services/literature/prompt6_endpoint_check.py @@ -0,0 +1,17 @@ +import urllib.request + +urls = [ + 'http://127.0.0.1:8082/healthz', + 'http://127.0.0.1:8082/metrics', + 'http://127.0.0.1:8082/api/v1/papers', +] + +for url in urls: + try: + with urllib.request.urlopen(url, timeout=10) as resp: + body = resp.read(2000).decode('utf-8', 'replace') + print(url) + print('STATUS', resp.status) + print(body) + except Exception as exc: + print(url, 'ERROR', repr(exc)) diff --git a/services/literature/requirements.txt b/services/literature/requirements.txt index 783c143..d435932 100644 --- a/services/literature/requirements.txt +++ b/services/literature/requirements.txt @@ -3,9 +3,8 @@ uvicorn[standard]==0.34.0 pydantic==2.10.4 pydantic-settings==2.7.1 asyncpg==0.30.0 -PyJWT==2.8.0 +httpx==0.28.1 python-json-logger==3.2.1 +spacy>=3.0.0,<4.0 pytest==8.3.4 pytest-asyncio==0.25.1 -croniter==1.4.1 -prometheus-client==0.21.0 diff --git a/services/literature/tests/test_pipeline_integration.py b/services/literature/tests/test_pipeline_integration.py new file mode 100644 index 0000000..4922eca --- /dev/null +++ b/services/literature/tests/test_pipeline_integration.py @@ -0,0 +1,23 @@ +from app.connectors.factory import ConnectorFactory +from app.nlp.pipeline import LiteratureNLP +from app.orchestrator.pipeline import PipelineRunner + + +def test_connector_factory_builds_pubmed_connector(): + connector = ConnectorFactory.create("pubmed", {"base_url": "https://example.com"}) + assert connector.name == "pubmed" + assert connector.config["base_url"] == "https://example.com" + + +def test_pipeline_runner_executes_stage_sequence(): + runner = PipelineRunner(stages=[lambda doc: {**doc, "stage": "parsed"}, lambda doc: {**doc, "stage": "indexed"}]) + result = runner.run({"id": "paper-1"}) + assert result["stage"] == "indexed" + + +def test_nlp_pipeline_extracts_entities_and_summary(): + text = "HER2-positive breast cancer is treated by trastuzumab in combination therapy." + result = LiteratureNLP().run(text) + assert "entities" in result + assert "summary" in result + assert any(entity.lower() in {"her2", "trastuzumab", "breast cancer"} for entity in result["entities"]) diff --git a/services/literature/tests/test_prompt6_comprehensive.py b/services/literature/tests/test_prompt6_comprehensive.py new file mode 100644 index 0000000..90ca19c --- /dev/null +++ b/services/literature/tests/test_prompt6_comprehensive.py @@ -0,0 +1,363 @@ +from unittest.mock import MagicMock, patch + +import httpx +from fastapi.testclient import TestClient + +from app.connectors.base import BaseConnector +from app.connectors.factory import ConnectorFactory +from app.connectors.sources import CompanyWebsiteConnector, PatentsConnector +from app.database.models import IngestionJobState, job_store +from app.integrations.kg_client import KGClient +from app.integrations.wiki_client import LLMWikiClient +from app.main import app +from app.nlp.deduplication import DuplicateDetector +from app.nlp.evidence_ranking import EvidenceRanker +from app.nlp.ner import RuleBasedEntityExtractor +from app.nlp.pipeline import LiteratureNLP +from app.nlp.relationships import RelationshipExtractor +from app.nlp.summarizer import DocumentSummarizer +from app.orchestrator.pipeline import PipelineRunner +from app.parsing.text_parser import TextParser + +client = TestClient(app) + + +def test_all_11_connectors_registered_and_instantiated(): + required_sources = [ + "pubmed", + "pmc", + "clinicaltrials", + "aacr", + "asco", + "sabcs", + "esmo", + "biorxiv", + "medrxiv", + "patents", + "company_websites", + ] + + for source in required_sources: + connector = ConnectorFactory.create(source, {"timeout": 0.1}) + assert connector.name == source + info = connector.connect() + assert info["connected"] is True + fetched = connector.fetch("HER2 breast cancer") + assert isinstance(fetched, list) + if fetched: + assert "title" in fetched[0] + assert "source" in fetched[0] + else: + assert connector.get_limitation() is not None or source in {"pubmed", "pmc", "clinicaltrials", "biorxiv", "medrxiv"} + + +def test_connectors_with_documented_limitations(): + limitation_sources = ["aacr", "asco", "sabcs", "esmo", "patents", "company_websites"] + for source in limitation_sources: + connector = ConnectorFactory.create(source) + limitation = connector.get_limitation() + assert limitation is not None + assert isinstance(limitation, str) + + +def test_conference_connectors_return_search_metadata(monkeypatch): + search_html = """ + Test Search + Result + """ + detail_html = """ + Detail Page +

Full content here.

+ """ + + def fake_request_text(self, url, **kwargs): + if url.endswith("/search-results?q=test") or "action/doSearch" in url or "?s=test" in url or "search?searchText=test" in url: + return search_html + return detail_html + + monkeypatch.setattr(BaseConnector, "_request_text", fake_request_text) + + for source in ["aacr", "asco", "sabcs", "esmo"]: + connector = ConnectorFactory.create(source) + results = connector.fetch("test", max_results=1) + assert isinstance(results, list) + assert len(results) == 1 + item = results[0] + assert item["source"] == source + assert item["url"] is not None + assert item["metadata"]["limitation"] == connector.get_limitation() + + +def test_patents_connector_falls_back_to_search_page(monkeypatch): + search_html = """ + Patent search +

Patent listing content.

+ """ + + def fake_request_json(self, url, **kwargs): + return {"patents": []} + + def fake_request_text(self, url, **kwargs): + return search_html + + monkeypatch.setattr(PatentsConnector, "_request_json", fake_request_json) + monkeypatch.setattr(PatentsConnector, "_request_text", fake_request_text) + + connector = PatentsConnector() + results = connector.fetch("cancer", max_results=1) + assert isinstance(results, list) + assert len(results) == 1 + assert results[0]["source"] == "patents" + assert results[0]["metadata"]["limitation"] == connector.get_limitation() + + +def test_company_website_connector_crawls_configured_urls(monkeypatch): + urls = ["https://example.com/press"] + html = """ + Press Release +

Launch announcement with trastuzumab.

+ """ + + class FakeCrawlResult: + def __init__(self, url, html): + self.url = url + self.success = True + self.status_code = 200 + self.error = None + self.metadata = {"title": "Press Release", "content": html} + self.html = html + + def fake_crawl(self, seed_urls): + return [FakeCrawlResult(seed_urls[0], html)] + + monkeypatch.setattr("app.crawling.crawler.WebCrawler.crawl", fake_crawl) + + connector = CompanyWebsiteConnector({"urls": urls}) + results = connector.fetch("trastuzumab", max_results=1) + assert isinstance(results, list) + assert len(results) == 1 + item = results[0] + assert item["source"] == "company_websites" + assert item["metadata"]["limitation"] == connector.get_limitation() + + +def test_connector_retry_behavior(monkeypatch): + connector = ConnectorFactory.create("pubmed", {"timeout": 0.1, "max_retries": 1, "backoff_seconds": 0.0}) + attempts = {"count": 0} + + def flaky_request(self, method, url, **kwargs): + attempts["count"] += 1 + if attempts["count"] == 1: + raise httpx.ReadTimeout("temporary timeout") + response = MagicMock() + response.status_code = 200 + response.json.return_value = {"esearchresult": {"idlist": []}} + return response + + monkeypatch.setattr(httpx.Client, "request", flaky_request) + assert connector.fetch("HER2") == [] + assert attempts["count"] == 2 + + +def test_parser_strips_html_and_handles_malformed_metadata(): + parser = TextParser() + raw_doc = { + "title": "

HER2 Therapy

", + "abstract": "

Trastuzumab shows efficacy.

", + "authors": "Alice Smith, Bob Jones", + "metadata": "not_a_dict", # malformed + } + normalized = parser.normalize_document(raw_doc) + assert normalized["title"] == "HER2 Therapy" + assert normalized["abstract"] == "Trastuzumab shows efficacy." + assert normalized["authors"] == ["Alice Smith", "Bob Jones"] + assert isinstance(normalized["metadata"], dict) + + +def test_ner_extracts_entities_across_categories(): + extractor = RuleBasedEntityExtractor() + text = "Trastuzumab targets HER2 protein in breast cancer clinical trial NCT01234567 conducted by AACR." + entities = extractor.extract_entities(text) + labels = {e["label"] for e in entities} + assert "drug" in labels + assert "gene" in labels or "protein" in labels + assert "disease" in labels + assert "clinical_trial" in labels + assert "organization" in labels + + +def test_summarizer_extractive_fallback(): + summarizer = DocumentSummarizer() + doc = { + "title": "HER2 Targeted Therapy in Breast Cancer", + "abstract": "Trastuzumab significantly improves progression-free survival in patients with HER2-positive breast cancer. Overall response rates were high.", + } + summary = summarizer.summarize(doc) + assert summary["summary_type"] == "extractive_fallback" + assert "Trastuzumab" in summary["concise_summary"] + assert summary["llm_used"] is False + + +def test_spacy_provider_falls_back_to_rule_based(monkeypatch): + def fake_init(self, model_name="en_core_web_sm"): + self.available = False + self.nlp = None + + monkeypatch.setattr( + "app.nlp.ner.SpaCyEntityExtractor.__init__", + fake_init, + ) + + nlp = LiteratureNLP({"ner_provider": "spacy"}) + assert isinstance(nlp.ner, RuleBasedEntityExtractor) + + +def test_summarizer_llm_fallback_retries_and_uses_extractive(monkeypatch): + summarizer = DocumentSummarizer( + { + "llm_api_key": "fake-key", + "llm_api_url": "http://invalid-llm", + "llm_max_retries": 1, + "llm_backoff_seconds": 0.0, + } + ) + + def always_fail(*args, **kwargs): + raise httpx.HTTPError("LLM service unavailable") + + monkeypatch.setattr(httpx, "post", always_fail) + + doc = { + "title": "HER2 Targeted Therapy in Breast Cancer", + "abstract": "Trastuzumab significantly improves progression-free survival in patients with HER2-positive breast cancer.", + } + summary = summarizer.summarize(doc) + assert summary["summary_type"] == "extractive_fallback" + assert summary["llm_used"] is False + + +def test_relationship_extraction(): + extractor = RelationshipExtractor() + doc = { + "title": "Trastuzumab clinical trial", + "abstract": "Trastuzumab targets HER2 and treats breast cancer.", + "source_id": "PMID999", + } + entities = [ + {"text": "trastuzumab", "label": "drug", "category": "drugs"}, + {"text": "her2", "label": "gene", "category": "genes"}, + {"text": "breast cancer", "label": "disease", "category": "diseases"}, + ] + relationships = extractor.extract(doc, entities) + predicates = [r["predicate"] for r in relationships] + assert "targets" in predicates + assert "treats" in predicates + + +def test_duplicate_detection_four_tiers(): + detector = DuplicateDetector() + + # Tier 1: DOI + d1 = detector.detect_duplicate({"doi": "10.1000/a"}, {"doi": "10.1000/a"}) + assert d1["duplicate"] is True and d1["reason"] == "doi_match" + + # Tier 2: Source ID + d2 = detector.detect_duplicate({"source": "pubmed", "source_id": "123"}, {"source": "pubmed", "source_id": "123"}) + assert d2["duplicate"] is True and d2["reason"] == "source_id_match" + + # Tier 3: Title + d3 = detector.detect_duplicate({"title": "Breast Cancer Study!"}, {"title": "breast cancer study"}) + assert d3["duplicate"] is True and d3["reason"] == "title_match" + + # Tier 4: Content Hash + d4 = detector.detect_duplicate({"abstract": "Unique content string for testing hashing"}, {"content": "Unique content string for testing hashing"}) + assert d4["duplicate"] is True and d4["reason"] == "content_hash_match" + + # Distinct + d5 = detector.detect_duplicate({"title": "Paper A"}, {"title": "Paper B"}) + assert d5["duplicate"] is False + + +def test_evidence_ranking_deterministic_scoring(): + ranker = EvidenceRanker() + high_doc = { + "source": "clinicaltrials", + "title": "Phase 3 Clinical Trial of Trastuzumab", + "published_date": "2024-01-01", + "metadata": {"overallStatus": "Completed"}, + } + low_doc = { + "source": "biorxiv", + "title": "Preprint abstract", + "published_date": "2015-01-01", + } + high_rank = ranker.rank_document(high_doc) + low_rank = ranker.rank_document(low_doc) + + assert high_rank["score"] > low_rank["score"] + assert high_rank["tier"] in ("High", "Medium") + + +def test_kg_integration_success_and_graceful_failure(): + kg_client = KGClient({"kg_service_url": "http://invalid-localhost-9999"}) + entities = [{"text": "trastuzumab", "label": "drug", "category": "drugs"}] + relationships = [{"subject": "trastuzumab", "predicate": "targets", "object": "her2"}] + + res = kg_client.update_knowledge_graph(entities, relationships) + assert res["success"] is False + assert res["retry_eligible"] is True + + with patch("httpx.Client.post") as mock_post: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_post.return_value = mock_response + + valid_client = KGClient({"kg_service_url": "http://localhost:8001"}) + res_ok = valid_client.update_knowledge_graph(entities, relationships) + assert res_ok["success"] is True + + +def test_wiki_integration_okf_volume_write(tmp_path): + wiki_client = LLMWikiClient({"wiki_dir": str(tmp_path)}) + doc = {"source": "pubmed", "source_id": "PMID123", "title": "HER2 Study", "doi": "10.1000/xyz", "url": "https://example.com"} + entities = [{"text": "her2", "category": "genes"}] + summary = {"concise_summary": "HER2 targeted therapy."} + + res = wiki_client.update_wiki(doc, entities, summary) + assert res["success"] is True + assert res["method"] == "okf_volume" + + concept_file = tmp_path / "wiki" / "genes" / "her2.md" + assert concept_file.exists() + assert "HER2 Study" in concept_file.read_text(encoding="utf-8") + + +def test_orchestrator_state_transitions_and_dead_letter(): + job = IngestionJobState(id="job-test-1", source="pubmed", query="oncology") + job_store.save(job) + + def failing_stage(payload): + raise ValueError("Stage failure simulation") + + runner = PipelineRunner(stages=[failing_stage], retries=1, delay_seconds=0.01) + result = runner.run_with_job(job, {"query": "oncology"}) + + assert result["status"] == "dead_letter" + updated_job = job_store.get("job-test-1") + assert updated_job.status == "dead_letter" + assert "Stage failure simulation" in updated_job.error + + +def test_end_to_end_ingestion_api(): + response = client.post("/api/v1/ingestion", json={"source": "pubmed", "query": "trastuzumab HER2"}) + assert response.status_code == 202 + data = response.json() + assert "id" in data + assert data["status"] in ("completed", "pending", "running") + + job_res = client.get(f"/api/v1/ingestion/{data['id']}") + assert job_res.status_code == 200 + job_data = job_res.json() + assert job_data["status"] == "completed" + assert "result" in job_data + assert "source_status" in job_data["result"] diff --git a/services/literature/tests/test_prompt6_requirements.py b/services/literature/tests/test_prompt6_requirements.py new file mode 100644 index 0000000..40671be --- /dev/null +++ b/services/literature/tests/test_prompt6_requirements.py @@ -0,0 +1,90 @@ +from app.connectors.factory import ConnectorFactory +from app.nlp.pipeline import LiteratureNLP +from app.orchestrator.pipeline import PipelineRunner +from app.services.literature_service import LiteratureService + + +def test_all_required_connectors_are_registered(): + required = { + "pubmed", + "pmc", + "clinicaltrials", + "aacr", + "asco", + "sabcs", + "esmo", + "biorxiv", + "medrxiv", + "patents", + "company_websites", + } + available = set(ConnectorFactory.registry) + assert required.issubset(available) + + +def test_connector_factory_instantiates_registered_source(): + connector = ConnectorFactory.create("clinicaltrials", {"base_url": "https://example.com"}) + assert connector.name == "clinicaltrials" + + +def test_parser_normalizes_fields(): + source = { + "title": "HER2-targeted therapy in breast cancer", + "abstract": "Trastuzumab improved outcomes in HER2-positive breast cancer.", + "authors": ["Alice", "Bob"], + "published_date": "2024-01-15", + "source": "pubmed", + "source_id": "PMID123", + "doi": "10.1000/example", + "url": "https://example.com/article", + } + parsed = LiteratureNLP().parse_document(source) + assert parsed["title"] == "HER2-targeted therapy in breast cancer" + assert parsed["source"] == "pubmed" + assert parsed["authors"] == ["Alice", "Bob"] + assert parsed["doi"] == "10.1000/example" + + +def test_duplicate_detector_flags_same_document(): + doc_a = {"doi": "10.1000/example", "title": "HER2 targeted therapy", "source": "pubmed", "source_id": "PMID123"} + doc_b = {"doi": "10.1000/example", "title": "HER2 targeted therapy", "source": "pubmed", "source_id": "PMID123"} + result = LiteratureNLP().detect_duplicate(doc_a, doc_b) + assert result["duplicate"] is True + + +def test_relationship_extraction_builds_triples(): + doc = { + "title": "Trastuzumab targets HER2 and treats breast cancer", + "abstract": "Trastuzumab is associated with HER2-positive breast cancer.", + "entities": [{"text": "trastuzumab", "label": "drug"}, {"text": "her2", "label": "gene"}, {"text": "breast cancer", "label": "disease"}], + } + relationships = LiteratureNLP().extract_relationships(doc) + assert relationships + assert any(r["subject"].lower() == "trastuzumab" for r in relationships) + + +def test_ranking_and_summary_are_deterministic(): + result = LiteratureNLP().run("Trastuzumab targets HER2 in breast cancer. HER2-positive disease is studied in clinical trials.") + assert "summary" in result + assert "evidence" in result + assert result["evidence"][0]["score"] >= 0 + + +def test_orchestrator_runs_complete_pipeline(): + runner = PipelineRunner( + stages=[ + lambda doc: {**doc, "parsed": True}, + lambda doc: {**doc, "deduplicated": True}, + lambda doc: {**doc, "processed": True}, + ] + ) + result = runner.run({"id": "paper-1"}) + assert result["processed"] is True + + +def test_service_ingest_uses_registry_and_returns_payload(): + service = LiteratureService() + outcome = service.ingest("pubmed", "HER2 breast cancer", base_url="https://example.com") + assert "source" in outcome + assert outcome["source"] == "pubmed" + assert "items" in outcome diff --git a/services/literature/wiki-root/wiki/diseases/breast_cancer.md b/services/literature/wiki-root/wiki/diseases/breast_cancer.md new file mode 100644 index 0000000..2d200b9 --- /dev/null +++ b/services/literature/wiki-root/wiki/diseases/breast_cancer.md @@ -0,0 +1,12 @@ +# Concept: breast cancer +- **Category**: diseases +- **Last Updated**: 2026-08-09T15:32:27.801230+00:00 +- **Source**: pubmed (pubmed-33880) + +## Primary Summary +Literature document retrieved for query: HER2 breast cancer + +## Literature Evidence +- **Title**: PubMed result for: HER2 breast cancer +- **DOI**: 10.1016/j.pubmed.3880 +- **URL**: https://pubmed.ncbi.nlm.nih.gov diff --git a/services/literature/wiki-root/wiki/drugs/trastuzumab.md b/services/literature/wiki-root/wiki/drugs/trastuzumab.md new file mode 100644 index 0000000..6e45473 --- /dev/null +++ b/services/literature/wiki-root/wiki/drugs/trastuzumab.md @@ -0,0 +1,12 @@ +# Concept: trastuzumab +- **Category**: drugs +- **Last Updated**: 2026-08-09T15:32:25.732174+00:00 +- **Source**: pubmed (pubmed-47258) + +## Primary Summary +Literature document retrieved for query: trastuzumab HER2 + +## Literature Evidence +- **Title**: PubMed result for: trastuzumab HER2 +- **DOI**: 10.1016/j.pubmed.7258 +- **URL**: https://pubmed.ncbi.nlm.nih.gov diff --git a/services/literature/wiki-root/wiki/genes/her2.md b/services/literature/wiki-root/wiki/genes/her2.md new file mode 100644 index 0000000..29ff98e --- /dev/null +++ b/services/literature/wiki-root/wiki/genes/her2.md @@ -0,0 +1,12 @@ +# Concept: her2 +- **Category**: genes +- **Last Updated**: 2026-08-09T15:32:27.801230+00:00 +- **Source**: pubmed (pubmed-33880) + +## Primary Summary +Literature document retrieved for query: HER2 breast cancer + +## Literature Evidence +- **Title**: PubMed result for: HER2 breast cancer +- **DOI**: 10.1016/j.pubmed.3880 +- **URL**: https://pubmed.ncbi.nlm.nih.gov diff --git a/services/literature/wiki-root/wiki/log.md b/services/literature/wiki-root/wiki/log.md new file mode 100644 index 0000000..0cb4796 --- /dev/null +++ b/services/literature/wiki-root/wiki/log.md @@ -0,0 +1,19 @@ +- [2026-08-09T15:15:25.750325+00:00] Updated 2 concept pages from pubmed:pubmed-17947 +- [2026-08-09T15:18:00.732256+00:00] Updated 0 concept pages from pubmed:PMID:42571049 +- [2026-08-09T15:18:04.374312+00:00] Updated 0 concept pages from pubmed:PMID:42569044 +- [2026-08-09T15:18:08.520940+00:00] Updated 0 concept pages from pubmed:PMID:42563503 +- [2026-08-09T15:18:12.280801+00:00] Updated 0 concept pages from pubmed:PMID:42562949 +- [2026-08-09T15:18:16.097921+00:00] Updated 0 concept pages from pubmed:PMID:42562610 +- [2026-08-09T15:18:22.833199+00:00] Updated 2 concept pages from pubmed:pubmed-60255 +- [2026-08-09T15:19:13.555271+00:00] Updated 2 concept pages from pubmed:pubmed-10318 +- [2026-08-09T15:19:20.568237+00:00] Updated 2 concept pages from pubmed:pubmed-18816 +- [2026-08-09T15:19:59.811735+00:00] Updated 2 concept pages from pubmed:pubmed-73101 +- [2026-08-09T15:20:07.696794+00:00] Updated 2 concept pages from pubmed:pubmed-83263 +- [2026-08-09T15:20:23.546550+00:00] Updated 2 concept pages from pubmed:pubmed-371 +- [2026-08-09T15:20:41.892568+00:00] Updated 2 concept pages from pubmed:pubmed-17592 +- [2026-08-09T15:21:56.960572+00:00] Updated 2 concept pages from pubmed:pubmed-22209 +- [2026-08-09T15:22:01.466167+00:00] Updated 2 concept pages from pubmed:pubmed-56201 +- [2026-08-09T15:22:02.204525+00:00] Updated 2 concept pages from pubmed:pubmed-53528 +- [2026-08-09T15:22:04.841350+00:00] Updated 2 concept pages from pubmed:pubmed-88637 +- [2026-08-09T15:32:25.732174+00:00] Updated 2 concept pages from pubmed:pubmed-47258 +- [2026-08-09T15:32:27.801230+00:00] Updated 2 concept pages from pubmed:pubmed-33880