literature

Literature search, citation extraction, and bibliography processing.

Examples:

>>> from ures.literature import CitationManager
>>> CitationManager().citations
[]

BibTypeRule dataclass

BibTypeRule(entry_type: str, standard_name: str = 'default', required_fields: List[str] = list(), suggested_fields: List[str] = list(), optional_fields: List[str] = list(), forbidden_fields: List[str] = list(), field_mappings: Dict[str, str] = dict(), formatting: FormattingRules = FormattingRules(), output: OutputRules = OutputRules(), version: str = '1.0')

Rules for a specific BibTeX entry type.

to_dict

to_dict() -> Dict[str, Any]

Convert to dictionary for serialization.

Source code in ures/literature/citation/rules/data_type.py

def to_dict(self) -> Dict[str, Any]:
    """Convert to dictionary for serialization."""
    return asdict(self)

FormattingRules dataclass

FormattingRules(year_range: tuple[int, int] = (1000, 2100), page_separator: str = '--', proceedings_style: str = 'full')

Formatting rules for bibliography entries.

OutputRules dataclass

OutputRules(field_order: List[str] = list(), conditional_fields: Dict[str, str] = dict(), required_middlewares: List[Type[BlockMiddleware]] = list(), max_authors: int = 5)

Output formatting rules.

CitationInfo dataclass

CitationInfo(key: str, sources: List[CitationSource] = list(), bibliography: Optional[Entry] = None)

The data structure to hold citation information.

OutputOnlyDesiredFieldsMiddleware

OutputOnlyDesiredFieldsMiddleware(rule_register: Optional[BibRuleRegister] = None)

Bases: CitationMiddleware

Keep only desired fields in the output.

Source code in ures/literature/citation/middlewares.py

def __init__(self, rule_register: Optional[BibRuleRegister] = None):
    super().__init__()
    self.rule_register = rule_register or BibRuleRegister()

OutputCleanupNoneResultMiddleware

OutputCleanupNoneResultMiddleware(rule_register: Optional[BibRuleRegister] = None)

Bases: CitationMiddleware

Cleanup entries that are invalid or have missing required fields.

Source code in ures/literature/citation/middlewares.py

def __init__(self, rule_register: Optional[BibRuleRegister] = None):
    super().__init__()
    self.rule_register = rule_register or BibRuleRegister()

OutputLimitMaxAuthors

OutputLimitMaxAuthors(rule_register: Optional[BibRuleRegister] = None)

Bases: CitationMiddleware

Keep only desired fields in the output.

Source code in ures/literature/citation/middlewares.py

def __init__(self, rule_register: Optional[BibRuleRegister] = None):
    super().__init__()
    self.rule_register = rule_register or BibRuleRegister()

TypeNormalizationMiddleware

TypeNormalizationMiddleware(rule_register: Optional[BibRuleRegister] = None)

Bases: CitationMiddleware

Normalize entry types (conference -> inproceedings, etc.)

Source code in ures/literature/citation/middlewares.py

def __init__(self, rule_register: Optional[BibRuleRegister] = None):
    super().__init__()
    self.rule_register = rule_register or BibRuleRegister()

RuleBasedValidationMiddleware

RuleBasedValidationMiddleware(rule_register: Optional[BibRuleRegister] = None)

Bases: CitationMiddleware

Validate entries against predefined rules.

Source code in ures/literature/citation/middlewares.py

def __init__(self, rule_register: Optional[BibRuleRegister] = None):
    super().__init__()
    self.rule_register = rule_register or BibRuleRegister()

transform_entry

transform_entry(entry: Entry, *args, **kwargs)

Validate entry using dataclass rules.

Source code in ures/literature/citation/middlewares.py

def transform_entry(self, entry: Entry, *args, **kwargs):
    """Validate entry using dataclass rules."""
    is_valid = True
    rule = self.rule_register.get_rule(entry.entry_type)
    missing_required = []
    for req_field in rule.required_fields:
        if not field_group_present(entry, req_field):
            missing_required.append(req_field)

    if len(missing_required) > 0:
        is_valid = False

    suggested_fields = getattr(rule, "suggested_fields", [])
    if not isinstance(suggested_fields, list):
        suggested_fields = []
    missing_suggested = []
    for spec in suggested_fields:
        if not field_group_present(entry, spec):
            missing_suggested.append(spec)

    is_valid_field = Field(key="is_valid", value=is_valid)
    missing_fields = Field(key="missing_fields", value=missing_required)
    missing_suggested_field = Field(
        key="missing_suggested", value=missing_suggested
    )
    entry.set_field(is_valid_field)
    entry.set_field(missing_fields)
    entry.set_field(missing_suggested_field)

    return entry

AcmConferenceVenueMiddleware

AcmConferenceVenueMiddleware(rule_register: Optional[BibRuleRegister] = None)

Bases: CitationMiddleware

Put the conference city on the field ACM actually prints.

ACM-Reference-Format appends location or city to the booktitle and prints address after the publisher. Zotero's BibTeX export has only address for Place, and that Place is the venue city. BibLaTeX venue is the event location. For ACM and the default style, move the venue city onto location when location and city are empty. A publisher address is left in place when venue already holds the city. Books and IEEE entries are unchanged.

Source code in ures/literature/citation/middlewares.py

def __init__(self, rule_register: Optional[BibRuleRegister] = None):
    super().__init__()
    self.rule_register = rule_register or BibRuleRegister()

FieldNormalizationMiddleware

FieldNormalizationMiddleware(rule_register: Optional[BibRuleRegister] = None)

Bases: CitationMiddleware

Normalize field names (journaltitle -> journal, etc.)

Source code in ures/literature/citation/middlewares.py

def __init__(self, rule_register: Optional[BibRuleRegister] = None):
    super().__init__()
    self.rule_register = rule_register or BibRuleRegister()

transform_entry

transform_entry(entry: Entry, *args, **kwargs) -> Entry

Transform entry fields.

Source code in ures/literature/citation/middlewares.py

def transform_entry(self, entry: Entry, *args, **kwargs) -> Entry:
    """Transform entry fields."""
    for field in entry.fields:
        # Handle special cases
        key = field.key
        value = field.value
        if key == "pages":
            # Normalize page ranges
            field.value = self._normalize_pages(value)
        else:
            field.value = value
        # Apply field mapping
        rules = self.rule_register.get_rule(entry.entry_type)
        field_mappings = copy.deepcopy(
            self.rule_register.get_default_field_mapping()
        )
        field_mappings.update(rules.field_mappings)
        new_key = field_mappings.get(field.key, field.key)
        field.key = new_key
    return entry

LanguageAsciiNormalizationMiddleware

LanguageAsciiNormalizationMiddleware(rule_register: Optional[BibRuleRegister] = None)

Bases: CitationMiddleware

Source code in ures/literature/citation/middlewares.py

def __init__(self, rule_register: Optional[BibRuleRegister] = None):
    super().__init__()
    self.rule_register = rule_register or BibRuleRegister()

normalize_language

normalize_language(language_str: str) -> Any

Normalize language string to ISO 639-1 code.

Source code in ures/literature/citation/middlewares.py

def normalize_language(self, language_str: str) -> Any:
    """Normalize language string to ISO 639-1 code."""
    special_cases = {
        "pinyin": "zh",
    }
    if language_str in special_cases:
        return special_cases[language_str]
    if not isinstance(language_str, str):
        logger.warning(f"Failed to normalize language '{language_str}'")
        return language_str
    language_str = language_str.strip().lower()
    try:
        lang = pycountry.languages.get(name=language_str)
        if lang and hasattr(lang, "alpha_2"):
            return lang.alpha_2
        # Try searching by common name
        for lang in pycountry.languages:
            if language_str in str(lang.name).lower():
                if hasattr(lang, "alpha_2"):
                    return lang.alpha_2
    except:
        logger.warning(f"Failed to normalize language '{language_str}'")
        return language_str
    return language_str

ProceedingsNormalizationMiddleware

ProceedingsNormalizationMiddleware(rule_register: Optional[BibRuleRegister] = None)

Bases: CitationMiddleware

Source code in ures/literature/citation/middlewares.py

def __init__(self, rule_register: Optional[BibRuleRegister] = None):
    super().__init__()
    self.rule_register = rule_register or BibRuleRegister()

normailize_proceedings

normailize_proceedings(proceedings_str: str) -> str

Normalize proceedings string to standard format.

Source code in ures/literature/citation/middlewares.py

def normailize_proceedings(self, proceedings_str: str) -> str:
    """Normalize proceedings string to standard format."""
    patterns = [
        r"\bIn\s+Proceedings\s+of\s+the\s+",
        r"\bIn\s+Proceedings\s+of\s+",
        r"\bIn\s+Proc\.\s+of\s+the\s+",
        r"\bIn\s+Proc\.\s+of\s+",
        r"\bProceedings\s+of\s+the\s+",
        r"\bProceedings\s+of\s+",
        r"\bProc\.\s+of\s+the\s+",
        r"\bProc\.\s+of\s+",
        r"\bIn\s+(?=\d|\w+\s+(International|Annual|ACM|IEEE))",
        r"\bProc\.\s+",
    ]
    proceedings_str = self._shorten_venue_year(proceedings_str.strip())
    for pattern in patterns:
        if re.search(pattern, proceedings_str, re.IGNORECASE):
            proceedings_str = re.sub(
                pattern, "", proceedings_str, flags=re.IGNORECASE
            )
            proceedings_str = re.sub(r"\s+", " ", proceedings_str).strip()
            break
    prefix_map = {
        "remove": "",
        "full": "In Proceedings of the ",
        "short": "In Proc. of the ",
        "proceedings": "Proceedings of the ",
        "minimal": "In ",
        "proc": "Proc. ",
    }

    style = self.rule_register.get_rule(
        "inproceedings"
    ).formatting.proceedings_style
    prefix = prefix_map.get(style, "full")
    return f"{prefix}{proceedings_str}".strip()

CitationManager

CitationManager(bibliography_files: Optional[Union[Union[str, Path], List[Union[str, Path]]]] = None, bibliography_style: str = 'default')

Load bibliography files and import citations from TeX or BBL sources.

Examples:

>>> from ures.literature import CitationManager
>>> manager = CitationManager(bibliography_style="acm")
>>> manager.citations
[]

Source code in ures/literature/citation/__init__.py

def __init__(
    self,
    bibliography_files: Optional[
        Union[Union[str, Path], List[Union[str, Path]]]
    ] = None,
    bibliography_style: str = "default",
):
    # Load bibliography files
    if bibliography_files is None:
        bibliography_files = []
    if not isinstance(bibliography_files, list):
        bibliography_files = [bibliography_files]
    self._bib_manager = BibManager(bibliography_style=bibliography_style)
    for bib_file in bibliography_files:
        self._bib_manager.append_bibliography(bib_file)
    # Store citation items for quick access
    self._citations: list[CitationInfo] = []

import_citations

import_citations(files: List[Union[str, Path]], cleanup: bool = False) -> dict[str, CitationInfo]

Import citations from the given files.

Parameters:

• files (List[Union[str, Path]]) –

  List of file paths to import citations from.

• cleanup (bool, default: False ) –

  Whether to clean up the citations previously stored. Defaults to False.

Returns:

• dict[str, CitationInfo] –

  List[CitationInfo]: List of imported citations.

Source code in ures/literature/citation/__init__.py

def import_citations(
    self, files: List[Union[str, Path]], cleanup: bool = False
) -> dict[str, CitationInfo]:
    """Import citations from the given files.

    Args:
                    files (List[Union[str, Path]]): List of file paths to import citations from.
                    cleanup (bool): Whether to clean up the citations previously stored. Defaults to False.

    Returns:
                    List[CitationInfo]: List of imported citations.

    """
    unique_citations = {}
    for file in files:
        file_path = Path(file)
        if not file_path.exists():
            continue

        if file_path.suffix == ".tex":
            extractor = TexCitationExtractor()
        elif file_path.suffix == ".bbl":
            extractor = BBLCitationExtractor()
        else:
            logger.warning(
                f"Unsupported file type: {file_path.suffix} for file {file_path}. Skipping."
            )
            continue

        for cite in extractor.extract_citations(file_path):
            if cite.key not in unique_citations:
                unique_citations[cite.key] = cite
            else:
                unique_citations[cite.key].sources.extend(cite.sources)

    # Merge citation information with existing citations
    if cleanup:
        self._citations.clear()
        self._add_citation(list(unique_citations.values()))
    else:
        for key, cite in unique_citations.items():
            is_found = False
            for stored_cite in self._citations:
                if stored_cite.key == key:
                    stored_cite.sources.extend(cite.sources)
                    is_found = True
                    break
            if not is_found:
                self._add_citation(cite)

    return unique_citations

display_invalid_citations

display_invalid_citations() -> None

Display citations that do not have corresponding bibliography entries.

Source code in ures/literature/citation/__init__.py

def display_invalid_citations(self) -> None:
    """Display citations that do not have corresponding bibliography entries."""
    for cite in self._citations:
        if (
            cite.bibliography is not None
            and cite.bibliography.get("is_valid", False) is False
        ):
            print(
                f"Citation Key: {cite.key} is invalid. Missing Fields: {cite.bibliography.get('missing_fields', [])}"
            )

to_library

to_library() -> bibtexparser.Library

Convert all citations to a bibtexparser Library.

Source code in ures/literature/citation/__init__.py

def to_library(self) -> bibtexparser.Library:
    """Convert all citations to a bibtexparser Library."""
    lib = bibtexparser.Library()
    if len(self.manager.bibliograph_library.strings) > 0:
        lib.add(self.manager.bibliograph_library.strings)
    for cite in self._citations:
        if cite.bibliography is not None:
            lib.add(copy.deepcopy(cite.bibliography))
    return lib

save_bibliography

save_bibliography(file_path: str, middlewares: Optional[List[Union[Type[BlockMiddleware], BlockMiddleware]]] = None) -> None

Save all bibliography entries to a BibTeX file.

Parameters:

• file_path (Union[str, Path]) –

  Path to save the BibTeX file.

• middlewares (Optional[List[Type[CitationMiddleware]]], default: None ) –

  List of middleware classes to process the entries before saving. If None, no additional middleware will be applied. Defaults to None.

Source code in ures/literature/citation/__init__.py

def save_bibliography(
    self,
    file_path: str,
    middlewares: Optional[
        List[
            Union[
                Type[bibtexparser.middlewares.BlockMiddleware],
                bibtexparser.middlewares.BlockMiddleware,
            ]
        ]
    ] = None,
) -> None:
    """Save all bibliography entries to a BibTeX file.

    Args:
        file_path (Union[str, Path]): Path to save the BibTeX file.
        middlewares (Optional[List[Type[CitationMiddleware]]]): List of middleware classes to process the entries before saving.
            If None, no additional middleware will be applied. Defaults to None.
    """
    self._bib_manager.export_to_file(
        file_path=file_path, library=self.to_library(), middlewares=middlewares
    )

AdapterFactory

Factory class for creating database adapters.

create_adapter staticmethod

create_adapter(database_name: str, **kwargs) -> Optional[DatabaseAdapter]

Create a database adapter instance.

Parameters:

• database_name (str) –

  Name of the database

• **kwargs (Any, default: {} ) –

  Adapter options such as api_key or rate_limit.

Returns:

• Optional[DatabaseAdapter] –

  DatabaseAdapter instance or None if not supported

Source code in ures/literature/search/adapters.py

@staticmethod
def create_adapter(database_name: str, **kwargs) -> Optional[DatabaseAdapter]:
    """
    Create a database adapter instance.

    Args:
        database_name: Name of the database
        **kwargs (Any): Adapter options such as api_key or rate_limit.

    Returns:
        DatabaseAdapter instance or None if not supported
    """
    adapters = AdapterFactory.SUPPORTED_DATABASES
    if database_name not in adapters:
        return None

    try:
        adapter_class = adapters[database_name]

        # Handle different parameter requirements
        if database_name == "arxiv":
            return adapter_class(rate_limit=kwargs.get("rate_limit", 3.0))
        elif database_name in ["ieee", "elsevier", "springer", "wiley"]:
            api_key = kwargs.get("api_key")
            if not api_key:
                return None
            return adapter_class(
                api_key=api_key, rate_limit=kwargs.get("rate_limit", 100.0)
            )
        elif database_name in ["acm", "google_scholar"]:
            return adapter_class(rate_limit=kwargs.get("rate_limit", 1.0))

    except Exception as e:
        logging.getLogger(__name__).error(
            f"Failed to create adapter for {database_name}: {e}"
        )
        return None

get_supported_databases staticmethod

get_supported_databases() -> List[str]

Get list of supported database names.

Source code in ures/literature/search/adapters.py

@staticmethod
def get_supported_databases() -> List[str]:
    """Get list of supported database names."""
    return list(AdapterFactory.SUPPORTED_DATABASES.keys())

get_adapter_requirements staticmethod

get_adapter_requirements(database_name: str) -> Dict[str, Any]

Get requirements for a specific adapter.

Source code in ures/literature/search/adapters.py

@staticmethod
def get_adapter_requirements(database_name: str) -> Dict[str, Any]:
    """Get requirements for a specific adapter."""
    requirements = {
        "arxiv": {
            "api_key_required": False,
            "rate_limit": 3.0,
            "free": True,
            "scraping_based": False,
            "reliability": "high",
            "database_url": "https://arxiv.org/",
            "api_docs_url": "https://info.arxiv.org/help/api/index.html",
            "note": "Official XML API with excellent reliability and comprehensive metadata.",
            "features": [
                "title",
                "authors",
                "abstract",
                "categories",
                "arxiv_id",
                "pdf_url",
                "publication_date",
            ],
            "limitations": [
                "preprints_only",
                "no_citation_counts",
                "academic_categories_only",
            ],
        },
        "ieee": {
            "api_key_required": True,
            "rate_limit": 100.0,
            "free": False,
            "scraping_based": False,
            "reliability": "high",
            "database_url": "https://ieeexplore.ieee.org/",
            "api_docs_url": "https://developer.ieee.org/",
            "api_signup_url": "https://developer.ieee.org/",
            "note": "Official REST API requiring subscription. High-quality engineering and computer science papers.",
            "features": [
                "title",
                "authors",
                "abstract",
                "doi",
                "citation_counts",
                "publication_info",
                "pdf_access",
            ],
            "limitations": [
                "requires_paid_subscription",
                "engineering_focus",
                "rate_limited_per_hour",
            ],
        },
        "elsevier": {
            "api_key_required": True,
            "rate_limit": 100.0,
            "free": False,
            "scraping_based": False,
            "reliability": "high",
            "database_url": "https://www.sciencedirect.com/",
            "api_docs_url": "https://dev.elsevier.com/",
            "api_signup_url": "https://dev.elsevier.com/",
            "note": "Official ScienceDirect API requiring subscription. Extensive collection across multiple disciplines.",
            "features": [
                "title",
                "authors",
                "abstract",
                "doi",
                "full_text_access",
                "publication_info",
                "subject_areas",
            ],
            "limitations": [
                "requires_paid_subscription",
                "complex_authentication",
                "usage_quotas",
            ],
        },
        "springer": {
            "api_key_required": True,
            "rate_limit": 100.0,
            "free": True,
            "scraping_based": False,
            "reliability": "high",
            "database_url": "https://link.springer.com/",
            "api_docs_url": "https://dev.springernature.com/",
            "api_signup_url": "https://dev.springernature.com/",
            "note": "Official API with free tier available. Good coverage of scientific literature.",
            "features": [
                "title",
                "authors",
                "abstract",
                "doi",
                "publication_info",
                "open_access_indicators",
            ],
            "limitations": [
                "rate_limits_on_free_tier",
                "some_content_requires_subscription",
            ],
        },
        "wiley": {
            "api_key_required": True,
            "rate_limit": 100.0,
            "free": False,
            "scraping_based": False,
            "reliability": "medium",
            "database_url": "https://onlinelibrary.wiley.com/",
            "api_docs_url": "https://onlinelibrary.wiley.com/library-info/resources/text-and-datamining",
            "api_signup_url": "https://onlinelibrary.wiley.com/library-info/resources/text-and-datamining",
            "note": "TDM (Text and Data Mining) API requiring institutional access. Focus on academic journals.",
            "features": [
                "title",
                "authors",
                "abstract",
                "doi",
                "publication_info",
                "full_text_mining",
            ],
            "limitations": [
                "requires_institutional_access",
                "complex_authentication",
                "limited_free_access",
            ],
        },
        "acm": {
            "api_key_required": False,
            "rate_limit": 0.5,
            "free": True,
            "scraping_based": True,
            "reliability": "medium",
            "database_url": "https://dl.acm.org/",
            "api_docs_url": None,
            "note": "Uses web scraping with robust live availability checks. May break if ACM changes website structure.",
            "features": [
                "title",
                "authors",
                "doi",
                "publication_date",
                "basic_metadata",
            ],
            "limitations": [
                "no_abstracts_in_search_results",
                "limited_to_20_results_per_page",
                "web_scraping_fragility",
            ],
        },
        "google_scholar": {
            "api_key_required": False,
            "rate_limit": 0.2,
            "free": True,
            "scraping_based": True,
            "reliability": "low",
            "database_url": "https://scholar.google.com/",
            "api_docs_url": None,
            "note": "Functional but actively blocks automated access. Use very conservative rate limits and expect occasional failures.",
            "features": [
                "title",
                "authors",
                "citations",
                "abstracts",
                "publication_info",
                "year_filtering",
            ],
            "limitations": [
                "frequent_blocking",
                "captcha_challenges",
                "rate_limiting_required",
                "results_limited_to_20",
            ],
            "recommendations": [
                "use_as_fallback_only",
                "implement_captcha_handling",
                "rotate_user_agents",
            ],
        },
    }

    return requirements.get(database_name, {})

QueryParser

QueryParser()

Parse complex Boolean search queries and convert to database-specific formats.

Source code in ures/literature/search/adapters.py

def __init__(self):
    self.logger = logging.getLogger(__name__)

parse_boolean_query

parse_boolean_query(query: str) -> Dict[str, Any]

Parse Boolean query into structured format.

Source code in ures/literature/search/adapters.py

def parse_boolean_query(self, query: str) -> Dict[str, Any]:
    """Parse Boolean query into structured format."""
    query = query.strip()

    # Extract quoted phrases first
    quoted_phrases = re.findall(r'"([^"]*)"', query)

    # Replace quoted phrases with placeholders to protect them
    temp_query = query
    for i, phrase in enumerate(quoted_phrases):
        temp_query = temp_query.replace(f'"{phrase}"', f"__PHRASE_{i}__")

    # Extract parenthetical groups
    groups = self._extract_parenthetical_groups(temp_query)

    # If no groups found, try to parse the whole query as a single group
    if not groups:
        if " OR " in temp_query:
            terms = [term.strip() for term in temp_query.split(" OR ")]
            groups.append({"type": "OR", "terms": terms})
        elif " AND " in temp_query:
            terms = [term.strip() for term in temp_query.split(" AND ")]
            groups.append({"type": "AND", "terms": terms})
        else:
            groups.append({"type": "SINGLE", "terms": [temp_query.strip()]})

    # Restore quoted phrases in terms
    for group in groups:
        for j, term in enumerate(group["terms"]):
            for i, phrase in enumerate(quoted_phrases):
                term = term.replace(f"__PHRASE_{i}__", f'"{phrase}"')
            group["terms"][j] = term

    return {
        "quoted_phrases": quoted_phrases,
        "groups": groups,
        "original_query": query,
    }

to_arxiv_query

to_arxiv_query(parsed_query: Dict) -> str

Convert parsed query to arXiv format.

Source code in ures/literature/search/adapters.py

def to_arxiv_query(self, parsed_query: Dict) -> str:
    """Convert parsed query to arXiv format."""
    terms = []

    for group in parsed_query["groups"]:
        if group["type"] == "OR":
            # Clean quotes from terms for arXiv
            clean_terms = [term.replace('"', "") for term in group["terms"]]
            group_terms = " OR ".join([f'all:"{term}"' for term in clean_terms])
            terms.append(f"({group_terms})")
        elif group["type"] == "AND":
            clean_terms = [term.replace('"', "") for term in group["terms"]]
            group_terms = " AND ".join([f'all:"{term}"' for term in clean_terms])
            terms.append(f"({group_terms})")
        else:
            clean_term = group["terms"][0].replace('"', "")
            terms.append(f'all:"{clean_term}"')

    final_query = " AND ".join(terms)
    self.logger.debug(f"Converted to arXiv query: {final_query}")
    return final_query

to_ieee_query

to_ieee_query(parsed_query: Dict) -> str

Convert parsed query to IEEE format.

Source code in ures/literature/search/adapters.py

def to_ieee_query(self, parsed_query: Dict) -> str:
    """Convert parsed query to IEEE format."""
    terms = []

    for group in parsed_query["groups"]:
        if group["type"] == "OR":
            group_terms = " OR ".join(group["terms"])
            terms.append(f"({group_terms})")
        elif group["type"] == "AND":
            group_terms = " AND ".join(group["terms"])
            terms.append(f"({group_terms})")
        else:
            terms.append(group["terms"][0])

    final_query = " AND ".join(terms)
    self.logger.debug(f"Converted to IEEE query: {final_query}")
    return final_query

to_simple_query

to_simple_query(parsed_query: Dict) -> str

Convert to simple query for basic APIs.

Source code in ures/literature/search/adapters.py

def to_simple_query(self, parsed_query: Dict) -> str:
    """Convert to simple query for basic APIs."""
    all_terms = []
    for group in parsed_query["groups"]:
        for term in group["terms"]:
            # Remove quotes and clean up
            clean_term = term.replace('"', "").strip()
            if clean_term and clean_term not in all_terms:
                all_terms.append(clean_term)

    simple_query = " ".join(all_terms)
    self.logger.debug(f"Converted to simple query: {simple_query}")
    return simple_query

to_google_scholar_query

to_google_scholar_query(parsed_query: Dict) -> str

Convert parsed query to Google Scholar format.

Source code in ures/literature/search/adapters.py

def to_google_scholar_query(self, parsed_query: Dict) -> str:
    """Convert parsed query to Google Scholar format."""
    terms = []

    for group in parsed_query["groups"]:
        if group["type"] == "OR":
            group_terms = " OR ".join(
                ['"' + term.replace('"', "") + '"' for term in group["terms"]]
            )
            terms.append(f"({group_terms})")
        elif group["type"] == "AND":
            group_terms = " ".join(
                ['"' + term.replace('"', "") + '"' for term in group["terms"]]
            )
            terms.append(group_terms)
        else:
            clean_term = group["terms"][0].replace('"', "")
            terms.append(f'"{clean_term}"')

    final_query = " ".join(terms)
    self.logger.debug(f"Converted to Google Scholar query: {final_query}")
    return final_query

Paper dataclass

Paper(title: str, authors: List[str], abstract: str = '', year: int = 0, venue: str = '', doi: str = '', arxiv_id: str = '', url: str = '', citations: int = 0, keywords: List[str] = None, database_source: str = '', pdf_url: str = '', publication_type: str = '', issue: str = '', volume: str = '', pages: str = '', publisher: str = '')

A research paper with standardized metadata.

Examples:

>>> from ures.literature import Paper
>>> paper = Paper(title="CXL memory", authors=["Doe, J."], year=2024)
>>> paper.title
'CXL memory'

get_canonical_id

get_canonical_id() -> str

Generate a canonical identifier for deduplication.

Source code in ures/literature/search/paper.py

def get_canonical_id(self) -> str:
    """Generate a canonical identifier for deduplication."""
    if self.doi:
        return f"doi:{self.doi}"
    elif self.arxiv_id:
        return f"arxiv:{self.arxiv_id}"
    else:
        title_norm = re.sub(r"[^\w\s]", "", self.title.lower()).strip()
        first_author = self.authors[0] if self.authors else ""
        content = f"{title_norm}|{first_author}|{self.year}"
        return f"hash:{hashlib.md5(content.encode()).hexdigest()}"

to_dict

to_dict() -> Dict

Convert to dictionary representation.

Source code in ures/literature/search/paper.py

def to_dict(self) -> Dict:
    """Convert to dictionary representation."""
    return asdict(self)

from_dict classmethod

from_dict(data: Dict) -> Paper

Create Paper instance from dictionary.

Source code in ures/literature/search/paper.py

@classmethod
def from_dict(cls, data: Dict) -> "Paper":
    """Create Paper instance from dictionary."""
    defaults = {
        "title": "",
        "authors": [],
        "abstract": "",
        "year": 0,
        "venue": "",
        "doi": "",
        "arxiv_id": "",
        "url": "",
        "citations": 0,
        "keywords": [],
        "database_source": "",
        "pdf_url": "",
        "publication_type": "",
        "issue": "",
        "volume": "",
        "pages": "",
        "publisher": "",
    }
    paper_data = {**defaults, **data}
    return cls(**paper_data)

similarity_score

similarity_score(other: Paper) -> float

Calculate similarity score with another paper (0-1).

Source code in ures/literature/search/paper.py

def similarity_score(self, other: "Paper") -> float:
    """Calculate similarity score with another paper (0-1)."""
    if not isinstance(other, Paper):
        return 0.0

    score = 0.0

    # Title similarity (highest weight)
    if self.title and other.title:
        title1 = re.sub(r"[^\w\s]", "", self.title.lower())
        title2 = re.sub(r"[^\w\s]", "", other.title.lower())
        if title1 == title2:
            score += 0.5
        elif title1 in title2 or title2 in title1:
            score += 0.3

    # Author overlap
    if self.authors and other.authors:
        common_authors = set(self.authors) & set(other.authors)
        if common_authors:
            score += 0.2 * (
                len(common_authors) / max(len(self.authors), len(other.authors))
            )

    # DOI or arXiv ID match
    if self.doi and other.doi and self.doi == other.doi:
        score += 0.3
    elif self.arxiv_id and other.arxiv_id and self.arxiv_id == other.arxiv_id:
        score += 0.3

    # Year proximity
    if self.year and other.year:
        year_diff = abs(self.year - other.year)
        if year_diff == 0:
            score += 0.1
        elif year_diff <= 1:
            score += 0.05

    return min(score, 1.0)

PaperFormatter

Unified formatter for normalizing papers from different database sources.

format_arxiv_paper staticmethod

format_arxiv_paper(entry_data: Dict) -> Paper

Format arXiv entry data into standardized Paper object.

Source code in ures/literature/search/paper.py

@staticmethod
def format_arxiv_paper(entry_data: Dict) -> Paper:
    """Format arXiv entry data into standardized Paper object."""
    title = entry_data.get("title", "").strip().replace("\n", " ")
    title = re.sub(r"\s+", " ", title)

    authors = []
    for author in entry_data.get("authors", []):
        if isinstance(author, dict):
            name = author.get("name", "")
        else:
            name = str(author)
        if name.strip():
            authors.append(name.strip())

    abstract = entry_data.get("summary", "").strip().replace("\n", " ")
    abstract = re.sub(r"\s+", " ", abstract)

    arxiv_id = entry_data.get("id", "").split("/")[-1]
    year = PaperFormatter._extract_arxiv_year(arxiv_id)

    keywords = []
    categories = entry_data.get("categories", [])
    if isinstance(categories, str):
        keywords = [categories]
    elif isinstance(categories, list):
        keywords = categories

    return Paper(
        title=title,
        authors=authors,
        abstract=abstract,
        year=year,
        venue="arXiv",
        arxiv_id=arxiv_id,
        url=entry_data.get("id", ""),
        keywords=keywords,
        database_source="arxiv",
        pdf_url=entry_data.get("pdf_url", ""),
        publication_type="preprint",
    )

format_ieee_paper staticmethod

format_ieee_paper(entry_data: Dict) -> Paper

Format IEEE entry data into standardized Paper object.

Source code in ures/literature/search/paper.py

@staticmethod
def format_ieee_paper(entry_data: Dict) -> Paper:
    """Format IEEE entry data into standardized Paper object."""
    title = entry_data.get("title", "")

    authors = []
    if "authors" in entry_data:
        if isinstance(entry_data["authors"], list):
            authors = [
                author.get("full_name", "") for author in entry_data["authors"]
            ]
        elif isinstance(entry_data["authors"], dict):
            authors = [entry_data["authors"].get("full_name", "")]

    return Paper(
        title=title,
        authors=authors,
        abstract=entry_data.get("abstract", ""),
        year=int(entry_data.get("publication_year", 0)),
        venue=entry_data.get("publication_title", ""),
        doi=entry_data.get("doi", ""),
        url=entry_data.get("pdf_url", ""),
        citations=int(entry_data.get("citing_paper_count", 0)),
        database_source="ieee",
        pdf_url=entry_data.get("pdf_url", ""),
        publication_type="article",
        volume=entry_data.get("volume", ""),
        issue=entry_data.get("issue", ""),
        pages=PaperFormatter._format_pages(
            entry_data.get("start_page", ""), entry_data.get("end_page", "")
        ),
        publisher="IEEE",
    )

format_elsevier_paper staticmethod

format_elsevier_paper(entry_data: Dict) -> Paper

Format Elsevier/ScienceDirect entry data.

Source code in ures/literature/search/paper.py

@staticmethod
def format_elsevier_paper(entry_data: Dict) -> Paper:
    """Format Elsevier/ScienceDirect entry data."""
    title = entry_data.get("dc:title", "")

    authors = []
    if "authors" in entry_data and "author" in entry_data["authors"]:
        author_list = entry_data["authors"]["author"]
        if isinstance(author_list, list):
            for author in author_list:
                if isinstance(author, dict):
                    given = author.get("ce:given-name", "")
                    surname = author.get("ce:surname", "")
                    full_name = f"{given} {surname}".strip()
                    if full_name:
                        authors.append(full_name)

    return Paper(
        title=title,
        authors=authors,
        abstract=entry_data.get("dc:description", ""),
        year=(
            int(entry_data.get("prism:coverDate", "").split("-")[0])
            if entry_data.get("prism:coverDate")
            else 0
        ),
        venue=entry_data.get("prism:publicationName", ""),
        doi=entry_data.get("prism:doi", ""),
        url=(
            entry_data.get("link", [{}])[0].get("@href", "")
            if entry_data.get("link")
            else ""
        ),
        database_source="elsevier",
        publication_type="article",
        volume=entry_data.get("prism:volume", ""),
        issue=entry_data.get("prism:issueIdentifier", ""),
        pages=entry_data.get("prism:pageRange", ""),
        publisher="Elsevier",
    )

format_springer_paper staticmethod

format_springer_paper(entry_data: Dict) -> Paper

Format Springer entry data.

Source code in ures/literature/search/paper.py

@staticmethod
def format_springer_paper(entry_data: Dict) -> Paper:
    """Format Springer entry data."""
    title = entry_data.get("title", "")

    authors = []
    if "creators" in entry_data:
        for creator in entry_data["creators"]:
            if isinstance(creator, dict):
                name = creator.get("creator", "")
            else:
                name = str(creator)
            if name:
                authors.append(name)

    return Paper(
        title=title,
        authors=authors,
        abstract=entry_data.get("abstract", ""),
        year=(
            int(entry_data.get("publicationDate", "").split("-")[0])
            if entry_data.get("publicationDate")
            else 0
        ),
        venue=entry_data.get("publicationName", ""),
        doi=entry_data.get("doi", ""),
        url=(
            entry_data.get("url", [{}])[0].get("value", "")
            if entry_data.get("url")
            else ""
        ),
        database_source="springer",
        publication_type=entry_data.get("contentType", "article").lower(),
        publisher="Springer",
    )

format_wiley_paper staticmethod

format_wiley_paper(entry_data: Dict) -> Paper

Format Wiley entry data.

Source code in ures/literature/search/paper.py

@staticmethod
def format_wiley_paper(entry_data: Dict) -> Paper:
    """Format Wiley entry data."""
    return Paper(
        title=entry_data.get("title", ""),
        authors=entry_data.get("authors", []),
        abstract=entry_data.get("abstract", ""),
        year=int(entry_data.get("publicationYear", 0)),
        venue=entry_data.get("source", ""),
        doi=entry_data.get("doi", ""),
        url=entry_data.get("onlineLibraryUrl", ""),
        database_source="wiley",
        publication_type="article",
        publisher="Wiley",
    )

LiteratureSearchEngine

LiteratureSearchEngine(config_dir: Optional[str] = None, app_name: str = 'literature-search')

Search one or more literature databases with optional caching.

Examples:

>>> from ures.literature import LiteratureSearchEngine
>>> engine = LiteratureSearchEngine()
>>> papers = engine.search("CXL memory", databases=["arxiv"], max_results=5)
>>> isinstance(papers, list)
True

Initialize the Literature Search Engine.

Parameters:

• config_dir (Optional[str], default: None ) –

  Path to a directory used to store configuration file

• app_name (str, default: 'literature-search' ) –

  Application name for key management

Source code in ures/literature/search/search.py

def __init__(
    self, config_dir: Optional[str] = None, app_name: str = "literature-search"
):
    """
    Initialize the Literature Search Engine.

    Args:
                    config_dir: Path to a directory used to store configuration file
                    app_name: Application name for key management
    """
    self.config = DatabaseConfig(config_dir, app_name)
    self.cache = CacheManager(self.config.config["cache"]["directory"])
    self.adapters: Dict[str, DatabaseAdapter] = {}
    self.logger = logging.getLogger(__name__)

    # Statistics
    self.stats = {
        "total_searches": 0,
        "cached_searches": 0,
        "total_papers_found": 0,
        "databases_used": set(),
        "last_search_time": None,
    }

    self._init_adapters()

search

search(query: str, databases: List[str] = None, max_results: int = None, use_cache: bool = True, year_min: int = None, **kwargs) -> List[Paper]

Search multiple databases with Boolean query support.

Parameters:

• query (str) –

  Boolean search query

• databases (List[str], default: None ) –

  List of database names to search

• max_results (int, default: None ) –

  Maximum results per database

• use_cache (bool, default: True ) –

  Whether to use cached results

• year_min (int, default: None ) –

  Minimum publication year

• **kwargs (Any, default: {} ) –

  Extra search options forwarded to adapters.

Returns:

• List[Paper] –

  List of Paper objects, deduplicated and unified

Source code in ures/literature/search/search.py

def search(
    self,
    query: str,
    databases: List[str] = None,
    max_results: int = None,
    use_cache: bool = True,
    year_min: int = None,
    **kwargs,
) -> List[Paper]:
    """
    Search multiple databases with Boolean query support.

    Args:
                    query: Boolean search query
                    databases: List of database names to search
                    max_results: Maximum results per database
                    use_cache: Whether to use cached results
                    year_min: Minimum publication year
                    **kwargs (Any): Extra search options forwarded to adapters.

    Returns:
                    List of Paper objects, deduplicated and unified
    """
    if not query or not query.strip():
        return []

    if databases is None:
        databases = list(self.adapters.keys())

    if max_results is None:
        max_results = self.config.config["search"]["max_results"]

    if year_min is None:
        year_min = self.config.config["search"]["min_year"]

    self.logger.info(f"Searching with Boolean query: {query}")
    self.logger.info(f"Target databases: {databases}")

    # Update statistics
    self.stats["total_searches"] += 1
    self.stats["last_search_time"] = datetime.now().isoformat()

    # Check cache first
    if use_cache:
        max_age_hours = self.config.config["cache"]["max_age_hours"]
        cached_results = self.cache.get_cached_search(
            query, databases, max_age_hours
        )
        if cached_results:
            self.logger.info(f"Retrieved {len(cached_results)} papers from cache")
            self.stats["cached_searches"] += 1
            return self._filter_by_year(cached_results, year_min)

    # Perform searches across all databases
    all_papers = []
    for db_name in databases:
        if db_name not in self.adapters:
            self.logger.warning(f"Database {db_name} not available")
            continue

        self.logger.info(f"Searching {db_name}...")
        try:
            adapter = self.adapters[db_name]

            # Database-specific parameters
            search_kwargs = {}
            if db_name == "arxiv":
                categories = self.config.get_database_config("arxiv").get(
                    "categories", []
                )
                search_kwargs["categories"] = categories
            if year_min:
                search_kwargs["year_min"] = year_min

            papers = adapter.search(query, max_results, **search_kwargs)

            # Apply year filter
            if year_min:
                papers = [p for p in papers if p.year >= year_min]

            all_papers.extend(papers)
            self.logger.info(f"Found {len(papers)} papers from {db_name}")
            self.stats["databases_used"].add(db_name)

            # Cache individual papers
            for paper in papers:
                self.cache.cache_paper(paper)

        except Exception as e:
            self.logger.error(f"Error searching {db_name}: {e}")

    # Deduplicate results using canonical IDs
    if self.config.config["search"]["deduplication"]:
        all_papers = self._deduplicate_papers(all_papers)

    # Cache search results
    if use_cache and all_papers:
        self.cache.cache_search_results(query, databases, all_papers)

    # Sort by relevance (citations and year)
    all_papers.sort(key=lambda p: (-p.citations, -p.year))

    self.stats["total_papers_found"] += len(all_papers)
    self.logger.info(f"Total unique papers found: {len(all_papers)}")
    return all_papers

get_cached_papers

get_cached_papers(query: str = None, year_min: int = None, database_source: str = None, limit: int = None) -> List[Paper]

Retrieve papers from local cache.

Source code in ures/literature/search/search.py

def get_cached_papers(
    self,
    query: str = None,
    year_min: int = None,
    database_source: str = None,
    limit: int = None,
) -> List[Paper]:
    """Retrieve papers from local cache."""
    return self.cache.get_cached_papers(query, year_min, database_source, limit)

export_results

export_results(papers: List[Paper], format: str = None, filename: str = None, include_abstracts: bool = None) -> Optional[str]

Export search results to various formats.

Parameters:

• papers (List[Paper]) –

  List of papers to export

• format (str, default: None ) –

  Export format ('json', 'csv', 'bibtex')

• filename (str, default: None ) –

  Output filename

• include_abstracts (bool, default: None ) –

  Whether to include abstracts

Returns:

• str ( Optional[str] ) –

  Path to exported file or None if failed

Source code in ures/literature/search/search.py

def export_results(
    self,
    papers: List[Paper],
    format: str = None,
    filename: str = None,
    include_abstracts: bool = None,
) -> Optional[str]:
    """
    Export search results to various formats.

    Args:
                    papers: List of papers to export
                    format: Export format ('json', 'csv', 'bibtex')
                    filename: Output filename
                    include_abstracts: Whether to include abstracts

    Returns:
                    str: Path to exported file or None if failed
    """
    if not papers:
        self.logger.warning("No papers to export")
        return None

    if format is None:
        format = self.config.config["export"]["default_format"]

    if include_abstracts is None:
        include_abstracts = self.config.config["export"]["include_abstracts"]

    if not filename:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"literature_search_{timestamp}.{format}"

    try:
        max_size = self.config.config["export"]["max_export_size"]
        if len(papers) > max_size:
            self.logger.warning(f"Limiting export to {max_size} papers")
            papers = papers[:max_size]

        if format == "json":
            return self._export_json(papers, filename, include_abstracts)
        elif format == "bibtex":
            return self._export_bibtex(papers, filename)
        elif format == "csv":
            return self._export_csv(papers, filename, include_abstracts)
        else:
            raise ValueError(f"Unsupported export format: {format}")

    except Exception as e:
        self.logger.error(f"Export failed: {e}")
        return None

analyze_search_coverage

analyze_search_coverage(query: str, databases: List[str] = None) -> Dict[str, Any]

Analyze search coverage across databases.

Source code in ures/literature/search/search.py

def analyze_search_coverage(
    self, query: str, databases: List[str] = None
) -> Dict[str, Any]:
    """Analyze search coverage across databases."""
    if databases is None:
        databases = list(self.adapters.keys())

    coverage_report = {
        "query": query,
        "databases_searched": [],
        "results_per_database": {},
        "total_papers": 0,
        "unique_papers": 0,
        "overlap_analysis": {},
        "adapter_stats": {},
    }

    all_papers = []
    db_papers = {}

    for db_name in databases:
        if db_name not in self.adapters:
            continue

        try:
            adapter = self.adapters[db_name]
            papers = adapter.search(query, 50)  # Limited for analysis
            db_papers[db_name] = papers
            all_papers.extend(papers)

            coverage_report["databases_searched"].append(db_name)
            coverage_report["results_per_database"][db_name] = len(papers)
            coverage_report["adapter_stats"][db_name] = adapter.get_stats()

        except Exception as e:
            self.logger.error(f"Error in coverage analysis for {db_name}: {e}")

    coverage_report["total_papers"] = len(all_papers)

    # Calculate unique papers
    unique_papers = self._deduplicate_papers(all_papers)
    coverage_report["unique_papers"] = len(unique_papers)

    # Calculate overlap between databases
    for db1 in db_papers:
        for db2 in db_papers:
            if db1 < db2:  # Avoid duplicate pairs
                overlap = self._calculate_overlap(db_papers[db1], db_papers[db2])
                coverage_report["overlap_analysis"][f"{db1}_vs_{db2}"] = overlap

    return coverage_report

get_engine_stats

get_engine_stats() -> Dict[str, Any]

Get comprehensive engine statistics.

Source code in ures/literature/search/search.py

def get_engine_stats(self) -> Dict[str, Any]:
    """Get comprehensive engine statistics."""
    stats = self.stats.copy()
    stats["databases_used"] = list(stats["databases_used"])
    stats["available_adapters"] = list(self.adapters.keys())
    stats["cache_stats"] = self.cache.get_cache_stats()

    # Add adapter statistics
    adapter_stats = {}
    for name, adapter in self.adapters.items():
        adapter_stats[name] = adapter.get_stats()
    stats["adapter_stats"] = adapter_stats

    return stats

cleanup_cache

cleanup_cache(days_old: int = None) -> bool

Clean up old cache entries.

Source code in ures/literature/search/search.py

def cleanup_cache(self, days_old: int = None) -> bool:
    """Clean up old cache entries."""
    if days_old is None:
        days_old = self.config.config["cache"]["expire_days"]

    return self.cache.cleanup_cache(days_old)

find_duplicates

find_duplicates(similarity_threshold: float = None) -> List[List[Paper]]

Find potential duplicate papers in cache.

Source code in ures/literature/search/search.py

def find_duplicates(self, similarity_threshold: float = None) -> List[List[Paper]]:
    """Find potential duplicate papers in cache."""
    if similarity_threshold is None:
        similarity_threshold = self.config.config["search"]["similarity_threshold"]

    return self.cache.find_duplicates(similarity_threshold)

reload_adapters

reload_adapters()

Reload all database adapters (useful after config changes).

Source code in ures/literature/search/search.py

def reload_adapters(self):
    """Reload all database adapters (useful after config changes)."""
    self.adapters.clear()
    self._init_adapters()
    self.logger.info("Reloaded database adapters")

DatabaseConfig

DatabaseConfig(config_dir: Optional[str] = None, app_name: str = 'literature-search')

Enhanced configuration management with secure API key integration.

Source code in ures/literature/search/search.py

def __init__(
    self, config_dir: Optional[str] = None, app_name: str = "literature-search"
):
    config_dir = config_dir or Path.home() / ".ures_lit_search"
    self.config_dir = config_dir
    self.app_name = app_name
    self.logger = logging.getLogger(__name__)
    self.key_manager = SecureKeyManager(app_name, config_dir=self.config_dir)
    self.config = self._load_config()

get_database_config

get_database_config(db_name: str) -> Dict

Get configuration for a specific database.

Source code in ures/literature/search/search.py

def get_database_config(self, db_name: str) -> Dict:
    """Get configuration for a specific database."""
    return self.config.get("databases", {}).get(db_name, {})

is_database_enabled

is_database_enabled(db_name: str) -> bool

Check if a database is enabled.

Source code in ures/literature/search/search.py

def is_database_enabled(self, db_name: str) -> bool:
    """Check if a database is enabled."""
    return self.get_database_config(db_name).get("enabled", False)

get_api_key

get_api_key(db_name: str) -> Optional[str]

Get API key for a database using secure storage.

Source code in ures/literature/search/search.py

def get_api_key(self, db_name: str) -> Optional[str]:
    """Get API key for a database using secure storage."""
    db_config = self.get_database_config(db_name)

    if "api_key" in db_config and db_config["api_key"]:
        return db_config["api_key"]

    # Use secure key manager
    return self.key_manager.get_key(db_name)

set_api_key

set_api_key(db_name: str, api_key: str, method: StorageMethod = StorageMethod.ENCRYPTED) -> bool

Set API key for a database using secure storage.

Parameters:

• db_name (str) –

  Database name

• api_key (str) –

  API key or reference (depends on method)

• method (StorageMethod, default: ENCRYPTED ) –

  Storage method ('encrypted', 'env', '1password', 'keychain')

Returns:

• bool ( bool ) –

  Success status

Source code in ures/literature/search/search.py

def set_api_key(
    self,
    db_name: str,
    api_key: str,
    method: StorageMethod = StorageMethod.ENCRYPTED,
) -> bool:
    """
    Set API key for a database using secure storage.

    Args:
                    db_name: Database name
                    api_key: API key or reference (depends on method)
                    method: Storage method ('encrypted', 'env', '1password', 'keychain')

    Returns:
                    bool: Success status
    """
    # Store the key securely
    success = self.key_manager.store_key(db_name, api_key, method)

    if success:
        # Update config to reflect the storage method
        if "databases" not in self.config:
            self.config["databases"] = {}
        if db_name not in self.config["databases"]:
            self.config["databases"][db_name] = {}

        self.config["databases"][db_name]["api_key_method"] = method
        self.config["databases"][db_name]["enabled"] = True

        self._save_config(self.config)

    return success

list_api_keys

list_api_keys() -> Dict[str, Dict]

List all stored API keys (without revealing actual keys).

Source code in ures/literature/search/search.py

def list_api_keys(self) -> Dict[str, Dict]:
    """List all stored API keys (without revealing actual keys)."""
    return self.key_manager.list_keys()

delete_api_key

delete_api_key(db_name: str) -> bool

Delete API key for a database.

Source code in ures/literature/search/search.py

def delete_api_key(self, db_name: str) -> bool:
    """Delete API key for a database."""
    success = self.key_manager.delete_key(db_name)

    if success:
        # Update config
        if db_name in self.config.get("databases", {}):
            self.config["databases"][db_name]["enabled"] = False
            self._save_config(self.config)

    return success

update_config

update_config(section: str, key: str, value: Any) -> bool

Update a configuration value.

Source code in ures/literature/search/search.py

def update_config(self, section: str, key: str, value: Any) -> bool:
    """Update a configuration value."""
    try:
        if section not in self.config:
            self.config[section] = {}

        self.config[section][key] = value
        self._save_config(self.config)
        return True
    except Exception as e:
        self.logger.error(f"Failed to update config: {e}")
        return False

get_config_value

get_config_value(section: str, key: str, default=None)

Get a configuration value.

Source code in ures/literature/search/search.py

def get_config_value(self, section: str, key: str, default=None):
    """Get a configuration value."""
    return self.config.get(section, {}).get(key, default)

LiteratureSearchCLI

LiteratureSearchCLI()

Enhanced command-line interface for literature search.

Source code in ures/literature/search_cli.py

def __init__(self):
    self.engine = None
    self.config = None
    self.logger = logging.getLogger(__name__)

setup_logging

setup_logging(level: str = 'INFO')

Setup logging configuration.

Source code in ures/literature/search_cli.py

def setup_logging(self, level: str = "INFO"):
    """Setup logging configuration."""
    log_level = getattr(logging, level.upper(), logging.INFO)
    logging.basicConfig(
        level=log_level,
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    )

init_engine

init_engine(config_dir: Optional[str] = None) -> bool

Initialize the search engine.

Source code in ures/literature/search_cli.py

def init_engine(self, config_dir: Optional[str] = None) -> bool:
    """Initialize the search engine."""
    try:
        self.config = DatabaseConfig(config_dir)
        self.engine = LiteratureSearchEngine(config_dir)
        return True
    except Exception as e:
        print(f"❌ Failed to initialize search engine: {e}")
        return False

cmd_init

cmd_init(args)

Initialize the literature search system interactively.

Source code in ures/literature/search_cli.py

def cmd_init(self, args):
    """Initialize the literature search system interactively."""
    print("🚀 Welcome to Literature Search System!")
    print("This wizard will help you set up everything you need.")
    print("=" * 60)

    # Initialize config and key manager
    config_dir = args.config_dir
    if not config_dir:
        config_dir = Path.home() / ".ures_lit_search"

    # Create directories
    Path(config_dir).mkdir(parents=True, exist_ok=True)

    # Initialize components
    key_manager = SecureKeyManager("literature-search", config_dir=config_dir)
    config = DatabaseConfig(config_dir)

    setup = InteractiveSetup(key_manager, config)

    # Run setup steps
    setup.setup_secrets()
    setup.setup_configuration()

    print(f"\n🎉 Setup completed!")
    print(f"Configuration stored in: {config.config_path}")
    print(f"API keys stored securely in: {key_manager.config_dir}")
    print(f"\nYou can now use: python {sys.argv[0]} search 'your query here'")

cmd_status

cmd_status(args)

Show comprehensive system status.

Source code in ures/literature/search_cli.py

def cmd_status(self, args):
    """Show comprehensive system status."""
    print("📊 Literature Search System Status")
    print("=" * 60)

    # Initialize to get config
    if not self.init_engine(args.config_dir):
        return

    # API Keys Status
    print("🔐 API Keys Status:")
    print("-" * 30)

    key_info = self.config.list_api_keys()
    if not key_info:
        print("  No API keys configured")
    else:
        for service, info in key_info.items():
            status = "✅" if info.get("has_key", False) else "❌"
            method = info.get("method", "unknown")
            print(f"  {service:15} {status} ({method})")

    # Database Status
    print(f"\n📚 Database Status:")
    print("-" * 30)

    databases = AdapterFactory.get_supported_databases()
    for db in databases:
        req = AdapterFactory.get_adapter_requirements(db)
        enabled = self.config.is_database_enabled(db)

        # Check if adapter is actually available
        available = False
        if enabled and db in self.engine.adapters:
            adapter = self.engine.adapters[db]
            available = adapter.is_available()

        # Status indicators
        config_status = "✅" if enabled else "⚪"
        avail_status = "🟢" if available else "🔴" if enabled else "⚪"
        api_status = "🔑" if req.get("api_key_required", False) else "🆓"

        print(f"  {db:15} {config_status}{avail_status}{api_status}")

    print(f"\nLegend:")
    print(f"  Config: ✅ Enabled ⚪ Disabled")
    print(f"  Status: 🟢 Available 🔴 Error ⚪ Not configured")
    print(f"  Access: 🔑 API Key Required 🆓 Free")

    # Cache Status
    cache_stats = self.engine.get_engine_stats().get("cache_stats", {})
    print(f"\n💾 Cache Status:")
    print("-" * 30)
    print(f"  Total papers: {cache_stats.get('total_papers', 0)}")
    print(f"  Recent searches: {cache_stats.get('recent_searches', 0)}")
    print(f"  Papers by source: {cache_stats.get('papers_by_source', {})}")

    # System Info
    engine_stats = self.engine.get_engine_stats()
    print(f"\n🔍 Search Engine Status:")
    print("-" * 30)
    print(f"  Available adapters: {len(self.engine.adapters)}")
    print(f"  Total searches: {engine_stats.get('total_searches', 0)}")
    print(f"  Cached searches: {engine_stats.get('cached_searches', 0)}")
    print(f"  Papers found: {engine_stats.get('total_papers_found', 0)}")

cmd_search

cmd_search(args)

Perform literature search with enhanced output.

Source code in ures/literature/search_cli.py

def cmd_search(self, args):
    """Perform literature search with enhanced output."""
    if not self.init_engine(args.config_dir):
        return

    query = args.query
    if not query:
        print("❌ No search query provided")
        return

    # Check if any adapters are available
    if not self.engine.adapters:
        print("❌ No database adapters available.")
        print("Run the initialization wizard first:")
        print(f"  python {sys.argv[0]} init")
        return

    print(f"🔍 Searching Literature")
    print("=" * 50)
    print(f"Query: {query}")

    # Show search parameters
    params_info = []
    if args.databases:
        params_info.append(f"Databases: {', '.join(args.databases)}")
    else:
        params_info.append(f"Databases: {', '.join(self.engine.adapters.keys())}")

    if args.max_results:
        params_info.append(f"Max results: {args.max_results}")
    if args.year_min:
        params_info.append(f"Min year: {args.year_min}")
    if args.no_cache:
        params_info.append("Cache: disabled")

    if params_info:
        print(f"Parameters: {' | '.join(params_info)}")

    print("-" * 50)

    # Perform search with timing
    start_time = datetime.now()

    try:
        papers = self.engine.search(
            query=query,
            databases=args.databases,
            max_results=args.max_results,
            year_min=args.year_min,
            use_cache=not args.no_cache,
        )

        search_time = (datetime.now() - start_time).total_seconds()

        if not papers:
            print("📭 No papers found matching your query.")
            self._show_suggestions(query)
            return

        # Display results summary
        print(f"📚 Found {len(papers)} papers in {search_time:.2f}s")

        # Group by database source
        by_source = {}
        for paper in papers:
            source = paper.database_source
            by_source.setdefault(source, 0)
            by_source[source] += 1

        source_summary = ", ".join(
            [f"{src}: {count}" for src, count in by_source.items()]
        )
        print(f"Sources: {source_summary}")
        print("=" * 50)

        # Display detailed results
        display_count = min(args.show, len(papers))
        for i, paper in enumerate(papers[:display_count], 1):
            self._display_paper(i, paper, args.show_abstracts, args.show_urls)

        if len(papers) > display_count:
            print(f"\n... and {len(papers) - display_count} more papers")
            print(f"Use --show {len(papers)} to see all results")

        # Export if requested
        if args.export:
            self._export_results(papers, args)

        # Show statistics if verbose
        if args.verbose:
            self._show_search_stats()

    except Exception as e:
        print(f"❌ Search failed: {e}")
        if args.verbose:
            import traceback

            traceback.print_exc()

cmd_config

cmd_config(args)

Configuration management.

Source code in ures/literature/search_cli.py

def cmd_config(self, args):
    """Configuration management."""
    if not self.init_engine(args.config_dir):
        return

    if args.config_action == "show":
        print("⚙️ Current Configuration")
        print("=" * 40)
        print(json.dumps(self.config.config, indent=2))

    elif args.config_action == "edit":
        # Interactive configuration editing
        setup = InteractiveSetup(self.config.key_manager, self.config)
        setup.setup_configuration()

    elif args.config_action == "keys":
        # API key management
        setup = InteractiveSetup(self.config.key_manager, self.config)
        setup.setup_secrets()

    elif args.config_action == "test":
        # Test all configured API keys
        print("🧪 Testing API Keys")
        print("-" * 30)

        key_info = self.config.list_api_keys()
        for service, info in key_info.items():
            if info.get("has_key", False):
                test_result = self.config.key_manager.test_key_access(service)
                status = "✅" if test_result["accessible"] else "❌"
                print(f"  {service:15} {status}")
                if test_result.get("error"):
                    print(f"                  Error: {test_result['error']}")

cmd_cache

cmd_cache(args)

Enhanced cache management.

Source code in ures/literature/search_cli.py

def cmd_cache(self, args):
    """Enhanced cache management."""
    if not self.init_engine(args.config_dir):
        return

    if args.cache_action == "stats":
        stats = self.engine.cache.get_cache_stats()
        print("📊 Cache Statistics")
        print("=" * 40)
        print(f"Total papers: {stats.get('total_papers', 0):,}")
        print(f"Unique sources: {stats.get('unique_sources', 0)}")
        print(f"Total searches: {stats.get('total_searches', 0)}")
        print(f"Recent searches (24h): {stats.get('recent_searches', 0)}")

        # Papers by source
        papers_by_source = stats.get("papers_by_source", {})
        if papers_by_source:
            print(f"\nPapers by database:")
            for source, count in sorted(
                papers_by_source.items(), key=lambda x: x[1], reverse=True
            ):
                print(f"  {source:15}: {count:,}")

        # Recent years distribution
        recent_years = stats.get("recent_years", {})
        if recent_years:
            print(f"\nPapers by year (top 5):")
            for year, count in list(recent_years.items())[:5]:
                print(f"  {year}: {count:,}")

    elif args.cache_action == "clean":
        days = args.days or 30
        print(f"🧹 Cleaning cache entries older than {days} days...")
        success = self.engine.cleanup_cache(days)
        print(
            "✅ Cache cleanup completed" if success else "❌ Cache cleanup failed"
        )

    elif args.cache_action == "export":
        format_type = args.format or "json"
        filename = args.output
        print(f"💾 Exporting cached papers to {format_type.upper()}...")

        result = self.engine.cache.export_papers(format_type, filename)
        if result:
            print(f"✅ Exported to: {result}")
        else:
            print("❌ Export failed")

    elif args.cache_action == "duplicates":
        threshold = args.similarity or 0.8
        print(f"🔍 Finding duplicates (similarity ≥ {threshold})...")

        duplicates = self.engine.find_duplicates(threshold)
        if duplicates:
            print(f"Found {len(duplicates)} groups of similar papers:")
            for i, group in enumerate(duplicates[:10], 1):  # Limit display
                print(f"\nGroup {i} ({len(group)} papers):")
                for paper in group:
                    print(
                        f"  • {paper.title[:80]}{'...' if len(paper.title) > 80 else ''}"
                    )
                    print(f"    {paper.year} | {paper.database_source}")

            if len(duplicates) > 10:
                print(f"\n... and {len(duplicates) - 10} more groups")
        else:
            print("No duplicates found")

run

run()

Main CLI entry point with comprehensive command structure.

Source code in ures/literature/search_cli.py

    def run(self):
        """Main CLI entry point with comprehensive command structure."""
        parser = argparse.ArgumentParser(
            description="Literature Search System - Advanced academic database search with Boolean queries",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog="""
Examples:
  # First time setup
  python literature_search.py init

  # Quick searches
  python literature_search.py search "machine learning performance"
  python literature_search.py search '"deep learning" AND optimization' --export

  # Advanced searches
  python literature_search.py search 'software AND (performance OR efficiency)' \\
      --databases arxiv ieee --year-min 2020 --max-results 50

  # System management
  python literature_search.py status
  python literature_search.py config keys
  python literature_search.py cache stats

Boolean Query Examples:
  • "machine learning" AND performance
  • software OR system AND "neural network"
  • ("deep learning" OR "artificial intelligence") AND optimization
  • cloud computing AND NOT "edge computing"
            """,
        )

        # Global options
        parser.add_argument(
            "--verbose", "-v", action="store_true", help="Enable verbose output"
        )
        parser.add_argument("--config-dir", help="Custom configuration directory")
        parser.add_argument(
            "--log-level",
            default="INFO",
            choices=["DEBUG", "INFO", "WARNING", "ERROR"],
            help="Set logging level",
        )

        subparsers = parser.add_subparsers(dest="command", help="Available commands")

        # Init command - Interactive setup
        init_parser = subparsers.add_parser("init", help="Interactive system setup")

        # Search command - Enhanced with quick access
        search_parser = subparsers.add_parser(
            "search", help="Search academic literature"
        )
        search_parser.add_argument(
            "query", help="Search query (supports Boolean operators)"
        )
        search_parser.add_argument(
            "--databases", nargs="+", help="Specific databases to search"
        )
        search_parser.add_argument(
            "--max-results", type=int, help="Maximum results per database"
        )
        search_parser.add_argument(
            "--year-min", type=int, help="Minimum publication year"
        )
        search_parser.add_argument(
            "--show", type=int, default=10, help="Number of results to display"
        )
        search_parser.add_argument(
            "--show-abstracts", action="store_true", help="Include abstracts in output"
        )
        search_parser.add_argument(
            "--show-urls", action="store_true", help="Include URLs in output"
        )
        search_parser.add_argument(
            "--no-cache", action="store_true", help="Disable cache usage"
        )
        search_parser.add_argument(
            "--export", action="store_true", help="Export results to file"
        )
        search_parser.add_argument(
            "--format",
            choices=["json", "csv", "bibtex"],
            default="json",
            help="Export format",
        )
        search_parser.add_argument("--output", help="Output filename for export")

        # Status command - System overview
        status_parser = subparsers.add_parser("status", help="Show system status")

        # Config command - Configuration management
        config_parser = subparsers.add_parser("config", help="Configuration management")
        config_subparsers = config_parser.add_subparsers(
            dest="config_action", help="Config actions"
        )

        config_show = config_subparsers.add_parser(
            "show", help="Show current configuration"
        )
        config_edit = config_subparsers.add_parser(
            "edit", help="Interactive configuration editor"
        )
        config_keys = config_subparsers.add_parser("keys", help="Manage API keys")
        config_test = config_subparsers.add_parser(
            "test", help="Test API key accessibility"
        )

        # Cache command - Enhanced cache management
        cache_parser = subparsers.add_parser("cache", help="Cache management")
        cache_subparsers = cache_parser.add_subparsers(
            dest="cache_action", help="Cache actions"
        )

        cache_stats = cache_subparsers.add_parser("stats", help="Show cache statistics")

        cache_clean = cache_subparsers.add_parser(
            "clean", help="Clean old cache entries"
        )
        cache_clean.add_argument(
            "--days", type=int, help="Remove entries older than N days"
        )

        cache_export = cache_subparsers.add_parser(
            "export", help="Export cached papers"
        )
        cache_export.add_argument("--format", choices=["json", "csv"], default="json")
        cache_export.add_argument("--output", help="Output filename")

        cache_duplicates = cache_subparsers.add_parser(
            "duplicates", help="Find duplicate papers"
        )
        cache_duplicates.add_argument(
            "--similarity",
            type=float,
            default=0.8,
            help="Similarity threshold (0.0-1.0)",
        )

        # Coverage command - Search coverage analysis
        coverage_parser = subparsers.add_parser(
            "coverage", help="Analyze search coverage"
        )
        coverage_parser.add_argument("query", help="Query for coverage analysis")
        coverage_parser.add_argument(
            "--databases", nargs="+", help="Databases to analyze"
        )

        # Quick commands for common operations
        quick_parser = subparsers.add_parser("quick", help="Quick operations")
        quick_subparsers = quick_parser.add_subparsers(
            dest="quick_action", help="Quick actions"
        )

        quick_arxiv = quick_subparsers.add_parser("arxiv", help="Quick arXiv search")
        quick_arxiv.add_argument("query", help="Search query")
        quick_arxiv.add_argument("--max", type=int, default=20, help="Max results")

        quick_all = quick_subparsers.add_parser(
            "all", help="Search all available databases"
        )
        quick_all.add_argument("query", help="Search query")
        quick_all.add_argument(
            "--max", type=int, default=10, help="Max results per database"
        )

        # Parse arguments and handle commands
        if len(sys.argv) == 1:
            parser.print_help()
            return

        args = parser.parse_args()

        # Setup logging
        self.setup_logging(args.log_level)

        # Handle commands
        if args.command == "init":
            self.cmd_init(args)
        elif args.command == "search":
            self.cmd_search(args)
        elif args.command == "status":
            self.cmd_status(args)
        elif args.command == "config":
            self.cmd_config(args)
        elif args.command == "cache":
            self.cmd_cache(args)
        elif args.command == "coverage":
            self.cmd_coverage(args)
        elif args.command == "quick":
            self.cmd_quick(args)
        else:
            parser.print_help()

cmd_coverage

cmd_coverage(args)

Analyze search coverage across databases.

Source code in ures/literature/search_cli.py

def cmd_coverage(self, args):
    """Analyze search coverage across databases."""
    if not self.init_engine(args.config_dir):
        return

    query = args.query
    print(f"📈 Search Coverage Analysis")
    print("=" * 50)
    print(f"Query: {query}")

    if args.databases:
        print(f"Analyzing databases: {', '.join(args.databases)}")

    print("-" * 50)

    try:
        coverage = self.engine.analyze_search_coverage(query, args.databases)

        # Results summary
        print(f"📊 Coverage Summary:")
        print(f"   Total papers found: {coverage['total_papers']}")
        print(f"   Unique papers: {coverage['unique_papers']}")
        print(f"   Databases searched: {len(coverage['databases_searched'])}")

        # Results per database
        if coverage["results_per_database"]:
            print(f"\n📚 Results by Database:")
            for db, count in coverage["results_per_database"].items():
                percentage = (
                    (count / coverage["total_papers"] * 100)
                    if coverage["total_papers"] > 0
                    else 0
                )
                print(f"   {db:15}: {count:3} papers ({percentage:5.1f}%)")

        # Overlap analysis
        if coverage["overlap_analysis"]:
            print(f"\n🔄 Database Overlaps:")
            for pair, overlap in coverage["overlap_analysis"].items():
                db1, db2 = pair.split("_vs_")
                total1 = coverage["results_per_database"].get(db1, 0)
                total2 = coverage["results_per_database"].get(db2, 0)

                if total1 > 0 and total2 > 0:
                    overlap_pct = overlap / min(total1, total2) * 100
                    print(
                        f"   {db1} ↔ {db2}: {overlap} papers ({overlap_pct:.1f}% overlap)"
                    )

        # Recommendations
        print(f"\n💡 Recommendations:")
        if coverage["unique_papers"] < coverage["total_papers"] * 0.8:
            print(
                "   • High overlap detected - consider focusing on fewer databases"
            )

        no_result_dbs = [
            db
            for db, count in coverage["results_per_database"].items()
            if count == 0
        ]
        if no_result_dbs:
            print(
                f"   • No results from: {', '.join(no_result_dbs)} - try different terms"
            )

        if coverage["unique_papers"] > 0:
            best_db = max(
                coverage["results_per_database"].items(), key=lambda x: x[1]
            )
            print(f"   • Best single source: {best_db[0]} ({best_db[1]} papers)")

    except Exception as e:
        print(f"❌ Coverage analysis failed: {e}")
        if args.verbose:
            import traceback

            traceback.print_exc()

cmd_quick

cmd_quick(args)

Quick search operations for common use cases.

Source code in ures/literature/search_cli.py

def cmd_quick(self, args):
    """Quick search operations for common use cases."""
    if not self.init_engine(args.config_dir):
        return

    if args.quick_action == "arxiv":
        print(f"🚀 Quick arXiv Search")
        print("=" * 30)

        if "arxiv" not in self.engine.adapters:
            print("❌ arXiv adapter not available")
            return

        papers = self.engine.search(
            query=args.query,
            databases=["arxiv"],
            max_results=args.max,
            use_cache=True,
        )

        self._display_quick_results(papers, "arXiv")

    elif args.quick_action == "all":
        print(f"🚀 Quick Multi-Database Search")
        print("=" * 35)

        if not self.engine.adapters:
            print("❌ No database adapters available")
            return

        papers = self.engine.search(
            query=args.query, max_results=args.max, use_cache=True
        )

        self._display_quick_results(papers, "All Databases")