|
|
|
import os |
|
from typing import Optional, Type |
|
|
|
from asyncer import asyncify |
|
|
|
from lagent.actions.base_action import AsyncActionMixin, BaseAction, tool_api |
|
from lagent.schema import ActionReturn, ActionStatusCode |
|
from .parser import BaseParser, JsonParser |
|
|
|
|
|
class GoogleScholar(BaseAction): |
|
"""Plugin for google scholar search. |
|
|
|
Args: |
|
api_key (str): API KEY to use serper google search API, |
|
You can create a free API key at https://serper.dev. |
|
description (dict): The description of the action. Defaults to ``None``. |
|
parser (Type[BaseParser]): The parser class to process the |
|
action's inputs and outputs. Defaults to :class:`JsonParser`. |
|
""" |
|
|
|
def __init__( |
|
self, |
|
api_key: Optional[str] = None, |
|
description: Optional[dict] = None, |
|
parser: Type[BaseParser] = JsonParser, |
|
): |
|
super().__init__(description, parser) |
|
api_key = os.environ.get('SERPER_API_KEY', api_key) |
|
if api_key is None: |
|
raise ValueError( |
|
'Please set Serper API key either in the environment ' |
|
'as SERPER_API_KEY or pass it as `api_key` parameter.' |
|
) |
|
self.api_key = api_key |
|
|
|
@tool_api(explode_return=True) |
|
def search_google_scholar( |
|
self, |
|
query: str, |
|
cites: Optional[str] = None, |
|
as_ylo: Optional[int] = None, |
|
as_yhi: Optional[int] = None, |
|
scisbd: Optional[int] = None, |
|
cluster: Optional[str] = None, |
|
hl: Optional[str] = None, |
|
lr: Optional[str] = None, |
|
start: Optional[int] = None, |
|
num: Optional[int] = None, |
|
as_sdt: Optional[str] = None, |
|
safe: Optional[str] = None, |
|
filter: Optional[str] = None, |
|
as_vis: Optional[str] = None, |
|
) -> dict: |
|
"""Search for scholarly articles based on a query according to the google scholar. |
|
|
|
Args: |
|
query (str): The query to search for. |
|
cites (Optional[str]): The unique ID of an article for triggering "Cited By" searches. |
|
as_ylo (Optional[int]): The starting year for results (e.g., if as_ylo=2018, results before this year will be omitted). |
|
as_yhi (Optional[int]): The ending year for results (e.g., if as_yhi=2018, results after this year will be omitted). |
|
scisbd (Optional[int]): Defines articles added in the last year, sorted by date. It can be set to 1 to include only abstracts, or 2 to include everything. |
|
cluster (Optional[str]): The unique ID of an article for triggering "All Versions" searches. |
|
hl (Optional[str]): The language to use for the Google Scholar search. |
|
lr (Optional[str]): One or multiple languages to limit the search to. |
|
start (Optional[int]): The result offset for pagination (0 is the first page of results, 10 is the 2nd page, etc.) |
|
num (Optional[int]): The maximum number of results to return, limited to 20. |
|
as_sdt (Optional[str]): Can be used either as a search type or a filter. |
|
safe (Optional[str]): The level of filtering for adult content. |
|
filter (Optional[str]): Defines if the filters for 'Similar Results' and 'Omitted Results' are on or off. |
|
as_vis (Optional[str]): Defines whether to include citations or not. |
|
|
|
Returns: |
|
:class:`dict`: article information |
|
- title: a list of the titles of the three selected papers |
|
- cited_by: a list of the citation numbers of the three selected papers |
|
- organic_id: a list of the organic results' ids of the three selected papers |
|
- pub_info: publication information of selected papers |
|
""" |
|
from serpapi import GoogleSearch |
|
|
|
params = { |
|
'q': query, |
|
'engine': 'google_scholar', |
|
'api_key': self.api_key, |
|
'cites': cites, |
|
'as_ylo': as_ylo, |
|
'as_yhi': as_yhi, |
|
'scisbd': scisbd, |
|
'cluster': cluster, |
|
'hl': hl, |
|
'lr': lr, |
|
'start': start, |
|
'num': num, |
|
'as_sdt': as_sdt, |
|
'safe': safe, |
|
'filter': filter, |
|
'as_vis': as_vis, |
|
} |
|
search = GoogleSearch(params) |
|
try: |
|
r = search.get_dict() |
|
results = r['organic_results'] |
|
title = [] |
|
snippets = [] |
|
cited_by = [] |
|
organic_id = [] |
|
pub_info = [] |
|
for item in results[:3]: |
|
title.append(item['title']) |
|
pub_info.append(item['publication_info']['summary']) |
|
citation = item['inline_links'].get('cited_by', {'total': ''}) |
|
cited_by.append(citation['total']) |
|
snippets.append(item['snippet']) |
|
organic_id.append(item['result_id']) |
|
return dict(title=title, cited_by=cited_by, organic_id=organic_id, snippets=snippets) |
|
except Exception as e: |
|
return ActionReturn(errmsg=str(e), state=ActionStatusCode.HTTP_ERROR) |
|
|
|
@tool_api(explode_return=True) |
|
def get_author_information( |
|
self, |
|
author_id: str, |
|
hl: Optional[str] = None, |
|
view_op: Optional[str] = None, |
|
sort: Optional[str] = None, |
|
citation_id: Optional[str] = None, |
|
start: Optional[int] = None, |
|
num: Optional[int] = None, |
|
no_cache: Optional[bool] = None, |
|
async_req: Optional[bool] = None, |
|
output: Optional[str] = None, |
|
) -> dict: |
|
"""Search for an author's information by author's id provided by get_author_id. |
|
|
|
Args: |
|
author_id (str): Required. The ID of an author. |
|
hl (Optional[str]): The language to use for the Google Scholar Author search. Default is 'en'. |
|
view_op (Optional[str]): Used for viewing specific parts of a page. |
|
sort (Optional[str]): Used for sorting and refining articles. |
|
citation_id (Optional[str]): Used for retrieving individual article citation. |
|
start (Optional[int]): Defines the result offset. Default is 0. |
|
num (Optional[int]): Defines the number of results to return. Default is 20. |
|
no_cache (Optional[bool]): Forces SerpApi to fetch the results even if a cached version is already present. Default is False. |
|
async_req (Optional[bool]): Defines the way you want to submit your search to SerpApi. Default is False. |
|
output (Optional[str]): Defines the final output you want. Default is 'json'. |
|
|
|
Returns: |
|
:class:`dict`: author information |
|
* name: author's name |
|
* affliation: the affliation of the author |
|
* articles: at most 3 articles by the author |
|
* website: the author's homepage url |
|
""" |
|
from serpapi import GoogleSearch |
|
|
|
params = { |
|
'engine': 'google_scholar_author', |
|
'author_id': author_id, |
|
'api_key': self.api_key, |
|
'hl': hl, |
|
'view_op': view_op, |
|
'sort': sort, |
|
'citation_id': citation_id, |
|
'start': start, |
|
'num': num, |
|
'no_cache': no_cache, |
|
'async': async_req, |
|
'output': output, |
|
} |
|
try: |
|
search = GoogleSearch(params) |
|
results = search.get_dict() |
|
author = results['author'] |
|
articles = results.get('articles', []) |
|
return dict( |
|
name=author['name'], |
|
affiliations=author.get('affiliations', ''), |
|
website=author.get('website', ''), |
|
articles=[dict(title=article['title'], authors=article['authors']) for article in articles[:3]], |
|
) |
|
except Exception as e: |
|
return ActionReturn(errmsg=str(e), state=ActionStatusCode.HTTP_ERROR) |
|
|
|
@tool_api(explode_return=True) |
|
def get_citation_format( |
|
self, |
|
q: str, |
|
no_cache: Optional[bool] = None, |
|
async_: Optional[bool] = None, |
|
output: Optional[str] = 'json', |
|
) -> dict: |
|
"""Function to get MLA citation format by an identification of organic_result's id provided by search_google_scholar. |
|
|
|
Args: |
|
q (str): ID of an individual Google Scholar organic search result. |
|
no_cache (Optional[bool]): If set to True, will force SerpApi to fetch the Google Scholar Cite results even if a cached version is already present. Defaults to None. |
|
async_ (Optional[bool]): If set to True, will submit search to SerpApi and retrieve results later. Defaults to None. |
|
output (Optional[str]): Final output format. Set to 'json' to get a structured JSON of the results, or 'html' to get the raw html retrieved. Defaults to 'json'. |
|
|
|
Returns: |
|
:class:`dict`: citation format |
|
* authors: the authors of the article |
|
* citation: the citation format of the article |
|
""" |
|
from serpapi import GoogleSearch |
|
|
|
params = { |
|
'q': q, |
|
'engine': 'google_scholar_cite', |
|
'api_key': self.api_key, |
|
'no_cache': no_cache, |
|
'async': async_, |
|
'output': output, |
|
} |
|
try: |
|
search = GoogleSearch(params) |
|
results = search.get_dict() |
|
citation = results['citations'] |
|
citation_info = citation[0]['snippet'] |
|
return citation_info |
|
except Exception as e: |
|
return ActionReturn(errmsg=str(e), state=ActionStatusCode.HTTP_ERROR) |
|
|
|
@tool_api(explode_return=True) |
|
def get_author_id( |
|
self, |
|
mauthors: str, |
|
hl: Optional[str] = 'en', |
|
after_author: Optional[str] = None, |
|
before_author: Optional[str] = None, |
|
no_cache: Optional[bool] = False, |
|
_async: Optional[bool] = False, |
|
output: Optional[str] = 'json', |
|
) -> dict: |
|
"""The getAuthorId function is used to get the author's id by his or her name. |
|
|
|
Args: |
|
mauthors (str): Defines the author you want to search for. |
|
hl (Optional[str]): Defines the language to use for the Google Scholar Profiles search. It's a two-letter language code. (e.g., 'en' for English, 'es' for Spanish, or 'fr' for French). Defaults to 'en'. |
|
after_author (Optional[str]): Defines the next page token. It is used for retrieving the next page results. The parameter has the precedence over before_author parameter. Defaults to None. |
|
before_author (Optional[str]): Defines the previous page token. It is used for retrieving the previous page results. Defaults to None. |
|
no_cache (Optional[bool]): Will force SerpApi to fetch the Google Scholar Profiles results even if a cached version is already present. Defaults to False. |
|
_async (Optional[bool]): Defines the way you want to submit your search to SerpApi. Defaults to False. |
|
output (Optional[str]): Defines the final output you want. It can be set to 'json' (default) to get a structured JSON of the results, or 'html' to get the raw html retrieved. Defaults to 'json'. |
|
|
|
Returns: |
|
:class:`dict`: author id |
|
* author_id: the author_id of the author |
|
""" |
|
from serpapi import GoogleSearch |
|
|
|
params = { |
|
'mauthors': mauthors, |
|
'engine': 'google_scholar_profiles', |
|
'api_key': self.api_key, |
|
'hl': hl, |
|
'after_author': after_author, |
|
'before_author': before_author, |
|
'no_cache': no_cache, |
|
'async': _async, |
|
'output': output, |
|
} |
|
try: |
|
search = GoogleSearch(params) |
|
results = search.get_dict() |
|
profile = results['profiles'] |
|
author_info = dict(author_id=profile[0]['author_id']) |
|
return author_info |
|
except Exception as e: |
|
return ActionReturn(errmsg=str(e), state=ActionStatusCode.HTTP_ERROR) |
|
|
|
|
|
class AsyncGoogleScholar(AsyncActionMixin, GoogleScholar): |
|
"""Plugin for google scholar search. |
|
|
|
Args: |
|
api_key (str): API KEY to use serper google search API, |
|
You can create a free API key at https://serper.dev. |
|
description (dict): The description of the action. Defaults to ``None``. |
|
parser (Type[BaseParser]): The parser class to process the |
|
action's inputs and outputs. Defaults to :class:`JsonParser`. |
|
""" |
|
|
|
@tool_api(explode_return=True) |
|
@asyncify |
|
def search_google_scholar( |
|
self, |
|
query: str, |
|
cites: Optional[str] = None, |
|
as_ylo: Optional[int] = None, |
|
as_yhi: Optional[int] = None, |
|
scisbd: Optional[int] = None, |
|
cluster: Optional[str] = None, |
|
hl: Optional[str] = None, |
|
lr: Optional[str] = None, |
|
start: Optional[int] = None, |
|
num: Optional[int] = None, |
|
as_sdt: Optional[str] = None, |
|
safe: Optional[str] = None, |
|
filter: Optional[str] = None, |
|
as_vis: Optional[str] = None, |
|
) -> dict: |
|
"""Search for scholarly articles based on a query according to the google scholar. |
|
|
|
Args: |
|
query (str): The query to search for. |
|
cites (Optional[str]): The unique ID of an article for triggering "Cited By" searches. |
|
as_ylo (Optional[int]): The starting year for results (e.g., if as_ylo=2018, results before this year will be omitted). |
|
as_yhi (Optional[int]): The ending year for results (e.g., if as_yhi=2018, results after this year will be omitted). |
|
scisbd (Optional[int]): Defines articles added in the last year, sorted by date. It can be set to 1 to include only abstracts, or 2 to include everything. |
|
cluster (Optional[str]): The unique ID of an article for triggering "All Versions" searches. |
|
hl (Optional[str]): The language to use for the Google Scholar search. |
|
lr (Optional[str]): One or multiple languages to limit the search to. |
|
start (Optional[int]): The result offset for pagination (0 is the first page of results, 10 is the 2nd page, etc.) |
|
num (Optional[int]): The maximum number of results to return, limited to 20. |
|
as_sdt (Optional[str]): Can be used either as a search type or a filter. |
|
safe (Optional[str]): The level of filtering for adult content. |
|
filter (Optional[str]): Defines if the filters for 'Similar Results' and 'Omitted Results' are on or off. |
|
as_vis (Optional[str]): Defines whether to include citations or not. |
|
|
|
Returns: |
|
:class:`dict`: article information |
|
- title: a list of the titles of the three selected papers |
|
- cited_by: a list of the citation numbers of the three selected papers |
|
- organic_id: a list of the organic results' ids of the three selected papers |
|
- pub_info: publication information of selected papers |
|
""" |
|
return super().search_google_scholar( |
|
query, |
|
cites, |
|
as_ylo, |
|
as_yhi, |
|
scisbd, |
|
cluster, |
|
hl, |
|
lr, |
|
start, |
|
num, |
|
as_sdt, |
|
safe, |
|
filter, |
|
as_vis, |
|
) |
|
|
|
@tool_api(explode_return=True) |
|
@asyncify |
|
def get_author_information( |
|
self, |
|
author_id: str, |
|
hl: Optional[str] = None, |
|
view_op: Optional[str] = None, |
|
sort: Optional[str] = None, |
|
citation_id: Optional[str] = None, |
|
start: Optional[int] = None, |
|
num: Optional[int] = None, |
|
no_cache: Optional[bool] = None, |
|
async_req: Optional[bool] = None, |
|
output: Optional[str] = None, |
|
) -> dict: |
|
"""Search for an author's information by author's id provided by get_author_id. |
|
|
|
Args: |
|
author_id (str): Required. The ID of an author. |
|
hl (Optional[str]): The language to use for the Google Scholar Author search. Default is 'en'. |
|
view_op (Optional[str]): Used for viewing specific parts of a page. |
|
sort (Optional[str]): Used for sorting and refining articles. |
|
citation_id (Optional[str]): Used for retrieving individual article citation. |
|
start (Optional[int]): Defines the result offset. Default is 0. |
|
num (Optional[int]): Defines the number of results to return. Default is 20. |
|
no_cache (Optional[bool]): Forces SerpApi to fetch the results even if a cached version is already present. Default is False. |
|
async_req (Optional[bool]): Defines the way you want to submit your search to SerpApi. Default is False. |
|
output (Optional[str]): Defines the final output you want. Default is 'json'. |
|
|
|
Returns: |
|
:class:`dict`: author information |
|
* name: author's name |
|
* affliation: the affliation of the author |
|
* articles: at most 3 articles by the author |
|
* website: the author's homepage url |
|
""" |
|
return super().get_author_information( |
|
author_id, hl, view_op, sort, citation_id, start, num, no_cache, async_req, output |
|
) |
|
|
|
@tool_api(explode_return=True) |
|
@asyncify |
|
def get_citation_format( |
|
self, |
|
q: str, |
|
no_cache: Optional[bool] = None, |
|
async_: Optional[bool] = None, |
|
output: Optional[str] = 'json', |
|
) -> dict: |
|
"""Function to get MLA citation format by an identification of organic_result's id provided by search_google_scholar. |
|
|
|
Args: |
|
q (str): ID of an individual Google Scholar organic search result. |
|
no_cache (Optional[bool]): If set to True, will force SerpApi to fetch the Google Scholar Cite results even if a cached version is already present. Defaults to None. |
|
async_ (Optional[bool]): If set to True, will submit search to SerpApi and retrieve results later. Defaults to None. |
|
output (Optional[str]): Final output format. Set to 'json' to get a structured JSON of the results, or 'html' to get the raw html retrieved. Defaults to 'json'. |
|
|
|
Returns: |
|
:class:`dict`: citation format |
|
* authors: the authors of the article |
|
* citation: the citation format of the article |
|
""" |
|
return super().get_citation_format(q, no_cache, async_, output) |
|
|
|
@tool_api(explode_return=True) |
|
@asyncify |
|
def get_author_id( |
|
self, |
|
mauthors: str, |
|
hl: Optional[str] = 'en', |
|
after_author: Optional[str] = None, |
|
before_author: Optional[str] = None, |
|
no_cache: Optional[bool] = False, |
|
_async: Optional[bool] = False, |
|
output: Optional[str] = 'json', |
|
) -> dict: |
|
"""The getAuthorId function is used to get the author's id by his or her name. |
|
|
|
Args: |
|
mauthors (str): Defines the author you want to search for. |
|
hl (Optional[str]): Defines the language to use for the Google Scholar Profiles search. It's a two-letter language code. (e.g., 'en' for English, 'es' for Spanish, or 'fr' for French). Defaults to 'en'. |
|
after_author (Optional[str]): Defines the next page token. It is used for retrieving the next page results. The parameter has the precedence over before_author parameter. Defaults to None. |
|
before_author (Optional[str]): Defines the previous page token. It is used for retrieving the previous page results. Defaults to None. |
|
no_cache (Optional[bool]): Will force SerpApi to fetch the Google Scholar Profiles results even if a cached version is already present. Defaults to False. |
|
_async (Optional[bool]): Defines the way you want to submit your search to SerpApi. Defaults to False. |
|
output (Optional[str]): Defines the final output you want. It can be set to 'json' (default) to get a structured JSON of the results, or 'html' to get the raw html retrieved. Defaults to 'json'. |
|
|
|
Returns: |
|
:class:`dict`: author id |
|
* author_id: the author_id of the author |
|
""" |
|
return super().get_author_id(mauthors, hl, after_author, before_author, no_cache, _async, output) |
|
|
