""" 邮件通知服务 使用 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)} --- ⚡ 快速通道： 请点击下方链接直接进入系统审批： https://172.16.0.198/outbound/approval --- 请登录仓库管理系统进行审批。 此邮件由系统自动发送，请勿回复。 """ 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}")