Tolan/sendEmailAsFeedBack/send_feedback_email.py

"""用户反馈邮件通知服务

每 N 分钟扫描 feedbacks 表，将本周期内新增的用户反馈信息汇总后，通过邮件发送给管理员。

Prerequisites:
- Python >= 3.7
- 配置文件：send_feedback.conf（与脚本同目录）
  - [smtp] 发件配置
  - [database] 数据库配置
  - [admin] admin_emails 收件人，多个逗号分隔

运行: python send_feedback_email.py
"""

from __future__ import annotations

import asyncio
import configparser
import logging
import os
import re
import smtplib
import sys
from datetime import datetime, timedelta
from email.header import Header
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import formataddr
from pathlib import Path
from typing import Optional

from sqlalchemy import select

config_file = Path(__file__).parent / "send_feedback.conf"
config = configparser.ConfigParser()
if config_file.exists():
    config.read(config_file, encoding="utf-8")
else:
    raise FileNotFoundError(f"配置文件不存在: {config_file}")

# backend 目录加入 sys.path，复用项目内 DatabaseManager 与模型定义
backend_dir = Path(__file__).parent.parent / "backend"
sys.path.insert(0, str(backend_dir))
from database import DatabaseManager  # type: ignore  # noqa: E402
from models_db import User, Feedback  # type: ignore  # noqa: E402

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()
DEFAULT_EMAIL_SUBJECT = "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="60"))
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 get_email_subject_from_config() -> str:
    """
    从配置读取邮件主题。
    兼容历史误配置：若仍包含“新用户注册”，自动替换为“用户反馈”语义。
    """
    raw = config.get("smtp", "email_subject", fallback=DEFAULT_EMAIL_SUBJECT).strip()
    if not raw:
        return DEFAULT_EMAIL_SUBJECT
    # 兼容旧模板/误配置
    if "新用户注册" in raw:
        return raw.replace("新用户注册", "用户反馈")
    return raw


def format_feedback_line(idx: int, fb: Feedback, user: Optional[User]) -> str:
    """将单条反馈+用户信息格式化为一行中文说明"""
    created = fb.created_at.strftime("%Y-%m-%d %H:%M:%S") if fb.created_at else "-"
    username = user.username if user and getattr(user, "username", None) else "-"
    email = user.email if user and getattr(user, "email", None) else "-"
    phone = user.phone if user and getattr(user, "phone", None) else "-"
    parts = [
        f"反馈{idx}：",
        f"反馈ID {fb.feedback_id}",
        f"用户ID {fb.user_id}",
        f"用户名 {username}",
        f"邮箱 {email or '-'}",
        f"手机 {phone or '-'}",
        f"状态 {fb.status}",
        f"创建时间 {created}",
        f"内容：{fb.content.strip() if fb.content else '-'}",
    ]
    return "，".join(parts)


def build_body(
    feedback_items: list[tuple[Feedback, Optional[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: list[str] = [
        "【Lyrria 用户反馈通知】",
        "",
        f"统计周期：{start_str} 至 {end_str}",
        f"本周期新增用户反馈数：{len(feedback_items)}",
        "",
    ]
    if not feedback_items:
        lines.append("本周期暂无新的用户反馈。")
        return "\n".join(lines)

    for idx, (fb, user) in enumerate(feedback_items, 1):
        lines.append(format_feedback_line(idx, fb, user))

    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_feedbacks(
    db_manager: DatabaseManager,
    period_start: datetime,
    period_end: datetime,
) -> list[tuple[Feedback, Optional[User]]]:
    """查询 (period_start, period_end] 内创建的反馈，附带用户信息"""
    async with db_manager.get_session() as session:
        result = await session.execute(
            select(Feedback, User)
            .join(User, Feedback.user_id == User.user_id, isouter=True)
            .where(
                Feedback.created_at > period_start,
                Feedback.created_at <= period_end,
            )
            .order_by(Feedback.created_at.asc())
        )
        rows = list(result.all())
        # SQLAlchemy 返回 Row 对象，转换为 (Feedback, User|None)
        feedback_items: list[tuple[Feedback, Optional[User]]] = []
        for row in rows:
            # row 可能是 (Feedback, User) 或具有属性 .Feedback / .User
            if isinstance(row, tuple) and len(row) >= 2:
                fb, user = row[0], row[1]
            else:
                fb = getattr(row, "Feedback", None)
                user = getattr(row, "User", None)
            if fb is None:
                continue
            feedback_items.append((fb, user))
        return feedback_items


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)

    feedback_items = await fetch_new_feedbacks(db_manager, period_start, period_end)
    # 无新反馈时不发邮件，仅更新扫描时间
    if not feedback_items:
        _write_last_scan_end(period_end)
        logger.info("本周期无新用户反馈，跳过发邮件，已更新 last_scan_end")
        return

    body = build_body(feedback_items, period_start, period_end)
    logger.info("📋 本周期新增用户反馈数: %d，正在发送通知邮件", len(feedback_items))
    subject = get_email_subject_from_config()
    ok, _ = await asyncio.to_thread(
        send_email_sync,
        ADMIN_EMAILS,
        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_feedback_email.log，错误 -> send_feedback_email_error.log"""
    log_dir = Path(__file__).parent
    log_file = log_dir / "send_feedback_email.log"
    error_log_file = log_dir / "send_feedback_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())