Spaces:

librarian-bots
/

recommend_similar_papers

Running

App Files Files Community

recommend_similar_papers / app.py

davanstrien HF staff

chore: Refactor comment posting functionality and improve performance

38c818a 15 days ago

raw history blame contribute delete

No virus

11.5 kB

	import gradio as gr
	import requests
	from cachetools import cached, TTLCache
	from bs4 import BeautifulSoup
	from httpx import Client
	import json
	from pathlib import Path
	from huggingface_hub import CommitScheduler
	from dotenv import load_dotenv
	import os
	from functools import lru_cache
	from typing import Tuple

	load_dotenv()

	HF_TOKEN = os.getenv("HF_TOKEN")

	CACHE_TIME = 60 * 60 * 6 # 6 hours

	client = Client()

	REPO_ID = "librarian-bots/paper-recommendations-v2"

	scheduler = CommitScheduler(
	repo_id=REPO_ID,
	repo_type="dataset",
	folder_path="comments",
	path_in_repo="data",
	every=5,
	token=HF_TOKEN,
	)


	def parse_arxiv_id_from_paper_url(url):
	return url.split("/")[-1]


	@cached(cache=TTLCache(maxsize=500, ttl=CACHE_TIME))
	def get_recommendations_from_semantic_scholar(semantic_scholar_id: str):
	try:
	r = requests.post(
	"https://api.semanticscholar.org/recommendations/v1/papers/",
	json={
	"positivePaperIds": [semantic_scholar_id],
	},
	params={"fields": "externalIds,title,year", "limit": 10},
	)
	return r.json()["recommendedPapers"]
	except KeyError as e:
	raise gr.Error(
	"Error getting recommendations, if this is a new paper it may not yet have"
	" been indexed by Semantic Scholar."
	) from e


	def filter_recommendations(recommendations, max_paper_count=5):
	# include only arxiv papers
	arxiv_paper = [
	r for r in recommendations if r["externalIds"].get("ArXiv", None) is not None
	]
	if len(arxiv_paper) > max_paper_count:
	arxiv_paper = arxiv_paper[:max_paper_count]
	return arxiv_paper


	@cached(cache=TTLCache(maxsize=500, ttl=CACHE_TIME))
	def get_paper_title_from_arxiv_id(arxiv_id):
	try:
	return requests.get(f"https://huggingface.co/api/papers/{arxiv_id}").json()[
	"title"
	]
	except Exception as e:
	print(f"Error getting paper title for {arxiv_id}: {e}")
	raise gr.Error("Error getting paper title for {arxiv_id}: {e}") from e


	def format_recommendation_into_markdown(arxiv_id, recommendations):
	# title = get_paper_title_from_arxiv_id(arxiv_id)
	# url = f"https://huggingface.co/papers/{arxiv_id}"
	# comment = f"Recommended papers for [{title}]({url})\n\n"
	comment = "The following papers were recommended by the Semantic Scholar API \n\n"
	for r in recommendations:
	hub_paper_url = f"https://huggingface.co/papers/{r['externalIds']['ArXiv']}"
	comment += f"* [{r['title']}]({hub_paper_url}) ({r['year']})\n"
	return comment


	def format_comment(result: str):
	result = (
	"This is an automated message from the [Librarian Bot](https://huggingface.co/librarian-bots). I found the following papers similar to this paper. \n\n"
	+ result
	)
	result += "\n\n Please give a thumbs up to this comment if you found it helpful!"
	result += "\n\n If you want recommendations for any Paper on Hugging Face checkout [this](https://huggingface.co/spaces/librarian-bots/recommend_similar_papers) Space"
	result += "\n\n You can directly ask Librarian Bot for paper recommendations by tagging it in a comment: `@librarian-bot recommend`"
	return result


	def post_comment(
	paper_url: str, comment: str, comment_id: str \| None = None, token: str = HF_TOKEN
	) -> Tuple[bool, str]:
	"""
	Post a comment on a paper or a reply to a comment using the Hugging Face API.

	Args:
	paper_url (str): The URL of the paper to post the comment on.
	comment (str): The text of the comment or reply to post.
	comment_id (str, optional): The ID of the comment to reply to. If provided, the function will post a reply to the specified comment. Defaults to None.
	token (str, optional): The authentication token to use for the API request. Defaults to HF_TOKEN.

	Returns:
	Tuple[bool, str]: A tuple containing two elements:
	- bool: True if the comment or reply was posted successfully, False otherwise.
	- str: The ID of the posted comment or reply if successful, an empty string otherwise.

	Raises:
	requests.exceptions.RequestException: If an error occurs while making the API request.
	"""
	try:
	paper_id = paper_url.split("/")[-1]
	if comment_id:
	url = f"https://huggingface.co/api/papers/{paper_id}/comment/{comment_id}/reply"
	gr.Info(f"Replying to comment {comment_id}")
	print(f"Replying to comment {comment_id}")
	else:
	url = f"https://huggingface.co/api/papers/{paper_id}/comment"
	print(f"Posting comment for {paper_url}")
	gr.Info(f"Posting comment for {paper_url}")
	headers = {
	"Authorization": f"Bearer {token}",
	"Content-Type": "application/json",
	}
	comment_data = {"comment": comment}

	response = requests.post(url, json=comment_data, headers=headers)

	if response.status_code == 201:
	posted_comment_id = response.json().get("id", "")
	if comment_id:
	print(
	f"Reply posted successfully to comment {comment_id} for {paper_url}. Reply ID: {posted_comment_id}"
	)
	else:
	print(
	f"Comment posted successfully for {paper_url}. Comment ID: {posted_comment_id}"
	)
	return True, posted_comment_id
	else:
	print(
	f"Failed to post {'reply' if comment_id else 'comment'} for {paper_url}. Status code: {response.status_code}"
	)
	print(f"Response text: {response.text}")
	return False, ""

	except requests.exceptions.RequestException as e:
	print(
	f"Error posting {'reply' if comment_id else 'comment'} for {paper_url}: {e}"
	)
	return False, ""


	# @lru_cache(maxsize=500)
	# def is_comment_from_librarian_bot(html: str) -> bool:
	# """
	# Checks if the given HTML contains a comment from the librarian-bot.

	# Args:
	# html (str): The HTML content to check.

	# Returns:
	# bool: True if a comment from the librarian-bot is found, False otherwise.
	# """
	# soup = BeautifulSoup(html, "lxml")
	# librarian_bot_links = soup.find_all("a", string="librarian-bot")
	# return any(librarian_bot_links)


	def check_if_lib_bot_comment_exists(paper_url: str) -> Tuple[bool, str]:
	"""
	Check if a comment or reply from the librarian-bot exists for a given paper URL using the Hugging Face API.

	Args:
	paper_url (str): The URL of the paper to check for librarian-bot comments.

	Returns:
	Tuple[bool, str]: A tuple containing two elements:
	- bool: True if a comment or reply from the librarian-bot is found, False otherwise.
	- str: The ID of the comment if a librarian-bot comment is found, an empty string otherwise.

	Raises:
	Exception: If an error occurs while retrieving comments from the API.
	"""
	try:
	paper_id = paper_url.split("/")[-1]
	url = f"https://huggingface.co/api/papers/{paper_id}/?field=comments"
	headers = {"Authorization": f"Bearer {HF_TOKEN}"}
	response = requests.get(url, headers=headers)

	if response.status_code == 200:
	paper_data = response.json()
	comments = paper_data.get("comments", [])
	for comment in comments:
	comment_author = comment.get("author", {}).get("name")
	if comment_author == "librarian-bot":
	return True, comment.get("id")
	replies = comment.get("replies", [])
	for reply in replies:
	reply_author = reply.get("author", {}).get("name")
	if reply_author == "librarian-bot":
	return True, comment.get("id")
	else:
	print(
	f"Failed to retrieve comments for {paper_url}. Status code: {response.status_code}"
	)

	return False, ""
	except Exception as e:
	print(f"Error checking if comment exists for {paper_url}: {e}")
	return True, "" # default to not posting comment


	def log_comments(paper_url: str, comment: str):
	"""
	Logs comments for a given paper URL.

	Args:
	paper_url (str): The URL of the paper.
	comment (str): The comment to be logged.

	Returns:
	None
	"""
	paper_id = paper_url.split("/")[-1]
	file_path = Path(f"comments/{paper_id}.json")
	if not file_path.exists():
	with scheduler.lock:
	with open(file_path, "w") as f:
	data = {"paper_url": paper_url, "comment": comment}
	json.dump(data, f)


	def return_recommendations(
	url: str, comment_id: str \| None, post_to_paper: bool = True
	) -> str:
	arxiv_id = parse_arxiv_id_from_paper_url(url)
	recommendations = get_recommendations_from_semantic_scholar(f"ArXiv:{arxiv_id}")
	filtered_recommendations = filter_recommendations(recommendations)
	formatted_recommendation = format_recommendation_into_markdown(
	arxiv_id, filtered_recommendations
	) # Assign early

	if post_to_paper:
	comment = format_comment(formatted_recommendation)

	# Check if a librarian-bot comment already exists.
	existing_comments, existing_comment_id = check_if_lib_bot_comment_exists(url)
	if existing_comments:
	gr.Info(
	f"Librarian-bot already commented on this paper. Comment ID: {existing_comment_id}. No further action will be taken."
	)
	else:
	# If no existing librarian-bot comment, check if a specific comment_id is provided for replying.
	if comment_id:
	comment_status, posted_comment_id = post_comment(
	url, comment, comment_id, token=HF_TOKEN
	)
	if comment_status:
	log_comments(url, comment)
	gr.Info(f"Posted reply to comment {posted_comment_id}")
	else:
	# If no comment_id is provided, post a new comment.
	comment_status, posted_comment_id = post_comment(
	url, comment, token=HF_TOKEN
	)
	if comment_status:
	log_comments(url, comment)
	gr.Info(f"Posted new comment {posted_comment_id}")

	if not comment_status:
	gr.Info("Failed to post comment")

	return formatted_recommendation


	title = "Semantic Scholar Paper Recommender"
	description = (
	"Paste a link to a paper on Hugging Face Papers and get recommendations for similar"
	" papers from Semantic Scholar. Note: Some papers may not have recommendations"
	" yet if they are new or have not been indexed by Semantic Scholar."
	)
	examples = [
	["https://huggingface.co/papers/2309.12307", None, False],
	["https://huggingface.co/papers/2211.10086", None, False],
	]
	interface = gr.Interface(
	return_recommendations,
	[
	gr.Textbox(lines=1),
	gr.Textbox(None, lines=1, label="Comment ID (only for API)", visible=False),
	gr.Checkbox(False, label="Post recommendations to Paper page?"),
	],
	gr.Markdown(),
	examples=examples,
	title=title,
	description=description,
	)
	interface.queue()
	interface.launch()