#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 找回密码工单处理脚本 - 按 scan_interval_seconds 扫描 password_reset_requests 表 - 仅处理 created_at > config 中 last_scan_time 的 pending 工单，处理完后将当前时间写回 config.ini - 为每条记录生成随机密码、更新 users 表、发送邮件、更新工单状态 """ import os import sys import time import random import string import configparser import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.header import Header from datetime import datetime, timezone, timedelta from pathlib import Path import bcrypt # 北京时间 UTC+8，与 backend-client 一致 BEIJING_TZ = timezone(timedelta(hours=8)) def beijing_now() -> datetime: """当前北京时间（naive）""" return datetime.now(BEIJING_TZ).replace(tzinfo=None) import pymysql # bcrypt 与 backend-client 保持一致：密码最长 72 字节 BCRYPT_MAX_PASSWORD_BYTES = 72 def load_config(): """从 config.ini 加载配置，返回 (config, config 文件路径)""" conf_path = Path(__file__).resolve().parent / "config.ini" if not conf_path.exists(): print(f"错误: 未找到配置文件 {conf_path}，请复制 config.ini.example 为 config.ini 并填写", file=sys.stderr) sys.exit(1) config = configparser.ConfigParser() config.read(conf_path, encoding="utf-8") return config, conf_path def get_last_scan_time(config) -> datetime: """从配置读取上次扫描时间；为空或无效则返回很早的时间以处理全部 pending""" try: raw = config.get("script", "last_scan_time", fallback="").strip() if not raw: return datetime(2000, 1, 1) return datetime.strptime(raw, "%Y-%m-%d %H:%M:%S") except Exception: return datetime(2000, 1, 1) def save_last_scan_time(conf_path: Path, dt: datetime) -> None: """将上次扫描时间写回 config.ini（只改写 last_scan_time 行，保留其余内容）""" text = conf_path.read_text(encoding="utf-8") new_line = f"last_scan_time = {dt.strftime('%Y-%m-%d %H:%M:%S')}" lines = [] for line in text.splitlines(): if line.strip().startswith("last_scan_time"): lines.append(new_line) else: lines.append(line) conf_path.write_text("\n".join(lines) + "\n", encoding="utf-8") def hash_password(password: str) -> str: """与 backend-client 一致的 bcrypt 哈希""" raw = password.encode("utf-8") if len(raw) > BCRYPT_MAX_PASSWORD_BYTES: raw = raw[:BCRYPT_MAX_PASSWORD_BYTES] return bcrypt.hashpw(raw, bcrypt.gensalt()).decode("utf-8") def generate_random_password(length: int = 12) -> str: """生成随机密码：字母+数字""" chars = string.ascii_letters + string.digits return "".join(random.choices(chars, k=length)) def send_email(config, to_email: str, username: str, new_password: str) -> None: """使用 126 SMTP 发送新密码邮件""" sender = config.get("email", "sender") smtp_host = config.get("email", "smtp_host") smtp_port = config.getint("email", "smtp_port") # 126 要求：登录名须为完整邮箱，密码须为授权码（非登录密码） smtp_user = config.get("email", "smtp_user").strip() if "@" not in smtp_user: smtp_user = sender # 未填完整邮箱时用发件人地址 smtp_password = config.get("email", "smtp_password") subject = "游铛铛 - 找回密码" body = f"""您好 {username}， 您已成功找回密码。新密码如下，请登录后及时修改： 新密码：{new_password} 请妥善保管，勿泄露给他人。 —— 游铛铛 """ msg = MIMEMultipart() msg["From"] = Header(sender, "utf-8") msg["To"] = Header(to_email, "utf-8") msg["Subject"] = Header(subject, "utf-8") msg.attach(MIMEText(body, "plain", "utf-8")) try: with smtplib.SMTP_SSL(smtp_host, smtp_port) as server: server.login(smtp_user, smtp_password) server.sendmail(sender, [to_email], msg.as_string()) except smtplib.SMTPAuthenticationError as e: # 535 多为：用了登录密码而非授权码，或 smtp_user 未填完整邮箱 err = f"{e}".replace("(535, b'Error: authentication failed')", "(535) SMTP 认证失败") raise RuntimeError( f"邮件发送认证失败: {err}\n" "126 邮箱请检查：1) smtp_user 填完整邮箱（如 xxx@126.com）；2) smtp_password 填「授权码」而非登录密码（在 126 设置-POP3/SMTP/IMAP 中开启服务并生成授权码）。" ) from e def get_db_connection(config): """创建数据库连接""" return pymysql.connect( host=config.get("database", "host"), port=config.getint("database", "port"), user=config.get("database", "user"), password=config.get("database", "password"), database=config.get("database", "database"), charset=config.get("database", "charset"), cursorclass=pymysql.cursors.DictCursor, ) def run_once(config, conf_path: Path) -> None: """执行一次扫描：仅处理 created_at > last_scan_time 的 pending 工单，处理完后更新 config 中的 last_scan_time""" last_scan = get_last_scan_time(config) now_dt = beijing_now() now_str = now_dt.strftime("%Y-%m-%d %H:%M:%S") last_scan_str = last_scan.strftime("%Y-%m-%d %H:%M:%S") conn = get_db_connection(config) try: with conn.cursor() as cursor: cursor.execute( "SELECT id, user_id, username, email FROM password_reset_requests " "WHERE status = %s AND created_at > %s ORDER BY id", ("pending", last_scan_str), ) rows = cursor.fetchall() finally: conn.close() if not rows: save_last_scan_time(conf_path, beijing_now()) return conn = get_db_connection(config) now = now_str for row in rows: req_id = row["id"] user_id = row["user_id"] username = row["username"] email = row["email"] new_password = generate_random_password(12) hashed = hash_password(new_password) try: # 标记为处理中 with conn.cursor() as cursor: cursor.execute( "UPDATE password_reset_requests SET status = %s WHERE id = %s", ("processing", req_id), ) conn.commit() # 先写历史表，再改/删：将当前用户行写入 users_his（操作类型 UPDATE），再更新密码 with conn.cursor() as cursor: cursor.execute( "SELECT id, username, password, email, machine_code, card_id, expire_time, " "last_login_time, last_login_ip, status, is_deleted, created_at, updated_at " "FROM users WHERE id = %s", (user_id,), ) user_row = cursor.fetchone() if user_row: cursor.execute( "INSERT INTO users_his (id, username, password, email, machine_code, card_id, " "expire_time, last_login_time, last_login_ip, status, is_deleted, created_at, " "updated_at, his_created_at, his_operation) " "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'UPDATE')", ( user_row["id"], user_row["username"], user_row["password"], user_row["email"], user_row["machine_code"], user_row["card_id"], user_row["expire_time"], user_row["last_login_time"], user_row["last_login_ip"], user_row["status"], user_row["is_deleted"], user_row["created_at"], user_row["updated_at"], now, ), ) conn.commit() # 更新用户密码 with conn.cursor() as cursor: cursor.execute( "UPDATE users SET password = %s, updated_at = %s WHERE id = %s", (hashed, now, user_id), ) conn.commit() # 发邮件 send_email(config, email, username, new_password) # 标记为已发送 with conn.cursor() as cursor: cursor.execute( "UPDATE password_reset_requests SET status = %s, processed_at = %s, error_message = NULL WHERE id = %s", ("sent", now, req_id), ) conn.commit() print(f"[{now}] 工单 {req_id} 已处理并发送邮件至 {email}") except Exception as e: err_msg = str(e)[:500] try: with conn.cursor() as cursor: cursor.execute( "UPDATE password_reset_requests SET status = %s, error_message = %s, processed_at = %s WHERE id = %s", ("failed", err_msg, now, req_id), ) conn.commit() except Exception: conn.rollback() print(f"[{now}] 工单 {req_id} 处理失败: {e}", file=sys.stderr) conn.close() # 本轮回笼结束，将当前时间写回 config.ini，下次只处理此时间之后创建的工单 save_last_scan_time(conf_path, beijing_now()) def main(): config, conf_path = load_config() interval = config.getint("script", "scan_interval_seconds", fallback=60) print(f"找回密码脚本已启动，每 {interval} 秒扫描一次；上次扫描时间与本次处理时间范围从 config.ini 的 last_scan_time 读取/更新") while True: try: run_once(config, conf_path) # 每轮后重新加载 config，以便下次 run_once 读到最新的 last_scan_time（由 run_once 写回） config, conf_path = load_config() except Exception as e: print(f"[{beijing_now()}] 本轮扫描异常: {e}", file=sys.stderr) time.sleep(interval) if __name__ == "__main__": main()