from typing import List, Dict, Tuple, Set, Any, Optional, Union """ mailer.py — Palladium 每日报告邮件发送 通过 SMTP 把"今日快照 + 24h 趋势"汇总发送到指定邮箱。 所有配置都从环境变量读 (生产建议用 systemd EnvironmentFile 或 .env)。 """ import os import smtplib import socket import ssl from datetime import datetime, timezone, timedelta from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import formataddr, format_datetime from app import ( ScanSnapshot, BoardStatus, JobInfo, db, app, DOMAIN_LABELS, DOMAIN_COLORS, ) # ═══════════════════════════════════════════════════════════════════════════ # 配置 (从环境变量读) # ═══════════════════════════════════════════════════════════════════════════ def _bool(name: str, default: bool = False) -> bool: v = os.environ.get(name, "").strip().lower() if v in ("1", "true", "yes", "y", "on"): return True if v in ("0", "false", "no", "n", "off", ""): return False return default def _int_env(name: str, default: int) -> int: try: return int(os.environ.get(name, "").strip()) except (TypeError, ValueError): return default # ═══════════════════════════════════════════════════════════════════════════ # 数据库配置持久化 # ═══════════════════════════════════════════════════════════════════════════ def _get_email_config_model(): """惰性获取 EmailConfig 模型 (避免循环导入)。""" try: from app import EmailConfig return EmailConfig except (ImportError, AttributeError): return None def _read_db_config() -> Optional[Dict[str, str]]: """从 EmailConfig 表读取配置, 返回 {key: value} dict, 无记录返回 None。""" Model = _get_email_config_model() if not Model: return None try: with app.app_context(): rows = Model.query.all() if not rows: return None return {r.key: r.value for r in rows} except Exception: return None def _build_config_from_env() -> dict: """从环境变量读取配置 (原始逻辑)。""" return { "enabled": _bool("MAIL_ENABLED", False), "smtp_host": os.environ.get("SMTP_HOST", "").strip(), "smtp_port": _int_env("SMTP_PORT", 587), "smtp_user": os.environ.get("SMTP_USER", "").strip(), "smtp_password_set": bool(os.environ.get("SMTP_PASSWORD", "").strip()), "smtp_use_tls": _bool("SMTP_USE_TLS", True), "smtp_use_ssl": _bool("SMTP_USE_SSL", False), "from_addr": os.environ.get("MAIL_FROM", "").strip(), "from_name": os.environ.get("MAIL_FROM_NAME", "Palladium Monitor").strip(), "to_addrs": [a.strip() for a in os.environ.get("MAIL_TO", "").split(",") if a.strip()], "send_hour": _int_env("MAIL_SEND_HOUR", 9), "send_minute": _int_env("MAIL_SEND_MINUTE", 0), "monitor_base_url": os.environ.get("MONITOR_BASE_URL", "http://127.0.0.1:5100").strip(), } def _build_config_from_dict(raw: Dict[str, str]) -> dict: """从数据库 {key: value} dict 构建配置。""" def _s(key: str, default: str = "") -> str: return raw.get(key, default).strip() def _b(key: str, default: bool) -> bool: v = raw.get(key, "").strip().lower() return v in ("1", "true", "yes", "y", "on") def _i(key: str, default: int) -> int: try: return int(raw.get(key, str(default))) except (TypeError, ValueError): return default return { "enabled": _b("enabled", False), "smtp_host": _s("smtp_host"), "smtp_port": _i("smtp_port", 587), "smtp_user": _s("smtp_user"), "smtp_password_set": bool(_s("smtp_password")), "smtp_use_tls": _b("smtp_use_tls", True), "smtp_use_ssl": _b("smtp_use_ssl", False), "from_addr": _s("from_addr"), "from_name": _s("from_name") or "Palladium Monitor", "to_addrs": [a.strip() for a in _s("to_addrs").split(",") if a.strip()], "send_hour": _i("send_hour", 9), "send_minute": _i("send_minute", 0), "monitor_base_url": _s("monitor_base_url") or "http://127.0.0.1:5100", } def save_config(data: dict) -> int: """保存配置到数据库。返回保存的键数。""" Model = _get_email_config_model() if not Model: raise RuntimeError("EmailConfig 模型不可用") to_addrs = data.get("to_addrs") # 统一处理：支持中英文逗号、分号分隔，自动规范化为英文逗号分隔 if isinstance(to_addrs, list): # 每个元素内部也可能含中英文逗号，全部拆开 parts = [] for item in to_addrs: s = str(item).replace("\uff0c", ",").replace("\u3001", ",").replace(";", ",") for p in s.split(","): p = p.strip() if p: parts.append(p) to_addrs_str = ",".join(parts) else: s = str(to_addrs or "").replace("\uff0c", ",").replace("\u3001", ",").replace(";", ",") parts = [p.strip() for p in s.split(",") if p.strip()] to_addrs_str = ",".join(parts) save_fields = { "enabled": str(data.get("enabled", False)).lower(), "smtp_host": str(data.get("smtp_host", "")).strip(), "smtp_port": str(data.get("smtp_port", 587)), "smtp_user": str(data.get("smtp_user", "")).strip(), "smtp_password": str(data.get("smtp_password", "")), "smtp_use_tls": str(data.get("smtp_use_tls", True)).lower(), "smtp_use_ssl": str(data.get("smtp_use_ssl", False)).lower(), "from_addr": str(data.get("from_addr", "")).strip(), "from_name": str(data.get("from_name", "Palladium Monitor")).strip(), "to_addrs": to_addrs_str.strip(), "send_hour": str(data.get("send_hour", 9)), "send_minute": str(data.get("send_minute", 0)), "monitor_base_url": str(data.get("monitor_base_url", "http://127.0.0.1:5100")).strip(), } count = 0 with app.app_context(): for key, value in save_fields.items(): row = Model.query.filter_by(key=key).first() if row: row.value = value else: db.session.add(Model(key=key, value=value)) count += 1 db.session.commit() return count def reset_config() -> bool: """清除数据库配置, 回退到环境变量。""" Model = _get_email_config_model() if not Model: return False with app.app_context(): Model.query.delete() db.session.commit() return True def get_config() -> dict: """读取邮件配置。优先使用数据库存储的配置, 回退到环境变量。""" db_raw = _read_db_config() if db_raw is not None: return _build_config_from_dict(db_raw) return _build_config_from_env() def get_config_full() -> dict: """读取完整配置 (含密码原始值)。优先 DB, 回退到环境变量。""" cfg = get_config() db_raw = _read_db_config() if db_raw is not None: cfg["smtp_password"] = db_raw.get("smtp_password", "").strip() else: cfg["smtp_password"] = os.environ.get("SMTP_PASSWORD", "").strip() return cfg def config_errors(cfg: dict) -> List[str]: """检查配置是否完整可发送。返回缺失/错误项的列表 (空=OK)。""" errs = [] if not cfg["enabled"]: return errs # 禁用时不做任何检查 if not cfg["smtp_host"]: errs.append("SMTP_HOST 未设置") if not cfg["smtp_user"]: errs.append("SMTP_USER 未设置") if not cfg["smtp_password_set"]: errs.append("SMTP_PASSWORD 未设置") if not cfg["from_addr"]: errs.append("MAIL_FROM 未设置") if not cfg["to_addrs"]: errs.append("MAIL_TO 未设置 (逗号分隔多个地址)") if not (0 <= cfg["send_hour"] <= 23): errs.append(f"MAIL_SEND_HOUR 越界: {cfg['send_hour']}") if not (0 <= cfg["send_minute"] <= 59): errs.append(f"MAIL_SEND_MINUTE 越界: {cfg['send_minute']}") return errs # ═══════════════════════════════════════════════════════════════════════════ # 数据收集 # ═══════════════════════════════════════════════════════════════════════════ def collect_report_data() -> dict: """ 拉取邮件所需的所有数据。 返回 dict 包含: latest, boards_offline, jobs_active, stats_24h, samples_24h。 """ now = datetime.now(timezone.utc) cutoff_24h = now - timedelta(hours=24) cutoff_yesterday = now - timedelta(hours=48) # ── 最新快照 ── latest = ScanSnapshot.query.order_by(ScanSnapshot.timestamp.desc()).first() if not latest: return { "latest": None, "boards_offline": [], "jobs_active": [], "stats_24h": None, "samples_24h": [], "now_utc": now, "now_local": datetime.now(), } # ── 离线板列表 ── boards_offline = BoardStatus.query.filter_by( snapshot_id=latest.id, status="OFFLINE" ).order_by(BoardStatus.rack, BoardStatus.cluster, BoardStatus.ld_index).all() # ── 活跃作业 ── jobs_active = JobInfo.query.filter_by(snapshot_id=latest.id).order_by(JobInfo.job_index).all() # Z1 详细输出把作业信息存在 DomainDetail 里, 若 JobInfo 为空则从 DomainDetail 回退 if not jobs_active: try: from app import DomainDetail domain_rows = DomainDetail.query.filter_by(snapshot_id=latest.id).all() if domain_rows: # 按 (owner, pid, design) 去重统计活跃作业数 active_set = {} # key: (owner, pid) -> {design, t_pod, elap_time} for dr in domain_rows: owner = dr.owner or "" pid = dr.pid or "" if owner and owner not in ("NONE", "") and pid and pid not in ("0", ""): key = (owner, pid) if key not in active_set: active_set[key] = { "design": dr.design or "—", "t_pod": dr.t_pod or "—", "elap_time": dr.elap_time or "—", } if active_set: # 构造虚拟 job 对象供渲染 (兼容 jobs_active 格式) class _VJob: pass for idx, ((o, p), info) in enumerate(sorted(active_set.items()), 1): j = _VJob() j.job_index = idx j.owner = o j.pid = p j.t_pod = info["t_pod"] if "--" not in info["t_pod"] and info["t_pod"] not in ("—",) else "—" j.design = info["design"] j.elap_time = info["elap_time"] if info["elap_time"] not in ("—", "") else "—" jobs_active.append(j) except (ImportError, AttributeError): pass # ── 24h 统计 ── snaps_24h = ScanSnapshot.query.filter( ScanSnapshot.timestamp >= cutoff_24h ).order_by(ScanSnapshot.timestamp.asc()).all() if snaps_24h: utils = [s.utilization for s in snaps_24h] used_counts = [s.used_boards for s in snaps_24h] stats_24h = { "count": len(snaps_24h), "util_min": min(utils), "util_max": max(utils), "util_avg": round(sum(utils) / len(utils), 1), "used_min": min(used_counts), "used_max": max(used_counts), "used_avg": round(sum(used_counts) / len(used_counts), 1), "first_ts": snaps_24h[0].timestamp, "last_ts": snaps_24h[-1].timestamp, } # 趋势样本: 最多 12 个等距点 (避免邮件太长) if len(snaps_24h) > 12: step = len(snaps_24h) // 12 samples = snaps_24h[::step] if samples[-1] != snaps_24h[-1]: samples.append(snaps_24h[-1]) else: samples = snaps_24h samples_24h = samples else: stats_24h = None samples_24h = [] # ── 按用户聚合 LD 板使用 ── user_board_agg = [] try: from app import DomainDetail domain_rows = DomainDetail.query.filter_by( snapshot_id=latest.id ).all() if domain_rows: # 按用户聚合: 统计每用户用了多少块不同 LD user_boards: dict[str, dict] = {} for dr in domain_rows: owner = dr.owner or "" if not owner or owner.upper() in ("NONE", "UNOWNED", "FREE", "AVAILABLE"): continue board_key = (dr.rack, dr.cluster, dr.ld_index) if owner not in user_boards: user_boards[owner] = { "user": owner, "boards": set(), "racks": set(), "pods": set(), "designs": set(), } entry = user_boards[owner] entry["boards"].add(board_key) if dr.rack is not None: entry["racks"].add(dr.rack) t_pod = dr.t_pod or "" # 过滤占位符 "--" (包括 "24 --" 这种含 -- 的) if t_pod and "--" not in t_pod and t_pod.strip() != "": entry["pods"].add(t_pod) design = dr.design or "" if design: entry["designs"].add(design) # 转为列表并排序 (按 LD 数降序) for owner, info in user_boards.items(): boards_sorted = sorted(info["boards"], key=lambda x: x[2]) user_board_agg.append({ "user": owner, "ld_count": len(boards_sorted), "boards": ", ".join(f"LD{b[2]}" for b in boards_sorted), "racks": ", ".join(str(r) for r in sorted(info["racks"])) if info["racks"] else "—", "pods": ", ".join(sorted(info["pods"])) if info["pods"] else "—", "designs": "; ".join(sorted(info["designs"])) if info["designs"] else "—", }) user_board_agg.sort(key=lambda x: (-x["ld_count"], x["user"])) except (ImportError, AttributeError): pass return { "latest": latest, "boards_offline": boards_offline, "jobs_active": jobs_active, "stats_24h": stats_24h, "samples_24h": samples_24h, "user_board_agg": user_board_agg, "now_utc": now, "now_local": datetime.now(), } # ═══════════════════════════════════════════════════════════════════════════ # 模板渲染 # ═══════════════════════════════════════════════════════════════════════════ def _fmt_ts(ts, fmt: str = "%Y-%m-%d %H:%M:%S") -> str: if ts is None: return "—" if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) return ts.astimezone().strftime(fmt) def _util_color_class(util: float) -> str: """根据利用率返回 CSS class 名 (用于 HTML 颜色条)。""" if util >= 80: return "util-high" if util >= 50: return "util-mid" return "util-low" def _render_text(data: dict, base_url: str) -> str: """纯文本版本 (HTML 不支持时回退)。""" latest = data["latest"] if not latest: return "Palladium Monitor — 当前无快照数据, 无法生成报告。\n" lines = [] title_date = _fmt_ts(latest.timestamp, "%Y-%m-%d") lines.append(f"Palladium 每日报告 — {title_date}") lines.append("=" * 60) lines.append(f"生成时间: {_fmt_ts(data['now_utc'])} UTC") lines.append(f"快照时间: {_fmt_ts(latest.timestamp)}") lines.append("") # 系统信息 lines.append("【系统信息】") lines.append(f" 仿真器: {latest.emulator or '—'}") lines.append(f" 硬件: {latest.hardware or '—'}") lines.append(f" ConfigMgr:{latest.configmgr or '—'}") lines.append(f" 系统状态: {latest.system_status or '—'}") lines.append(f" 数据来源: {latest.source or '—'}") lines.append("") # 当前指标 lines.append("【当前指标】") lines.append(f" 逻辑板总数: {latest.total_boards}") lines.append(f" 在线: {latest.online_boards}") lines.append(f" 离线: {latest.offline_boards}") lines.append(f" 利用率: {latest.utilization:.1f}% (使用 {latest.used_boards}/{latest.total_boards})") lines.append(f" 活跃作业: {latest.active_jobs}") lines.append("") # 用户-LD 板统计 if data.get("user_board_agg"): lines.append("【用户-LD 板统计】") for ub in data["user_board_agg"]: lines.append(f" {ub['user']}: {ub['ld_count']} 块 LD") lines.append(f" LD 板: {ub['boards']}") if ub["racks"] != "—": lines.append(f" Rack: {ub['racks']}") if ub["pods"] != "—": lines.append(f" T-Pod: {ub['pods']}") if ub["designs"] != "—": lines.append(f" 设计: {ub['designs']}") lines.append("") # 24h 趋势 if data["stats_24h"]: s = data["stats_24h"] lines.append("【近 24 小时趋势】") lines.append(f" 快照数: {s['count']}") lines.append(f" 利用率: min={s['util_min']:.1f}% max={s['util_max']:.1f}% avg={s['util_avg']:.1f}%") lines.append(f" 使用板: min={s['used_min']} max={s['used_max']} avg={s['used_avg']:.1f}") lines.append("") lines.append(" 趋势样本 (最多 12 个):") for s in data["samples_24h"]: bar = "█" * int(round(s.utilization / 5)) # 1 个 █ = 5% lines.append(f" {_fmt_ts(s.timestamp, '%m-%d %H:%M')} {s.utilization:5.1f}% {bar}") lines.append("") # 离线板 if data["boards_offline"]: lines.append(f"【离线板列表】 ({len(data['boards_offline'])} 块)") for b in data["boards_offline"]: lines.append(f" Rack {b.rack} / Cluster {b.cluster} / LD {b.ld_index} CCD={b.ccd or '?'}") lines.append("") else: lines.append("【离线板列表】 无 (全部在线)") lines.append("") # 活跃作业 if data["jobs_active"]: lines.append(f"【活跃作业】 ({len(data['jobs_active'])} 个)") for j in data["jobs_active"]: t_pod = j.t_pod or "—" lines.append(f" #{j.job_index} {j.owner} {j.pid} T-Pod: {t_pod} {j.design or '—'} ({j.elap_time or '—'})") lines.append("") lines.append("─" * 60) lines.append(f"查看完整仪表板: {base_url}/") return "\n".join(lines) def _render_html(data: dict, base_url: str) -> str: """HTML 版本 (multipart/alternative 中优先使用)。""" latest = data["latest"] if not latest: return ( '' '

Palladium Monitor

' '

当前无快照数据, 无法生成报告。

' '' ) title_date = _fmt_ts(latest.timestamp, "%Y-%m-%d") util_class = _util_color_class(latest.utilization) util_bar_width = f"{min(latest.utilization, 100):.1f}%" # 离线板 HTML if data["boards_offline"]: offline_rows = "".join( f"Rack {b.rack}Cluster {b.cluster}" f"LD {b.ld_index}{b.ccd or '—'}" for b in data["boards_offline"] ) offline_html = f""" {offline_rows}
Rack Cluster LD CCD
""" else: offline_html = '

✓ 无离线板, 全部在线

' # 作业 HTML if data["jobs_active"]: job_rows = "".join( f"#{j.job_index}{j.owner}{j.pid}" f"{j.t_pod or '—'}{j.design or '—'}{j.elap_time or '—'}" for j in data["jobs_active"] ) jobs_html = f""" {job_rows}
# 用户 PID T-Pod 设计 已运行
""" else: jobs_html = '

无活跃作业

' # 用户-LD 板统计 HTML user_board_html = "" if data.get("user_board_agg"): ub_rows = "".join( f'' f'{ub["user"]}' f'{ub["ld_count"]}' f'{ub["boards"]}' f'{ub["racks"]}' f'{ub["pods"]}' f'{ub["designs"]}' f'' for ub in data["user_board_agg"] ) user_board_html = f"""

👤 用户-LD 板统计

{ub_rows}
用户 LD 数 LD 板 Rack T-Pod 设计
""" # 24h 趋势 HTML if data["stats_24h"]: s = data["stats_24h"] # 进度条: 用 table 实现 (Gmail 兼容), 宽 200px, 蓝色按利用率百分比填充 sample_rows = "".join( f'{_fmt_ts(smp.timestamp, "%m-%d %H:%M")}' f'{smp.utilization:.1f}%' f'' f'' f'' f'
' f'' for smp in data["samples_24h"] ) trend_html = f"""
快照数{s['count']}
利用率最小/最大/平均 {s['util_min']:.1f}% / {s['util_max']:.1f}% / {s['util_avg']:.1f}%
使用板数最小/最大/平均 {s['used_min']} / {s['used_max']} / {s['used_avg']:.1f}

趋势样本

{sample_rows}
时间 利用率 图示
""" else: trend_html = '

近 24 小时无快照数据

' return f"""

⚡ Palladium 每日报告

{title_date}

📋 系统信息

仿真器{latest.emulator or '—'}
硬件{latest.hardware or '—'}
ConfigMgr{latest.configmgr or '—'}
系统状态{latest.system_status or '—'}
数据来源{latest.source or '—'}
快照时间{_fmt_ts(latest.timestamp)}

📊 当前指标

逻辑板总数
{latest.total_boards}
在线 {latest.online_boards}
离线 {latest.offline_boards}
逻辑板利用率
{latest.utilization:.1f}%
使用 {latest.used_boards} / 空闲 {latest.online_boards - latest.used_boards}
活跃作业
{len(data.get('jobs_active', []))}
{user_board_html}

📈 近 24 小时趋势

{trend_html}

🔴 离线板

{offline_html}

⚙️ 活跃作业

{jobs_html}
由 Palladium Monitor 自动生成 · {_fmt_ts(data['now_utc'], '%Y-%m-%d %H:%M:%S')} UTC
""" # ═══════════════════════════════════════════════════════════════════════════ # 发送 # ═══════════════════════════════════════════════════════════════════════════ def render_report(data: dict, cfg: dict) -> Tuple[str, str, str]: """返回 (subject, html, text)。""" latest = data["latest"] if latest: title_date = _fmt_ts(latest.timestamp, "%Y-%m-%d") else: title_date = _fmt_ts(data["now_utc"], "%Y-%m-%d") util = f"{latest.utilization:.1f}%" if latest else "—" subject = f"[Palladium] 每日报告 {title_date} — 利用率 {util}" html = _render_html(data, cfg["monitor_base_url"]) text = _render_text(data, cfg["monitor_base_url"]) return subject, html, text def _send_smtp(cfg: dict, subject: str, html: str, text: str) -> Tuple[bool, str]: """实际 SMTP 发送, 返回 (ok, error_msg)。""" msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = formataddr((cfg["from_name"], cfg["from_addr"])) msg["To"] = ", ".join(cfg["to_addrs"]) msg["Date"] = format_datetime(datetime.now(timezone.utc)) msg.attach(MIMEText(text, "plain", "utf-8")) msg.attach(MIMEText(html, "html", "utf-8")) password = cfg.get("smtp_password", "") if not password: password = os.environ.get("SMTP_PASSWORD", "") try: if cfg["smtp_use_ssl"]: # SMTP_SSL (port 465) ctx = ssl.create_default_context() with smtplib.SMTP_SSL(cfg["smtp_host"], cfg["smtp_port"], context=ctx, timeout=30) as s: s.login(cfg["smtp_user"], password) s.sendmail(cfg["from_addr"], cfg["to_addrs"], msg.as_string()) else: # STARTTLS (port 587) 或 明文 (port 25) with smtplib.SMTP(cfg["smtp_host"], cfg["smtp_port"], timeout=30) as s: s.ehlo() if cfg["smtp_use_tls"]: s.starttls() s.ehlo() s.login(cfg["smtp_user"], password) s.sendmail(cfg["from_addr"], cfg["to_addrs"], msg.as_string()) return (True, "") except smtplib.SMTPAuthenticationError as e: # Python 3.12+: smtp_error is str; older: bytes smtp_err = e.smtp_error.decode() if isinstance(e.smtp_error, bytes) else (e.smtp_error or "") return (False, f"SMTP 认证失败: {e.smtp_code} {smtp_err}") except smtplib.SMTPException as e: return (False, f"SMTP 错误: {e}") except (socket.timeout, ConnectionError, OSError) as e: return (False, f"网络错误: {e}") except Exception as e: return (False, f"{type(e).__name__}: {e}") def send_report(cfg: Optional[Dict] = None) -> dict: """ 收集数据 → 渲染 → 发送。 返回: {"ok": bool, "subject": str, "recipients": [...], "error": str, "preview_text": str} """ cfg = cfg or get_config_full() if not cfg["enabled"]: return {"ok": False, "error": "邮件功能未启用 (MAIL_ENABLED 未设 true)", "subject": "", "recipients": []} errs = config_errors(cfg) if errs: return {"ok": False, "error": "配置不完整: " + "; ".join(errs), "subject": "", "recipients": []} with app.app_context(): data = collect_report_data() subject, html, text = render_report(data, cfg) ok, err = _send_smtp(cfg, subject, html, text) return { "ok": ok, "subject": subject, "recipients": cfg["to_addrs"], "error": err, "preview_text": text[:500] + ("..." if len(text) > 500 else ""), "bytes": len(text) + len(html), } def write_preview(output_path: str) -> str: """把渲染好的报告写到文件 (不发送, 仅供调试)。返回写入路径。""" cfg = get_config() with app.app_context(): data = collect_report_data() subject, html, text = render_report(data, cfg) with open(output_path, "w", encoding="utf-8") as f: f.write(f"Subject: {subject}\n") f.write(f"To: {', '.join(cfg['to_addrs'])}\n") f.write(f"From: {cfg['from_name']} <{cfg['from_addr']}>\n") f.write("\n--- TEXT VERSION ---\n") f.write(text) f.write("\n\n--- HTML VERSION ---\n") f.write(html) return output_path