251 lines
9.9 KiB
Python
251 lines
9.9 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()
|
||
|
||
print(f"[DEBUG send_email] cfg = {cfg}")
|
||
|
||
# 发送总开关
|
||
if not cfg.get('enabled'):
|
||
print(f"[Email] 邮件功能已禁用 (MAIL_ENABLED=false)，跳过发送: {subject}")
|
||
logger.info(f"[Email] 邮件功能已禁用 (MAIL_ENABLED=false)，跳过发送: {subject}")
|
||
return
|
||
|
||
# 配置完整性检查
|
||
if not cfg.get('server') or not cfg.get('username') or not cfg.get('password'):
|
||
print(f"[Email] 邮件配置不完整 server={cfg.get('server')} username={cfg.get('username')} password={'已设' if cfg.get('password') else '空'}，跳过发送")
|
||
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:
|
||
print("[Email] 收件人地址为空，跳过发送")
|
||
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'))
|
||
|
||
print(f"DEBUG: 准备向服务器提交发信请求，收件人: {recipients} 发件人: {cfg['username']}")
|
||
|
||
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:
|
||
print(f"!!! 邮件发送核心报错: SMTPAuthenticationError - 邮箱认证失败，请检查 MAIL_USERNAME / MAIL_PASSWORD（授权码）")
|
||
logger.error("[Email] 邮箱认证失败，请检查 MAIL_USERNAME / MAIL_PASSWORD（授权码）")
|
||
except smtplib.SMTPRecipientsRefused as e:
|
||
print(f"!!! 邮件发送核心报错: SMTPRecipientsRefused - 收件人被服务器拒绝: {e}")
|
||
logger.error(f"[Email] 收件人被服务器拒绝: {e}")
|
||
except smtplib.SMTPException as e:
|
||
print(f"!!! 邮件发送核心报错: SMTPException - {e}")
|
||
logger.error(f"[Email] SMTP 异常: {e}")
|
||
except Exception as e:
|
||
import traceback
|
||
traceback.print_exc()
|
||
print(f"!!! 邮件发送核心报错: {type(e).__name__} - {e}")
|
||
logger.error(f"[Email] 发送邮件时发生未知异常: {e}")
|
||
|
||
|
||
def send_new_request_notify(to_emails: List[str], request_no: str,
|
||
applicant_name: str = '', remark: str = '',
|
||
items: list = None):
|
||
"""
|
||
通知审批人有新的出库申请单待审批（可附带物料清单）
|
||
|
||
Args:
|
||
to_emails: 审批人邮箱列表
|
||
request_no: 审批单号
|
||
applicant_name: 申请人姓名
|
||
remark: 申请备注
|
||
items: 物料明细列表（可选）
|
||
"""
|
||
print(f"[DEBUG send_new_request_notify] 入参 items={items}")
|
||
print(f"[DEBUG send_new_request_notify] items 类型={type(items)}, 长度={len(items) if items else 0}")
|
||
|
||
# 拼装物料表格
|
||
rows = []
|
||
rows.append("名称 | 规格 | 计划数量")
|
||
rows.append("-" * 40)
|
||
if items:
|
||
for item in items:
|
||
name = item.get('name', '-') or '-'
|
||
spec = item.get('spec_model', '-') or '-'
|
||
qty = item.get('quantity', '-') or '-'
|
||
rows.append(f"{name} | {spec} | {qty}")
|
||
else:
|
||
rows.append("(无物料明细)")
|
||
|
||
subject = f"【待审批】出库申请单 {request_no}"
|
||
content = f"""您好，
|
||
|
||
您有一笔新的出库审批申请待处理：
|
||
|
||
申请单号：{request_no}
|
||
申请人：{applicant_name or '未知'}
|
||
备注说明：{remark or '无'}
|
||
|
||
物料清单如下：
|
||
{chr(10).join(rows)}
|
||
|
||
请登录仓库管理系统进行审批。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
send_email(to_emails, subject, content)
|
||
|
||
|
||
def send_approval_result_notify(to_emails: List[str], request_no: str,
|
||
is_passed: bool, reject_reason: str = '',
|
||
applicant_name: str = ''):
|
||
"""
|
||
通知审批结果
|
||
|
||
Args:
|
||
to_emails: 收件人邮箱列表
|
||
request_no: 审批单号
|
||
is_passed: 是否通过（通过时发给库管，驳回时发给申请人）
|
||
reject_reason: 驳回原因（仅 is_passed=False 时使用）
|
||
applicant_name: 申请人姓名（仅驳回通知时使用）
|
||
"""
|
||
if is_passed:
|
||
# ★ 发给申请人：告知已通过，去领料
|
||
subject = f"【已通过】出库申请单 {request_no}"
|
||
content = f"""{"尊敬的 " + applicant_name + "，您好" if applicant_name else "您好"}，
|
||
|
||
您的出库申请单 {request_no} 已审批通过，请联系仓库管理员领取物料。
|
||
|
||
请登录仓库管理系统查看详情。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
else:
|
||
# ★ 发给申请人：告知被驳回
|
||
subject = f"【已驳回】出库申请单 {request_no}"
|
||
content = f"""{"尊敬的 " + applicant_name + "，您好" if applicant_name else "您好"}，
|
||
|
||
出库申请单 {request_no} 已被审批驳回。
|
||
|
||
驳回原因：{reject_reason or '未填写'}
|
||
|
||
请登录仓库管理系统查看详情，并根据驳回原因调整后重新提交申请。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
send_email(to_emails, subject, content)
|
||
|
||
|
||
def send_warehouse_dispatch_notify(to_emails: List[str], request_no: str,
|
||
applicant_name: str = '', items: list = None):
|
||
"""
|
||
通知库管备货出库（包含完整物料清单）
|
||
|
||
Args:
|
||
to_emails: 库管邮箱列表
|
||
request_no: 审批单号
|
||
applicant_name: 申请人姓名
|
||
items: 物料明细列表，每个元素包含 name/spec_model/warehouse_location/quantity
|
||
"""
|
||
print(f"[DEBUG send_warehouse_dispatch_notify] 入参 items={items}")
|
||
print(f"[DEBUG send_warehouse_dispatch_notify] items 类型={type(items)}, 长度={len(items) if items else 0}")
|
||
|
||
rows = []
|
||
rows.append("名称 | 规格 | 库位 | 计划数量")
|
||
rows.append("-" * 50)
|
||
if items:
|
||
for item in items:
|
||
name = item.get('name', '-') or '-'
|
||
spec = item.get('spec_model', '-') or '-'
|
||
loc = item.get('warehouse_location', '-') or '-'
|
||
qty = item.get('quantity', '-') or '-'
|
||
rows.append(f"{name} | {spec} | {loc} | {qty}")
|
||
else:
|
||
rows.append("(无物料明细)")
|
||
|
||
subject = f"【待出库】出库申请单 {request_no} 已审批通过"
|
||
content = f"""您好，
|
||
|
||
出库申请单 {request_no} 已审批通过，请按以下清单准备备货：
|
||
|
||
{chr(10).join(rows)}
|
||
|
||
申请人：{applicant_name or '未知'}
|
||
|
||
请登录仓库管理系统执行"按单出库"操作。
|
||
|
||
此邮件由系统自动发送，请勿回复。
|
||
"""
|
||
send_email(to_emails, subject, content)
|
||
print(f"DEBUG: 准备向服务器提交发信请求，收件人: {to_emails}")
|