Metaphor Research Digest Slack App

metaphor-research-digest.py

"""
This script uses the Metaphor (https://metaphor.systems) API to fetch the latest research results for a query of choice.
Given the popularity of Slack as a communication channel, it also provides the option to forward this information
to Slack using a web hook URL.

Usage:
1. Set the environment variables METAPHOR_API_KEY and SLACK_URL
2. Create a python>=3.7 environment
3. Install dependencies: `pip install metaphor-python requests`
4. Run: `python metaphor-research-digest.py --days 7 --num-results 10 --publish-to-slack`
"""

from argparse import ArgumentParser, Namespace
from datetime import datetime, timedelta
from metaphor_python import Metaphor, SearchResponse
import os
import requests


METAPHOR_API_KEY = os.environ["METAPHOR_API_KEY"]
SLACK_URL = os.environ["SLACK_URL"]
DEFAULT_QUERY = "Neural audio codec, generative AI audio, neural audio synthesis"


def format_response(response: SearchResponse) -> list[str]:
    urls = []
    for result in response.results:
        author_last, author_first = result.author.split(";")[:2]
        urls.append(
            f"{result.title} by {author_first} {author_last} et al. | {result.url}"
        )
    return urls


def parse_args() -> Namespace:
    parser = ArgumentParser()
    parser.add_argument("--days", type=int, default=7)
    parser.add_argument("--num-results", type=int, default=2)
    parser.add_argument("--query", type=str, default=DEFAULT_QUERY)
    parser.add_argument("--publish-to-slack", action="store_true")
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()
    metaphor = Metaphor(METAPHOR_API_KEY)

    # start date should be one week ago formatted as YYYY-MM-DD
    start_date = (datetime.now() - timedelta(days=args.days)).strftime("%Y-%m-%d")
    end_date = datetime.now().strftime("%Y-%m-%d")

    # fetch all research papers published in the last week
    response = metaphor.search(
        args.query,
        num_results=args.num_results,
        include_domains=["scholar.google.com", "jstor.org", "arxiv.org", "nature.com"],
        start_crawl_date=start_date,
        end_crawl_date=end_date,
        start_published_date=start_date,
        end_published_date=end_date,
        use_autoprompt=True,
    )

    results = format_response(response)
    msg = (
        f"Here are {args.num_results} papers from the past {args.days} days on the topic: "
        f"'{args.query}'\n" + "\n".join(results)
    )

    # send message to slack
    if args.publish_to_slack:
        res = requests.post(
            SLACK_URL,
            json={"text": msg},
        )

        if res.status_code == 200:
            print(f"Success: {res.status_code}")
        else:
            print(f"Failed to send message: {res.status_code}")

    print(f"Message: {msg}")