Tolan/sendEmailAsNewUser/send_new_user_email.py

"""新用户注册邮件通知服务

每 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] 内创建的非 guest 用户"""
    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,
            )
            .where(User.user_type != "guest")
            .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())