adapters

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

DatabaseAdapter

DatabaseAdapter(rate_limit: float = 1.0, api_key: str = None)

Bases: ABC

Abstract base class for all database adapters.

Initialize database adapter.

Parameters:

• rate_limit (float, default: 1.0 ) –

  Requests per second limit

• api_key (str, default: None ) –

  API key for the database (if required)

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

def __init__(self, rate_limit: float = 1.0, api_key: str = None):
    """
    Initialize database adapter.

    Args:
                                    rate_limit: Requests per second limit
                                    api_key: API key for the database (if required)
    """
    self.rate_limit = rate_limit
    self.api_key = api_key
    self.last_request = 0
    self.logger = logging.getLogger(self.__class__.__name__)
    self.query_parser = QueryParser()
    self._availability_cache = None
    self._availability_check_time = 0
    self._availability_cache_duration = 300  # 5 minutes

    # Statistics
    self.stats = {
        "total_requests": 0,
        "successful_requests": 0,
        "failed_requests": 0,
        "total_papers_found": 0,
        "last_request_time": None,
        "rate_limit_hits": 0,
        "availability_checks": 0,
        "last_availability_check": None,
    }

search abstractmethod

search(query: str, max_results: int = 100, **kwargs) -> List[Paper]

Search the database for papers.

Parameters:

• query (str) –

  Search query (supports Boolean operations)

• max_results (int, default: 100 ) –

  Maximum number of results to return

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

  Database-specific search options.

Returns:

• List[Paper] –

  List[Paper]: List of found papers

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

@abstractmethod
def search(self, query: str, max_results: int = 100, **kwargs) -> List[Paper]:
    """
    Search the database for papers.

    Args:
                                    query: Search query (supports Boolean operations)
                                    max_results: Maximum number of results to return
                                    **kwargs (Any): Database-specific search options.

    Returns:
                                    List[Paper]: List of found papers
    """
    pass

get_database_name abstractmethod

get_database_name() -> str

Get the name of this database.

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

@abstractmethod
def get_database_name(self) -> str:
    """Get the name of this database."""
    pass

get_stats

get_stats() -> Dict[str, Any]

Get adapter statistics.

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

def get_stats(self) -> Dict[str, Any]:
    """Get adapter statistics."""
    return self.stats.copy()

reset_stats

reset_stats()

Reset adapter statistics.

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

def reset_stats(self):
    """Reset adapter statistics."""
    self.stats = {
        "total_requests": 0,
        "successful_requests": 0,
        "failed_requests": 0,
        "total_papers_found": 0,
        "last_request_time": None,
        "rate_limit_hits": 0,
        "availability_checks": 0,
        "last_availability_check": None,
    }

is_available

is_available() -> bool

Check if the database adapter is available and configured with caching.

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

def is_available(self) -> bool:
    """Check if the database adapter is available and configured with caching."""
    current_time = time.time()

    # Use cached result if still valid
    if (
        self._availability_cache is not None
        and current_time - self._availability_check_time
        < self._availability_cache_duration
    ):
        return self._availability_cache

    # Perform live check
    self._log_availability_check()
    try:
        self._availability_cache = self._check_live_availability()
    except Exception as e:
        self.logger.warning(f"Availability check failed: {e}")
        self._availability_cache = False

    self._availability_check_time = current_time
    return self._availability_cache

validate_query

validate_query(query: str) -> bool

Validate if the query is supported by this adapter.

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

def validate_query(self, query: str) -> bool:
    """Validate if the query is supported by this adapter."""
    return bool(query and query.strip())

preprocess_query

preprocess_query(query: str) -> str

Preprocess query for this specific database.

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

def preprocess_query(self, query: str) -> str:
    """Preprocess query for this specific database."""
    return query.strip()

ArxivAdapter

ArxivAdapter(rate_limit: float = 3.0)

Bases: DatabaseAdapter

Adapter for arXiv API with Boolean query support.

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

def __init__(self, rate_limit: float = 3.0):
    super().__init__(rate_limit=rate_limit)
    self.base_url = "http://export.arxiv.org/api/query"

search

search(query: str, max_results: int = 100, categories: List[str] = None, **kwargs) -> List[Paper]

Search arXiv with Boolean query support.

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

def search(
    self, query: str, max_results: int = 100, categories: List[str] = None, **kwargs
) -> List[Paper]:
    """Search arXiv with Boolean query support."""
    self._rate_limit_wait()

    try:
        # Parse Boolean query
        parsed_query = self.query_parser.parse_boolean_query(query)
        self.logger.info(f"Parsed query: {parsed_query}")

        arxiv_query = self.query_parser.to_arxiv_query(parsed_query)
        self.logger.info(f"ArXiv query: {arxiv_query}")

        # Add category filter if specified
        if categories:
            cat_filter = " OR ".join([f"cat:{cat}" for cat in categories])
            arxiv_query = f"({arxiv_query}) AND ({cat_filter})"
            self.logger.info(f"ArXiv query with categories: {arxiv_query}")

        params = {
            "search_query": arxiv_query,
            "start": 0,
            "max_results": max_results,
            "sortBy": "relevance",
            "sortOrder": "descending",
        }

        url = f"{self.base_url}?{urllib.parse.urlencode(params)}"
        self.logger.debug(f"ArXiv URL: {url}")

        with urllib.request.urlopen(url, timeout=30) as response:
            xml_data = response.read().decode("utf-8")

        papers = self._parse_arxiv_response(xml_data)
        self.logger.info(f"ArXiv returned {len(papers)} papers")
        self._log_request(success=True, papers_found=len(papers))
        return papers

    except Exception as e:
        self.logger.error(f"ArXiv search failed: {e}")
        self._log_request(success=False)
        return []

IEEEAdapter

IEEEAdapter(api_key: str, rate_limit: float = 100.0)

Bases: DatabaseAdapter

Adapter for IEEE Xplore API with Boolean query support.

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

def __init__(self, api_key: str, rate_limit: float = 100.0):
    super().__init__(rate_limit=rate_limit, api_key=api_key)
    self.base_url = "http://ieeexploreapi.ieee.org/api/v1/search/articles"

search

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

Search IEEE Xplore with Boolean query support.

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

def search(
    self, query: str, max_results: int = 100, year_min: int = None, **kwargs
) -> List[Paper]:
    """Search IEEE Xplore with Boolean query support."""
    if not self.is_available():
        self.logger.warning("IEEE API key not provided")
        return []

    self._rate_limit_wait()

    try:
        parsed_query = self.query_parser.parse_boolean_query(query)
        ieee_query = self.query_parser.to_ieee_query(parsed_query)

        params = {
            "apikey": self.api_key,
            "querytext": ieee_query,
            "max_records": min(max_results, 200),
            "start_record": 1,
            "sort_field": "publication_year",
            "sort_order": "desc",
        }

        if year_min:
            params["start_year"] = year_min

        url = f"{self.base_url}?{urllib.parse.urlencode(params)}"
        with urllib.request.urlopen(url, timeout=30) as response:
            data = json.loads(response.read().decode("utf-8"))

        papers = []
        for article in data.get("articles", []):
            paper = PaperFormatter.format_ieee_paper(article)
            papers.append(paper)

        self._log_request(success=True, papers_found=len(papers))
        return papers

    except Exception as e:
        self.logger.error(f"IEEE search failed: {e}")
        self._log_request(success=False)
        return []

ElsevierAdapter

ElsevierAdapter(api_key: str, rate_limit: float = 100.0)

Bases: DatabaseAdapter

Adapter for Elsevier/ScienceDirect API.

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

def __init__(self, api_key: str, rate_limit: float = 100.0):
    super().__init__(rate_limit=rate_limit, api_key=api_key)
    self.base_url = "https://api.elsevier.com/content/search/sciencedirect"

search

search(query: str, max_results: int = 100, **kwargs) -> List[Paper]

Search Elsevier ScienceDirect.

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

def search(self, query: str, max_results: int = 100, **kwargs) -> List[Paper]:
    """Search Elsevier ScienceDirect."""
    if not self.is_available():
        self.logger.warning("Elsevier API key not provided")
        return []

    self._rate_limit_wait()

    try:
        parsed_query = self.query_parser.parse_boolean_query(query)
        simple_query = self.query_parser.to_simple_query(parsed_query)

        headers = {"X-ELS-APIKey": self.api_key, "Accept": "application/json"}

        params = {"query": simple_query, "count": min(max_results, 100)}

        url = f"{self.base_url}?{urllib.parse.urlencode(params)}"
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=30) as response:
            data = json.loads(response.read().decode("utf-8"))

        papers = []
        for entry in data.get("search-results", {}).get("entry", []):
            paper = PaperFormatter.format_elsevier_paper(entry)
            papers.append(paper)

        self._log_request(success=True, papers_found=len(papers))
        return papers

    except Exception as e:
        self.logger.error(f"Elsevier search failed: {e}")
        self._log_request(success=False)
        return []

SpringerAdapter

SpringerAdapter(api_key: str, rate_limit: float = 100.0)

Bases: DatabaseAdapter

Adapter for Springer Nature API.

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

def __init__(self, api_key: str, rate_limit: float = 100.0):
    super().__init__(rate_limit=rate_limit, api_key=api_key)
    self.base_url = "http://api.springernature.com/meta/v1/json"

search

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

Search Springer Nature.

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

def search(
    self, query: str, max_results: int = 100, year_min: int = None, **kwargs
) -> List[Paper]:
    """Search Springer Nature."""
    if not self.is_available():
        self.logger.warning("Springer API key not provided")
        return []

    self._rate_limit_wait()

    try:
        parsed_query = self.query_parser.parse_boolean_query(query)
        simple_query = self.query_parser.to_simple_query(parsed_query)

        if year_min:
            simple_query += f" year:{year_min}-{datetime.now().year}"

        params = {
            "api_key": self.api_key,
            "q": simple_query,
            "s": min(max_results, 100),
            "p": 1,
        }

        url = f"{self.base_url}?{urllib.parse.urlencode(params)}"
        with urllib.request.urlopen(url, timeout=30) as response:
            data = json.loads(response.read().decode("utf-8"))

        papers = []
        for record in data.get("records", []):
            paper = PaperFormatter.format_springer_paper(record)
            papers.append(paper)

        self._log_request(success=True, papers_found=len(papers))
        return papers

    except Exception as e:
        self.logger.error(f"Springer search failed: {e}")
        self._log_request(success=False)
        return []

WileyAdapter

WileyAdapter(api_key: str, rate_limit: float = 100.0)

Bases: DatabaseAdapter

Adapter for Wiley Online Library API.

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

def __init__(self, api_key: str, rate_limit: float = 100.0):
    super().__init__(rate_limit=rate_limit, api_key=api_key)
    self.base_url = "https://api.wiley.com/onlinelibrary/tdm/v1/articles"

search

search(query: str, max_results: int = 100, **kwargs) -> List[Paper]

Search Wiley Online Library.

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

def search(self, query: str, max_results: int = 100, **kwargs) -> List[Paper]:
    """Search Wiley Online Library."""
    if not self.is_available():
        self.logger.warning("Wiley API key not provided")
        return []

    self._rate_limit_wait()

    try:
        parsed_query = self.query_parser.parse_boolean_query(query)
        simple_query = self.query_parser.to_simple_query(parsed_query)

        headers = {
            "Wiley-TDM-Client-Token": self.api_key,
            "Accept": "application/json",
        }

        params = {"query": simple_query, "count": min(max_results, 100)}

        url = f"{self.base_url}?{urllib.parse.urlencode(params)}"
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=30) as response:
            data = json.loads(response.read().decode("utf-8"))

        papers = []
        for item in data.get("items", []):
            paper = PaperFormatter.format_wiley_paper(item)
            papers.append(paper)

        self._log_request(success=True, papers_found=len(papers))
        return papers

    except Exception as e:
        self.logger.error(f"Wiley search failed: {e}")
        self._log_request(success=False)
        return []

ACMAdapter

ACMAdapter(rate_limit: float = 0.5)

Bases: DatabaseAdapter

Database adapter for the ACM Digital Library.

NOTE: This adapter uses web scraping as ACM does not provide a public search API. It is fragile and may break if ACM changes its website layout.

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

def __init__(self, rate_limit: float = 0.5):  # A gentler rate limit for scraping
    super().__init__(rate_limit=rate_limit)
    self.session = requests.Session()
    self.session.headers.update(
        {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
        }
    )

get_database_name

get_database_name() -> str

Get the name of this database.

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

def get_database_name(self) -> str:
    """Get the name of this database."""
    return "ACM Digital Library"

preprocess_query

preprocess_query(query: str) -> str

URL-encode the query for the search endpoint.

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

def preprocess_query(self, query: str) -> str:
    """URL-encode the query for the search endpoint."""
    return urllib.parse.quote_plus(query.strip())

search

search(query: str, max_results: int = 20, **kwargs) -> List[Paper]

Search the ACM Digital Library by scraping its search results page.

Parameters:

• query (str) –

  The search term.

• max_results (int, default: 20 ) –

  Maximum number of results to return (default is 20 per page).

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

  Unused.

Returns:

• List[Paper] –

  A list of Paper objects.

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

def search(self, query: str, max_results: int = 20, **kwargs) -> List[Paper]:
    """
    Search the ACM Digital Library by scraping its search results page.

    Args:
            query: The search term.
            max_results: Maximum number of results to return (default is 20 per page).
            **kwargs (Any): Unused.

    Returns:
            A list of Paper objects.
    """
    if not self.is_available():
        self.logger.warning("ACM Digital Library is not available")
        return []

    if not self.validate_query(query):
        self.logger.warning("Search query is empty or invalid.")
        return []

    processed_query = self.preprocess_query(query)
    # ACM uses 'AllField' for a general search
    search_url = urllib.parse.urljoin(
        self.BASE_URL, f"action/doSearch?AllField={processed_query}"
    )

    self._rate_limit_wait()
    papers = []
    try:
        self.logger.info(f"Searching ACM for: '{query}'")
        response = self.session.get(search_url, timeout=15)
        response.raise_for_status()  # Raise an exception for bad status codes

        soup = BeautifulSoup(response.text, "html.parser")

        # Find all list items that contain search results
        results = soup.select("div.issue-item__content")

        if not results:
            self.logger.info("No search results found on the page.")

        for item in results[:max_results]:
            try:
                title_tag = item.select_one("h5.issue-item__title a")
                title = title_tag.get_text(strip=True) if title_tag else "N/A"
                url = (
                    urllib.parse.urljoin(self.BASE_URL, title_tag["href"])
                    if title_tag and title_tag.has_attr("href")
                    else None
                )

                # Extract authors from the author list
                author_tags = item.select("ul[aria-label='authors'] li a")
                authors = [author.get_text(strip=True) for author in author_tags]

                # Extract DOI
                doi_tag = item.select_one("a.issue-item__doi")
                doi = doi_tag.get_text(strip=True) if doi_tag else None

                # Extract publication date
                date_tag = item.select_one("span.epub-section__date")
                pub_date = date_tag.get_text(strip=True) if date_tag else None

                # Extract year from publication date
                year = 0
                if pub_date:
                    year_match = re.search(r"\b(19|20)\d{2}\b", pub_date)
                    if year_match:
                        year = int(year_match.group())

                # Abstract/snippet is not reliably available on the search page
                abstract = "Abstract not available on search results page."

                paper = Paper(
                    title=title,
                    authors=authors,
                    doi=doi,
                    url=url,
                    abstract=abstract,
                    year=year,
                    database_source=self.get_database_name(),
                    publication_type="article",
                    venue="ACM Digital Library",
                    publisher="ACM",
                )
                papers.append(paper)
            except Exception as e:
                self.logger.error(f"Error parsing a paper item: {e}", exc_info=True)

        self._log_request(success=True, papers_found=len(papers))
        self.logger.info(f"Found {len(papers)} papers on ACM for query '{query}'.")

    except requests.RequestException as e:
        self.logger.error(f"Failed to fetch search results from ACM: {e}")
        self._log_request(success=False)

    return papers

GoogleScholarAdapter

GoogleScholarAdapter(rate_limit: float = 0.2)

Bases: DatabaseAdapter

Enhanced Google Scholar adapter with actual functionality.

Note: Google Scholar actively blocks automated requests and requires careful rate limiting and proper headers to work reliably.

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

def __init__(self, rate_limit: float = 0.2):  # Very conservative rate limit
    super().__init__(rate_limit=rate_limit)
    self.session = requests.Session()
    self.session.headers.update(
        {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.5",
            "Accept-Encoding": "gzip, deflate",
            "Connection": "keep-alive",
        }
    )

search

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

Search Google Scholar with careful rate limiting and error handling.

Parameters:

• query (str) –

  Search query

• max_results (int, default: 20 ) –

  Maximum number of results (limited to reduce blocking risk)

• year_min (int, default: None ) –

  Minimum publication year

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

  Extra search options.

Returns:

• List[Paper] –

  List[Paper]: Found papers (may be empty if blocked)

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

def search(
    self, query: str, max_results: int = 20, year_min: int = None, **kwargs
) -> List[Paper]:
    """
    Search Google Scholar with careful rate limiting and error handling.

    Args:
            query: Search query
            max_results: Maximum number of results (limited to reduce blocking risk)
            year_min: Minimum publication year
            **kwargs (Any): Extra search options.

    Returns:
            List[Paper]: Found papers (may be empty if blocked)
    """
    if not self.is_available():
        self.logger.warning("Google Scholar is not available")
        return []

    # Limit max results to reduce blocking risk
    max_results = min(max_results, 20)

    self._rate_limit_wait()

    try:
        # Parse and format query
        parsed_query = self.query_parser.parse_boolean_query(query)
        scholar_query = self.query_parser.to_google_scholar_query(parsed_query)

        # Add year filter if specified
        if year_min:
            scholar_query += f" after:{year_min}"

        # Build search URL
        params = {"q": scholar_query, "hl": "en", "num": max_results, "start": 0}

        search_url = f"{self.BASE_URL}scholar?" + urllib.parse.urlencode(params)
        self.logger.info(f"Searching Google Scholar: {search_url}")

        # Add random delay to appear more human-like
        import random

        time.sleep(random.uniform(1, 3))

        response = self.session.get(search_url, timeout=15)
        response.raise_for_status()

        # Check if we've been blocked
        if self._is_blocked(response.text):
            self.logger.warning("Google Scholar has blocked our request")
            self._log_request(success=False)
            return []

        soup = BeautifulSoup(response.text, "html.parser")

        # Find search results
        results = soup.select(
            "div.gs_r, div[data-lid]"
        )  # Different possible selectors

        papers = []
        for result in results[:max_results]:
            paper = self._parse_scholar_result(result)
            if paper:
                papers.append(paper)

        self.logger.info(f"Google Scholar returned {len(papers)} papers")
        self._log_request(success=True, papers_found=len(papers))

        return papers

    except Exception as e:
        self.logger.error(f"Google Scholar search failed: {e}")
        self._log_request(success=False)
        return []

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, {})