Citation sources

def arxiv_id(reference: Node) -> str | None:
    """Return the arXiv identifier a bibliography entry names, if it names one.

    Args:
        reference: The reference node.

    Returns:
        The first identifier in its statement, or ``None``.
    """
    found = _ARXIV.search(reference.statement or "")
    return found[1] if found else None

def entry_doi(reference: Node) -> str | None:
    """Return the DOI a bibliography entry names, in lower case, if it names one.

    Args:
        reference: The reference node.

    Returns:
        The first DOI in its statement, without trailing punctuation, or ``None``.
    """
    found = _DOI.search(reference.statement or "")
    return found[1].rstrip(".,;:)]").lower() if found else None

def key_file(reference: Node, names: Mapping[str, str] | set[str]) -> str | None:
    """Return the file of the cited works that a reference's key names: the first of its suffixes there.

    Args:
        reference: The reference node.
        names: The names of the files of the run's cited works.

    Returns:
        ``<key>.txt``, ``<key>.tex`` or ``<key>.pdf``, the first there, or ``None``.
    """
    key = reference.attributes.get("key")
    if not isinstance(key, str) or not _KEY.fullmatch(key):
        return None
    return next((f"{key}{suffix}" for suffix in SUFFIXES if f"{key}{suffix}" in names), None)

def read_cited(run_dir: Path, source: str, passage: str) -> tuple[dict[str, Any], bytes]:
    """Read a file of the run's copy of the cited works.

    Args:
        run_dir: The run directory.
        source: The file's name, or a path to it in ``<run>/cited/``.
        passage: The passage the agent quotes.

    Returns:
        The observation (``kind`` ``cited``, ``file``, ``digest``, ``text_digest``, ``found``) and the bytes.

    Raises:
        SourceError: If the path is not a regular file directly in the run's ``cited/``, or cannot be read.
    """
    cited = run_dir / CITED
    path = Path(source)
    name = path.name
    inside = path == Path(name) or (path.is_absolute() and path.parent.resolve() == cited.resolve())
    target = cited / name
    if not inside or not name or target.is_symlink() or not target.is_file():
        msg = f"{source!r} is not a file of the run's copy of the cited works; a local source is read only from there"
        raise SourceError("unreadable-source", msg)
    data = target.read_bytes()
    observation = {"kind": "cited", "file": name, "digest": bytes_digest(data)}
    return _located(observation, text_of(data, target.suffix), passage), data

def read_url(reference: Node, source: str, passage: str, fetcher: Fetch = fetch) -> tuple[dict[str, Any], bytes]:
    """Fetch a URL source, and what ties it to the reference.

    An arXiv URL of the identifier the entry names is read from arXiv at the entry's version, or the
    current one; a doi.org URL of the entry's DOI is read through doi.org. Any other URL is read as
    given, and, when the entry names a DOI, the DOI is resolved too, so the two final URLs can be compared.

    Args:
        reference: The reference node.
        source: The URL the agent gave.
        passage: The passage the agent quotes.
        fetcher: Fetches a URL under the fetch policy.

    Returns:
        The observation (``kind`` ``url``, the fetch's URLs and content type, ``digest``, ``tie``,
        ``text_digest``, ``found``) and the bytes.

    Raises:
        SourceError: If a URL is refused, cannot be fetched, or the body cannot be read.
    """
    tie: dict[str, Any] | None = None
    entry_arxiv = arxiv_id(reference)
    doi = entry_doi(reference)
    given = _ARXIV_URL.fullmatch(source)
    if entry_arxiv is not None and given is not None and given["id"] == _split_arxiv(entry_arxiv)[0]:
        fetched, tie = _arxiv(entry_arxiv, fetcher)
    elif doi is not None and _doi_of_url(source) == doi:
        fetched = _fetched(DOI_URL.format(doi), fetcher)
        tie = {"by": "doi", "doi": doi, "final_url": fetched.final_url}
    else:
        fetched = _fetched(source, fetcher)
        if doi is not None:
            try:
                resolved = _fetched(DOI_URL.format(doi), fetcher)
            except SourceError:
                tie = None
            else:
                tie = {"by": "doi", "doi": doi, "final_url": resolved.final_url}
    kind = fetched.content_type or ("application/pdf" if fetched.data.startswith(b"%PDF") else "")
    observation = {"kind": "url", **describe(fetched), "digest": bytes_digest(fetched.data), "tie": tie}
    return _located(observation, text_of(fetched.data, kind), passage), fetched.data

def text_of(data: bytes, kind: str) -> str:
    """Return the normalised text of a source's bytes.

    Args:
        data: The bytes.
        kind: A file suffix (``.txt``, ``.tex``, ``.pdf``) or a media type (``text/plain``,
            ``text/html``, ``application/xhtml+xml``, ``application/pdf``).

    Returns:
        The text, normalised as a quoted passage is.

    Raises:
        SourceError: If the kind is not one of those, or the bytes cannot be read as it.
    """
    try:
        if kind in {".txt", "text/plain"}:
            text = data.decode("utf-8")
        elif kind in {"text/html", "application/xhtml+xml"}:
            parser = _Text()
            parser.feed(data.decode("utf-8", errors="replace"))
            parser.close()
            text = "".join(parser.parts)
        elif kind in {".pdf", "application/pdf", ".tex"}:
            suffix = ".tex" if kind == ".tex" else ".pdf"
            with tempfile.TemporaryDirectory() as directory:
                path = Path(directory) / f"source{suffix}"
                path.write_bytes(data)
                chunks = read_latex(path) if suffix == ".tex" else read_pdf(path)
            text = " ".join(chunk.text for chunk in chunks)
        else:
            raise SourceError("unreadable-source", f"a source of type {kind!r} cannot be read")
    except (UnicodeDecodeError, LatexError, PdfError, OSError) as error:
        raise SourceError("unreadable-source", f"the source cannot be read as {kind}: {type(error).__name__}") from None
    normalised = normalise_text(text)
    if not normalised:
        raise SourceError("unreadable-source", "the source has no text")
    return normalised

def tie(reference: Node, pins: Mapping[str, str], observation: Mapping[str, Any]) -> str | None:
    """Decide whether an observed source is tied to the reference; return why not, or ``None`` when it is.

    Args:
        reference: The reference node.
        pins: The digest of each file of the cited works, as ``start_run`` recorded it.
        observation: What the tool read.

    Returns:
        ``None`` for a tied source, else the reason it is not, in words.
    """
    if observation.get("kind") == "cited":
        return _cited_tie(reference, pins, observation)
    found = observation.get("tie")
    if isinstance(found, Mapping) and found.get("by") == "arxiv":
        return _arxiv_tie(reference, found, observation)
    if isinstance(found, Mapping) and found.get("by") == "doi":
        return _doi_tie(reference, found, observation)
    return _NOT_TIED

def verify_observation(observation: Mapping[str, Any], data: bytes, passage: str) -> list[str]:
    """Return the fields of an observation that the source's bytes do not give again.

    Args:
        observation: What the tool recorded.
        data: The bytes, found by their digest.
        passage: The passage the agent quoted.

    Returns:
        The names of the fields that differ: ``digest``, ``text_digest`` or ``found``; or ``unreadable``.
    """
    if bytes_digest(data) != observation.get("digest"):
        return ["digest"]
    kind = (
        Path(str(observation.get("file"))).suffix
        if observation.get("kind") == "cited"
        else observation.get("content_type")
    )
    if observation.get("kind") == "url" and not kind:
        kind = "application/pdf" if data.startswith(b"%PDF") else ""
    try:
        again = _located({}, text_of(data, str(kind)), passage)
    except SourceError:
        return ["unreadable"]
    return [name for name in ("text_digest", "found") if again[name] != observation.get(name)]