Spaces:
Runtime error
Runtime error
| import asyncio | |
| from typing import Optional | |
| from backend.providers.base import fetch_json, normalize_result | |
| async def fetch_metadata(doi: str = None, url: str = None, title: str = None) -> dict: | |
| """Fetch metadata for a single paper.""" | |
| if doi: | |
| # Try OpenAlex by DOI | |
| data = await fetch_json(f"https://api.openalex.org/works/https://doi.org/{doi}") | |
| if "error" not in data: | |
| return normalize_result( | |
| title=data.get("title"), | |
| authors=[a.get("author", {}).get("display_name", "") for a in data.get("authorships", [])], | |
| year=data.get("publication_year"), | |
| abstract=None, # Need to decode inverted index | |
| doi=doi, | |
| pdf_url=data.get("open_access", {}).get("oa_url"), | |
| source="openalex", | |
| university=data.get("authorships", [{}])[0].get("institutions", [{}])[0].get("display_name") if data.get("authorships") else None, | |
| citation_count=data.get("cited_by_count"), | |
| ) | |
| if title: | |
| data = await fetch_json("https://api.openalex.org/works", params={"search": title, "per-page": 1}) | |
| if "error" not in data and data.get("results"): | |
| work = data["results"][0] | |
| return normalize_result( | |
| title=work.get("title"), | |
| authors=[a.get("author", {}).get("display_name", "") for a in work.get("authorships", [])], | |
| year=work.get("publication_year"), | |
| abstract=None, | |
| doi=work.get("ids", {}).get("doi", "").replace("https://doi.org/", ""), | |
| pdf_url=work.get("open_access", {}).get("oa_url"), | |
| source="openalex", | |
| ) | |
| return {"error": "No metadata found"} | |
| async def recover_metadata(doi: str = None, url: str = None, title: str = None) -> dict: | |
| """Deep metadata recovery from multiple sources.""" | |
| result = {} | |
| sources_tried = [] | |
| # Try OpenAlex | |
| if doi: | |
| sources_tried.append("OpenAlex") | |
| data = await fetch_json(f"https://api.openalex.org/works/https://doi.org/{doi}") | |
| if "error" not in data: | |
| result["title"] = data.get("title") | |
| result["year"] = data.get("publication_year") | |
| result["doi"] = doi | |
| result["authors"] = [a.get("author", {}).get("display_name", "") for a in data.get("authorships", [])] | |
| # Try Crossref for abstract | |
| if doi and not result.get("abstract"): | |
| sources_tried.append("Crossref") | |
| data = await fetch_json(f"https://api.crossref.org/works/{doi}") | |
| if "error" not in data: | |
| item = data.get("message", {}) | |
| if item.get("abstract"): | |
| result["abstract"] = item["abstract"][:500] | |
| # Try Semantic Scholar for PDF | |
| if doi and not result.get("pdfUrl"): | |
| sources_tried.append("Semantic Scholar") | |
| data = await fetch_json(f"https://api.semanticscholar.org/graph/v1/paper/DOI:{doi}?fields=openAccessPdf") | |
| if "error" not in data: | |
| result["pdfUrl"] = data.get("openAccessPdf", {}).get("url") | |
| result["sourcesTried"] = sources_tried | |
| return result | |