本文目录导读：
Python + Gmail API（最推荐）
import os
import pickle
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']
def check_new_emails():
creds = None
if os.path.exists('token.pickle'):
with open('token.pickle', 'rb') as token:
creds = pickle.load(token)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES)
creds = flow.run_local_server(port=0)
with open('token.pickle', 'wb') as token:
pickle.dump(creds, token)
service = build('gmail', 'v1', credentials=creds)
# 检查未读邮件
results = service.users().messages().list(
userId='me',
q='is:unread',
maxResults=10
).execute()
messages = results.get('messages', [])
if messages:
print(f"你有 {len(messages)} 封新邮件")
for msg in messages[:5]: # 显示前5封
message = service.users().messages().get(
userId='me',
id=msg['id']
).execute()
headers = message['payload']['headers']
subject = next(h['value'] for h in headers if h['name'] == 'Subject')
from_addr = next(h['value'] for h in headers if h['name'] == 'From')
print(f"发件人: {from_addr}")
print(f"主题: {subject}")
print("-" * 50)
return len(messages)
if __name__ == "__main__":
new_count = check_new_emails()
Python + IMAP（更通用）
import imaplib
import email
from email.header import decode_header
import time
import os
def check_email_imap():
# 邮箱配置
IMAP_HOST = "imap.gmail.com" # 或你的邮箱IMAP服务器
IMAP_PORT = 993
USERNAME = "your_email@gmail.com"
PASSWORD = "your_password" # 建议使用应用专用密码
try:
# 连接IMAP服务器
mail = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT)
mail.login(USERNAME, PASSWORD)
mail.select("inbox")
# 搜索未读邮件
status, messages = mail.search(None, 'UNSEEN')
if status == "OK":
mail_ids = messages[0].split()
print(f"发现 {len(mail_ids)} 封新邮件")
if mail_ids:
for mail_id in mail_ids[-5:]: # 显示最后5封
status, msg_data = mail.fetch(mail_id, '(RFC822)')
if status == "OK":
msg = email.message_from_bytes(msg_data[0][1])
# 解码主题
subject = decode_header(msg["Subject"])[0]
if isinstance(subject[0], bytes):
subject = subject[0].decode(subject[1] or 'utf-8')
from_addr = msg["From"]
print(f"发件人: {from_addr}")
print(f"主题: {subject}")
print("-" * 50)
# 可选：标记为已读
# mail.store(mail_id, '+FLAGS', '\\Seen')
mail.logout()
return len(mail_ids)
except Exception as e:
print(f"检查邮件失败: {e}")
return 0
if __name__ == "__main__":
check_email_imap()
Bash 脚本 + curl（简单版）
#!/bin/bash
# 邮箱配置
EMAIL="your_email@gmail.com"
PASSWORD="your_app_password"
IMAP_SERVER="imap.gmail.com"
IMAP_PORT="993"
# 使用curl检查未读邮件（需要curl支持IMAP）
check_email() {
# 方法1：使用curl
curl --url "imaps://$IMAP_SERVER/INBOX" \
--user "$EMAIL:$PASSWORD" \
--request "UNSEEN" 2>/dev/null | head -20
# 方法2：使用openssl（更简单但功能有限）
# echo -e "a1 LOGIN $EMAIL $PASSWORD\na2 SELECT INBOX\na3 SEARCH UNSEEN\n" | \
# openssl s_client -connect $IMAP_SERVER:$IMAP_PORT 2>/dev/null
}
# 主循环，定时检查
while true; do
echo "[$(date)] 检查新邮件..."
check_email
sleep 300 # 每5分钟检查一次
done
Node.js + 第三方库
const Imap = require('imap');
const { simpleParser } = require('mailparser');
const imapConfig = {
user: 'your_email@gmail.com',
password: 'your_app_password',
host: 'imap.gmail.com',
port: 993,
tls: true
};
const imap = new Imap(imapConfig);
function checkNewEmails() {
imap.once('ready', () => {
imap.openBox('INBOX', false, (err, box) => {
if (err) throw err;
// 搜索未读邮件
imap.search(['UNSEEN'], (err, results) => {
if (err) throw err;
console.log(`发现 ${results.length} 封新邮件`);
if (results.length > 0) {
// 获取最新邮件
const latest = results.slice(-5);
const f = imap.fetch(latest, { bodies: '' });
f.on('message', (msg, seqno) => {
msg.on('body', (stream, info) => {
simpleParser(stream, (err, parsed) => {
console.log(`发件人: ${parsed.from.text}`);
console.log(`主题: ${parsed.subject}`);
console.log('---');
});
});
});
}
imap.end();
});
});
});
imap.on('error', (err) => {
console.log('连接错误:', err);
});
imap.connect();
}
checkNewEmails();
定时自动检查脚本
import time
import schedule
from datetime import datetime
def job():
print(f"[{datetime.now()}] 检查新邮件...")
# 调用上面的检查函数
check_email_imap() # 或 check_new_emails()
# 每天特定时间检查
schedule.every().day.at("09:00").do(job)
schedule.every().hour.do(job) # 每小时检查
# 持续运行
while True:
schedule.run_pending()
time.sleep(60)
使用建议：
-
安全性：
- 不要直接在脚本中写密码，使用环境变量
- Gmail等邮箱使用应用专用密码
- 考虑使用OAuth2.0认证
-
部署方式：
- 可以设置cron job或Windows任务计划程序
- 或使用守护进程方式持续运行
-
扩展功能：
- 添加新邮件通知（邮件、短信、钉钉等）
- 自动回复或分类
- 将附件保存到本地
你需要哪种邮件服务,我可以提供更具体的配置说明。
