|
|
@@ -0,0 +1,295 @@
|
|
|
+"""新用户注册邮件通知服务
|
|
|
+
|
|
|
+每 N 分钟扫描 user 表，将本周期内新注册用户信息汇总后通过邮件发送给管理员。
|
|
|
+
|
|
|
+Prerequisites:
|
|
|
+- Python >= 3.7
|
|
|
+- 配置文件：send_new_user.conf（与脚本同目录）
|
|
|
+ - [smtp] 发件配置
|
|
|
+ - [database] 数据库配置
|
|
|
+ - [admin] admin_emails 收件人，多个逗号分隔
|
|
|
+
|
|
|
+运行: python send_new_user_email.py
|
|
|
+"""
|
|
|
+
|
|
|
+import asyncio
|
|
|
+import configparser
|
|
|
+import logging
|
|
|
+import os
|
|
|
+import re
|
|
|
+import smtplib
|
|
|
+import sys
|
|
|
+from datetime import datetime, timedelta
|
|
|
+from pathlib import Path
|
|
|
+from typing import Optional
|
|
|
+
|
|
|
+from email.header import Header
|
|
|
+from email.mime.multipart import MIMEMultipart
|
|
|
+from email.mime.text import MIMEText
|
|
|
+from email.utils import formataddr
|
|
|
+from sqlalchemy import select
|
|
|
+
|
|
|
+config_file = Path(__file__).parent / "send_new_user.conf"
|
|
|
+config = configparser.ConfigParser()
|
|
|
+if config_file.exists():
|
|
|
+ config.read(config_file, encoding="utf-8")
|
|
|
+else:
|
|
|
+ raise FileNotFoundError(f"配置文件不存在: {config_file}")
|
|
|
+
|
|
|
+backend_dir = Path(__file__).parent.parent / "backend"
|
|
|
+sys.path.insert(0, str(backend_dir))
|
|
|
+from database import DatabaseManager
|
|
|
+from models_db import User
|
|
|
+
|
|
|
+logger = logging.getLogger(__name__)
|
|
|
+
|
|
|
+# ------- 从配置文件读取参数 -------
|
|
|
+SMTP_HOST = config.get("smtp", "smtp_host", fallback="smtpdm.aliyun.com")
|
|
|
+SMTP_PORT_SSL = int(config.get("smtp", "smtp_port_ssl", fallback="465"))
|
|
|
+SMTP_USERNAME = config.get("smtp", "smtp_username", fallback="").strip()
|
|
|
+SMTP_PASSWORD = config.get("smtp", "smtp_password", fallback="").strip()
|
|
|
+EMAIL_SUBJECT = config.get("smtp", "email_subject", fallback="Lyrria 新用户注册通知")
|
|
|
+
|
|
|
+MYSQL_HOST = config.get("database", "mysql_host", fallback="").strip()
|
|
|
+MYSQL_PORT = config.get("database", "mysql_port", fallback="").strip()
|
|
|
+MYSQL_USER = config.get("database", "mysql_user", fallback="").strip()
|
|
|
+MYSQL_PASSWORD = config.get("database", "mysql_password", fallback="").strip()
|
|
|
+MYSQL_DATABASE = config.get("database", "mysql_database", fallback="").strip()
|
|
|
+
|
|
|
+POLL_INTERVAL = float(config.get("service", "poll_interval", fallback="600"))
|
|
|
+ADMIN_EMAILS_RAW = config.get("admin", "admin_emails", fallback="").strip()
|
|
|
+ADMIN_EMAILS = [e.strip() for e in ADMIN_EMAILS_RAW.split(",") if e.strip()]
|
|
|
+
|
|
|
+
|
|
|
+def build_message(
|
|
|
+ sender: str,
|
|
|
+ recipients: list[str],
|
|
|
+ subject: str,
|
|
|
+ body_text: str,
|
|
|
+) -> MIMEMultipart:
|
|
|
+ """构建纯文本邮件消息"""
|
|
|
+ msg = MIMEMultipart("alternative")
|
|
|
+ msg["Subject"] = Header(subject, "utf-8")
|
|
|
+ msg["From"] = formataddr(["Lyrria 管理员通知", sender])
|
|
|
+ msg["To"] = ",".join(recipients)
|
|
|
+ msg.attach(MIMEText(body_text, "plain", "utf-8"))
|
|
|
+ return msg
|
|
|
+
|
|
|
+
|
|
|
+def format_user_line(idx: int, u: User) -> str:
|
|
|
+ """将单条用户格式化为一行中文说明"""
|
|
|
+ created = u.created_at.strftime("%Y-%m-%d %H:%M:%S") if u.created_at else "-"
|
|
|
+ reg = u.registered_at.strftime("%Y-%m-%d %H:%M:%S") if u.registered_at else "-"
|
|
|
+ parts = [
|
|
|
+ f"用户{idx}：",
|
|
|
+ f"昵称 {u.username or '-'}",
|
|
|
+ f"用户ID {u.user_id}",
|
|
|
+ f"类型 {u.user_type or 'guest'}",
|
|
|
+ f"邮箱 {u.email or '-'}",
|
|
|
+ f"手机 {u.phone or '-'}",
|
|
|
+ f"区号 {u.area_code or '-'}",
|
|
|
+ f"注册时间 {reg}",
|
|
|
+ f"创建时间 {created}",
|
|
|
+ ]
|
|
|
+ return "，".join(parts)
|
|
|
+
|
|
|
+
|
|
|
+def build_body(users: list[User], period_start: datetime, period_end: datetime) -> str:
|
|
|
+ """组合新用户信息为邮件正文（汉字）"""
|
|
|
+ start_str = period_start.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
+ end_str = period_end.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
+ lines = [
|
|
|
+ "【Lyrria 新用户注册通知】",
|
|
|
+ "",
|
|
|
+ f"统计周期：{start_str} 至 {end_str}",
|
|
|
+ f"本周期新注册用户数：{len(users)}",
|
|
|
+ "",
|
|
|
+ ]
|
|
|
+ if not users:
|
|
|
+ lines.append("本周期暂无新用户。")
|
|
|
+ return "\n".join(lines)
|
|
|
+ for i, u in enumerate(users, 1):
|
|
|
+ lines.append(format_user_line(i, u))
|
|
|
+ lines.append("")
|
|
|
+ lines.append("本邮件由系统自动发送，请勿直接回复。")
|
|
|
+ return "\n".join(lines)
|
|
|
+
|
|
|
+
|
|
|
+def send_email_sync(
|
|
|
+ recipients: list[str],
|
|
|
+ subject: str,
|
|
|
+ body: str,
|
|
|
+ smtp_username: str,
|
|
|
+ smtp_password: str,
|
|
|
+) -> tuple[bool, Optional[str]]:
|
|
|
+ """同步发送邮件。返回 (成功与否, 错误信息)。"""
|
|
|
+ if not recipients:
|
|
|
+ return False, "收件人列表为空"
|
|
|
+ if not smtp_username or not smtp_password:
|
|
|
+ return False, "SMTP 未配置 smtp_username / smtp_password"
|
|
|
+ try:
|
|
|
+ msg = build_message(smtp_username, recipients, subject, body)
|
|
|
+ with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT_SSL) as client:
|
|
|
+ client.login(smtp_username, smtp_password)
|
|
|
+ client.sendmail(smtp_username, recipients, msg.as_string())
|
|
|
+ logger.info("✅ 新用户通知邮件已发送，收件人: %s", recipients)
|
|
|
+ return True, None
|
|
|
+ except smtplib.SMTPAuthenticationError as e:
|
|
|
+ err = f"SMTP 认证失败: {e}"
|
|
|
+ logger.error("❌ %s", err)
|
|
|
+ return False, err
|
|
|
+ except smtplib.SMTPException as e:
|
|
|
+ err = f"SMTP 错误: {e}"
|
|
|
+ logger.error("❌ %s", err)
|
|
|
+ return False, err
|
|
|
+ except Exception as e:
|
|
|
+ err = f"发送异常: {e}"
|
|
|
+ logger.error("❌ %s", err)
|
|
|
+ return False, err
|
|
|
+
|
|
|
+
|
|
|
+async def fetch_new_users(
|
|
|
+ db_manager: DatabaseManager,
|
|
|
+ period_start: datetime,
|
|
|
+ period_end: datetime,
|
|
|
+) -> list[User]:
|
|
|
+ """查询 (period_start, period_end] 内创建的用户"""
|
|
|
+ async with db_manager.get_session() as session:
|
|
|
+ result = await session.execute(
|
|
|
+ select(User)
|
|
|
+ .where(
|
|
|
+ User.created_at > period_start,
|
|
|
+ User.created_at <= period_end,
|
|
|
+ )
|
|
|
+ .order_by(User.created_at.asc())
|
|
|
+ )
|
|
|
+ return list(result.scalars().all())
|
|
|
+
|
|
|
+
|
|
|
+def _parse_last_scan_end() -> Optional[datetime]:
|
|
|
+ """从配置文件读取并解析 last_scan_end，无效或为空返回 None"""
|
|
|
+ raw = config.get("service", "last_scan_end", fallback="").strip()
|
|
|
+ if not raw:
|
|
|
+ return None
|
|
|
+ try:
|
|
|
+ return datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
|
|
+ except (ValueError, TypeError):
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def _write_last_scan_end(period_end: datetime) -> None:
|
|
|
+ """将本次扫描结束时间写回配置文件（仅修改 last_scan_end 行，保留注释）"""
|
|
|
+ try:
|
|
|
+ value = period_end.isoformat()
|
|
|
+ path = Path(config_file)
|
|
|
+ text = path.read_text(encoding="utf-8")
|
|
|
+ new_text = re.sub(
|
|
|
+ r"^(\s*last_scan_end\s*=\s*).*$",
|
|
|
+ rf"\g<1>{value}",
|
|
|
+ text,
|
|
|
+ count=1,
|
|
|
+ flags=re.MULTILINE,
|
|
|
+ )
|
|
|
+ if new_text != text:
|
|
|
+ path.write_text(new_text, encoding="utf-8")
|
|
|
+ logger.debug("已更新 last_scan_end = %s", value)
|
|
|
+ except Exception as e:
|
|
|
+ logger.warning("写回 last_scan_end 失败: %s", e)
|
|
|
+
|
|
|
+
|
|
|
+async def run_once(db_manager: DatabaseManager) -> None:
|
|
|
+ """执行一次：按 last_scan_end 确定扫描范围，查新用户并发邮件，成功后写回 last_scan_end"""
|
|
|
+ if not ADMIN_EMAILS:
|
|
|
+ logger.warning("⚠️ 未配置 admin_emails，跳过发送")
|
|
|
+ return
|
|
|
+ config.read(config_file, encoding="utf-8")
|
|
|
+ period_end = datetime.now()
|
|
|
+ last_end = _parse_last_scan_end()
|
|
|
+ if last_end is not None:
|
|
|
+ period_start = last_end
|
|
|
+ logger.info("📅 从上次扫描结束时间继续: %s", last_end.strftime("%Y-%m-%d %H:%M:%S"))
|
|
|
+ else:
|
|
|
+ period_start = period_end - timedelta(seconds=POLL_INTERVAL)
|
|
|
+ users = await fetch_new_users(db_manager, period_start, period_end)
|
|
|
+ if not users:
|
|
|
+ _write_last_scan_end(period_end)
|
|
|
+ logger.debug("本周期无新用户，未发邮件，已更新 last_scan_end")
|
|
|
+ return
|
|
|
+ body = build_body(users, period_start, period_end)
|
|
|
+ logger.info("📋 本周期新用户数: %d，正在发送通知邮件", len(users))
|
|
|
+ ok, _ = await asyncio.to_thread(
|
|
|
+ send_email_sync,
|
|
|
+ ADMIN_EMAILS,
|
|
|
+ EMAIL_SUBJECT,
|
|
|
+ body,
|
|
|
+ SMTP_USERNAME,
|
|
|
+ SMTP_PASSWORD,
|
|
|
+ )
|
|
|
+ if ok:
|
|
|
+ _write_last_scan_end(period_end)
|
|
|
+
|
|
|
+
|
|
|
+async def main_loop() -> None:
|
|
|
+ """主循环：每 POLL_INTERVAL 秒扫描并发邮件"""
|
|
|
+ if MYSQL_HOST:
|
|
|
+ os.environ["MYSQL_HOST"] = MYSQL_HOST
|
|
|
+ if MYSQL_PORT:
|
|
|
+ os.environ["MYSQL_PORT"] = MYSQL_PORT
|
|
|
+ if MYSQL_USER:
|
|
|
+ os.environ["MYSQL_USER"] = MYSQL_USER
|
|
|
+ if MYSQL_PASSWORD:
|
|
|
+ os.environ["MYSQL_PASSWORD"] = MYSQL_PASSWORD
|
|
|
+ if MYSQL_DATABASE:
|
|
|
+ os.environ["MYSQL_DATABASE"] = MYSQL_DATABASE
|
|
|
+
|
|
|
+ db_manager = DatabaseManager()
|
|
|
+ try:
|
|
|
+ await db_manager.connect()
|
|
|
+ logger.info("✅ 数据库连接成功")
|
|
|
+ logger.info("🚀 新用户通知服务已启动，扫描间隔: %s 秒", POLL_INTERVAL)
|
|
|
+ while True:
|
|
|
+ try:
|
|
|
+ await run_once(db_manager)
|
|
|
+ except Exception as e:
|
|
|
+ logger.error("❌ 本周期处理异常: %s", e, exc_info=True)
|
|
|
+ await asyncio.sleep(POLL_INTERVAL)
|
|
|
+ except KeyboardInterrupt:
|
|
|
+ logger.info("🛑 收到中断信号，正在关闭...")
|
|
|
+ except Exception as e:
|
|
|
+ logger.error("❌ 服务异常: %s", e, exc_info=True)
|
|
|
+ sys.exit(1)
|
|
|
+ finally:
|
|
|
+ await db_manager.disconnect()
|
|
|
+ logger.info("✅ 服务已关闭")
|
|
|
+
|
|
|
+
|
|
|
+def setup_logging() -> None:
|
|
|
+ """日志：正常 -> send_new_user_email.log，错误 -> send_new_user_email_error.log"""
|
|
|
+ log_dir = Path(__file__).parent
|
|
|
+ log_file = log_dir / "send_new_user_email.log"
|
|
|
+ error_log_file = log_dir / "send_new_user_email_error.log"
|
|
|
+ formatter = logging.Formatter(
|
|
|
+ "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
|
+ datefmt="%Y-%m-%d %H:%M:%S",
|
|
|
+ )
|
|
|
+ root = logging.getLogger()
|
|
|
+ root.setLevel(logging.INFO)
|
|
|
+ root.handlers.clear()
|
|
|
+ console = logging.StreamHandler(sys.stdout)
|
|
|
+ console.setLevel(logging.INFO)
|
|
|
+ console.setFormatter(formatter)
|
|
|
+ root.addHandler(console)
|
|
|
+ fh = logging.FileHandler(log_file, encoding="utf-8")
|
|
|
+ fh.setLevel(logging.INFO)
|
|
|
+ fh.setFormatter(formatter)
|
|
|
+ root.addHandler(fh)
|
|
|
+ efh = logging.FileHandler(error_log_file, encoding="utf-8")
|
|
|
+ efh.setLevel(logging.ERROR)
|
|
|
+ efh.setFormatter(formatter)
|
|
|
+ root.addHandler(efh)
|
|
|
+ logger.info("✅ 日志已配置: %s, %s", log_file, error_log_file)
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ setup_logging()
|
|
|
+ asyncio.run(main_loop())
|
