agent_search_gateway.paper_search_parser

  1"""Strict restricted-format parser for LLM academic paper search results."""
  2
  3from __future__ import annotations
  4
  5from datetime import date
  6
  7from .errors import InputFailure, ParserFailure
  8from .providers.contracts import PaperSearchHit
  9from .url_normalization import normalize_url
 10
 11_HEADING = "## Paper"
 12_FIELDS = (
 13    "Title",
 14    "Authors",
 15    "Abstract",
 16    "DOI",
 17    "arXiv",
 18    "Published",
 19    "Updated",
 20    "URL",
 21    "PDF",
 22    "Venue",
 23    "Topics",
 24    "Citations",
 25    "Open Access",
 26    "OA Status",
 27    "License",
 28)
 29_BLOCK_LINES = 1 + len(_FIELDS)
 30
 31
 32def parse_paper_markdown(markdown: str, *, provider: str) -> list[PaperSearchHit]:
 33    """Parse only exact repeated ``## Paper`` blocks into provider candidates."""
 34
 35    if not isinstance(markdown, str) or not markdown.strip():
 36        raise ParserFailure("LLM paper response contained no Paper blocks")
 37    provider_name = provider.strip() if isinstance(provider, str) else ""
 38    if not provider_name:
 39        raise ParserFailure("LLM paper provider name is invalid")
 40
 41    lines = markdown.strip().splitlines()
 42    if len(lines) % _BLOCK_LINES != 0:
 43        raise ParserFailure("LLM paper response does not match the required block grammar")
 44
 45    hits: list[PaperSearchHit] = []
 46    for start in range(0, len(lines), _BLOCK_LINES):
 47        block = lines[start : start + _BLOCK_LINES]
 48        hits.append(_parse_block(block, provider_name))
 49    return hits
 50
 51
 52def _parse_block(lines: list[str], provider: str) -> PaperSearchHit:
 53    if len(lines) != _BLOCK_LINES or lines[0] != _HEADING:
 54        raise ParserFailure("Paper block heading is invalid")
 55
 56    values: dict[str, str] = {}
 57    for index, field in enumerate(_FIELDS, start=1):
 58        prefix = f"{field}:"
 59        line = lines[index]
 60        if not line.startswith(prefix):
 61            raise ParserFailure("Paper block fields do not match the required grammar")
 62        values[field] = line[len(prefix) :].strip()
 63
 64    title = values["Title"]
 65    if not title:
 66        raise ParserFailure("Paper block title is empty")
 67    url = _parse_required_url(values["URL"])
 68    pdf_url = _parse_optional_url(values["PDF"])
 69    published = _parse_optional_date(values["Published"], "published date")
 70    updated = _parse_optional_date(values["Updated"], "updated date")
 71    citations = _parse_optional_citations(values["Citations"])
 72    is_open_access = _parse_open_access(values["Open Access"])
 73
 74    return PaperSearchHit(
 75        source=f"llm:{provider}",
 76        source_id=str(url),
 77        title=title,
 78        authors=_split_semicolon(values["Authors"]),
 79        abstract=values["Abstract"],
 80        doi=values["DOI"],
 81        arxiv_id=values["arXiv"],
 82        published_date=published,
 83        updated_date=updated,
 84        url=str(url),
 85        pdf_url=str(pdf_url) if pdf_url is not None else "",
 86        venue=values["Venue"],
 87        topics=_split_semicolon(values["Topics"]),
 88        citation_count=citations,
 89        is_open_access=is_open_access,
 90        oa_status=values["OA Status"],
 91        license=values["License"],
 92    )
 93
 94
 95def _split_semicolon(value: str) -> tuple[str, ...]:
 96    if not value:
 97        return ()
 98    items = tuple(item.strip() for item in value.split(";") if item.strip())
 99    if not items:
100        return ()
101    return items
102
103
104def _parse_required_url(value: str) -> str:
105    if not value:
106        raise ParserFailure("Paper block URL is empty")
107    try:
108        return str(normalize_url(value))
109    except InputFailure as exc:
110        raise ParserFailure("Paper block URL is invalid") from exc
111
112
113def _parse_optional_url(value: str) -> str | None:
114    if not value:
115        return None
116    try:
117        return str(normalize_url(value))
118    except InputFailure as exc:
119        raise ParserFailure("Paper block PDF URL is invalid") from exc
120
121
122def _parse_optional_date(value: str, label: str) -> date | None:
123    if not value:
124        return None
125    if len(value) != 10:
126        raise ParserFailure(f"Paper block {label} is invalid")
127    try:
128        parsed = date.fromisoformat(value)
129    except ValueError as exc:
130        raise ParserFailure(f"Paper block {label} is invalid") from exc
131    if parsed.isoformat() != value:
132        raise ParserFailure(f"Paper block {label} is invalid")
133    return parsed
134
135
136def _parse_optional_citations(value: str) -> int | None:
137    if not value:
138        return None
139    if not value.isascii() or not value.isdecimal():
140        raise ParserFailure("Paper block citations value is invalid")
141    return int(value)
142
143
144def _parse_open_access(value: str) -> bool | None:
145    normalized = value.casefold()
146    if normalized in {"", "unknown"}:
147        return None
148    if normalized == "true":
149        return True
150    if normalized == "false":
151        return False
152    raise ParserFailure("Paper block open-access value is invalid")