260 lines
8.4 KiB
Python
260 lines
8.4 KiB
Python
|
#!/usr/bin/env python
|
|||
|
|
"""
|
|||
|
|
PDF邮件发送调试脚本
|
|||
|
|
用于在生产服务器上直接调试PDF生成和邮件发送功能，无需依赖Celery
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import time
|
|||
|
|
from pathlib import Path
|
|||
|
|
from loguru import logger
|
|||
|
|
|
|||
|
|
# 配置日志
|
|||
|
|
logger.remove()
|
|||
|
|
logger.add(
|
|||
|
|
sys.stdout,
|
|||
|
|
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
|
|||
|
|
level="DEBUG"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
logger.add(
|
|||
|
|
"debug_pdf_email.log",
|
|||
|
|
format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
|
|||
|
|
level="DEBUG",
|
|||
|
|
rotation="1 day",
|
|||
|
|
retention="7 days",
|
|||
|
|
encoding="utf-8"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
logger.info("=== PDF邮件发送调试脚本开始 ===")
|
|||
|
|
logger.info(f"Python版本: {sys.version}")
|
|||
|
|
logger.info(f"当前目录: {os.getcwd()}")
|
|||
|
|
|
|||
|
|
# 初始化Django环境
|
|||
|
|
def init_django():
|
|||
|
|
"""初始化Django环境"""
|
|||
|
|
logger.info("初始化Django环境...")
|
|||
|
|
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'diary_family.settings')
|
|||
|
|
import django
|
|||
|
|
django.setup()
|
|||
|
|
logger.info("✅ Django环境初始化成功")
|
|||
|
|
|
|||
|
|
# 测试WeasyPrint安装
|
|||
|
|
def test_weasyprint():
|
|||
|
|
"""测试WeasyPrint是否正确安装"""
|
|||
|
|
logger.info("测试WeasyPrint安装...")
|
|||
|
|
try:
|
|||
|
|
from weasyprint import HTML
|
|||
|
|
logger.info("✅ WeasyPrint导入成功")
|
|||
|
|
return True
|
|||
|
|
except ImportError as e:
|
|||
|
|
logger.error(f"❌ WeasyPrint导入失败: {e}")
|
|||
|
|
logger.error("解决方案: 运行 'pip install weasyprint' 安装WeasyPrint库")
|
|||
|
|
return False
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"❌ WeasyPrint测试失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
# 测试生成PDF
|
|||
|
|
def test_pdf_generation():
|
|||
|
|
"""测试生成PDF"""
|
|||
|
|
logger.info("测试生成PDF...")
|
|||
|
|
try:
|
|||
|
|
from django.conf import settings
|
|||
|
|
from django.template.loader import render_to_string
|
|||
|
|
from weasyprint import HTML
|
|||
|
|
from core.models import SystemConfig, ReadingRecord, InsightRecord, TodayPlan, FamilyTask
|
|||
|
|
from django.utils import timezone
|
|||
|
|
from datetime import timedelta
|
|||
|
|
|
|||
|
|
# 准备数据
|
|||
|
|
today = timezone.now().date()
|
|||
|
|
yesterday = today - timedelta(days=1)
|
|||
|
|
|
|||
|
|
logger.info(f"报告日期: {today}, 昨日: {yesterday}")
|
|||
|
|
|
|||
|
|
# 获取数据
|
|||
|
|
yesterday_reading = ReadingRecord.objects.filter(date=yesterday)
|
|||
|
|
yesterday_insight = InsightRecord.objects.filter(date=yesterday)
|
|||
|
|
today_plan = TodayPlan.objects.filter(date=today)
|
|||
|
|
|
|||
|
|
logger.info(f"昨日阅读记录: {yesterday_reading.count()} 条")
|
|||
|
|
logger.info(f"昨日感悟记录: {yesterday_insight.count()} 条")
|
|||
|
|
logger.info(f"今日计划: {today_plan.count()} 项")
|
|||
|
|
|
|||
|
|
# 准备上下文
|
|||
|
|
context = {
|
|||
|
|
'today': today,
|
|||
|
|
'yesterday': yesterday,
|
|||
|
|
'yesterday_reading': yesterday_reading,
|
|||
|
|
'yesterday_insight': yesterday_insight,
|
|||
|
|
'today_plan': today_plan,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 渲染模板
|
|||
|
|
logger.info("渲染HTML模板...")
|
|||
|
|
html_string = render_to_string('core/report_pdf.html', context)
|
|||
|
|
logger.info("✅ HTML模板渲染成功")
|
|||
|
|
|
|||
|
|
# 生成PDF
|
|||
|
|
pdf_file = f"debug_report_{today.strftime('%Y-%m-%d')}.pdf"
|
|||
|
|
pdf_path = os.path.join(settings.REPORTS_ROOT, pdf_file)
|
|||
|
|
|
|||
|
|
logger.info(f"生成PDF文件: {pdf_path}")
|
|||
|
|
os.makedirs(settings.REPORTS_ROOT, exist_ok=True)
|
|||
|
|
|
|||
|
|
HTML(string=html_string).write_pdf(pdf_path, timeout=60)
|
|||
|
|
logger.info(f"✅ PDF生成成功，文件大小: {os.path.getsize(pdf_path) / 1024:.2f} KB")
|
|||
|
|
|
|||
|
|
return pdf_path
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"❌ PDF生成失败: {e}")
|
|||
|
|
import traceback
|
|||
|
|
logger.error(f"错误详情:\n{traceback.format_exc()}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
# 测试邮件发送
|
|||
|
|
def test_email_sending(pdf_path=None):
|
|||
|
|
"""测试邮件发送"""
|
|||
|
|
logger.info("测试邮件发送...")
|
|||
|
|
try:
|
|||
|
|
from django.core.mail import EmailMessage
|
|||
|
|
from django.core.mail.backends.smtp import EmailBackend
|
|||
|
|
from core.models import SystemConfig
|
|||
|
|
from django.utils import timezone
|
|||
|
|
|
|||
|
|
# 获取配置
|
|||
|
|
config = SystemConfig.get_config()
|
|||
|
|
|
|||
|
|
# 验证配置
|
|||
|
|
required_fields = ['smtp_server', 'smtp_username', 'smtp_password', 'recipient_email']
|
|||
|
|
missing_fields = []
|
|||
|
|
for field in required_fields:
|
|||
|
|
if not getattr(config, field):
|
|||
|
|
missing_fields.append(field)
|
|||
|
|
|
|||
|
|
if missing_fields:
|
|||
|
|
logger.error(f"❌ 缺少必要的邮件配置: {', '.join(missing_fields)}")
|
|||
|
|
logger.error("解决方案: 登录管理后台，配置邮件设置")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
# 准备邮件
|
|||
|
|
from_email = config.sender_email or config.smtp_username
|
|||
|
|
to_email = config.recipient_email
|
|||
|
|
subject = f"[调试] 家庭日报 {timezone.now().date()} - PDF报告"
|
|||
|
|
body = f"""
|
|||
|
|
这是一封调试邮件，用于测试PDF生成和邮件发送功能。
|
|||
|
|
|
|||
|
|
发送时间: {timezone.now().strftime('%Y-%m-%d %H:%M:%S')}
|
|||
|
|
报告日期: {timezone.now().date()}
|
|||
|
|
|
|||
|
|
这封邮件是从调试脚本直接发送的，没有经过Celery队列。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
# 创建SMTP连接
|
|||
|
|
logger.info(f"创建SMTP连接: {config.smtp_server}:{config.smtp_port}")
|
|||
|
|
logger.info(f"发件人: {from_email}")
|
|||
|
|
logger.info(f"收件人: {to_email}")
|
|||
|
|
|
|||
|
|
backend = EmailBackend(
|
|||
|
|
host=config.smtp_server,
|
|||
|
|
port=config.smtp_port or 587,
|
|||
|
|
username=config.smtp_username,
|
|||
|
|
password=config.smtp_password,
|
|||
|
|
use_tls=True,
|
|||
|
|
fail_silently=False,
|
|||
|
|
timeout=30
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 创建邮件
|
|||
|
|
email = EmailMessage(
|
|||
|
|
subject=subject,
|
|||
|
|
body=body,
|
|||
|
|
from_email=from_email,
|
|||
|
|
to=[to_email],
|
|||
|
|
connection=backend
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 添加附件
|
|||
|
|
if pdf_path and os.path.exists(pdf_path):
|
|||
|
|
logger.info(f"添加PDF附件: {pdf_path}")
|
|||
|
|
with open(pdf_path, 'rb') as f:
|
|||
|
|
email.attach(os.path.basename(pdf_path), f.read(), 'application/pdf')
|
|||
|
|
|
|||
|
|
# 发送邮件
|
|||
|
|
logger.info("发送邮件...")
|
|||
|
|
sent_count = email.send(fail_silently=False)
|
|||
|
|
logger.info(f"邮件发送返回: {sent_count}")
|
|||
|
|
|
|||
|
|
if sent_count > 0:
|
|||
|
|
logger.success("✅ 邮件发送成功！")
|
|||
|
|
return True
|
|||
|
|
else:
|
|||
|
|
logger.error("❌ 邮件发送失败")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"❌ 邮件发送失败: {e}")
|
|||
|
|
import traceback
|
|||
|
|
logger.error(f"错误详情:\n{traceback.format_exc()}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
# 测试Redis连接
|
|||
|
|
def test_redis_connection():
|
|||
|
|
"""测试Redis连接"""
|
|||
|
|
logger.info("测试Redis连接...")
|
|||
|
|
try:
|
|||
|
|
import redis
|
|||
|
|
from django.conf import settings
|
|||
|
|
|
|||
|
|
redis_url = settings.CELERY_BROKER_URL
|
|||
|
|
logger.info(f"Redis URL: {redis_url}")
|
|||
|
|
|
|||
|
|
# 测试连接
|
|||
|
|
client = redis.from_url(redis_url, socket_connect_timeout=10, socket_timeout=10)
|
|||
|
|
pong = client.ping()
|
|||
|
|
logger.info(f"Redis ping: {pong}")
|
|||
|
|
logger.info(f"Redis版本: {client.info().get('redis_version', '未知')}")
|
|||
|
|
client.close()
|
|||
|
|
logger.info("✅ Redis连接成功")
|
|||
|
|
return True
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"❌ Redis连接失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
# 主函数
|
|||
|
|
def main():
|
|||
|
|
"""主函数"""
|
|||
|
|
logger.info("=== PDF邮件发送调试开始 ===")
|
|||
|
|
|
|||
|
|
# 初始化Django
|
|||
|
|
init_django()
|
|||
|
|
|
|||
|
|
# 测试Redis连接
|
|||
|
|
logger.info("\n1. 测试Redis连接")
|
|||
|
|
test_redis_connection()
|
|||
|
|
|
|||
|
|
# 测试WeasyPrint
|
|||
|
|
logger.info("\n2. 测试WeasyPrint")
|
|||
|
|
if not test_weasyprint():
|
|||
|
|
logger.error("WeasyPrint测试失败，无法继续")
|
|||
|
|
return 1
|
|||
|
|
|
|||
|
|
# 测试生成PDF
|
|||
|
|
logger.info("\n3. 测试生成PDF")
|
|||
|
|
pdf_path = test_pdf_generation()
|
|||
|
|
if not pdf_path:
|
|||
|
|
logger.error("PDF生成失败，无法继续邮件发送测试")
|
|||
|
|
else:
|
|||
|
|
# 测试发送邮件
|
|||
|
|
logger.info("\n4. 测试发送邮件")
|
|||
|
|
test_email_sending(pdf_path)
|
|||
|
|
|
|||
|
|
logger.info("\n=== PDF邮件发送调试结束 ===")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
sys.exit(main())
|