当前位置：首页 > news >正文

大模型应用：一个基于AI大模型的自动邮件简报系统 - Flask + HTML 方案

news 2025/9/29 6:50:20

大模型应用：一个基于AI大模型的自动邮件简报系统 - Flask + HTML 方案

项目概述

这是一个简化的邮件自动收取和总结系统，使用 Flask 作为后端，简单的 HTML 页面作为前端，无需 Docker，部署简单。
在这里插入图片描述

项目结构

email-digest-simple/
├── app.py
├── config.py
├── email_fetcher.py
├── glm_client.py
├── digest_generator.py
├── models.py
├── templates/
│   ├── base.html
│   ├── index.html
│   └── digest.html
├── static/
│   └── style.css
├── data/
│   └── emails.db
├── .env
├── requirements.txt
└── README.md

核心代码实现

1. 配置文件

config.py

import os
from pathlib import PathBASE_DIR = Path(__file__).resolve().parentclass Config:# 邮箱配置EMAIL_HOST = os.getenv('EMAIL_HOST', 'imap.gmail.com')EMAIL_PORT = int(os.getenv('EMAIL_PORT', '993'))EMAIL_ADDRESS = os.getenv('EMAIL_ADDRESS')EMAIL_PASSWORD = os.getenv('EMAIL_PASSWORD')# GLM配置GLM_API_KEY = os.getenv('GLM_API_KEY')GLM_MODEL = os.getenv('GLM_MODEL', 'glm-4')GLM_BASE_URL = os.getenv('GLM_BASE_URL', 'https://open.bigmodel.cn/api/paas/v4')# 应用配置SECRET_KEY = os.getenv('SECRET_KEY', 'your-secret-key-change-in-production')DEBUG = os.getenv('DEBUG', 'False').lower() == 'true'# 调度配置CHECK_INTERVAL_MINUTES = int(os.getenv('CHECK_INTERVAL_MINUTES', '30'))MAX_EMAILS_PER_RUN = int(os.getenv('MAX_EMAILS_PER_RUN', '20'))# 数据库DATABASE_PATH = BASE_DIR / 'data' / 'emails.db'DATABASE_PATH.parent.mkdir(exist_ok=True)

2. 数据模型

models.py

import sqlite3
from datetime import datetime
import json
from config import Configclass Database:def __init__(self):self.db_path = Config.DATABASE_PATHself.init_database()def get_connection(self):return sqlite3.connect(self.db_path)def init_database(self):conn = self.get_connection()cursor = conn.cursor()# 创建邮件表cursor.execute('''CREATE TABLE IF NOT EXISTS emails (id INTEGER PRIMARY KEY AUTOINCREMENT,email_id TEXT UNIQUE,subject TEXT,sender TEXT,recipients TEXT,date TEXT,body TEXT,body_html TEXT,summary TEXT,processed BOOLEAN DEFAULT 0,created_at TEXT DEFAULT CURRENT_TIMESTAMP,updated_at TEXT DEFAULT CURRENT_TIMESTAMP)''')# 创建简报表cursor.execute('''CREATE TABLE IF NOT EXISTS digests (id INTEGER PRIMARY KEY AUTOINCREMENT,date TEXT,content TEXT,email_count INTEGER,created_at TEXT DEFAULT CURRENT_TIMESTAMP)''')conn.commit()conn.close()def save_email(self, email_data):conn = self.get_connection()cursor = conn.cursor()cursor.execute('''INSERT OR REPLACE INTO emails (email_id, subject, sender, recipients, date, body, body_html, summary, processed, updated_at)VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (email_data['email_id'],email_data['subject'],email_data['sender'],json.dumps(email_data['recipients']),email_data['date'].isoformat(),email_data['body'],email_data['body_html'],email_data['summary'],email_data['processed'],datetime.now().isoformat()))conn.commit()conn.close()def get_processed_email_ids(self):conn = self.get_connection()cursor = conn.cursor()cursor.execute('SELECT email_id FROM emails WHERE processed = 1')results = cursor.fetchall()conn.close()return {row[0] for row in results}def get_latest_digest(self):conn = self.get_connection()cursor = conn.cursor()cursor.execute('''SELECT id, date, content, email_count, created_at FROM digests ORDER BY date DESC LIMIT 1''')result = cursor.fetchone()conn.close()if result:return {'id': result[0],'date': result[1],'content': json.loads(result[2]),'email_count': result[3],'created_at': result[4]}return Nonedef save_digest(self, digest_data):conn = self.get_connection()cursor = conn.cursor()cursor.execute('''INSERT INTO digests (date, content, email_count)VALUES (?, ?, ?)''', (digest_data['date'].isoformat(),json.dumps(digest_data['content'], ensure_ascii=False),digest_data['email_count']))conn.commit()conn.close()def get_all_emails(self, limit=50):conn = self.get_connection()cursor = conn.cursor()cursor.execute('''SELECT id, subject, sender, date, summary, processedFROM emailsORDER BY date DESCLIMIT ?''', (limit,))results = cursor.fetchall()conn.close()emails = []for row in results:emails.append({'id': row[0],'subject': row[1],'sender': row[2],'date': row[3],'summary': row[4],'processed': bool(row[5])})return emails

3. 邮件获取器

email_fetcher.py

import imaplib
import email
from email.header import decode_header
from datetime import datetime, timedelta
import loggingfrom config import Config
from models import Databaselogger = logging.getLogger(__name__)class EmailFetcher:def __init__(self):self.db = Database()def connect(self):"""连接到IMAP服务器"""try:self.imap_server = imaplib.IMAP4_SSL(Config.EMAIL_HOST, Config.EMAIL_PORT)self.imap_server.login(Config.EMAIL_ADDRESS, Config.EMAIL_PASSWORD)logger.info(f"成功连接到邮箱: {Config.EMAIL_ADDRESS}")return Trueexcept Exception as e:logger.error(f"连接邮箱失败: {e}")return Falsedef disconnect(self):"""断开连接"""if hasattr(self, 'imap_server'):try:self.imap_server.logout()except:passdef _decode_mime_words(self, s):"""解码MIME编码的字符串"""if not s:return ""decoded_fragments = decode_header(s)fragments = []for fragment, encoding in decoded_fragments:if isinstance(fragment, bytes):if encoding:fragment = fragment.decode(encoding)else:fragment = fragment.decode('utf-8', errors='ignore')fragments.append(fragment)return ''.join(fragments)def _get_email_body(self, msg):"""提取邮件正文"""body = ""body_html = ""if msg.is_multipart():for part in msg.walk():content_type = part.get_content_type()content_disposition = str(part.get("Content-Disposition"))if "attachment" not in content_disposition:charset = part.get_content_charset() or 'utf-8'try:payload = part.get_payload(decode=True)if payload:text = payload.decode(charset, errors='ignore')if content_type == "text/plain":body += textelif content_type == "text/html":body_html += textexcept Exception as e:logger.warning(f"解析邮件正文失败: {e}")else:charset = msg.get_content_charset() or 'utf-8'try:payload = msg.get_payload(decode=True)if payload:body = payload.decode(charset, errors='ignore')except Exception as e:logger.warning(f"解析邮件正文失败: {e}")return body, body_htmldef fetch_new_emails(self, since_days=1):"""获取新的未处理邮件"""if not self.connect():return []try:self.imap_server.select('INBOX')# 计算日期范围since_date = (datetime.now() - timedelta(days=since_days)).strftime("%d-%b-%Y")search_criteria = f'(SINCE "{since_date}")'status, messages = self.imap_server.search(None, search_criteria)if status != 'OK':logger.error("搜索邮件失败")return []email_ids = messages[0].split()logger.info(f"找到 {len(email_ids)} 封邮件")# 限制处理数量email_ids = email_ids[-Config.MAX_EMAILS_PER_RUN:] if email_ids else []# 获取已处理的邮件IDprocessed_ids = self.db.get_processed_email_ids()new_emails = []for email_id in email_ids:email_id_str = email_id.decode('utf-8')if email_id_str in processed_ids:continuetry:status, msg_data = self.imap_server.fetch(email_id, '(RFC822)')if status != 'OK':continuemsg = email.message_from_bytes(msg_data[0][1])subject = self._decode_mime_words(msg.get("Subject", ""))sender = self._decode_mime_words(msg.get("From", ""))recipients = [self._decode_mime_words(r) for r in msg.get_all("To", [])]date_str = msg.get("Date", "")try:email_date = email.utils.parsedate_to_datetime(date_str)except:email_date = datetime.now()body, body_html = self._get_email_body(msg)email_data = {'email_id': email_id_str,'subject': subject[:200],'sender': sender[:100],'recipients': recipients[:5],'date': email_date,'body': body[:10000],'body_html': body_html[:10000] if body_html else None,'summary': None,'processed': False}new_emails.append(email_data)except Exception as e:logger.error(f"解析邮件 {email_id_str} 失败: {e}")continuereturn new_emailsexcept Exception as e:logger.error(f"获取邮件失败: {e}")return []finally:self.disconnect()

4. GLM客户端

glm_client.py

import requests
import json
import loggingfrom config import Configlogger = logging.getLogger(__name__)class GLMClient:def __init__(self):self.api_key = Config.GLM_API_KEYself.base_url = Config.GLM_BASE_URLself.model = Config.GLM_MODELdef summarize_email(self, email_content, subject=""):"""使用GLM模型总结邮件内容"""if not self.api_key:logger.error("GLM API key 未配置")return Noneheaders = {"Authorization": f"Bearer {self.api_key}","Content-Type": "application/json"}# 构建提示词prompt = f"""请为以下邮件生成简洁的中文摘要，突出重点信息：邮件主题：{subject}邮件内容：{email_content}要求：1. 摘要控制在100字以内2. 突出关键信息和行动项3. 语言简洁明了"""payload = {"model": self.model,"messages": [{"role": "user", "content": prompt}],"temperature": 0.3,"max_tokens": 150}try:response = requests.post(f"{self.base_url}/chat/completions",headers=headers,json=payload,timeout=30)if response.status_code == 200:result = response.json()summary = result['choices'][0]['message']['content'].strip()return summaryelse:error_text = response.textlogger.error(f"GLM API 调用失败: {response.status_code} - {error_text}")return Noneexcept Exception as e:logger.error(f"GLM API 调用异常: {e}")return Nonedef batch_summarize(self, emails):"""批量总结邮件"""for email_data in emails:if email_data['body'].strip():summary = self.summarize_email(email_data['body'], email_data['subject'])email_data['summary'] = summary or "总结失败"email_data['processed'] = Trueelse:email_data['summary'] = "邮件内容为空"email_data['processed'] = Truereturn emails

5. 简报生成器

digest_generator.py

from datetime import datetime
import jsonclass DigestGenerator:def generate_digest_content(self, emails):"""生成简报内容"""digest_data = {"generated_at": datetime.now().isoformat(),"email_count": len(emails),"emails": []}for email in emails:digest_data["emails"].append({"subject": email['subject'],"sender": email['sender'],"date": email['date'].isoformat(),"summary": email['summary']})return digest_datadef create_digest(self, emails):"""创建简报记录"""content = self.generate_digest_content(emails)digest = {'date': datetime.now(),'content': content,'email_count': len(emails)}return digest

6. 主应用

app.py

import os
import threading
import time
from datetime import datetime, timedelta
import logging
from flask import Flask, render_template, request, jsonify, redirect, url_for
from apscheduler.schedulers.background import BackgroundSchedulerfrom config import Config
from email_fetcher import EmailFetcher
from glm_client import GLMClient
from digest_generator import DigestGenerator
from models import Database# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)# 加载环境变量
from dotenv import load_dotenv
load_dotenv()app = Flask(__name__)
app.config.from_object(Config)# 初始化组件
db = Database()
scheduler = BackgroundScheduler()
email_fetcher = EmailFetcher()
glm_client = GLMClient()
digest_generator = DigestGenerator()def process_emails():"""处理邮件并生成简报"""logger = logging.getLogger(__name__)logger.info("开始处理邮件...")# 获取新邮件new_emails = email_fetcher.fetch_new_emails(since_days=1)if not new_emails:logger.info("没有新邮件需要处理")returnlogger.info(f"发现 {len(new_emails)} 封新邮件，开始生成摘要...")# 生成摘要summarized_emails = glm_client.batch_summarize(new_emails)# 保存到数据库for email_data in summarized_emails:db.save_email(email_data)logger.info(f"成功保存 {len(summarized_emails)} 封邮件")# 生成简报digest = digest_generator.create_digest(summarized_emails)db.save_digest(digest)logger.info(f"简报生成完成")@app.route('/')
def index():"""首页 - 显示最新简报"""latest_digest = db.get_latest_digest()return render_template('index.html', digest=latest_digest)@app.route('/emails')
def emails():"""邮件列表页面"""all_emails = db.get_all_emails(limit=100)return render_template('digest.html', emails=all_emails)@app.route('/trigger', methods=['POST'])
def trigger_processing():"""手动触发邮件处理"""try:process_emails()return jsonify({'success': True, 'message': '邮件处理完成'})except Exception as e:return jsonify({'success': False, 'message': f'处理失败: {str(e)}'})@app.route('/health')
def health():"""健康检查"""return jsonify({'status': 'healthy','scheduler_running': scheduler.running if scheduler else False,'current_time': datetime.now().isoformat()})def start_scheduler():"""启动定时任务"""if not scheduler.running:scheduler.add_job(func=process_emails,trigger="interval",minutes=Config.CHECK_INTERVAL_MINUTES,id='email_processing')scheduler.start()logging.info(f"定时任务已启动，每 {Config.CHECK_INTERVAL_MINUTES} 分钟检查一次")if __name__ == '__main__':# 启动定时任务start_scheduler()try:app.run(host='0.0.0.0',port=5000,debug=Config.DEBUG)except KeyboardInterrupt:logging.info("收到中断信号，正在停止...")finally:if scheduler.running:scheduler.shutdown()

7. HTML 模板

templates/base.html

<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>{% block title %}邮件简报系统{% endblock %}</title><link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body><div class="container"><header><h1>📧 邮件简报系统</h1><nav><a href="{{ url_for('index') }}" class="{% if request.endpoint == 'index' %}active{% endif %}">最新简报</a><a href="{{ url_for('emails') }}" class="{% if request.endpoint == 'emails' %}active{% endif %}">邮件列表</a></nav></header><main>{% block content %}{% endblock %}</main><footer><p>&copy; 2024 邮件简报系统</p></footer></div><script>// 手动触发处理function triggerProcessing() {const button = document.getElementById('trigger-btn');button.disabled = true;button.textContent = '处理中...';fetch('/trigger', {method: 'POST',headers: {'Content-Type': 'application/json',}}).then(response => response.json()).then(data => {if (data.success) {alert('邮件处理完成！');location.reload();} else {alert('处理失败: ' + data.message);button.disabled = false;button.textContent = '🔄 手动处理邮件';}}).catch(error => {alert('请求失败: ' + error.message);button.disabled = false;button.textContent = '🔄 手动处理邮件';});}</script>
</body>
</html>

templates/index.html

{% extends "base.html" %}{% block content %}
<div class="dashboard"><div class="controls"><button id="trigger-btn" onclick="triggerProcessing()" class="btn btn-primary">🔄 手动处理邮件</button><button onclick="location.reload()" class="btn btn-secondary">📥 刷新数据</button></div>{% if digest %}<div class="digest-card"><div class="digest-header"><h2>📅 {{ digest.date[:10] }}</h2><p class="email-count">📧 {{ digest.email_count }} 封邮件</p></div><div class="email-list">{% for email in digest.content.emails %}<div class="email-item"><div class="email-header"><h3>{{ email.subject }}</h3><span class="sender">{{ email.sender }}</span></div><div class="email-summary"><p>{{ email.summary or '暂无摘要' }}</p></div><div class="email-time">{{ email.date[11:16] }}</div></div>{% endfor %}</div></div>{% else %}<div class="no-data"><h3>暂无邮件简报</h3><p>系统会自动处理新邮件，或点击上方按钮手动处理</p></div>{% endif %}<div class="system-info"><h4>系统信息</h4><div class="info-grid"><div class="info-item"><strong>检查间隔:</strong> {{ config.CHECK_INTERVAL_MINUTES }} 分钟</div><div class="info-item"><strong>最大处理数:</strong> {{ config.MAX_EMAILS_PER_RUN }} 封/次</div><div class="info-item"><strong>邮箱地址:</strong> {{ config.EMAIL_ADDRESS }}</div></div></div>
</div>
{% endblock %}

templates/digest.html

{% extends "base.html" %}{% block content %}
<div class="emails-page"><h2>邮件列表 (共 {{ emails|length }} 封)</h2><div class="email-list">{% for email in emails %}<div class="email-item {% if email.processed %}processed{% endif %}"><div class="email-header"><h3>{{ email.subject }}</h3><span class="sender">{{ email.sender }}</span></div><div class="email-summary"><p>{{ email.summary or '暂无摘要' }}</p></div><div class="email-meta"><span class="date">{{ email.date[:10] }} {{ email.date[11:16] }}</span><span class="status {% if email.processed %}success{% else %}warning{% endif %}">{{ '已处理' if email.processed else '未处理' }}</span></div></div>{% endfor %}</div>
</div>
{% endblock %}

8. CSS 样式

static/style.css

* {margin: 0;padding: 0;box-sizing: border-box;
}body {font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;background-color: #f5f7fa;color: #333;line-height: 1.6;
}.container {max-width: 1200px;margin: 0 auto;padding: 20px;
}header {display: flex;justify-content: space-between;align-items: center;margin-bottom: 30px;padding-bottom: 20px;border-bottom: 2px solid #e0e0e0;
}header h1 {color: #2c3e50;font-size: 2.5rem;
}nav a {margin-left: 20px;text-decoration: none;color: #666;padding: 8px 16px;border-radius: 4px;transition: all 0.3s ease;
}nav a:hover {color: #3498db;background-color: #f8f9fa;
}nav a.active {color: #3498db;background-color: #e3f2fd;font-weight: bold;
}.btn {padding: 10px 20px;border: none;border-radius: 5px;cursor: pointer;font-size: 14px;font-weight: bold;transition: all 0.3s ease;margin-right: 10px;
}.btn-primary {background-color: #3498db;color: white;
}.btn-primary:hover:not(:disabled) {background-color: #2980b9;
}.btn-secondary {background-color: #95a5a6;color: white;
}.btn-secondary:hover {background-color: #7f8c8d;
}.btn:disabled {opacity: 0.6;cursor: not-allowed;
}.controls {margin-bottom: 30px;display: flex;gap: 10px;
}.digest-card {background: white;border-radius: 10px;box-shadow: 0 2px 10px rgba(0,0,0,0.1);padding: 25px;margin-bottom: 30px;
}.digest-header {display: flex;justify-content: space-between;align-items: center;margin-bottom: 25px;padding-bottom: 15px;border-bottom: 1px solid #eee;
}.digest-header h2 {color: #2c3e50;font-size: 2rem;
}.email-count {color: #666;font-size: 1.2rem;
}.email-item {padding: 20px 0;border-bottom: 1px solid #eee;position: relative;
}.email-item:last-child {border-bottom: none;
}.email-item.processed {background-color: #f8f9fa;
}.email-header {display: flex;justify-content: space-between;margin-bottom: 12px;
}.email-header h3 {color: #2c3e50;font-size: 1.3rem;font-weight: 500;
}.sender {color: #666;font-size: 0.9rem;
}.email-summary {color: #555;margin-bottom: 12px;line-height: 1.5;
}.email-summary p {margin: 0;
}.email-time, .date {color: #999;font-size: 0.85rem;
}.email-meta {display: flex;justify-content: space-between;align-items: center;margin-top: 8px;
}.status {font-size: 0.8rem;padding: 4px 8px;border-radius: 12px;font-weight: bold;
}.status.success {background-color: #d4edda;color: #155724;
}.status.warning {background-color: #fff3cd;color: #856404;
}.no-data {text-align: center;padding: 60px 20px;background: white;border-radius: 10px;box-shadow: 0 2px 10px rgba(0,0,0,0.1);
}.no-data h3 {color: #666;margin-bottom: 15px;
}.system-info {background: white;border-radius: 10px;box-shadow: 0 2px 10px rgba(0,0,0,0.1);padding: 20px;
}.system-info h4 {margin-bottom: 15px;color: #2c3e50;
}.info-grid {display: grid;grid-template-columns: repeat(auto-fit, minmax(250px, 1fr));gap: 15px;
}.info-item {padding: 10px;background-color: #f8f9fa;border-radius: 5px;
}.info-item strong {color: #666;margin-right: 5px;
}footer {margin-top: 40px;text-align: center;color: #666;padding-top: 20px;border-top: 1px solid #eee;
}/* 响应式设计 */
@media (max-width: 768px) {.container {padding: 15px;}header {flex-direction: column;align-items: flex-start;}nav {margin-top: 15px;}nav a {margin-left: 0;margin-right: 15px;}.controls {flex-direction: column;}.btn {width: 100%;margin-right: 0;margin-bottom: 10px;}.digest-header {flex-direction: column;align-items: flex-start;gap: 10px;}.email-header {flex-direction: column;gap: 5px;}.email-meta {flex-direction: column;align-items: flex-start;gap: 5px;}
}

9. 依赖文件

requirements.txt

Flask==2.3.3
python-dotenv==1.0.0
requests==2.31.0
APScheduler==3.10.4

10. 环境变量配置

.env 示例

# 邮箱配置
EMAIL_HOST=imap.gmail.com
EMAIL_PORT=993
EMAIL_ADDRESS=your_email@gmail.com
EMAIL_PASSWORD=your_app_password# GLM配置
GLM_API_KEY=your_glm_api_key
GLM_MODEL=glm-4# 应用配置
SECRET_KEY=your-very-secret-key-change-this-in-production
DEBUG=False# 调度配置
CHECK_INTERVAL_MINUTES=30
MAX_EMAILS_PER_RUN=20

部署说明

1. 环境准备

# 克隆或创建项目目录
mkdir email-digest-simple
cd email-digest-simple# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows# 安装依赖
pip install -r requirements.txt

2. 配置环境变量

# 复制环境变量模板
cp .env.example .env  # 如果有示例文件
# 或者直接创建 .env 文件# 编辑 .env 文件
nano .env

重要配置说明：

EMAIL_PASSWORD: 对于 Gmail，需要在 Google 账户设置中启用"两步验证"，然后生成"应用专用密码"
GLM_API_KEY: 从智谱AI开放平台获取 API 密钥

3. 运行应用

# 开发模式（自动重载）
python app.py# 生产模式
# 安装 gunicorn
pip install gunicorn# 使用 gunicorn 运行
gunicorn -w 4 -b 0.0.0.0:5000 app:app

4. 系统服务部署（Linux）

创建 systemd 服务文件：

sudo nano /etc/systemd/system/email-digest.service

服务配置内容：

[Unit]
Description=Email Digest Simple Service
After=network.target[Service]
Type=simple
User=your-username
WorkingDirectory=/path/to/email-digest-simple
Environment=PATH=/path/to/email-digest-simple/venv/bin
ExecStart=/path/to/email-digest-simple/venv/bin/python app.py
Restart=always
RestartSec=10[Install]
WantedBy=multi-user.target

启用并启动服务：

sudo systemctl daemon-reload
sudo systemctl enable email-digest.service
sudo systemctl start email-digest.service
sudo systemctl status email-digest.service

5. Nginx 反向代理（可选）

如果需要通过域名访问，可以配置 Nginx：

server {listen 80;server_name your-domain.com;location / {proxy_pass http://127.0.0.1:5000;proxy_set_header Host $host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto $scheme;}
}