fastapi_app/lib/utils/doi_utils.py

"""
DOI metadata utilities for fetching and parsing bibliographic information.

Provides:
- DOI validation and normalization
- Filesystem-safe encoding/decoding for doc_ids
- Metadata fetching from CrossRef and DataCite APIs
"""

import re
import requests
from typing import Dict, Any, Optional
import logging

logger = logging.getLogger(__name__)

# Type alias for bibliographic metadata to avoid circular imports
# This should match the BibliographicMetadata TypedDict in metadata_extraction.py
BibliographicMetadata = Dict[str, Any]

# DOI validation regex (from CrossRef specification)
# Matches: 10.{4-9 digits}/{suffix with allowed characters}
# Allowed suffix characters: A-Z 0-9 -._;()/
DOI_REGEX = r"^10\.\d{4,9}/[-._;()/:A-Z0-9]+$"


def validate_doi(doi: str) -> bool:
    """
    Check if a DOI string is valid according to CrossRef specifications.

    Args:
        doi: The DOI string to validate

    Returns:
        True if DOI matches the valid pattern, False otherwise

    Examples:
        >>> validate_doi("10.5771/2699-1284-2024-3-149")
        True
        >>> validate_doi("not-a-doi")
        False
        >>> validate_doi("10.1234/valid-doi_123")
        True
    """
    if not doi:
        return False
    return bool(re.match(DOI_REGEX, doi, flags=re.IGNORECASE))


def fetch_doi_metadata(doi: str, timeout: int = 10) -> Dict[str, Any]:
    """
    Fetch metadata for DOI from CrossRef or DataCite APIs.

    Tries CrossRef first (most common), falls back to DataCite.

    Args:
        doi: The DOI string to fetch metadata for
        timeout: Request timeout in seconds (default: 10)

    Returns:
        Dictionary with metadata fields:
            - title: Article/document title
            - authors: List of dicts with 'given' and 'family' name
            - date: Publication year
            - publisher: Publisher name
            - journal: Journal/container title
            - volume: Volume number
            - issue: Issue number
            - pages: Page range

    Raises:
        ValueError: If DOI format is invalid
        requests.exceptions.HTTPError: If both APIs fail
        requests.exceptions.Timeout: If request times out

    Examples:
        >>> metadata = fetch_doi_metadata("10.5771/2699-1284-2024-3-149")
        >>> metadata['title']
        "Legal status of Derived Text Formats..."
    """
    if not validate_doi(doi):
        raise ValueError(f"{doi} is not a valid DOI string")

    # Try CrossRef first (more common for academic papers)
    try:
        logger.debug(f"Fetching metadata from CrossRef for DOI: {doi}")
        return parse_crossref_metadata(doi, timeout)
    except requests.exceptions.HTTPError as e:
        logger.warning(f"CrossRef API failed for {doi}: {e}")
        # Fall back to DataCite
        try:
            logger.debug(f"Falling back to DataCite for DOI: {doi}")
            return parse_datacite_metadata(doi, timeout)
        except requests.exceptions.HTTPError as e2:
            logger.error(f"Both CrossRef and DataCite APIs failed for {doi}")
            raise


def parse_crossref_metadata(doi: str, timeout: int = 10) -> Dict[str, Any]:
    """
    Parse metadata from CrossRef API.

    Args:
        doi: The DOI to look up
        timeout: Request timeout in seconds

    Returns:
        Dictionary with parsed metadata

    Raises:
        requests.exceptions.HTTPError: If API request fails
    """
    url = f"https://api.crossref.org/works/{doi}"
    response = requests.get(url, timeout=timeout)
    response.raise_for_status()
    data = response.json()

    message = data.get("message", {})

    # Extract title
    title = message.get("title", [None])[0]

    # Extract authors
    authors_data = message.get("author", [])
    authors = [
        {"given": author.get("given"), "family": author.get("family")}
        for author in authors_data
    ]

    # Extract publication date
    issued_data = message.get("issued", {})
    date_parts = issued_data.get("date-parts", [[]])
    date = date_parts[0][0] if date_parts and date_parts[0] else None

    # Extract ISSN (take first one if multiple)
    issn_list = message.get("ISSN", [])
    issn = issn_list[0] if issn_list else None

    # Extract ISBN (take first one if multiple)
    isbn_list = message.get("ISBN", [])
    isbn = isbn_list[0] if isbn_list else None

    return {
        "title": title,
        "authors": authors,
        "date": date,
        "publisher": message.get("publisher"),
        "journal": message.get("container-title", [None])[0],
        "volume": message.get("volume"),
        "issue": message.get("issue"),
        "pages": message.get("page"),
        "doi": message.get("DOI"),
        "issn": issn,
        "isbn": isbn,
        "url": message.get("URL"),
    }


def parse_datacite_metadata(doi: str, timeout: int = 10) -> Dict[str, Any]:
    """
    Parse metadata from DataCite API.

    Args:
        doi: The DOI to look up
        timeout: Request timeout in seconds

    Returns:
        Dictionary with parsed metadata

    Raises:
        requests.exceptions.HTTPError: If API request fails
    """
    url = f"https://api.datacite.org/dois/{doi}"
    response = requests.get(url, timeout=timeout)
    response.raise_for_status()
    data = response.json()

    attributes = data.get("data", {}).get("attributes", {})

    # Extract title
    title = None
    if attributes.get("titles"):
        title = attributes["titles"][0].get("title")

    # Extract authors (creators in DataCite)
    authors_data = attributes.get("creators", [])
    authors = [
        {"given": author.get("givenName"), "family": author.get("familyName")}
        for author in authors_data
    ]

    # Extract journal/container
    journal = None
    if attributes.get("container"):
        journal = attributes["container"].get("title")

    # Extract DOI from the data object
    doi_value = data.get("data", {}).get("id")

    # Extract related identifiers (ISSN, ISBN)
    issn = None
    isbn = None
    related_identifiers = attributes.get("relatedIdentifiers", [])
    for identifier in related_identifiers:
        if identifier.get("relatedIdentifierType") == "ISSN" and not issn:
            issn = identifier.get("relatedIdentifier")
        elif identifier.get("relatedIdentifierType") == "ISBN" and not isbn:
            isbn = identifier.get("relatedIdentifier")

    # Extract URL
    url = attributes.get("url")

    return {
        "title": title,
        "authors": authors,
        "date": attributes.get("publicationYear"),
        "publisher": attributes.get("publisher"),
        "journal": journal,
        "volume": attributes.get("volume", ""),
        "issue": attributes.get("issue", ""),
        "pages": attributes.get("page"),
        "doi": doi_value,
        "issn": issn,
        "isbn": isbn,
        "url": url,
    }


def normalize_doi(doi: str) -> str:
    """
    Normalize a DOI string.

    - Removes leading/trailing whitespace
    - Removes common prefixes (doi:, http://doi.org/, https://doi.org/)
    - Converts to lowercase (DOIs are case-insensitive)

    Args:
        doi: DOI string to normalize

    Returns:
        Normalized DOI string

    Examples:
        >>> normalize_doi("  10.5771/2699-1284-2024-3-149  ")
        "10.5771/2699-1284-2024-3-149"
        >>> normalize_doi("doi:10.5771/2699-1284-2024-3-149")
        "10.5771/2699-1284-2024-3-149"
        >>> normalize_doi("https://doi.org/10.5771/2699-1284-2024-3-149")
        "10.5771/2699-1284-2024-3-149"
    """
    if not doi:
        return doi

    # Strip whitespace
    doi = doi.strip()

    # Remove common DOI prefixes
    prefixes = [
        "doi:",
        "DOI:",
        "http://doi.org/",
        "https://doi.org/",
        "http://dx.doi.org/",
        "https://dx.doi.org/",
    ]

    for prefix in prefixes:
        if doi.startswith(prefix):
            doi = doi[len(prefix):]
            break

    # DOIs are case-insensitive, but conventionally lowercase
    # However, keep original case for compatibility
    return doi




# Filesystem encoding utilities
# ==============================

def is_filename_encoded(filename: str) -> bool:
    """
    Check if a filename is already encoded.

    Detects encoding markers:
    - Double underscore (__) where original had forward slash
    - _xXX_ patterns for special characters (current encoding)
    - $XX$ patterns for special characters (legacy encoding, BC)

    Args:
        filename: Filename to check

    Returns:
        True if filename appears to be already encoded, False otherwise

    Examples:
        >>> is_filename_encoded("10.1111__1467-6478.00040")
        True
        >>> is_filename_encoded("10.1234__test_x3A_file")
        True
        >>> is_filename_encoded("10.1111/1467-6478.00040")
        False
        >>> is_filename_encoded("simple-filename.txt")
        False
    """
    # Check for _xXX_ encoding pattern (current)
    if re.search(r'_x[0-9A-F]{2}_', filename):
        return True

    # Check for $XX$ encoding pattern (legacy BC)
    if re.search(r'\$[0-9A-F]{2}\$', filename):
        return True

    # Check for __ (encoded slash) combined with DOI-like pattern
    # If it has __ and looks like a DOI prefix, it's likely encoded
    if '__' in filename and re.match(r'^10\.\d+__', filename):
        return True

    return False


def encode_filename(doc_id: str) -> str:
    """
    Encode a document ID candidate (e.g., DOI) to a filesystem-safe filename.

    Encoding rules:
    - Forward slashes (/) → double underscore (__)
    - Other filesystem-incompatible characters → _xXX_ encoding
      where XX is the uppercase hexadecimal representation of the character code
    - The _xXX_ pattern is also NCName-safe, so encoded filenames can be used
      as xml:id values directly (with a leading _ prepended for digit-starting IDs).

    Args:
        doc_id: Document identifier to encode (e.g., DOI, file reference)

    Returns:
        Filesystem-safe encoded string

    Raises:
        ValueError: If doc_id is empty

    Examples:
        >>> encode_filename("10.1111/1467-6478.00040")
        "10.1111__1467-6478.00040"
        >>> encode_filename("10.1234/test:file")
        "10.1234__test_x3A_file"
        >>> encode_filename("doc<name>")
        "doc_x3C_name_x3E_"
    """
    if not doc_id:
        raise ValueError("doc_id cannot be empty")

    # First handle forward slashes specially (common in DOIs)
    result = doc_id.replace("/", "__")

    # Characters that are unsafe on various filesystems
    # Windows: < > : " | ? * and control chars
    # Unix/Mac: / (already handled) and control chars
    unsafe_chars = set('<>:"|?*\\')

    # Build encoded string
    encoded = []
    for char in result:
        if char in unsafe_chars or ord(char) < 32:
            # Encode as _xXX_ where XX is uppercase hex
            encoded.append(f"_x{ord(char):02X}_")
        else:
            encoded.append(char)

    return ''.join(encoded)


_LEGACY_ENCODING_RE = re.compile(r'\$([0-9A-Fa-f]{2})\$')


def normalize_legacy_encoding(s: str) -> str:
    """Convert legacy ``$XX$`` percent-style encoding to the current ``_xXX_`` format."""
    return _LEGACY_ENCODING_RE.sub(lambda m: f"_x{m.group(1).upper()}_", s)


_NCNAME_INVALID_RE = re.compile(r'[^a-zA-Z0-9._\-]')


def encode_for_xml_id(file_id: str) -> str:
    """
    Convert any file identifier to a valid XML NCName for use as ``xml:id``.

    Handles input in any encoding state:
    - Legacy ``$XX$`` patterns → ``_xXX_``
    - URL ``%XX`` percent-encoding → ``_xXX_``
    - ``/`` (raw or unencoded DOI slash) → ``__``
    - Any remaining character invalid in NCName → ``_xXX_``
    - Prepends ``_`` if the result starts with a digit

    Args:
        file_id: Any file identifier string

    Returns:
        NCName-safe string suitable for use as ``xml:id``

    Examples:
        >>> encode_for_xml_id("10.5771__2699-1284-2024-3-149")
        "_10.5771__2699-1284-2024-3-149"
        >>> encode_for_xml_id("test_x3A_file")
        "test_x3A_file"
        >>> encode_for_xml_id("test$3A$value")
        "test_x3A_value"
        >>> encode_for_xml_id("10.1023/a%3A1015833415224")
        "_10.1023__a_x3A_1015833415224"
    """
    # 1. Legacy $XX$ → _xXX_
    result = normalize_legacy_encoding(file_id)
    # 2. URL %XX → _xXX_
    result = re.sub(r'%([0-9A-Fa-f]{2})', lambda m: f"_x{m.group(1).upper()}_", result)
    # 3. / → __ (DOI slash convention, must come before generic char encoding)
    result = result.replace("/", "__")
    # 4. Any remaining NCName-invalid character → _xXX_
    result = _NCNAME_INVALID_RE.sub(lambda m: f"_x{ord(m.group()):02X}_", result)
    # 5. Prepend _ if starts with digit
    if result and result[0].isdigit():
        result = '_' + result
    return result


def decode_from_xml_id(xml_id: str) -> str:
    """
    Decode an xml:id value back to encode_filename() format.

    Since encode_filename() now uses _xXX_ patterns (which are NCName-safe),
    xml:id values produced by encode_for_xml_id() differ only in the optional
    leading '_' prepended for digit-starting IDs. This function strips it.

    Args:
        xml_id: An xml:id value encoded by encode_for_xml_id()

    Returns:
        The original encode_filename()-encoded file_id

    Examples:
        >>> decode_from_xml_id("_10.5771__2699-1284-2024-3-149")
        "10.5771__2699-1284-2024-3-149"
        >>> decode_from_xml_id("test_x3A_file")
        "test_x3A_file"
    """
    if xml_id and len(xml_id) > 1 and xml_id[0] == '_' and xml_id[1].isdigit():
        xml_id = xml_id[1:]
    return xml_id


def decode_filename(filename: str) -> str:
    """
    Decode a filesystem-safe filename back to the original document ID.

    Reverses the encoding applied by encode_filename():
    - __ → / (forward slash)
    - _xXX_ → original character (current encoding)
    - $XX$ → original character (legacy BC encoding)

    Args:
        filename: Encoded filename to decode

    Returns:
        Original document ID string

    Raises:
        ValueError: If filename is empty

    Examples:
        >>> decode_filename("10.1111__1467-6478.00040")
        "10.1111/1467-6478.00040"
        >>> decode_filename("10.1234__test_x3A_file")
        "10.1234/test:file"
        >>> decode_filename("10.1234__test$3A$file")
        "10.1234/test:file"
        >>> decode_filename("doc_x3C_name_x3E_")
        "doc<name>"
    """
    if not filename:
        raise ValueError("filename cannot be empty")

    # Restore forward slashes from double underscores
    result = filename.replace("__", "/")

    # Decode _xXX_ patterns (current encoding)
    result = re.sub(r'_x([0-9A-F]{2})_', lambda m: chr(int(m.group(1), 16)), result)

    # Decode $XX$ patterns (legacy BC encoding)
    result = re.sub(r'\$([0-9A-F]{2})\$', lambda m: chr(int(m.group(1), 16)), result)

    return result