169 lines
6.2 KiB
Python
169 lines
6.2 KiB
Python
"""
|
||
邮件通知服务
|
||
使用 Python smtplib + email.mime 实现，支持 TLS/SSL SMTP 连接
|
||
从环境变量或 Flask config 读取邮件配置
|
||
"""
|
||
import os
|
||
import smtplib
|
||
import ssl
|
||
import logging
|
||
from email.mime.text import MIMEText
|
||
from email.mime.multipart import MIMEMultipart
|
||
from email.header import Header
|
||
from typing import List, Union
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _get_config():
|
||
"""
|
||
读取邮件配置，优先从 Flask app config，回退到环境变量
|
||
"""
|
||
try:
|
||
from flask import current_app
|
||
return {
|
||
'server': current_app.config.get('MAIL_SERVER', os.getenv('MAIL_SERVER')),
|
||
'port': current_app.config.get('MAIL_PORT', int(os.getenv('MAIL_PORT', 587))),
|
||
'username': current_app.config.get('MAIL_USERNAME', os.getenv('MAIL_USERNAME')),
|
||
'password': current_app.config.get('MAIL_PASSWORD', os.getenv('MAIL_PASSWORD')),
|
||
'sender': current_app.config.get('MAIL_DEFAULT_SENDER', os.getenv('MAIL_DEFAULT_SENDER')),
|
||
'use_tls': current_app.config.get('MAIL_USE_TLS', os.getenv('MAIL_USE_TLS', 'true').lower() in ('true', '1', 'yes')),
|
||
'use_ssl': current_app.config.get('MAIL_USE_SSL', os.getenv('MAIL_USE_SSL', 'false').lower() in ('true', '1', 'yes')),
|
||
'enabled': current_app.config.get('MAIL_ENABLED', os.getenv('MAIL_ENABLED', 'false').lower() in ('true', '1', 'yes')),
|
||
}
|
||
except RuntimeError:
|
||
# 不在 Flask 上下文时，直接读环境变量
|
||
return {
|
||
'server': os.getenv('MAIL_SERVER'),
|
||
'port': int(os.getenv('MAIL_PORT', 587)),
|
||
'username': os.getenv('MAIL_USERNAME'),
|
||
'password': os.getenv('MAIL_PASSWORD'),
|
||
'sender': os.getenv('MAIL_DEFAULT_SENDER'),
|
||
'use_tls': os.getenv('MAIL_USE_TLS', 'true').lower() in ('true', '1', 'yes'),
|
||
'use_ssl': os.getenv('MAIL_USE_SSL', 'false').lower() in ('true', '1', 'yes'),
|
||
'enabled': os.getenv('MAIL_ENABLED', 'false').lower() in ('true', '1', 'yes'),
|
||
}
|
||
|
||
|
||
def send_email(to_email: Union[str, List[str]], subject: str, content: str):
|
||
"""
|
||
通用邮件发送函数
|
||
|
||
Args:
|
||
to_email: 收件人，单个邮箱字符串或列表
|
||
subject: 邮件主题
|
||
content: 邮件正文（纯文本）
|
||
|
||
发送失败时打印日志，不抛出异常
|
||
"""
|
||
cfg = _get_config()
|
||
|
||
# 发送总开关
|
||
if not cfg.get('enabled'):
|
||
logger.info(f"[Email] 邮件功能已禁用 (MAIL_ENABLED=false)，跳过发送: {subject}")
|
||
return
|
||
|
||
# 配置完整性检查
|
||
if not cfg.get('server') or not cfg.get('username') or not cfg.get('password'):
|
||
logger.warning("[Email] 邮件配置不完整 (MAIL_SERVER/USERNAME/PASSWORD 缺失)，跳过发送")
|
||
return
|
||
|
||
# 标准化收件人列表
|
||
recipients = [to_email] if isinstance(to_email, str) else [r.strip() for r in to_email if r.strip()]
|
||
if not recipients:
|
||
logger.warning("[Email] 收件人地址为空，跳过发送")
|
||
return
|
||
|
||
try:
|
||
msg = MIMEMultipart()
|
||
msg['From'] = cfg['sender']
|
||
msg['To'] = ', '.join(recipients)
|
||
msg['Subject'] = Header(subject, 'utf-8')
|
||
msg.attach(MIMEText(content, 'plain', 'utf-8'))
|
||
|
||
if cfg.get('use_ssl'):
|
||
context = ssl.create_default_context()
|
||
with smtplib.SMTP_SSL(cfg['server'], cfg.get('port', 465), context=context) as server:
|
||
server.login(cfg['username'], cfg['password'])
|
||
server.sendmail(cfg['username'], recipients, msg.as_string())
|
||
else:
|
||
with smtplib.SMTP(cfg['server'], cfg.get('port', 587)) as server:
|
||
if cfg.get('use_tls'):
|
||
server.starttls(context=ssl.create_default_context())
|
||
server.login(cfg['username'], cfg['password'])
|
||
server.sendmail(cfg['username'], recipients, msg.as_string())
|
||
|
||
logger.info(f"[Email] 发送成功 -> {recipients}: {subject}")
|
||
|
||
except smtplib.SMTPAuthenticationError:
|
||
logger.error("[Email] 邮箱认证失败，请检查 MAIL_USERNAME / MAIL_PASSWORD（授权码）")
|
||
except smtplib.SMTPRecipientsRefused as e:
|
||
logger.error(f"[Email] 收件人被服务器拒绝: {e}")
|
||
except smtplib.SMTPException as e:
|
||
logger.error(f"[Email] SMTP 异常: {e}")
|
||
except Exception as e:
|
||
logger.error(f"[Email] 发送邮件时发生未知异常: {e}")
|
||
|
||
|
||
def send_new_request_notify(to_emails: List[str], request_no: str,
|
||
applicant_name: str = '', remark: str = ''):
|
||
"""
|
||
通知审批人有新的出库申请单待审批
|
||
|
||
Args:
|
||
to_emails: 审批人邮箱列表
|
||
request_no: 审批单号
|
||
applicant_name: 申请人姓名
|
||
remark: 申请备注
|
||
"""
|
||
subject = f"【待审批】出库申请单 {request_no}"
|
||
content = f"""您好，
|
||
|
||
您有一笔新的出库审批申请待处理：
|
||
|
||
申请单号：{request_no}
|
||
申请人：{applicant_name or '未知'}
|
||
备注说明：{remark or '无'}
|
||
|
||
请登录仓库管理系统进行审批。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
send_email(to_emails, subject, content)
|
||
|
||
|
||
def send_approval_result_notify(to_emails: List[str], request_no: str,
|
||
is_passed: bool, reject_reason: str = ''):
|
||
"""
|
||
通知库管和申请人审批结果
|
||
|
||
Args:
|
||
to_emails: 收件人邮箱列表（库管 + 申请人）
|
||
request_no: 审批单号
|
||
is_passed: 是否通过
|
||
reject_reason: 驳回原因（仅 is_passed=False 时使用）
|
||
"""
|
||
if is_passed:
|
||
subject = f"【已通过】出库申请单 {request_no}"
|
||
content = f"""您好，
|
||
|
||
出库申请单 {request_no} 已审批通过，请准备备货。
|
||
|
||
请尽快安排出库操作。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
else:
|
||
subject = f"【已驳回】出库申请单 {request_no}"
|
||
content = f"""您好，
|
||
|
||
出库申请单 {request_no} 已被驳回。
|
||
|
||
驳回原因：{reject_reason or '未填写'}
|
||
|
||
请登录仓库管理系统查看详情。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
send_email(to_emails, subject, content)
|