全自动 arXiv 论文监控 + Gemini 多模态解读 + 邮箱推送机器人
在科研日常中,每天追跟踪最新的前沿文献(例如图像超分辨率 Image Super-Resolution)是一件极其耗费精力的事。
为了解决“文献读不完、找不到重点”的痛点,我用 Python 写了一个自动化论文监控与 AI 解读工具。它不仅能定时抓取 arXiv 最新文献,还能将 PDF 首页转为图像,调用 Google Gemini 2.0 多模态模型 进行“图文联合分析”,最后将带有星级评级和核心创新的解读报告精准发送到你的邮箱。
🌟 核心功能亮点
-
“追新 + 补旧”推送机制:
-
优先推送当天最新的论文。
-
若当天无更新,系统会自动从未读历史缓存库中,筛选出星级评分最高的优质老论文进行补推送。
-
-
多模态图文联合解读:
-
使用
PyMuPDF(fitz) 自动提取 PDF 第一页(通常包含核心 Model 架构图/实验对比图)。 -
将图片与文本摘要一并输入 Gemini 2.0 Flash 展开深度剖析。
-
-
稳健的自动降级机制(Fallback):
-
若网络问题或 API 限流导致 AI 分析失败,系统会自动降级提取论文基础原文,保证推送不中断。
-
-
轻量化本地缓存:
-
利用 JSON 与 TXT 实现轻量去重与已读历史追踪,零数据库依赖。
-
🛠️ 环境依赖与配置
1. 基础依赖库
运行脚本前,请确保安装了以下依赖:
Bash
pip install pymupdf pillow requests google-genai
2. 环境变量配置
为了保证密钥安全,脚本采用环境变量读取配置。你可以在本地系统环境变量或 GitHub Actions 中进行如下设置:
| 环境变量名 | 说明 | 示例 |
MAIL_USER | 发件人 QQ 邮箱 | your_qq@qq.com |
MAIL_PASS | 邮箱 SMTP 授权码(非登录密码) | abcdefghijklmnop |
RECEIVER | 主接收邮箱 | target1@qq.com |
RECEIVER_2 | 备用接收邮箱(可选) | target2@gmail.com |
GEMINI_API_KEY | 谷歌双子座 API 密钥 | AIzaSy... |
💻 完整源码实现
源码简洁高效,开箱即用:
Python
import os, time, json, requests, smtplib, fitz
import xml.etree.ElementTree as ET
from datetime import datetime
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
from email.utils import formataddr
from google import genai
from PIL import Image
import io
# ================= 配置信息 =================
SMTP_SERVER = "smtp.qq.com"
MAIL_USER = os.environ.get("MAIL_USER")
MAIL_PASS = os.environ.get("MAIL_PASS")
RECEIVERS = [os.environ.get("RECEIVER"), os.environ.get("RECEIVER_2")]
RECEIVERS = [r for r in RECEIVERS if r]
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")
HISTORY_FILE = "seen_papers.txt"
CACHE_FILE = "paper_cache.json"
TITLES_LOG = "titles_history.txt"
KEYWORD = 'all:"image super-resolution"'
if GEMINI_API_KEY:
client = genai.Client(api_key=GEMINI_API_KEY)
# ===========================================
def log(msg):
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}")
def download_pdf_first_page_as_image(pdf_url):
"""渲染 PDF 首页为 PIL Image 格式供多模态模型读取"""
try:
headers = {'User-Agent': 'Mozilla/5.0'}
response = requests.get(pdf_url, headers=headers, timeout=20)
if response.status_code == 200:
doc = fitz.open(stream=response.content, filetype="pdf")
page = doc.load_page(0)
pix = page.get_pixmap(matrix=fitz.Matrix(1.5, 1.5))
return Image.open(io.BytesIO(pix.tobytes("png")))
except Exception as e:
log(f"⚠️ 视觉信息提取跳过: {e}")
return None
def ai_evaluate_and_analyze(title, abstract, pdf_url=None):
"""使用 Gemini 分析论文,若失败则降级返回基础信息"""
prompt = f"""你是一个顶尖CV专家。请深度分析这篇视觉论文并给出:
1.【星级评分】: 必须包含 'X星' 格式(1-5星)。
2.【核心创新点】: 总结3点。
3.【架构解读】: 简要描述Model流程。
论文标题: {title}
摘要内容: {abstract}"""
try:
if not GEMINI_API_KEY: raise Exception("未配置 API KEY")
contents = [prompt]
if pdf_url:
img = download_pdf_first_page_as_image(pdf_url)
if img: contents.append(img)
# 调用 Google GenAI SDK (Gemini 2.0 Flash)
response = client.models.generate_content(
model='gemini-2.0-flash',
contents=contents
)
if not response.text: raise Exception("AI 返回内容为空")
res = response.text.strip()
score = 1
for i in range(5, 0, -1):
if f"{i}星" in res:
score = i
break
return score, res
except Exception as e:
log(f"❌ AI分析不可用: {e}")
# 【降级方案】: 当 AI 失效时,手动拼接中文概览
fallback_res = f"""⚠️ 【AI分析暂时不可用,自动生成中文概览】
【论文标题】: {title}
【英文摘要】: {abstract[:500]}...
【提示】: 系统当前无法连接至AI模型,以上为论文原文信息,请点击下方链接查看PDF详情。"""
return 1, fallback_res
def send_email(subject, body):
"""SMTP 邮件发送服务"""
if not MAIL_USER or not MAIL_PASS: return
for rec in RECEIVERS:
try:
msg = MIMEMultipart()
msg['From'] = formataddr((Header("AI科研看板", 'utf-8').encode(), MAIL_USER))
msg['To'] = formataddr((Header("科研人", 'utf-8').encode(), rec))
msg['Subject'] = Header(subject, 'utf-8')
msg.attach(MIMEText(body, 'plain', 'utf-8'))
server = smtplib.SMTP_SSL(SMTP_SERVER, 465)
server.login(MAIL_USER, MAIL_PASS)
server.sendmail(MAIL_USER, [rec], msg.as_string())
server.quit()
log(f"✅ 已推送至: {rec}")
except Exception as e:
log(f"❌ 邮件发送失败: {e}")
def monitor_arxiv():
log("🚀 启动“追新补旧”监控程序...")
sent_ids = set()
if os.path.exists(HISTORY_FILE):
with open(HISTORY_FILE, 'r') as f: sent_ids = {l.strip() for l in f}
cache = {}
if os.path.exists(CACHE_FILE):
try:
with open(CACHE_FILE, 'r', encoding='utf-8') as f: cache = json.load(f)
except: pass
# 查询 arXiv API
api_url = f'http://export.arxiv.org/api/query?search_query={KEYWORD}&max_results=30&sortBy=submittedDate&sortOrder=descending'
try:
res = requests.get(api_url, timeout=30)
root = ET.fromstring(res.content)
except Exception as e:
log(f"❌ 获取 ArXiv 失败: {e}")
return
ns = {'atom': 'http://www.w3.org/2005/Atom'}
entries = root.findall('atom:entry', ns)
today_str = datetime.now().strftime("%Y-%m-%d")
new_paper_today = None
for entry in entries:
pid = entry.find('atom:id', ns).text.split('/abs/')[-1]
pub_date = entry.find('atom:published', ns).text[:10]
title = entry.find('atom:title', ns).text.strip().replace('\n', ' ')
with open(TITLES_LOG, 'a', encoding='utf-8') as f: f.write(title + " ")
if pid not in sent_ids and pid not in cache:
log(f"🔍 正在处理: {title[:40]}...")
summary = entry.find('atom:summary', ns).text.strip().replace('\n', ' ')
pdf_link = entry.find('atom:link[@title="pdf"]', ns).attrib['href']
score, analysis = ai_evaluate_and_analyze(title, summary, pdf_link)
cache[pid] = {
"title": title, "analysis": analysis, "score": score,
"link": pdf_link, "date": pub_date
}
with open(CACHE_FILE, 'w', encoding='utf-8') as f:
json.dump(cache, f, ensure_ascii=False, indent=4)
if pid not in sent_ids and pub_date == today_str and not new_paper_today:
new_paper_today = pid
# 决策推送对象:优先当天新论文,无新论文则推送历史高分论文
target_id = None
if new_paper_today:
target_id = new_paper_today
else:
pending = {k: v for k, v in cache.items() if k not in sent_ids}
if pending:
sorted_pending = sorted(pending.items(), key=lambda x: (x[1]['score'], x[1]['date']), reverse=True)
target_id = sorted_pending[0][0]
if target_id:
p = cache[target_id]
stars = "⭐" * p['score']
subject = f"{stars} {p['title'][:60]}"
body = f"🔹 标题: {p['title']}\n📅 日期: {p['date']}\n\n{p['analysis']}\n\n🔗 链接: {p['link']}"
send_email(subject, body)
with open(HISTORY_FILE, 'a') as f: f.write(target_id + "\n")
if __name__ == "__main__":
monitor_arxiv()
🔍 核心代码解析
-
PDF 转图像 (PyMuPDF):
download_pdf_first_page_as_image函数通过fitz.open将下载的二进制 PDF 文件加载到内存,提取第 0 页并将其渲染为 PNG 图像。这使得 Gemini 能直接识别第一页的框架图。 -
Gemini 2.0 SDK 调用:
使用最新的
google-genai客户端,将文本 Prompt 与 PIL Image 组合放入contents列表中,一次性完成多模态推演。 -
“追新补旧”排序逻辑:
通过
sorted(pending.items(), key=lambda x: (x[1]['score'], x[1]['date']), reverse=True),对未推送的缓存论文按照“AI 评分 > 发布日期”进行组合排序,确保每次推送都是当前最高质量的论文。
🚀 进阶部署建议(GitHub Actions 定时运行)
配合 GitHub Actions,可以实现完全免费且无需买服务器的定时自动推送。
在项目根目录下创建 .github/workflows/main.yml:
YAML
name: ArXiv Bot Service
on:
schedule:
- cron: '0 0 * * *' # 每天 UTC 0 点(北京时间 8 点)自动运行
workflow_dispatch: # 支持手动触发
jobs:
run-bot:
runs-on: ubuntu-latest
steps:
- name: Checkout Code
uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.10'
- name: Install Dependencies
run: |
pip install pymupdf pillow requests google-genai
- name: Run Monitor
env:
MAIL_USER: ${{ secrets.MAIL_USER }}
MAIL_PASS: ${{ secrets.MAIL_PASS }}
RECEIVER: ${{ secrets.RECEIVER }}
RECEIVER_2: ${{ secrets.RECEIVER_2 }}
GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
run: python 1.py
- name: Commit History
run: |
git config --local user.email "github-actions[bot]@users.noreply.github.com"
git config --local user.name "github-actions[bot]"
git add seen_papers.txt paper_cache.json titles_history.txt
git commit -m "Auto-update paper cache & history" || exit 0
git push
修改检索关键字 KEYWORD 为你关注的方向(如 all:"diffusion model"),即可搭建属于你自己的 AI 科研助理!
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
所有评论(0)