Spaces:
Sleeping
Sleeping
| # Copyright 2023 by Jan Philip Wahle, https://jpwahle.com/ | |
| # All rights reserved. | |
| import asyncio | |
| import datetime | |
| import os | |
| from collections import Counter | |
| from concurrent.futures import ThreadPoolExecutor, as_completed | |
| from typing import List, Tuple | |
| import aiohttp | |
| import requests | |
| from aclanthology import ( | |
| async_match_acl_id_to_s2_paper, | |
| extract_author_info, | |
| extract_paper_info, | |
| extract_venue_info, | |
| ) | |
| from metrics import calculate_gini, calculate_gini_simpson | |
| from pdf import parse_pdf_to_artcile_dict | |
| def get_or_create_eventloop(): | |
| """ | |
| Get the current event loop or create a new one if there is no current event loop in the thread. | |
| Returns: | |
| The current event loop. | |
| """ | |
| try: | |
| return asyncio.get_event_loop() | |
| except RuntimeError as ex: | |
| if "There is no current event loop in thread" in str(ex): | |
| loop = asyncio.new_event_loop() | |
| asyncio.set_event_loop(loop) | |
| return asyncio.get_event_loop() | |
| def send_s2_request(request_url): | |
| """ | |
| Sends a GET request to the specified URL with the S2 API key in the headers. | |
| Args: | |
| request_url (str): The URL to send the request to. | |
| Returns: | |
| requests.Response: The response object returned by the request. | |
| """ | |
| return requests.get( | |
| request_url, | |
| headers={"x-api-key": os.environ["s2apikey"]}, | |
| timeout=10, | |
| ) | |
| def check_s2_id_type(semantic_scholar_id): | |
| """ | |
| Check whether a given Semantic Scholar ID is valid for a paper or an author. | |
| Args: | |
| semantic_scholar_id (str): The Semantic Scholar ID to check. | |
| Returns: | |
| tuple: A tuple containing the type of the ID ("paper" or "author") and | |
| the name of the author (if the ID is valid for an author), or "invalid" | |
| if the ID is not valid for either a paper or an author. | |
| """ | |
| # First, check if it's a paper ID | |
| paper_response = requests.get( | |
| f"https://api.semanticscholar.org/v1/paper/{semantic_scholar_id}", | |
| timeout=5, | |
| ) | |
| # If the response status code is 200, it means the ID is valid for a paper | |
| if paper_response.status_code == 200: | |
| return "paper", None | |
| # Next, check if it's an author ID | |
| author_response = requests.get( | |
| f"https://api.semanticscholar.org/v1/author/{semantic_scholar_id}", | |
| timeout=5, | |
| ) | |
| # If the response status code is 200, it means the ID is valid for an author | |
| return ( | |
| "author", | |
| author_response.json()["name"] | |
| if author_response.status_code == 200 | |
| else "invalid", | |
| ) | |
| def get_papers_from_author(ssid_author_id): | |
| """Retrieves all papers for a given author | |
| Args: | |
| ssid_author_id (str): semantic scholar id | |
| Returns: | |
| list: a list of all papers for the given author | |
| """ | |
| # Create request URL for an author | |
| request_url = f"https://api.semanticscholar.org/graph/v1/author/{ssid_author_id}?fields=papers" | |
| r = send_s2_request(request_url) | |
| if r.status_code == 200: | |
| papers = r.json().get("papers", []) | |
| return [paper["paperId"] for paper in papers] | |
| return [] | |
| def compute_stats_for_references(s2_ref_paper_keys, year): | |
| """ | |
| Computes various statistics for a list of reference paper keys. | |
| Args: | |
| s2_ref_paper_keys (list): A list of Semantic Scholar paper keys for the references. | |
| year (int): The year of the paper. | |
| Returns: | |
| tuple: A tuple containing the following statistics: | |
| - num_references (int): The number of references. | |
| - fields_of_study_counts (dict): A dictionary containing the count of each field of study. | |
| - year_to_title_dict (dict): A dictionary mapping the year of each reference to its title. | |
| - cfdi (float): The CFDI (Cumulative Field Diversity Index) of the references. | |
| - cadi (float): The CADI (Cumulative Age Diversity Index) of the references. | |
| - output_maoc (float): The MAOC (Mean Age of Citation) of the references. | |
| If there are no valid references, returns a tuple of None values. | |
| """ | |
| # Go over the references of the paper | |
| reference_year_list = [] | |
| reference_title_list = [] | |
| reference_fos_list = [] | |
| with ThreadPoolExecutor() as executor: | |
| request_url_refs = [ | |
| f"https://api.semanticscholar.org/graph/v1/paper/{ref_paper_key}?fields=title,year,s2FieldsOfStudy" | |
| for ref_paper_key in s2_ref_paper_keys | |
| ] | |
| futures = [ | |
| executor.submit(send_s2_request, request_url_ref) | |
| for request_url_ref in request_url_refs | |
| ] | |
| for future in as_completed(futures): | |
| r_ref = future.result() | |
| if r_ref.status_code == 200: | |
| result_ref = r_ref.json() | |
| (title_ref, year_ref, fields_ref) = ( | |
| result_ref["title"], | |
| result_ref["year"], | |
| result_ref["s2FieldsOfStudy"], | |
| ) | |
| reference_year_list.append(year_ref) | |
| reference_title_list.append(title_ref) | |
| reference_fos_list.extend( | |
| field["category"] | |
| for field in fields_ref | |
| if field["source"] == "s2-fos-model" | |
| ) | |
| else: | |
| print( | |
| f"Error retrieving reference {r_ref.status_code} for" | |
| f" paper {s2_ref_paper_keys}" | |
| ) | |
| # Remove all None from reference_year_list and reference_title_list | |
| reference_year_list = [ | |
| year_ref for year_ref in reference_year_list if year_ref is not None | |
| ] | |
| reference_title_list = [ | |
| title_ref | |
| for title_ref in reference_title_list | |
| if title_ref is not None | |
| ] | |
| # Count references | |
| num_references = len(reference_year_list) | |
| # Flatten list and count occurrences | |
| fields_of_study_counts = dict( | |
| Counter( | |
| [ | |
| field | |
| for field in reference_fos_list | |
| ] | |
| ) | |
| ) | |
| # Citation age list | |
| aoc_list = [ | |
| year - year_ref | |
| for year_ref in reference_year_list | |
| if year_ref and year | |
| ] | |
| if not aoc_list: | |
| return None, None, None, None, None, None | |
| # Compute citation age | |
| output_maoc = sum(aoc_list) / len(aoc_list) | |
| cadi = calculate_gini(aoc_list) | |
| # Create a dictionary of year to title | |
| year_to_title_dict = dict(zip(reference_year_list, reference_title_list)) | |
| # Compute CFDI | |
| cfdi = calculate_gini_simpson(fields_of_study_counts) | |
| # Return the results | |
| return ( | |
| num_references, | |
| fields_of_study_counts, | |
| year_to_title_dict, | |
| cfdi, | |
| cadi, | |
| output_maoc, | |
| ) | |
| def compute_stats_for_s2_paper(ssid_paper_id): | |
| """ | |
| Computes statistics for a given paper ID using the Semantic Scholar API. | |
| Args: | |
| ssid_paper_id (str): The Semantic Scholar ID of the paper to compute statistics for. | |
| Returns: | |
| Tuple containing the following statistics: | |
| - title_authors (str): The title and authors of the paper. | |
| - num_references (int): The number of references in the paper. | |
| - fields_of_study_counts (dict): A dictionary containing the count of each field of study in the paper's references. | |
| - year_to_title_dict (dict): A dictionary mapping the year of each reference to its title. | |
| - cfdi (float): The CFDI (Cumulative Field Diversity Index) of the paper's references. | |
| - cadi (float): The CADI (Citation Age Diversity Index) of the paper's references. | |
| - output_maoc (float): The MAOC (Mean Age of Citation) of the paper's references. | |
| """ | |
| # Get the paper and its references | |
| request_url = f"https://api.semanticscholar.org/graph/v1/paper/{ssid_paper_id}?fields=references,title,year,authors" | |
| r = send_s2_request(request_url) | |
| if r.status_code == 200: # if successful request | |
| result = r.json() | |
| if not result.get("references") or result.get("references") == []: | |
| return None, None, None, None, None, None, None, None | |
| s2_ref_paper_keys = [ | |
| reference_paper_tuple["paperId"] | |
| for reference_paper_tuple in r.json()["references"] | |
| ] | |
| filtered_s2_ref_paper_keys = [ | |
| s2_ref_paper_key | |
| for s2_ref_paper_key in s2_ref_paper_keys | |
| if s2_ref_paper_key is not None | |
| ] | |
| title, year, authors = ( | |
| result["title"], | |
| result["year"], | |
| result["authors"], | |
| ) | |
| title_authors = ( | |
| title + "\n" + ", ".join([author["name"] for author in authors]) | |
| ) | |
| ( | |
| num_references, | |
| fields_of_study_counts, | |
| year_to_title_dict, | |
| cfdi, | |
| cadi, | |
| output_maoc, | |
| ) = compute_stats_for_references(filtered_s2_ref_paper_keys, year) | |
| # Return the results | |
| return ( | |
| title_authors, | |
| num_references, | |
| fields_of_study_counts, | |
| year_to_title_dict, | |
| cfdi, | |
| cadi, | |
| output_maoc, | |
| ) | |
| def compute_stats_for_s2_author(ssid_author_id, author_name): | |
| """ | |
| Computes statistics for an author based on their papers in the Semantic Scholar database. | |
| Args: | |
| ssid_author_id (str): The Semantic Scholar author ID. | |
| author_name (str): The name of the author. | |
| Returns: | |
| dict: A dictionary containing statistics for the author, or None if no papers were found. | |
| """ | |
| if papers := get_papers_from_author(ssid_author_id): | |
| return compute_stats_for_multiple_s2_papers(papers, author_name) | |
| return None | |
| def compute_stats_for_acl_paper(url): | |
| """ | |
| Computes statistics for a paper based on its ACL Anthology URL. | |
| Args: | |
| url (str): The URL of the paper on the ACL Anthology website. | |
| Returns: | |
| dict: A dictionary containing statistics for the paper, or None if the paper was not found. | |
| """ | |
| if paper_info := extract_paper_info(url): | |
| loop = get_or_create_eventloop() | |
| # Match paper ID to Semantic Scholar ID | |
| s2_paper = loop.run_until_complete( | |
| async_match_acl_id_to_s2_paper(paper_info["acl_id"]) | |
| ) | |
| return compute_stats_for_s2_paper(s2_paper["paperId"]) | |
| return None | |
| def compute_stats_for_acl_author(url): | |
| """ | |
| Computes statistics for an author's papers in the ACL anthology. | |
| Args: | |
| url (str): The URL of the author's page on the ACL anthology website. | |
| Returns: | |
| dict: A dictionary containing statistics for the author's papers, including | |
| the number of papers, the number of citations, and the h-index. | |
| Returns None if the author's page cannot be accessed or no papers are found. | |
| """ | |
| if paper_info := extract_author_info(url): | |
| loop = get_or_create_eventloop() | |
| tasks = [ | |
| async_match_acl_id_to_s2_paper(paper["url"].split("/")[-2]) | |
| for paper in paper_info["papers"] | |
| ] | |
| papers = loop.run_until_complete(asyncio.gather(*tasks)) | |
| return compute_stats_for_multiple_s2_papers( | |
| [paper["paperId"] for paper in papers if "paperId" in paper], | |
| paper_info["author"], | |
| ) | |
| return None | |
| def compute_stats_for_acl_venue(url): | |
| """ | |
| Computes statistics for papers in a given ACL venue. | |
| Args: | |
| url (str): The URL of the ACL venue. | |
| Returns: | |
| dict: A dictionary containing statistics for the papers in the venue. | |
| """ | |
| if paper_info := extract_venue_info(url): | |
| loop = get_or_create_eventloop() | |
| tasks = [ | |
| async_match_acl_id_to_s2_paper(paper["url"].split("/")[-2]) | |
| for paper in paper_info["papers"] | |
| ] | |
| papers = loop.run_until_complete(asyncio.gather(*tasks)) | |
| return compute_stats_for_multiple_s2_papers( | |
| [paper["paperId"] for paper in papers if "paperId" in paper], | |
| paper_info["venue"], | |
| ) | |
| return None | |
| def compute_stats_for_multiple_s2_papers( | |
| papers: List[dict], title: str | |
| ) -> Tuple[str, int, dict, dict, float, float, float]: | |
| """ | |
| Computes statistics for multiple S2 papers. | |
| Args: | |
| papers (List[dict]): A list of S2 papers. | |
| title (str): The title of the papers. | |
| Returns: | |
| A tuple containing the following statistics: | |
| - title (str): The title of the papers. | |
| - num_references (int): The total number of references in all papers. | |
| - top_fields (dict): A dictionary containing the top fields and their counts. | |
| - oldest_paper_dict (dict): A dictionary containing the oldest paper for each year. | |
| - cfdi (float): The average CFDI score for all papers. | |
| - cadi (float): The average CADI score for all papers. | |
| - output_maoc (float): The average output MAOC score for all papers. | |
| """ | |
| num_references = 0 | |
| top_fields = {} | |
| oldest_paper_dict = {} | |
| cfdi = 0 | |
| cadi = 0 | |
| output_maoc = 0 | |
| def process_paper(paper): | |
| return compute_stats_for_s2_paper(paper) | |
| with ThreadPoolExecutor() as executor: | |
| results_list = list(executor.map(process_paper, papers)) | |
| for results in results_list: | |
| if not results or results[0] is None: | |
| continue | |
| num_references += results[1] | |
| for field, count in results[2].items(): | |
| top_fields[field] = top_fields.get(field, 0) + count | |
| for year, ref_title in results[3].items(): | |
| oldest_paper_dict[year] = ref_title | |
| cfdi += results[4] | |
| cadi += results[5] | |
| output_maoc += results[6] | |
| return ( | |
| title, | |
| num_references, | |
| top_fields, | |
| oldest_paper_dict, | |
| cfdi / len(papers), | |
| cadi / len(papers), | |
| output_maoc / len(papers), | |
| ) | |
| async def send_s2_async_request(url): | |
| """ | |
| Sends an asynchronous request to the specified URL and returns the response as a JSON object. | |
| Args: | |
| url (str): The URL to send the request to. | |
| Returns: | |
| dict: The response from the URL as a JSON object. | |
| """ | |
| async with aiohttp.ClientSession() as session: | |
| async with session.get(url) as response: | |
| return await response.json() | |
| async def match_title_to_s2_paper(title, authors=None): | |
| """ | |
| Matches a given paper title (and authors) to Semantic Scholar to retrieve its S2 paper ID. | |
| Args: | |
| title (str): The title of the paper. | |
| authors (List[str], optional): List of authors of the paper. Defaults to None. | |
| Returns: | |
| str or None: Returns the S2 paper ID if found, otherwise None. | |
| """ | |
| # Send a request to the Semantic Scholar API to search for the paper by its title | |
| search_url = ( | |
| f"http://api.semanticscholar.org/graph/v1/paper/search?query={title}" | |
| ) | |
| # Send request | |
| response = await send_s2_async_request(search_url) | |
| results = response.get("data", []) | |
| if len(results) > 0: | |
| result = results[0] # Ranked by relevance | |
| return result.get("paperId") | |
| async def compute_stats_for_pdf(pdf_file): | |
| """ | |
| Computes statistics for a given PDF file. | |
| Args: | |
| pdf_file (file): The PDF file to compute statistics for. | |
| Returns: | |
| tuple: A tuple containing the title of the article and the computed statistics. | |
| """ | |
| s2_paper_ids = [] | |
| article_dict = parse_pdf_to_artcile_dict(pdf_file.name) | |
| references = article_dict["references"] | |
| # Get S2 paper IDs asynchronously | |
| tasks = [ | |
| match_title_to_s2_paper(reference["title"], reference["authors"]) | |
| for reference in references | |
| if reference["title"] | |
| ] | |
| s2_paper_ids = await asyncio.gather(*tasks) | |
| # Remove all None values from s2paperids | |
| s2_paper_ids = [s2_id for s2_id in s2_paper_ids if s2_id is not None] | |
| # Compute the current year | |
| today = datetime.date.today() | |
| year = int(today.strftime("%Y")) | |
| results = compute_stats_for_references(s2_paper_ids, year) | |
| results = (article_dict["title"],) + results | |
| return results | |