search

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)