34047007@qq.com
73f9468384
CRITICAL:
- invalidate_search_cache 清理 atm:* 缓存（MeSH ATM扩展不再用过期结果）
- Pro 方案 api_quota_per_day 1000→10000（修复低于 Free 的数据错误）
- CacheService/RateLimitMiddleware Redis 连接失败60秒自动重试（原永久降级）
- 普通搜索中文输入自动匹配 GlobalTag.name_zh（如"肺癌"通过MeSH标签关联文献）
- AdvancedPubSearchView resolveQuery 添加 seen Set 检测交替 #N 循环引用
HIGH:
- 限速器 _burst_windows 每500请求清理过期条目（防止内存泄漏）
- cron daily_ftp_update 末尾调用 invalidate_search_cache()（自动管道不再用过期缓存）
MEDIUM:
- _apply_order_by ASC 排序加 id tiebreaker（title/journal/first_author翻页跳行/重复）
- _keyset_condition 所有 is_(None) 加 id tiebreaker + __NULL__ 哨兵值
- _field_condition("all") 默认tsvector路径加 journal/journal_iso ILIKE 兜底
- SearchView restoreFromQuery date_preset/year_from/year_to 优先顺序修复
docs: 更新 12/13 搜索文档，移除 CLAUDE.md 陈旧 SQLite 提及
1317 lines
56 KiB
Python
1317 lines
56 KiB
Python
"""PubMed E-utilities API 管道：搜索 + 获取 + 解析 + 打标 + 入库"""
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import logging
|
||
import uuid
|
||
import xml.etree.ElementTree as ET
|
||
from datetime import date, datetime, timezone
|
||
|
||
import httpx
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.compat import UTC
|
||
from app.config import settings
|
||
|
||
# 预印本检测常量
|
||
_PREPRINT_DOI_PREFIXES = ("10.1101/", "10.21203/", "10.22541/", "10.20944/", "10.26434/", "10.31219/", "10.2139/", "10.48550/")
|
||
_PREPRINT_JOURNALS = frozenset({"medrxiv", "biorxiv", "arxiv", "research square", "preprints.org", "ssrn", "chemrxiv", "authorea"})
|
||
|
||
# 动态计算搜索年范围（覆盖最近两年，非硬编码） — 每次调用时实时计算
|
||
def _year_filter() -> str:
|
||
now = datetime.now()
|
||
return f"{now.year - 2}:{now.year}[dp]"
|
||
|
||
from app.db import async_session
|
||
from app.models.literature import GlobalLiterature
|
||
from app.services.journal_utils import ensure_journal_async
|
||
from app.services.feed_engine import generate_feeds_for_literature
|
||
from app.services.tag_service import tag_article
|
||
from app.services.pmc_oa import fetch_pmc_xml
|
||
from app.services.cos_client import download_json, make_full_text_key, upload_json
|
||
from app.config import settings
|
||
from app.services.jats_parser import extract_baseline_data, extract_license_from_xml, parse_full_text
|
||
from app.constants.study_design import classify_study_design
|
||
from app.services.rct_detector import detect_rct
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# NCBI E-utilities 配置
|
||
ESEARCH_URL = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi"
|
||
EFETCH_URL = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi"
|
||
ELINK_URL = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/elink.fcgi"
|
||
|
||
# Europe PMC RESTful API（支持 cursor 分页，无 API Key 限制）
|
||
EUROPE_PMC_SEARCH_URL = "https://www.ebi.ac.uk/europepmc/webservices/rest/search"
|
||
|
||
API_KEY = settings.PUBMED_API_KEY or None # 空字符串视为 None
|
||
TOOL_NAME = "scilit_oncology"
|
||
TOOL_EMAIL = "dev@scilit-oncology.com"
|
||
|
||
# 重试配置
|
||
MAX_RETRIES = 3
|
||
RETRYABLE_CODES = {429, 500, 502, 503, 504}
|
||
|
||
|
||
async def _fetch_with_retry(client: httpx.AsyncClient, url: str, params: dict) -> httpx.Response:
|
||
"""带指数退避重试的 HTTP GET"""
|
||
last_exc = None
|
||
for attempt in range(1, MAX_RETRIES + 1):
|
||
try:
|
||
r = await client.get(url, params=params)
|
||
if r.status_code in RETRYABLE_CODES:
|
||
raise httpx.HTTPStatusError(f"Retryable status {r.status_code}", request=r.request, response=r)
|
||
r.raise_for_status()
|
||
return r
|
||
except (httpx.TimeoutException, httpx.NetworkError) as e:
|
||
last_exc = e
|
||
logger.warning("NCBI %s attempt %d/%d failed: %s", url, attempt, MAX_RETRIES, e)
|
||
if attempt == MAX_RETRIES:
|
||
raise
|
||
await asyncio.sleep(2 ** attempt)
|
||
except httpx.HTTPStatusError as e:
|
||
last_exc = e
|
||
if e.response.status_code in RETRYABLE_CODES and attempt < MAX_RETRIES:
|
||
logger.warning("NCBI %s attempt %d/%d got %d, retrying", url, attempt, MAX_RETRIES, e.response.status_code)
|
||
await asyncio.sleep(2 ** attempt)
|
||
else:
|
||
raise
|
||
raise last_exc or RuntimeError("Unexpected retry exit")
|
||
|
||
# 肿瘤学搜索策略（两组）
|
||
|
||
# 1. 高精度：MeSH Major Topic，仅 indexed 记录
|
||
# 精度模式：majr（默认，Major Topic 优先）| mesh（普通 MeSH，更高召回）
|
||
_PRECISION_MODE: str = "majr"
|
||
|
||
def _build_oncology_queries(precision_mode: str = "majr") -> list[str]:
|
||
"""根据精度模式生成肿瘤学 MeSH 查询列表。"""
|
||
tag = "MAJR" if precision_mode == "majr" else "MeSH"
|
||
year_f = _year_filter()
|
||
queries = [
|
||
f'("Lung Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Breast Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Colorectal Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Stomach Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Liver Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Prostatic Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Leukemia"[{tag}]) AND {year_f}',
|
||
f'("Lymphoma"[{tag}]) AND {year_f}',
|
||
f'("Melanoma"[{tag}]) AND {year_f}',
|
||
f'("Pancreatic Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Ovarian Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Esophageal Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Glioma"[{tag}]) AND {year_f}',
|
||
f'("Head and Neck Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Sarcoma"[{tag}]) AND {year_f}',
|
||
f'("Immunotherapy"[{tag}]) AND ("Neoplasms"[{tag}]) AND {year_f}',
|
||
f'("Molecular Targeted Therapy"[{tag}]) AND ("Neoplasms"[{tag}]) AND {year_f}',
|
||
# 兜底：任何 Neoplasm 相关（覆盖甲状腺/肾/膀胱/宫颈/骨髓瘤等非白名单癌种）
|
||
f'("Neoplasms"[{tag}]) AND {year_f}',
|
||
]
|
||
return queries
|
||
|
||
|
||
# 保持 ONCOLOGY_SEARCH_QUERIES 向后兼容（不破坏已有 import）
|
||
ONCOLOGY_SEARCH_QUERIES = _build_oncology_queries("majr")
|
||
|
||
# 2. 高召回：Title/Abstract 关键词，覆盖 in-process + publisher（尚未 MeSH-indexed）
|
||
def _build_broad_queries() -> list[str]:
|
||
year_f = _year_filter()
|
||
return [
|
||
f'("lung cancer"[Title/Abstract] OR "lung carcinoma"[Title/Abstract] OR "lung adenocarcinoma"[Title/Abstract]) AND {year_f}',
|
||
f'("breast cancer"[Title/Abstract] OR "breast carcinoma"[Title/Abstract]) AND {year_f}',
|
||
f'("colorectal cancer"[Title/Abstract] OR "colon cancer"[Title/Abstract] OR "rectal cancer"[Title/Abstract]) AND {year_f}',
|
||
f'("gastric cancer"[Title/Abstract] OR "stomach cancer"[Title/Abstract]) AND {year_f}',
|
||
f'("hepatocellular"[Title/Abstract] OR "liver cancer"[Title/Abstract] OR "hepatic"[Title/Abstract]) AND {year_f}',
|
||
f'("prostate cancer"[Title/Abstract] OR "prostatic cancer"[Title/Abstract]) AND {year_f}',
|
||
f'("leukemia"[Title/Abstract] OR "leukaemia"[Title/Abstract]) AND {year_f}',
|
||
f'("lymphoma"[Title/Abstract] OR "Hodgkin"[Title/Abstract] OR "non-Hodgkin"[Title/Abstract]) AND {year_f}',
|
||
f'("melanoma"[Title/Abstract]) AND {year_f}',
|
||
f'("pancreatic cancer"[Title/Abstract]) AND {year_f}',
|
||
f'("ovarian cancer"[Title/Abstract] OR "ovarian carcinoma"[Title/Abstract]) AND {year_f}',
|
||
f'("esophageal cancer"[Title/Abstract] OR "oesophageal"[Title/Abstract]) AND {year_f}',
|
||
f'("glioma"[Title/Abstract] OR "glioblastoma"[Title/Abstract]) AND {year_f}',
|
||
f'("head and neck cancer"[Title/Abstract] OR "oral cancer"[Title/Abstract] OR "squamous cell carcinoma"[Title/Abstract]) AND {year_f}',
|
||
f'("sarcoma"[Title/Abstract] OR "osteosarcoma"[Title/Abstract]) AND {year_f}',
|
||
# 兜底：非白名单癌种（甲状腺/肾/膀胱/宫颈/骨髓瘤等）
|
||
f'("cancer"[Title/Abstract] OR "tumor"[Title/Abstract] OR "tumour"[Title/Abstract] OR "neoplasm"[Title/Abstract] OR "malignan*"[Title/Abstract]) AND {year_f}',
|
||
]
|
||
|
||
|
||
def _uid() -> uuid.UUID:
|
||
return uuid.uuid4()
|
||
|
||
|
||
async def search_pmids(query: str, max_results: int = 50) -> list[int]:
|
||
"""搜索 PubMed，返回 PMID 列表"""
|
||
params = {
|
||
"db": "pubmed",
|
||
"term": query,
|
||
"retmax": max_results,
|
||
"retmode": "json",
|
||
"sort": "date",
|
||
"tool": TOOL_NAME,
|
||
"email": TOOL_EMAIL,
|
||
}
|
||
if API_KEY:
|
||
params["api_key"] = API_KEY
|
||
|
||
async with httpx.AsyncClient(timeout=30) as client:
|
||
r = await _fetch_with_retry(client, ESEARCH_URL, params)
|
||
data = r.json()
|
||
return [int(x) for x in data.get("esearchresult", {}).get("idlist", [])]
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# Europe PMC API — cursor-based pagination
|
||
# 文档: https://europepmc.org/RestfulWebService
|
||
# ═══════════════════════════════════════════════
|
||
|
||
_EUROPE_PMC_SEM = asyncio.Semaphore(5) # 并发限制
|
||
|
||
|
||
def _parse_europe_pmc_article(art: dict) -> dict | None:
|
||
"""将 Europe PMC JSON 文献转换为标准 dict 格式（与 `_parse_pubmed_xml()` 一致）"""
|
||
pmid_str = art.get("id") or art.get("pmid")
|
||
if not pmid_str:
|
||
return None
|
||
try:
|
||
pmid = int(pmid_str)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
# Authors
|
||
authors = []
|
||
author_list = art.get("authorList", {})
|
||
for au in (author_list.get("author") or []):
|
||
authors.append({
|
||
"family": au.get("lastName", ""),
|
||
"given": au.get("firstName", ""),
|
||
"affiliation": ((au.get("affiliation") or [None])[0]
|
||
if isinstance(au.get("affiliation"), list)
|
||
else (au.get("affiliation") or "")),
|
||
})
|
||
if not authors:
|
||
author_str = art.get("authorString", "")
|
||
if author_str:
|
||
for name in author_str.split(", "):
|
||
parts = name.strip().split(" ", 1)
|
||
authors.append({
|
||
"family": parts[0],
|
||
"given": parts[-1] if len(parts) > 1 else "",
|
||
"affiliation": "",
|
||
})
|
||
|
||
# DOI
|
||
doi = art.get("doi") or art.get("DOI")
|
||
|
||
# pub_date
|
||
pub_date = None
|
||
pub_year = None
|
||
fpd = art.get("firstPublicationDate")
|
||
if fpd:
|
||
try:
|
||
parts = fpd.split("-")
|
||
pub_year = int(parts[0])
|
||
if len(parts) >= 3:
|
||
pub_date = date(int(parts[0]), int(parts[1]), int(parts[2]))
|
||
elif len(parts) == 2:
|
||
pub_date = date(int(parts[0]), int(parts[1]), 1)
|
||
# 仅年份精度不生成伪日期，pub_year 已足够
|
||
except (ValueError, TypeError):
|
||
pass
|
||
elif art.get("pubYear"):
|
||
try:
|
||
pub_year = int(art["pubYear"])
|
||
# 仅年份精度，不生成误导性的 YYYY-01-01
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# MeSH Headings
|
||
mesh_headings = []
|
||
for mh in (art.get("meshHeadingList", {}).get("meshHeading") or []):
|
||
desc = mh.get("descriptorName", "")
|
||
ui = mh.get("descriptorUI", "")
|
||
major = mh.get("majorTopic_YN", "") == "Y"
|
||
qualifiers = [q.get("qualifierName", "") for q in
|
||
(mh.get("qualifierList", {}).get("qualifier") or []) if q.get("qualifierName")]
|
||
if desc:
|
||
mesh_headings.append({
|
||
"descriptor": desc, "ui": ui, "major": major, "qualifiers": qualifiers,
|
||
})
|
||
|
||
# Keywords（Europe PMC 可能返回单字符串而非数组）
|
||
_kw_val = art.get("keywordList", {}).get("keyword")
|
||
if isinstance(_kw_val, str):
|
||
_kw_val = [_kw_val]
|
||
keywords = [kw for kw in (_kw_val or []) if kw]
|
||
|
||
# Publication types
|
||
pub_types = []
|
||
pt = art.get("publicationType", "")
|
||
if pt:
|
||
pub_types.append(pt)
|
||
for ptype in (art.get("pubTypeList", {}).get("publicationType") or []):
|
||
if ptype and ptype not in pub_types:
|
||
pub_types.append(ptype)
|
||
|
||
# Cache citedByCount for later use
|
||
cbc = art.get("citedByCount", 0)
|
||
|
||
return {
|
||
"pmid": pmid,
|
||
"title": (art.get("title") or "").strip(),
|
||
"abstract": (art.get("abstractText") or "").strip() or None,
|
||
"authors": authors,
|
||
"doi": doi,
|
||
"pmc_id": (art.get("pmcid") or "").removeprefix("PMC"),
|
||
"is_oa": bool(art.get("pmcid")),
|
||
"pubmed_revised": None,
|
||
"citation_status": None,
|
||
"date_completed": None,
|
||
"meshed_date": None,
|
||
"grants": [],
|
||
"journal": art.get("journalTitle", ""),
|
||
"journal_issn": art.get("journalIssn", "") or art.get("journalISSN", ""),
|
||
"journal_iso": art.get("journalAbbreviation") or "",
|
||
"volume": art.get("journalVolume") or art.get("volume", ""),
|
||
"issue": art.get("journalIssue") or art.get("issue", ""),
|
||
"pages": art.get("pageInfo"),
|
||
"pub_date": pub_date,
|
||
"pub_year": pub_year,
|
||
"pub_types": pub_types,
|
||
"mesh_headings": mesh_headings,
|
||
"language": art.get("language", "en"),
|
||
"trial_reg": None,
|
||
"retracted": art.get("isRetracted", False),
|
||
"retraction_details": None,
|
||
"chemical_list": [{"name": c.get("name", ""), "registry_number": c.get("registryNumber", ""), "mesh_ui": c.get("ui", "")}
|
||
for c in (art.get("chemicalList") or []) if isinstance(c, dict) and c.get("name")],
|
||
"gene_symbols": [g for g in (art.get("geneSymbolList") or []) if g],
|
||
"keywords": keywords,
|
||
"num_refs": art.get("numReferences"),
|
||
"publication_status": art.get("publicationStatus"),
|
||
"article_date": None,
|
||
"print_date": None,
|
||
"create_date": None,
|
||
"entrez_date": None,
|
||
"databank_list": [{"databank_name": d.get("dataBankName", ""),
|
||
"accession_numbers": d.get("accessionNumberList", []) or []}
|
||
for d in (art.get("dataBankList") or []) if isinstance(d, dict) and d.get("dataBankName")],
|
||
"suppl_mesh_list": [{"descriptor": s.get("name", s.get("supplMeshName", "")),
|
||
"ui": s.get("ui", s.get("supplMeshUI", "")),
|
||
"type": s.get("type", s.get("supplMeshType", ""))}
|
||
for s in (art.get("supplMeshList") or []) if isinstance(s, dict) and (s.get("name") or s.get("supplMeshName"))],
|
||
"pharmacological_actions": [],
|
||
"cois_statement": None,
|
||
"vernacular_title": None,
|
||
"is_preprint": (
|
||
(doi or "").lower().startswith(_PREPRINT_DOI_PREFIXES)
|
||
or (art.get("journalTitle") or "").lower() in _PREPRINT_JOURNALS
|
||
),
|
||
"source": "europe_pmc",
|
||
}
|
||
|
||
|
||
async def search_europe_pmc_articles(
|
||
query: str,
|
||
max_results: int = 1000,
|
||
page_size: int = 1000,
|
||
) -> list[dict]:
|
||
"""搜索 Europe PMC，返回完整文献 dict 列表（与 `fetch_articles()` 输出格式一致）。
|
||
|
||
- 使用 cursor-based 分页，适合大数据量场景（每日增量 >5000 篇时）
|
||
- 无 API Key 限制，10+ req/s
|
||
- 单步返回完整元数据（无需分两步 esearch + efetch）
|
||
- 返回结果的 citedByCount 自动缓存供后续引用更新
|
||
"""
|
||
if not query.strip():
|
||
return []
|
||
|
||
results = []
|
||
cursor = "*"
|
||
ps = min(page_size, 1000)
|
||
actual_limit = min(max_results, 100_000)
|
||
|
||
async with httpx.AsyncClient(timeout=30) as client:
|
||
while len(results) < actual_limit:
|
||
params = {
|
||
"query": query,
|
||
"resultType": "core",
|
||
"pageSize": ps,
|
||
"cursor": cursor,
|
||
"format": "json",
|
||
}
|
||
try:
|
||
async with _EUROPE_PMC_SEM:
|
||
r = await _fetch_with_retry(client, EUROPE_PMC_SEARCH_URL, params)
|
||
data = r.json()
|
||
except Exception:
|
||
logger.warning("europe_pmc_search_failed query=%s cursor=%s", query[:60], cursor, exc_info=True)
|
||
break
|
||
|
||
hit_count = data.get("hitCount", 0)
|
||
if hit_count == 0:
|
||
break
|
||
|
||
articles = data.get("resultList", {}).get("result", [])
|
||
if not articles:
|
||
break
|
||
|
||
for art in articles:
|
||
if len(results) >= actual_limit:
|
||
break
|
||
parsed = _parse_europe_pmc_article(art)
|
||
if parsed:
|
||
results.append(parsed)
|
||
|
||
next_cursor = data.get("nextCursorMark")
|
||
if not next_cursor or next_cursor == cursor:
|
||
break
|
||
cursor = next_cursor
|
||
|
||
# 避免无限循环——Europe PMC 的 nextCursorMark 不会变空
|
||
if len(results) >= hit_count:
|
||
break
|
||
|
||
logger.info("europe_pmc_search: query=%s results=%d", query[:60], len(results))
|
||
return results
|
||
|
||
|
||
async def fetch_articles(pmids: list[int]) -> list[dict]:
|
||
"""批量获取文献详情（XML → 结构化 dict）"""
|
||
if not pmids:
|
||
return []
|
||
|
||
params = {
|
||
"db": "pubmed",
|
||
"id": ",".join(str(p) for p in pmids),
|
||
"retmode": "xml",
|
||
"tool": TOOL_NAME,
|
||
"email": TOOL_EMAIL,
|
||
}
|
||
if API_KEY:
|
||
params["api_key"] = API_KEY
|
||
|
||
async with httpx.AsyncClient(timeout=30) as client:
|
||
r = await _fetch_with_retry(client, EFETCH_URL, params)
|
||
return _parse_pubmed_xml(r.text)
|
||
|
||
|
||
def _parse_pubmed_xml(xml_text: str) -> list[dict]:
|
||
"""解析 PubMed XML，返回文献 dict 列表"""
|
||
root = ET.fromstring(xml_text)
|
||
articles = []
|
||
|
||
for article_elem in root.findall(".//PubmedArticle"):
|
||
try:
|
||
medline = article_elem.find(".//MedlineCitation")
|
||
if medline is None:
|
||
continue
|
||
article = medline.find(".//Article")
|
||
if article is None:
|
||
continue
|
||
|
||
# PMID
|
||
pmid_elem = medline.find(".//PMID")
|
||
pmid = int(pmid_elem.text) if pmid_elem is not None and pmid_elem.text else None
|
||
if pmid is None:
|
||
continue
|
||
|
||
# Title
|
||
title_elem = article.find(".//ArticleTitle")
|
||
title = "".join(title_elem.itertext()).strip() if title_elem is not None else ""
|
||
|
||
# Abstract
|
||
abstract_parts = []
|
||
for at in article.findall(".//Abstract/AbstractText"):
|
||
label = at.get("Label", "")
|
||
text = "".join(at.itertext()).strip()
|
||
if label:
|
||
abstract_parts.append(f"{label}: {text}")
|
||
else:
|
||
abstract_parts.append(text)
|
||
abstract = "\n".join(abstract_parts) if abstract_parts else None
|
||
|
||
# Authors
|
||
authors = []
|
||
for au in article.findall(".//Author"):
|
||
last = au.findtext("LastName", "")
|
||
fore = au.findtext("ForeName", "")
|
||
aff_elem = au.find(".//AffiliationInfo/Affiliation")
|
||
aff = aff_elem.text if aff_elem is not None else ""
|
||
# Author Identifiers (ORCID etc.)
|
||
identifiers = []
|
||
for id_elem in au.findall("Identifier"):
|
||
id_source = id_elem.get("Source", "")
|
||
id_val = id_elem.text or ""
|
||
if id_source and id_val:
|
||
identifiers.append({"type": id_source, "value": id_val})
|
||
entry = {"family": last, "given": fore, "affiliation": aff}
|
||
if identifiers:
|
||
entry["identifiers"] = identifiers
|
||
# Type="editor" in AuthorList
|
||
if au.get("Type") == "editor":
|
||
entry["type"] = "editor"
|
||
authors.append(entry)
|
||
|
||
# InvestigatorList（多中心试验研究者）
|
||
investigators = []
|
||
for inv in article.findall(".//InvestigatorList/Investigator"):
|
||
last = inv.findtext("LastName", "")
|
||
fore = inv.findtext("ForeName", "")
|
||
aff_elem = inv.find(".//AffiliationInfo/Affiliation")
|
||
aff = aff_elem.text if aff_elem is not None else ""
|
||
identifiers = []
|
||
for id_elem in inv.findall("Identifier"):
|
||
id_source = id_elem.get("Source", "")
|
||
id_val = id_elem.text or ""
|
||
if id_source and id_val:
|
||
identifiers.append({"type": id_source, "value": id_val})
|
||
entry = {"family": last, "given": fore, "affiliation": aff}
|
||
if identifiers:
|
||
entry["identifiers"] = identifiers
|
||
investigators.append(entry)
|
||
|
||
# DOI
|
||
doi = None
|
||
for eid in article.findall(".//ELocationID"):
|
||
if eid.get("EIdType") == "doi":
|
||
doi = eid.text
|
||
break
|
||
|
||
# PMCID
|
||
pmc_id = None
|
||
for aid in article_elem.findall(".//PubmedData/ArticleIdList/ArticleId"):
|
||
if aid.get("IdType") == "pmc":
|
||
pmc_id = (aid.text or "").removeprefix("PMC")
|
||
break
|
||
|
||
# 临床试验注册号
|
||
trial_reg = {}
|
||
for aid in article_elem.findall(".//PubmedData/ArticleIdList/ArticleId"):
|
||
id_type = aid.get("IdType")
|
||
if id_type == "ClinicalTrials.gov":
|
||
trial_reg["nct"] = aid.text
|
||
elif id_type == "EU-CTR":
|
||
trial_reg["eudract"] = aid.text
|
||
elif id_type == "ChiCTR":
|
||
trial_reg["chictr"] = aid.text
|
||
|
||
# DateRevised（PubMed 修订日期）
|
||
pubmed_revised = None
|
||
dr_elem = medline.find(".//DateRevised")
|
||
if dr_elem is not None:
|
||
try:
|
||
yr = int(dr_elem.findtext("Year", "0"))
|
||
mo = int(dr_elem.findtext("Month", "1"))
|
||
dy = int(dr_elem.findtext("Day", "1"))
|
||
pubmed_revised = date(yr, mo, dy)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# CitationStatus（文献状态：publisher / in-data-review / in-process / medline / ...）
|
||
citation_status = medline.get("Status")
|
||
|
||
# DateCompleted（MeSH 标引完成日，即"成为成品"的日期）
|
||
date_completed = None
|
||
dc_elem = medline.find(".//DateCompleted")
|
||
if dc_elem is not None:
|
||
try:
|
||
yr = int(dc_elem.findtext("Year", "0"))
|
||
mo = int(dc_elem.findtext("Month", "1"))
|
||
dy = int(dc_elem.findtext("Day", "1"))
|
||
date_completed = date(yr, mo, dy)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# ─── CommentsCorrections（撤稿 / 修正 / 关注）───
|
||
retracted = False
|
||
retraction_details = None
|
||
for cc in medline.findall(".//CommentsCorrections"):
|
||
ref_type = cc.findtext("RefType")
|
||
ref_pmid_elem = cc.find("PMID")
|
||
ref_pmid = int(ref_pmid_elem.text) if ref_pmid_elem is not None and ref_pmid_elem.text else None
|
||
if ref_type in ("RetractedBy", "RetractionOf"):
|
||
retraction_details = {
|
||
"ref_type": ref_type,
|
||
"ref_pmid": ref_pmid,
|
||
"ref_source": cc.findtext("RefSource"),
|
||
}
|
||
if ref_type == "RetractedBy":
|
||
retracted = True
|
||
|
||
# Keywords（作者关键词）
|
||
keywords = []
|
||
for kw in article.findall(".//KeywordList/Keyword"):
|
||
if kw.text:
|
||
keywords.append(kw.text)
|
||
|
||
# Grants（基金资助）
|
||
grants = []
|
||
for gr in article.findall(".//GrantList/Grant"):
|
||
gr_abbrev = gr.findtext("Acronym") or ""
|
||
gr_agency = gr.findtext("Agency") or ""
|
||
gr_country = gr.get("Country", "")
|
||
gr_number = gr.findtext("GrantID") or ""
|
||
if gr_abbrev or gr_agency or gr_number:
|
||
grants.append({
|
||
"abbreviation": gr_abbrev,
|
||
"agency": gr_agency,
|
||
"country": gr_country,
|
||
"grant_id": gr_number,
|
||
})
|
||
|
||
# Journal
|
||
journal_elem = article.find(".//Journal")
|
||
journal_title = None
|
||
issn = None
|
||
journal_iso = None
|
||
volume = None
|
||
issue = None
|
||
pub_date = None
|
||
print_date = None
|
||
pub_year = None
|
||
pages = ""
|
||
|
||
if journal_elem is not None:
|
||
journal_title = journal_elem.findtext("Title")
|
||
journal_iso = journal_elem.findtext("ISOAbbreviation")
|
||
issn_elem = journal_elem.find(".//ISSN")
|
||
issn = issn_elem.text if issn_elem is not None else None
|
||
|
||
ji = journal_elem.find(".//JournalIssue")
|
||
if ji is not None:
|
||
volume = ji.findtext("Volume")
|
||
issue = ji.findtext("Issue")
|
||
pd_elem = ji.find(".//PubDate")
|
||
if pd_elem is not None:
|
||
y = pd_elem.findtext("Year")
|
||
if y:
|
||
pub_year = int(y)
|
||
m = pd_elem.findtext("Month", "Jan")
|
||
d = pd_elem.findtext("Day", "01")
|
||
try:
|
||
month_map = {"jan":1,"feb":2,"mar":3,"apr":4,"may":5,"jun":6,
|
||
"jul":7,"aug":8,"sep":9,"oct":10,"nov":11,"dec":12}
|
||
mi = month_map.get(m.lower()[:3], 1)
|
||
di = min(int(d) if d.isdigit() else 1, 31)
|
||
pub_date = date(int(y), mi, di)
|
||
print_date = pub_date # 原始 PubDate 作为 PPDAT
|
||
except (ValueError, TypeError):
|
||
pub_date = date(int(y), 1, 1)
|
||
print_date = pub_date
|
||
else:
|
||
md = pd_elem.findtext("MedlineDate")
|
||
if md:
|
||
import re
|
||
ym = re.search(r'(\d{4})', md)
|
||
if ym:
|
||
pub_year = int(ym.group(1))
|
||
|
||
pagination = article.findtext(".//Pagination/MedlinePgn")
|
||
if pagination:
|
||
pages = pagination
|
||
|
||
# Publication types
|
||
pub_types = []
|
||
for pt in article.findall(".//PublicationTypeList/PublicationType"):
|
||
if pt.text:
|
||
pub_types.append(pt.text)
|
||
|
||
# MeSH Headings（核心：用于后续打标）
|
||
mesh_headings = []
|
||
for mh in medline.findall(".//MeshHeadingList/MeshHeading"):
|
||
desc = mh.findtext("DescriptorName")
|
||
dn = mh.find("DescriptorName")
|
||
ui = dn.get("UI", "") if dn is not None else ""
|
||
major = dn.get("MajorTopicYN", "N") == "Y" if dn is not None else False
|
||
qualifiers = [q.text for q in mh.findall("QualifierName") if q.text]
|
||
mesh_headings.append({
|
||
"descriptor": desc,
|
||
"ui": ui,
|
||
"major": major,
|
||
"qualifiers": qualifiers,
|
||
})
|
||
|
||
# Language
|
||
lang_elem = article.find(".//Language")
|
||
language = lang_elem.text if lang_elem is not None else "en"
|
||
|
||
# ─── ChemicalList（化合物列表）───
|
||
chemical_list = []
|
||
for ch in medline.findall(".//ChemicalList/Chemical"):
|
||
rn = ch.findtext("RegistryNumber", "")
|
||
ns = ch.find("NameOfSubstance")
|
||
name = ns.text if ns is not None else ""
|
||
ui = ns.get("UI", "") if ns is not None else ""
|
||
chemical_list.append({
|
||
"name": name,
|
||
"registry_number": rn,
|
||
"mesh_ui": ui,
|
||
})
|
||
|
||
# ─── PharmacologicalAction（药理作用）───
|
||
pharmacological_actions = []
|
||
for pa in medline.findall(".//PharmacologicalAction/NameOfSubstance"):
|
||
name = pa.text or ""
|
||
ui = pa.get("UI", "")
|
||
if name:
|
||
pharmacological_actions.append({"name": name, "ui": ui})
|
||
|
||
# ─── GeneSymbolList（基因符号）───
|
||
gene_symbols = [gs.text for gs in article.findall(".//GeneSymbolList/GeneSymbol") if gs.text]
|
||
|
||
# ─── NumberOfReferences（参考文献数）───
|
||
num_refs = None
|
||
nr_elem = medline.find(".//NumberOfReferences")
|
||
if nr_elem is not None and nr_elem.text:
|
||
try:
|
||
num_refs = int(nr_elem.text)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# ─── PublicationStatus（出版阶段）───
|
||
publication_status = None
|
||
ps_elem = article_elem.find(".//PubmedData/PublicationStatus")
|
||
if ps_elem is not None:
|
||
publication_status = ps_elem.text
|
||
|
||
# ─── ArticleDate（电子出版日期）───
|
||
article_date = None
|
||
ad_elem = article.find(".//ArticleDate")
|
||
if ad_elem is not None:
|
||
try:
|
||
ayr = int(ad_elem.findtext("Year", "0"))
|
||
amo = int(ad_elem.findtext("Month", "1"))
|
||
ady = int(ad_elem.findtext("Day", "1"))
|
||
article_date = date(ayr, amo, ady)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# ─── 修复：无 ArticleDate 时，PubDate 的 Day 可能是 NCBI 默认（当月最后一天）───
|
||
if article_date is None and pub_date is not None:
|
||
import calendar
|
||
last = calendar.monthrange(pub_date.year, pub_date.month)[1]
|
||
if pub_date.day == last:
|
||
pub_date = date(pub_date.year, pub_date.month, 1)
|
||
|
||
# ─── PubmedData/History → create_date [CRDT] + entrez_date [EDAT] ───
|
||
create_date = None
|
||
entrez_date = None
|
||
for hist in article_elem.findall(".//PubmedData/History/PubMedPubDate"):
|
||
status = hist.get("PubStatus", "")
|
||
h_year = hist.findtext("Year")
|
||
if h_year and h_year.isdigit():
|
||
h_mo = int(hist.findtext("Month", "1"))
|
||
h_dy = int(hist.findtext("Day", "1"))
|
||
h_dt = date(int(h_year), h_mo, h_dy)
|
||
if status == "pubmed":
|
||
create_date = h_dt
|
||
elif status == "entrez":
|
||
entrez_date = h_dt
|
||
|
||
# ─── DataBankList（数据库引用）───
|
||
databank_list = []
|
||
for db in medline.findall(".//DataBankList/DataBank"):
|
||
db_name = db.findtext("DataBankName", "")
|
||
accessions = [an.text for an in db.findall(".//AccessionNumberList/AccessionNumber") if an.text]
|
||
if db_name:
|
||
databank_list.append({
|
||
"databank_name": db_name,
|
||
"accession_numbers": accessions,
|
||
})
|
||
|
||
# ─── SupplMeshList（补充 MeSH）───
|
||
suppl_mesh_list = []
|
||
for sm in medline.findall(".//SupplMeshList/SupplMeshName"):
|
||
name = sm.text or ""
|
||
ui = sm.get("UI", "")
|
||
stype = sm.get("Type", "")
|
||
suppl_mesh_list.append({
|
||
"descriptor": name,
|
||
"ui": ui,
|
||
"type": stype,
|
||
})
|
||
|
||
# ─── CoiStatement（利益冲突声明）───
|
||
cois_elem = article.find(".//CoiStatement")
|
||
cois_statement = cois_elem.text.strip() if cois_elem is not None and cois_elem.text else None
|
||
|
||
# ─── VernacularTitle（非英文标题的拉丁转写）───
|
||
vt_elem = article.find(".//VernacularTitle")
|
||
vernacular_title = vt_elem.text.strip() if vt_elem is not None and vt_elem.text else None
|
||
|
||
# ─── PersonalNameSubjectList（作为主题的人名）───
|
||
personal_name_subjects = []
|
||
for pns in medline.findall(".//PersonalNameSubjectList/PersonalNameSubject"):
|
||
last = pns.findtext("LastName", "")
|
||
fore = pns.findtext("ForeName", "")
|
||
if last or fore:
|
||
personal_name_subjects.append({"family": last, "given": fore})
|
||
|
||
# ─── PubmedData/PublicationNote（出版注释）───
|
||
publication_notes = []
|
||
for pn in article_elem.findall(".//PubmedData/PublicationNote"):
|
||
if pn.text and pn.text.strip():
|
||
publication_notes.append(pn.text.strip())
|
||
|
||
articles.append({
|
||
"pmid": pmid,
|
||
"title": title.strip(),
|
||
"abstract": abstract.strip() if abstract else None,
|
||
"authors": authors,
|
||
"doi": doi,
|
||
"pmc_id": pmc_id,
|
||
"is_oa": pmc_id is not None,
|
||
"pubmed_revised": pubmed_revised,
|
||
"citation_status": citation_status,
|
||
"meshed_date": date_completed, # 首次导入=date_completed，年更后刷新
|
||
"keywords": keywords,
|
||
"grants": grants,
|
||
"journal": journal_title,
|
||
"journal_issn": issn,
|
||
"journal_iso": journal_iso,
|
||
"volume": volume,
|
||
"issue": issue,
|
||
"pages": pages or None,
|
||
"pub_date": pub_date,
|
||
"pub_year": pub_year,
|
||
"pub_types": pub_types,
|
||
"mesh_headings": mesh_headings,
|
||
"language": language,
|
||
"trial_reg": trial_reg or None,
|
||
"retracted": retracted,
|
||
"retraction_details": retraction_details,
|
||
"chemical_list": chemical_list,
|
||
"gene_symbols": gene_symbols,
|
||
"num_refs": num_refs,
|
||
"publication_status": publication_status,
|
||
"article_date": article_date,
|
||
"print_date": print_date,
|
||
"create_date": create_date,
|
||
"entrez_date": entrez_date,
|
||
"databank_list": databank_list,
|
||
"suppl_mesh_list": suppl_mesh_list,
|
||
"pharmacological_actions": pharmacological_actions,
|
||
"cois_statement": cois_statement,
|
||
"vernacular_title": vernacular_title,
|
||
"is_preprint": (
|
||
(doi or "").lower().startswith(_PREPRINT_DOI_PREFIXES)
|
||
or (journal_title or "").lower() in _PREPRINT_JOURNALS
|
||
),
|
||
"investigators": investigators,
|
||
"personal_name_subjects": personal_name_subjects,
|
||
"publication_notes": publication_notes,
|
||
"source": "pubmed_api",
|
||
})
|
||
except Exception:
|
||
logger.warning("parse_article_failed", exc_info=True)
|
||
continue
|
||
|
||
return articles
|
||
|
||
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# 引用计数 — PubMed elink (免费, 3/10 req/s)
|
||
# 文档: https://www.ncbi.nlm.nih.gov/books/NBK25499/#chapter4.ElLink
|
||
# ═══════════════════════════════════════════════
|
||
|
||
ELINK_RATE_LIMIT = 0.34 # ≈ 3 req/s (有 API key 则 0.1 = 10 req/s)
|
||
|
||
|
||
async def fetch_citedby_counts(pmids: list[int]) -> dict[int, int]:
|
||
"""批量查询引用次数，返回 {pmid: count}"""
|
||
if not pmids:
|
||
return {}
|
||
|
||
# 去重并保留顺序
|
||
unique = list(dict.fromkeys(pmids))
|
||
result: dict[int, int] = {}
|
||
client_kw = {"timeout": 30}
|
||
|
||
async with httpx.AsyncClient(**client_kw) as client:
|
||
for i in range(0, len(unique), 200):
|
||
batch = unique[i:i + 200]
|
||
params = {
|
||
"dbfrom": "pubmed",
|
||
"db": "pubmed",
|
||
"linkname": "pubmed_pubmed_citedin",
|
||
"id": ",".join(str(p) for p in batch),
|
||
"retmode": "xml",
|
||
"tool": TOOL_NAME,
|
||
"email": TOOL_EMAIL,
|
||
}
|
||
if API_KEY:
|
||
params["api_key"] = API_KEY
|
||
|
||
try:
|
||
r = await _fetch_with_retry(client, ELINK_URL, params)
|
||
|
||
# 解析 XML：elink 返回 <LinkSetSet> 包含多个 <LinkSet>
|
||
root = ET.fromstring(r.text)
|
||
for linkset in root.iter("LinkSet"):
|
||
# 取源 PMID：IdUrlList/IdUrl/Id（旧格式）或 IdList/Id（新格式，2025年起 NCBI 简化了 XML）
|
||
src_el = linkset.findtext("IdUrlList/IdUrl/Id") or linkset.findtext("IdList/Id")
|
||
if not src_el:
|
||
continue
|
||
source_pmid = int(src_el.strip())
|
||
|
||
# 统计 cited-by 数：LinkSetDb[LinkName=pubmed_pubmed_citedin] -> Link -> Id
|
||
count = 0
|
||
for lsdb in linkset.findall("LinkSetDb"):
|
||
ln = lsdb.findtext("LinkName", "")
|
||
if ln == "pubmed_pubmed_citedin":
|
||
count = len([e for e in lsdb.findall("Link/Id") if e.text])
|
||
break
|
||
result[source_pmid] = max(result.get(source_pmid, 0), count)
|
||
|
||
except Exception:
|
||
logger.warning("elink_failed batch size=%d offset=%d", len(batch), i, exc_info=True)
|
||
# 部分失败不中断，已获取的结果保留
|
||
|
||
# 限速
|
||
delay = ELINK_RATE_LIMIT if not API_KEY else 0.11
|
||
await asyncio.sleep(delay)
|
||
|
||
return result
|
||
|
||
|
||
async def update_citation_counts(
|
||
db: AsyncSession,
|
||
pmids: list[int] | None = None,
|
||
*,
|
||
limit: int = 0,
|
||
) -> dict:
|
||
"""更新被引次数。pmids=None 时更新全库 (limit=0=全部)。返回 {updated, total, errors}"""
|
||
if pmids:
|
||
targets = pmids
|
||
else:
|
||
q = select(GlobalLiterature.pmid).order_by(GlobalLiterature.created_at.desc())
|
||
if limit > 0:
|
||
q = q.limit(limit)
|
||
result = await db.execute(q)
|
||
targets = [r for (r,) in result]
|
||
|
||
if not targets:
|
||
return {"updated": 0, "total": 0, "errors": 0}
|
||
|
||
counts = await fetch_citedby_counts(targets)
|
||
updated = 0
|
||
errors = 0
|
||
now = datetime.now(UTC)
|
||
for pmid, count in counts.items():
|
||
try:
|
||
await db.execute(
|
||
GlobalLiterature.__table__.update()
|
||
.where(GlobalLiterature.pmid == pmid)
|
||
.values(cited_by_count=count, updated_at=now)
|
||
)
|
||
updated += 1
|
||
except Exception:
|
||
errors += 1
|
||
logger.warning("citation_update_failed pmid=%d", pmid, exc_info=True)
|
||
|
||
if updated:
|
||
await db.commit()
|
||
|
||
return {"updated": updated, "total": len(targets), "errors": errors}
|
||
|
||
|
||
def _update_lit_from_article(lit: GlobalLiterature, article: dict):
|
||
"""用 PubMed 返回的最新数据更新已存在的文献记录"""
|
||
fields = [
|
||
"title", "abstract", "authors", "doi", "journal", "journal_issn", "journal_iso",
|
||
"volume", "issue", "pages", "pub_date", "pub_year", "pub_types",
|
||
"mesh_headings", "language", "pmc_id", "keywords", "grants",
|
||
"pubmed_revised", "trial_reg", "citation_status", "date_completed", "meshed_date",
|
||
"retracted", "retraction_details",
|
||
"chemical_list", "gene_symbols", "num_refs", "publication_status",
|
||
"article_date", "print_date", "create_date", "entrez_date",
|
||
"databank_list", "suppl_mesh_list",
|
||
"pharmacological_actions",
|
||
"cois_statement", "vernacular_title",
|
||
"investigators", "personal_name_subjects", "publication_notes",
|
||
"is_preprint",
|
||
]
|
||
for f in fields:
|
||
val = article.get(f)
|
||
if val is not None and val != "" and val != [] and val != {}:
|
||
setattr(lit, f, val)
|
||
lit.is_oa = article.get("pmc_id") is not None
|
||
# 同步更新研究设计分类
|
||
if article.get("pub_types"):
|
||
lit.study_design = classify_study_design(article["pub_types"])
|
||
# RCT 检测（两档可信度）
|
||
lit.rct_detection = detect_rct(
|
||
article.get("pub_types"), article.get("title"),
|
||
article.get("abstract"), article.get("pub_year"),
|
||
)
|
||
# 阴性结果检测
|
||
if article.get("abstract") or article.get("title"):
|
||
from app.services.negative_detector import detect_negative_result
|
||
# 每次重新抓取后重置 is_negative_result，避免旧标记永不清零
|
||
lit.is_negative_result = False
|
||
lit.negative_result_details = None
|
||
neg = detect_negative_result(article.get("abstract"), article.get("title", ""))
|
||
if neg.get("is_negative"):
|
||
lit.is_negative_result = True
|
||
lit.negative_result_details = neg
|
||
|
||
|
||
async def _run_pipeline(
|
||
db: AsyncSession,
|
||
max_per_query: int,
|
||
use_majr: bool,
|
||
use_broad: bool,
|
||
precision_mode: str = "majr",
|
||
update_citations: bool = True,
|
||
use_europe_pmc: bool = False,
|
||
) -> dict:
|
||
"""管道核心：执行查询列表 + 入库/更新 + 打标 + Feed + 被引。返回统计 dict
|
||
|
||
当 use_europe_pmc=True 时使用 Europe PMC API（cursor 分页，适合大数据量）；
|
||
False 时使用 NCBI E-utilities（默认，向后兼容）。
|
||
"""
|
||
# 记录管线开始时间，供 PipelineRun duration 计算
|
||
_started_at = datetime.now(UTC)
|
||
stats = {"searched": 0, "fetched": 0, "new": 0, "updated": 0, "tagged": 0, "feeds": 0, "citations": 0, "oa_fetched": 0, "started_at": _started_at}
|
||
|
||
all_new_pmids: list[int] = []
|
||
oa_pending: dict[int, str] = {} # pmid -> pmc_id（待抓全文的新 OA 文献）
|
||
seen_pmids: set[int] = set()
|
||
|
||
async def _process_article(article: dict, existing_set: set, pmid_to_lit: dict) -> None:
|
||
"""处理单条文献：更新已存在的或创建新记录
|
||
|
||
DOI 跨源去重：当 PMID 不存在但 DOI 已在数据库中时，
|
||
更新已有记录（补充 PMID/PMCID 等），避免同文献多源重复。
|
||
"""
|
||
nonlocal stats
|
||
if article["pmid"] in existing_set:
|
||
lit = pmid_to_lit[article["pmid"]]
|
||
_update_lit_from_article(lit, article)
|
||
await tag_article(db, lit.id, article.get("mesh_headings", []))
|
||
stats["updated"] += 1
|
||
else:
|
||
# 二次检查：防止并发/间发的 duplicate INSERT
|
||
dup = await db.execute(
|
||
select(GlobalLiterature).where(GlobalLiterature.pmid == article["pmid"])
|
||
)
|
||
dup_lit = dup.scalar()
|
||
if dup_lit:
|
||
_update_lit_from_article(dup_lit, article)
|
||
await tag_article(db, dup_lit.id, article.get("mesh_headings", []))
|
||
stats["updated"] += 1
|
||
return
|
||
|
||
# DOI 跨源去重：检查数据库中是否存在同 DOI 但 PMID 不同的记录
|
||
doi_dedup_lit = None
|
||
doi = article.get("doi")
|
||
if doi:
|
||
doi_result = await db.execute(
|
||
select(GlobalLiterature).where(
|
||
GlobalLiterature.doi == doi,
|
||
GlobalLiterature.pmid != article["pmid"],
|
||
)
|
||
)
|
||
doi_dedup_lit = doi_result.scalar()
|
||
if doi_dedup_lit:
|
||
# DOI 相同但 PMID 不同——视为同篇文献，保留已有记录并补充新数据
|
||
_update_lit_from_article(doi_dedup_lit, article)
|
||
await tag_article(db, doi_dedup_lit.id, article.get("mesh_headings", []))
|
||
existing_set.add(article["pmid"])
|
||
pmid_to_lit[article["pmid"]] = doi_dedup_lit
|
||
stats["updated"] += 1
|
||
return
|
||
|
||
lit = GlobalLiterature(
|
||
id=_uid(), pmid=article["pmid"], title=article["title"],
|
||
abstract=article.get("abstract"), authors=article.get("authors", []),
|
||
doi=article.get("doi"), journal=article.get("journal"),
|
||
journal_issn=article.get("journal_issn"),
|
||
journal_iso=article.get("journal_iso"),
|
||
volume=article.get("volume"), issue=article.get("issue"),
|
||
pages=article.get("pages"), pub_date=article.get("pub_date"),
|
||
pub_year=article.get("pub_year"),
|
||
pub_types=article.get("pub_types", []),
|
||
mesh_headings=article.get("mesh_headings", []),
|
||
language=article.get("language", "en"),
|
||
pmc_id=article.get("pmc_id"), is_oa=article.get("is_oa", False),
|
||
keywords=article.get("keywords", []),
|
||
pubmed_revised=article.get("pubmed_revised"),
|
||
citation_status=article.get("citation_status"),
|
||
date_completed=article.get("date_completed"),
|
||
meshed_date=article.get("meshed_date"),
|
||
grants=article.get("grants", []),
|
||
trial_reg=article.get("trial_reg"),
|
||
study_design=classify_study_design(article.get("pub_types", [])),
|
||
rct_detection=detect_rct(
|
||
article.get("pub_types"), article.get("title"),
|
||
article.get("abstract"), article.get("pub_year"),
|
||
),
|
||
chemical_list=article.get("chemical_list", []),
|
||
gene_symbols=article.get("gene_symbols", []),
|
||
num_refs=article.get("num_refs"),
|
||
publication_status=article.get("publication_status"),
|
||
article_date=article.get("article_date"),
|
||
print_date=article.get("print_date"),
|
||
create_date=article.get("create_date"),
|
||
entrez_date=article.get("entrez_date"),
|
||
databank_list=article.get("databank_list", []),
|
||
suppl_mesh_list=article.get("suppl_mesh_list", []),
|
||
pharmacological_actions=article.get("pharmacological_actions", []),
|
||
cois_statement=article.get("cois_statement"),
|
||
vernacular_title=article.get("vernacular_title"),
|
||
is_preprint=article.get("is_preprint", False),
|
||
investigators=article.get("investigators", []),
|
||
personal_name_subjects=article.get("personal_name_subjects", []),
|
||
publication_notes=article.get("publication_notes", []),
|
||
source=article.get("source", "pubmed_api"),
|
||
raw_xml_hash=hashlib.sha256(article["title"].encode()).hexdigest()[:16],
|
||
)
|
||
db.add(lit)
|
||
await db.flush()
|
||
tagged = await tag_article(db, lit.id, article.get("mesh_headings", []))
|
||
stats["tagged"] += tagged
|
||
stats["new"] += 1
|
||
all_new_pmids.append(article["pmid"])
|
||
pmc_id = article.get("pmc_id")
|
||
if pmc_id:
|
||
oa_pending[article["pmid"]] = pmc_id
|
||
feeds = await generate_feeds_for_literature(db, lit.id)
|
||
stats["feeds"] += feeds
|
||
if feeds > 0:
|
||
try:
|
||
from app.core.websocket import ws_manager
|
||
await ws_manager.send_new_literature_alert(
|
||
article["title"], article["pmid"], article.get("journal", ""))
|
||
except Exception:
|
||
logger.warning("ws_push_failed pmid=%s", article.get("pmid"), exc_info=True)
|
||
|
||
# 确保期刊在 global_journals 中存在
|
||
try:
|
||
await ensure_journal_async(
|
||
db, article.get("journal_issn"),
|
||
article.get("journal"),
|
||
)
|
||
except Exception:
|
||
logger.warning("journal_ensure_failed pmid=%s", article.get("pmid"), exc_info=True)
|
||
|
||
async def _process_queries(query_list: list[str]):
|
||
nonlocal stats
|
||
for query in query_list:
|
||
try:
|
||
if use_europe_pmc:
|
||
articles = await search_europe_pmc_articles(query, max_results=max_per_query)
|
||
pmids = [a["pmid"] for a in articles if a.get("pmid")]
|
||
else:
|
||
pmids = await search_pmids(query, max_results=max_per_query)
|
||
fresh = [p for p in pmids if p not in seen_pmids]
|
||
seen_pmids.update(pmids)
|
||
if not fresh:
|
||
continue
|
||
stats["searched"] += len(fresh)
|
||
|
||
existing_set = set()
|
||
pmid_to_lit: dict[int, GlobalLiterature] = {}
|
||
result = await db.execute(
|
||
select(GlobalLiterature).where(GlobalLiterature.pmid.in_(fresh))
|
||
)
|
||
for lit in result.scalars().all():
|
||
existing_set.add(lit.pmid)
|
||
pmid_to_lit[lit.pmid] = lit
|
||
|
||
new_pmids = [p for p in fresh if p not in existing_set]
|
||
old_pmids = [p for p in fresh if p in existing_set]
|
||
fetch_targets = new_pmids + old_pmids
|
||
|
||
if use_europe_pmc:
|
||
# search_europe_pmc_articles 已返回完整数据，按 pmid 索引即可
|
||
art_by_pmid = {a["pmid"]: a for a in articles if a.get("pmid")}
|
||
# 只处理 fetch_targets 中的文献
|
||
for batch_start in range(0, len(fetch_targets), 50):
|
||
batch = fetch_targets[batch_start:batch_start+50]
|
||
batch_articles = [art_by_pmid[p] for p in batch if p in art_by_pmid]
|
||
stats["fetched"] += len(batch_articles)
|
||
for article in batch_articles:
|
||
await _process_article(article, existing_set, pmid_to_lit)
|
||
await db.commit()
|
||
await asyncio.sleep(0.4)
|
||
else:
|
||
for batch_start in range(0, len(fetch_targets), 50):
|
||
batch = fetch_targets[batch_start:batch_start+50]
|
||
articles = await fetch_articles(batch)
|
||
stats["fetched"] += len(articles)
|
||
for article in articles:
|
||
await _process_article(article, existing_set, pmid_to_lit)
|
||
await db.commit()
|
||
await asyncio.sleep(0.4)
|
||
|
||
except (httpx.TimeoutException, httpx.HTTPStatusError, httpx.NetworkError) as e:
|
||
logger.warning("Query error (%s): %s", query[:60], e)
|
||
continue
|
||
|
||
if use_majr:
|
||
await _process_queries(_build_oncology_queries(precision_mode))
|
||
if use_broad:
|
||
await _process_queries(_build_broad_queries())
|
||
|
||
# OA 全文抓取（新创建的 OA 文献）
|
||
if oa_pending:
|
||
oa_ok = await _fetch_oa_fulltexts(db, oa_pending)
|
||
stats["oa_fetched"] = oa_ok
|
||
|
||
if all_new_pmids and update_citations:
|
||
try:
|
||
cit = await update_citation_counts(db, all_new_pmids)
|
||
stats["citations"] = cit["updated"]
|
||
except Exception:
|
||
logger.warning("citation_update_skipped", exc_info=True)
|
||
|
||
return stats
|
||
|
||
|
||
async def _fetch_oa_fulltexts(db: AsyncSession, pending: dict[int, str]) -> int:
|
||
"""批量抓取 PMC OA 全文 XML → 解析 → 写入 COS（或 PG 降级）"""
|
||
use_cos = bool(settings.COS_SECRET_ID)
|
||
ok = 0
|
||
for pmid, pmc_id in pending.items():
|
||
xml = await fetch_pmc_xml(pmc_id)
|
||
if not xml:
|
||
continue
|
||
lit = (await db.execute(
|
||
select(GlobalLiterature).where(GlobalLiterature.pmid == pmid)
|
||
)).scalar()
|
||
if not lit:
|
||
continue
|
||
try:
|
||
parsed = parse_full_text(xml)
|
||
if not parsed:
|
||
lit.pre_extracted_data = {
|
||
"_extract_failed": True,
|
||
"_extract_error": "XML 解析为空（可能为非标准 DTD）",
|
||
"_note": "自动识别失败，请人工核对原始全文",
|
||
}
|
||
continue
|
||
|
||
if use_cos:
|
||
# 写入 COS，PG 只存路径引用
|
||
key = make_full_text_key(pmid)
|
||
if await upload_json(key, parsed):
|
||
lit.full_text_path = key
|
||
else:
|
||
# COS 上传失败，降级到 PG
|
||
lit.full_text_sections = parsed
|
||
else:
|
||
# 开发环境无 COS，继续存 PG
|
||
lit.full_text_sections = parsed
|
||
|
||
# 提取许可信息
|
||
lic = parsed.get("license")
|
||
if lic:
|
||
lit.license_info = lic
|
||
# 预抽取基线特征（Table 1 → pre_extracted_data）
|
||
baseline = extract_baseline_data(parsed)
|
||
if baseline:
|
||
lit.pre_extracted_data = baseline
|
||
else:
|
||
lit.pre_extracted_data = {
|
||
"_extract_failed": False,
|
||
"_note": "未识别到基线特征表（可能无 Table 1 或格式非常规），请人工核对",
|
||
}
|
||
ok += 1
|
||
except Exception:
|
||
logger.warning("oa_save_failed pmid=%s", pmid, exc_info=True)
|
||
lit.pre_extracted_data = {
|
||
"_extract_failed": True,
|
||
"_extract_error": "基线抽取异常崩溃",
|
||
"_note": "自动识别失败，请人工核对原始全文",
|
||
}
|
||
if ok:
|
||
await db.commit()
|
||
logger.info("oa_fulltext_fetched: %d/%d (cos=%s)", ok, len(pending), use_cos)
|
||
return ok
|
||
|
||
|
||
async def pull_oncology_literature(db: AsyncSession, max_per_query: int = 30, precision_mode: str = "majr") -> dict:
|
||
"""手动全量管道：MeSH 精搜 + Title/Abstract 宽搜，完整覆盖。返回统计 dict"""
|
||
stats = await _run_pipeline(db, max_per_query, use_majr=True, use_broad=True, precision_mode=precision_mode)
|
||
return stats
|
||
|
||
|
||
async def pull_daily_oncology(precision_mode: str = "majr"):
|
||
"""每日精搜（供 ARQ 定时任务），仅 MeSH 高精度查询"""
|
||
async with async_session() as db:
|
||
stats = await _run_pipeline(db, max_per_query=20, use_majr=True, use_broad=False, precision_mode=precision_mode)
|
||
from app.models.operations import PipelineRun
|
||
run = PipelineRun(
|
||
id=_uid(), run_type="daily_update", status="success",
|
||
articles_total=stats["fetched"], articles_new=stats["new"],
|
||
articles_filtered=stats["tagged"], feeds_generated=stats["feeds"],
|
||
started_at=stats["started_at"],
|
||
completed_at=datetime.now(UTC),
|
||
)
|
||
db.add(run)
|
||
await db.commit()
|
||
# pipeline 结束后刷新 stats 缓存
|
||
try:
|
||
from app.api.v1.public import refresh_stats_cache
|
||
await refresh_stats_cache()
|
||
except Exception:
|
||
logger.warning("refresh_stats_cache 失败", exc_info=True)
|
||
return stats
|
||
|
||
|
||
async def pull_broad_oncology():
|
||
"""宽搜补充（供 ARQ 定时任务），仅 Title/Abstract 覆盖 in-process + publisher"""
|
||
async with async_session() as db:
|
||
stats = await _run_pipeline(db, max_per_query=50, use_majr=False, use_broad=True)
|
||
from app.models.operations import PipelineRun
|
||
run = PipelineRun(
|
||
id=_uid(), run_type="broad_update", status="success",
|
||
articles_total=stats["fetched"], articles_new=stats["new"],
|
||
articles_filtered=stats["tagged"], feeds_generated=stats["feeds"],
|
||
started_at=stats["started_at"],
|
||
completed_at=datetime.now(UTC),
|
||
)
|
||
db.add(run)
|
||
await db.commit()
|
||
try:
|
||
from app.api.v1.public import refresh_stats_cache
|
||
await refresh_stats_cache()
|
||
except Exception:
|
||
logger.warning("refresh_stats_cache 失败", exc_info=True)
|
||
return stats
|
||
|
||
|
||
async def pull_europe_pmc_oncology(max_results_per_query: int = 5000):
|
||
"""Europe PMC 全量管道（cursor 分页，适合初始回填或大数据量）。
|
||
|
||
供管理后台手动触发。使用 Europe PMC API：
|
||
- cursor-based 分页，无 10K 上限
|
||
- 含 MAJR 精搜 + Title/Abstract 宽搜
|
||
- 无 API Key 依赖，10+ req/s
|
||
"""
|
||
async with async_session() as db:
|
||
stats = await _run_pipeline(
|
||
db, max_per_query=max_results_per_query,
|
||
use_majr=True, use_broad=True,
|
||
use_europe_pmc=True,
|
||
)
|
||
from app.models.operations import PipelineRun
|
||
run = PipelineRun(
|
||
id=_uid(), run_type="europe_pmc_full", status="success",
|
||
articles_total=stats["fetched"], articles_new=stats["new"],
|
||
articles_filtered=stats["tagged"], feeds_generated=stats["feeds"],
|
||
started_at=stats["started_at"],
|
||
completed_at=datetime.now(UTC),
|
||
)
|
||
db.add(run)
|
||
await db.commit()
|
||
return stats
|