自动化每日Arxiv纸摘要和松弛通知

<code class="python">import datetime
import logging
import os
import time

import arxiv
import google.generativeai as genai
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

# Configuration (best practice to use environment variables for sensitive data)
PAPER_TYPES = ["cs.ai", "cs.cy", "cs.ma"]
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")
GEMINI_MODEL = "gemini-2.0-flash"
SLACK_BOT_TOKEN = os.environ.get("SLACK_BOT_TOKEN")
SLACK_CHANNEL = os.environ.get("SLACK_CHANNEL")
MAX_RESULTS = 30

# Logging setup (highly recommended for debugging and monitoring)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


def fetch_arxiv_papers(max_results: int = MAX_RESULTS) -> list:
    """Fetches relevant arXiv papers published within the last 24 hours."""
    query = " OR ".join([f"cat:{paper_type}" for paper_type in PAPER_TYPES])
    client = arxiv.Client()
    search = client.search(query=query, max_results=max_results, sort_by=arxiv.SortCriterion.SubmittedDate, sort_order=arxiv.SortOrder.Descending)
    papers = list(client.results(search))

    if not papers:
        logger.warning("No papers found.")
        return []

    latest_published = papers[0].published
    threshold = latest_published - datetime.timedelta(hours=24)
    filtered_papers = [paper for paper in papers if paper.published >= threshold]

    return [
        {
            "title": paper.title,
            "summary": paper.summary,
            "pdf_url": paper.pdf_url,
            "published": paper.published,
        } for paper in filtered_papers
    ]


def summarize_paper(abstract_text: str) -> str:
    """Generates a summary of the paper abstract using Google Gemini."""
    try:
        genai.configure(api_key=GEMINI_API_KEY)
        model = genai.GenerativeModel(GEMINI_MODEL)
        prompt = (
            "Summarize the following paper abstract concisely (under 300 characters) for beginners, "
            "including significance and results.  Output only the summary.\n---\n\n"
            f"{abstract_text}"
        )
        response = model.generate_content(prompt)
        return response.text.strip()
    except Exception as e:
        logger.error(f"Error summarizing paper: {e}")
        return "Error generating summary."


def post_to_slack(papers: list) -> None:
    """Posts the paper summaries to the specified Slack channel."""
    if not papers:
        return

    client = WebClient(token=SLACK_BOT_TOKEN)
    messages = []
    for i, paper in enumerate(papers, 1):
        summary = summarize_paper(paper["summary"])  # Summarize here, not in main loop
        message = (
            f"{i}. *{paper['title']}*\n\n"
            f"{summary}\n\n"
            f"PDF: {paper['pdf_url']}\n"
            f"Published: {paper['published']}\n"
            f"────────────────────────"
        )
        messages.append(message)

    all_messages = "\n".join(messages)

    try:
        result = client.chat_postMessage(channel=SLACK_CHANNEL, text=all_messages)
        logger.info(f"Slack message sent successfully: {result}")
    except SlackApiError as e:
        logger.error(f"Error posting to Slack: {e}")


def lambda_handler(event, context):
    """AWS Lambda handler function."""
    papers = fetch_arxiv_papers()
    post_to_slack(papers)
    return {
        'statusCode': 200,
        'body': "Successfully processed arXiv papers and posted to Slack."
    }
</code>