webapp/app/services/email_authorizations_service.py

import threading
import socket
import socks
import imaplib
import smtplib
import email
import asyncio
import re
import time
from datetime import datetime, timedelta, timezone
from email.message import EmailMessage
from email.header import decode_header
from typing import List, Optional

from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, text
from starlette.concurrency import run_in_threadpool

from app.core.logger import logger
from app.core.biz_exception import NotFoundError, BizLogicError
from app.models.email_authorizations import EmailAuthorization
from app.schemas.email_authorizations import EmailAuthorizationCreate, EmailAuthorizationUpdate

# 保持锁逻辑不变
_PROXY_LOCK = threading.Lock()

class EmailAuthorizationService:
    
    DEFAULT_READ_TOP_N_EMAIL = 10       
    RETRY_DELAY_SECONDS = 5             
    
    # =================================================================
    # 数据库操作 (DB CRUD) - 使用 AsyncSession
    # =================================================================

    @staticmethod
    async def get_all(db: AsyncSession) -> List[EmailAuthorization]:
        # AsyncSession 不支持 db.query，需要用 select(Model)
        stmt = select(EmailAuthorization).order_by(EmailAuthorization.id.desc())
        result = await db.execute(stmt)
        return result.scalars().all()

    @staticmethod
    async def get_by_id(db: AsyncSession, id: int) -> Optional[EmailAuthorization]:
        stmt = select(EmailAuthorization).where(EmailAuthorization.id == id)
        result = await db.execute(stmt)
        # scalar_one_or_none 类似于 .first()，但更严格（如果有多个会报错，这里id是主键所以没问题）
        obj = result.scalar_one_or_none()
        if not obj:
            raise NotFoundError("Email authorization not found")
        return obj
        
    @staticmethod
    async def get_by_email(db: AsyncSession, email: str) -> Optional[EmailAuthorization]:
        stmt = select(EmailAuthorization).where(EmailAuthorization.email == email)
        result = await db.execute(stmt)
        obj = result.scalar_one_or_none()
        if not obj:
            raise NotFoundError("Email authorization not found")
        return obj
        
    @staticmethod
    async def create(db: AsyncSession, obj_in: EmailAuthorizationCreate) -> EmailAuthorization:
        # 先检查是否存在
        stmt = select(EmailAuthorization).where(EmailAuthorization.email == obj_in.email)
        result = await db.execute(stmt)
        if result.scalar_one_or_none():
            raise BizLogicError(f"Email {obj_in.email} already exist")
        
        # 创建对象
        db_obj = EmailAuthorization(**obj_in.dict(exclude_unset=True))
        
        # db.add 是同步方法（只是添加到 session 上下文）
        db.add(db_obj)
        
        # commit 和 refresh 是异步的
        await db.commit()
        await db.refresh(db_obj)
        return db_obj

    @staticmethod
    async def update(db: AsyncSession, id: int, obj_in: EmailAuthorizationUpdate) -> Optional[EmailAuthorization]:
        stmt = select(EmailAuthorization).where(EmailAuthorization.id == id)
        result = await db.execute(stmt)
        db_obj = result.scalar_one_or_none()
        
        if not db_obj:
            raise NotFoundError("Email authorization not found")
            
        for field, value in obj_in.dict(exclude_unset=True).items():
            setattr(db_obj, field, value)
            
        db.add(db_obj)
        await db.commit()
        await db.refresh(db_obj)
        return db_obj

    @staticmethod
    async def delete(db: AsyncSession, id: int) -> Optional[EmailAuthorization]:
        stmt = select(EmailAuthorization).where(EmailAuthorization.id == id)
        result = await db.execute(stmt)
        db_obj = result.scalar_one_or_none()
        
        if not db_obj:
            raise NotFoundError("Email authorization not found")
            
        # delete 也是同步标记
        await db.delete(db_obj)
        await db.commit()
        return db_obj

    @staticmethod
    def _connect_imap_with_proxy(
        host: str,
        port: int,
        proxy_host: Optional[str] = None,
        proxy_port: Optional[int] = None,
        proxy_user: Optional[str] = None,
        proxy_password: Optional[str] = None,
    ) -> imaplib.IMAP4_SSL:
        """
        创建连接 (同步方法，将在线程中运行)
        使用 Lock 确保 socket patching 不会影响其他并发请求
        """
        if proxy_host and proxy_port and proxy_port > 0:
            with _PROXY_LOCK:  # 加锁，防止多线程同时修改全局 socket
                original_socket = socket.socket
                socks.setdefaultproxy(
                    proxy_type=socks.SOCKS5,
                    addr=proxy_host,
                    port=proxy_port,
                    username=proxy_user or None,
                    password=proxy_password or None,
                )
                socket.socket = socks.socksocket
                try:
                    imap = imaplib.IMAP4_SSL(host, port)
                finally:
                    socket.socket = original_socket
        else:
            imap = imaplib.IMAP4_SSL(host, port)
        return imap
    
    @staticmethod
    def _connect_smtp_with_proxy(
        host: str,
        port: int,
        proxy_host: Optional[str] = None,
        proxy_port: Optional[int] = None,
        proxy_user: Optional[str] = None,
        proxy_password: Optional[str] = None,
    ) -> smtplib.SMTP_SSL:
        """
        创建连接 (同步方法，将在线程中运行)
        """
        if proxy_host and proxy_port and proxy_port > 0:
            with _PROXY_LOCK: # 加锁
                original_socket = socket.socket
                socks.setdefaultproxy(
                    proxy_type=socks.SOCKS5,
                    addr=proxy_host,
                    port=proxy_port,
                    username=proxy_user or None,
                    password=proxy_password or None,
                )
                socket.socket = socks.socksocket
                try:
                    smtp = smtplib.SMTP_SSL(host, port)
                finally:
                    socket.socket = original_socket
        else:
            smtp = smtplib.SMTP_SSL(host, port)
        return smtp
    
    @staticmethod
    async def fetch_email_authorizations2(
            db: Session,
            auth,
            sender: str,
            recipient: str,
            subject_keywords: str,
            body_keywords: str
    ):
        # =========================================================
        # 第一步：在数据库中查找最新的 UID (主线程/DB线程执行)
        # =========================================================
        
        # 1. 构建动态 SQL
        # 假设表名为 emails，字段为 uid, sender, recipient, subject, body_text
        sql = "SELECT uid, subject, body_text FROM emails WHERE 1=1"
        params = {}

        # 2. 处理发件人 (模糊匹配)
        if sender.strip():
            sql += " AND sender LIKE :sender"
            params['sender'] = f"%{sender.strip()}%"

        # 3. 处理收件人 (模糊匹配)
        if recipient.strip():
            sql += " AND recipient LIKE :recipient"
            params['recipient'] = f"%{recipient.strip()}%"

        # 4. 处理主题关键词 (OR 关系)
        subj_keys = [k.strip() for k in subject_keywords.split(',') if k.strip()]
        if subj_keys:
            for i, k in enumerate(subj_keys):
                key_name = f"subj_{i}"
                # 直接拼接到主 SQL 中，要求同时满足
                sql += f" AND subject LIKE :{key_name}"
                params[key_name] = f"%{k}%"

        # 5. 处理内容关键词 (OR 关系)
        body_keys = [k.strip() for k in body_keywords.split(',') if k.strip()]
        if body_keys:
            for i, k in enumerate(body_keys):
                key_name = f"body_{i}"
                # 直接拼接到主 SQL 中，要求同时满足
                sql += f" AND body_text LIKE :{key_name}"
                params[key_name] = f"%{k}%"

        # 6. 获取最新的一条
        sql += " ORDER BY uid DESC LIMIT 1"
 
        # 执行查询
        result_proxy = await db.execute(text(sql), params)
        result = result_proxy.fetchone()
        
        if not result:
            logger.info(f"DB Search: No email found for {sender} -> {recipient}")
            return None
            
        target_uid = result.uid
        target_subject = result.subject
        target_body_text = result.body_text
        logger.info(f"DB Search: Found UID {target_uid} matching criteria. Subject: {target_subject}")
        return f'{target_subject}\n{target_body_text}'

    @staticmethod
    async def fetch_email_authorizations(
        auth,
        sender: str,
        recipient: str,
        subject_keywords: str,
        body_keywords: str,
        sent_date: str,
        expiry: int = 300,
        only_text: bool = True
    ) -> Optional[str]:
        """
        在有效期内循环读取邮箱，找到符合条件的邮件（使用最后一条 Received 头作为收件时间）
        """
        def _worker():
            EMAIL_ACCOUNT = auth.email
            EMAIL_PASSWORD = auth.authorization_code
            IMAP_SERVER = auth.imap_server
            IMAP_PORT = auth.imap_port

            subject_keys = [s.strip() for s in subject_keywords.split(",") if s.strip()]
            body_keys = [s.strip() for s in body_keywords.split(",") if s.strip()]

            # === 时间计算 ===
            sent_dt = datetime.strptime(sent_date, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
            max_wait_time = min(5 * 60, expiry)   # 最长等待5分钟
            expiry_at = time.time() + max_wait_time

            def get_received_time(msg):
                """
                使用最后一条 Received 头解析收件时间
                """
                received_headers = msg.get_all("Received", [])
                if not received_headers:
                    return None
                for i, header in enumerate(received_headers, 1):
                    logger.debug(f"  [{i}] {header}")
                last_received = received_headers[-1]
                if ";" not in last_received:
                    return None
                time_str = last_received.split(";")[-1].strip()
                dt_tuple = email.utils.parsedate_tz(time_str)
                if not dt_tuple:
                    return None
                return datetime.fromtimestamp(email.utils.mktime_tz(dt_tuple), tz=timezone.utc)

            mail = EmailAuthorizationService._connect_imap_with_proxy(
                IMAP_SERVER,
                IMAP_PORT,
                auth.proxy_host,
                auth.proxy_port,
                auth.proxy_username,
                auth.proxy_password,
            )
            mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
            mail.select("INBOX")

            while time.time() < expiry_at:
                mail.noop()  # 刷新邮箱状态
                _, data = mail.search(None, "ALL")
                mail_ids = data[0].split()
                if not mail_ids:
                    time.sleep(EmailAuthorizationService.RETRY_DELAY_SECONDS)
                    continue

                recent_ids = mail_ids[-EmailAuthorizationService.DEFAULT_READ_TOP_N_EMAIL:]
                messages = []
                debug = True

                for email_id in reversed(recent_ids):
                    res, msg_data = mail.fetch(email_id, "(RFC822)")
                    if res != "OK" or not msg_data:
                        if debug:
                            logger.debug(f"[WARN] 邮件 ID={email_id.decode()} 获取失败")
                        continue

                    msg_bytes = None
                    for part in msg_data:
                        if isinstance(part, tuple):
                            msg_bytes = part[1]

                    if not msg_bytes:
                        if debug:
                            logger.debug(f"[WARN] 邮件 ID={email_id.decode()} 无正文")
                        continue

                    msg = email.message_from_bytes(msg_bytes)
                    received_dt = get_received_time(msg)
                    if not received_dt:
                        if debug:
                            logger.debug(f"[WARN] 邮件 ID={email_id.decode()} 未解析出 Received 时间")
                        continue

                    messages.append((msg, received_dt))

                if debug:
                    logger.debug(f"[DEBUG] 成功解析邮件数: {len(messages)}")
                    logger.debug(f"[DEBUG] 收件时间列表: {[m[1] for m in messages]}")

                # 按收件时间降序排序
                messages.sort(key=lambda x: x[1], reverse=True)

                for msg, received_dt in messages:
                    # 判断是否在发送时间后的有效窗口内
                    if received_dt < sent_dt:
                        if debug:
                            logger.debug(f"[INFO] 邮件太旧: {received_dt}")
                        continue
                    if received_dt > sent_dt + timedelta(seconds=expiry):
                        if debug:
                            logger.debug(f"[INFO] 邮件太新: {received_dt}")
                        continue

                    # 匹配发件人/收件人
                    msg_from = msg.get("From", "")
                    msg_to = msg.get("To", "")
                    if sender.lower() not in msg_from.lower():
                        if debug:
                            logger.debug("发件人不匹配")
                        continue
                    if recipient.lower() not in msg_to.lower():
                        if debug:
                            logger.debug("收件人不匹配")
                        continue

                    # 匹配主题
                    subject, enc = decode_header(msg.get("Subject"))[0]
                    if isinstance(subject, bytes):
                        subject = subject.decode(enc or "utf-8", errors="ignore")
                    if subject_keys and not any(k.lower() in subject.lower() for k in subject_keys):
                        continue

                    # 提取正文
                    body = EmailAuthorizationService._extract_body(msg, only_text)
                    if body_keys and not any(k.lower() in body.lower() for k in body_keys):
                        continue

                    # 找到匹配邮件 → 返回内容
                    mail.close()
                    mail.logout()
                    return body.strip()

                # 未匹配到 → 等待重试
                time.sleep(EmailAuthorizationService.RETRY_DELAY_SECONDS)

            mail.close()
            mail.logout()
            raise NotFoundError("Get email timeout")
        
        return await run_in_threadpool(_worker)
        

    @staticmethod
    def _process_recent_emails_sync(
        mail, recent_ids, sent_dt, expiry, sender, recipient, subject_keys, body_keys, only_text
    ) -> Optional[str]:
        """
        同步辅助函数：处理邮件解析逻辑。
        将这段繁琐的逻辑放入线程运行，避免阻塞 Async Loop。
        """
        messages = []
        debug = True

        for email_id in reversed(recent_ids):
            res, msg_data = mail.fetch(email_id, "(RFC822)")
            if res != "OK" or not msg_data:
                continue

            msg_bytes = None
            for part in msg_data:
                if isinstance(part, tuple):
                    msg_bytes = part[1]

            if not msg_bytes:
                continue

            msg = email.message_from_bytes(msg_bytes)
            
            # 解析时间
            received_dt = None
            received_headers = msg.get_all("Received", [])
            if received_headers:
                last_received = received_headers[-1]
                if ";" in last_received:
                    time_str = last_received.split(";")[-1].strip()
                    dt_tuple = email.utils.parsedate_tz(time_str)
                    if dt_tuple:
                        received_dt = datetime.fromtimestamp(email.utils.mktime_tz(dt_tuple), tz=timezone.utc)

            if not received_dt:
                continue

            messages.append((msg, received_dt))

        # 排序
        messages.sort(key=lambda x: x[1], reverse=True)

        for msg, received_dt in messages:
            # 时间判定
            if received_dt < sent_dt:
                continue
            if received_dt > sent_dt + timedelta(seconds=expiry):
                continue

            # 匹配逻辑
            msg_from = msg.get("From", "")
            msg_to = msg.get("To", "")
            
            if sender.lower() not in msg_from.lower():
                continue
            if recipient.lower() not in msg_to.lower():
                continue

            subject_raw = msg.get("Subject")
            subject = ""
            if subject_raw:
                decoded_list = decode_header(subject_raw)
                if decoded_list:
                    sub_bytes, enc = decoded_list[0]
                    if isinstance(sub_bytes, bytes):
                        subject = sub_bytes.decode(enc or "utf-8", errors="ignore")
                    else:
                        subject = str(sub_bytes)
            
            if subject_keys and not any(k.lower() in subject.lower() for k in subject_keys):
                continue

            body = EmailAuthorizationService._extract_body(msg, only_text)
            if body_keys and not any(k.lower() in body.lower() for k in body_keys):
                continue

            return body.strip()
        
        return None

    @staticmethod
    async def fetch_email_authorizations_from_top_n(
        auth,
        sender: str,
        recipient: str,
        subject_keywords: str,
        body_keywords: str,
        top: int = 10,
        only_text: bool = True
    ) -> Optional[str]:
        
        # 定义一个纯同步的 worker 函数来执行所有 IMAP 逻辑
        def _worker():
            subject_keys = [s.strip() for s in subject_keywords.split(",") if s.strip()]
            body_keys = [s.strip() for s in body_keywords.split(",") if s.strip()]

            mail = EmailAuthorizationService._connect_imap_with_proxy(
                auth.imap_server,
                auth.imap_port,
                auth.proxy_host,
                auth.proxy_port,
                auth.proxy_username,
                auth.proxy_password,
            )
            try:
                mail.login(auth.email, auth.authorization_code)
                mail.select("INBOX")
                _, data = mail.search(None, "ALL")
                mail_ids = data[0].split()
                if not mail_ids:
                    return None

                recent_ids = mail_ids[-top:]
                
                # 复用上面的解析逻辑，但稍作调整，因为这个方法不需要时间过滤
                # 这里为了简单，直接写精简版解析
                for email_id in reversed(recent_ids):
                    res, msg_data = mail.fetch(email_id, "(RFC822)")
                    if res != "OK" or not msg_data: continue

                    msg_bytes = None
                    for part in msg_data:
                        if isinstance(part, tuple): msg_bytes = part[1]
                    
                    if not msg_bytes: continue
                    msg = email.message_from_bytes(msg_bytes)

                    # 匹配逻辑
                    msg_from = msg.get("From", "")
                    msg_to = msg.get("To", "")
                    if sender.lower() not in msg_from.lower(): continue
                    if recipient.lower() not in msg_to.lower(): continue

                    subject_raw = msg.get("Subject")
                    subject = ""
                    if subject_raw:
                        d = decode_header(subject_raw)[0]
                        subject = d[0].decode(d[1] or "utf-8", errors="ignore") if isinstance(d[0], bytes) else str(d[0])

                    if subject_keys and not any(k.lower() in subject.lower() for k in subject_keys): continue
                    
                    body = EmailAuthorizationService._extract_body(msg, only_text)
                    if body_keys and not any(k.lower() in body.lower() for k in body_keys): continue
                    
                    return body.strip()
                
                return None
            finally:
                try:
                    mail.close()
                    mail.logout()
                except:
                    pass

        return await run_in_threadpool(_worker)

    @staticmethod
    async def forward_first_matching_email(
            auth,
            forward_to: str,
            sender: str,
            recipient: str,
            subject_keywords: str,
            body_keywords: str
    ):
        # 整个逻辑比较复杂且都是 IO，直接打包扔进线程池
        def _worker():
            subject_keys = [s.strip() for s in subject_keywords.split(",") if s.strip()]
            body_keys = [s.strip() for s in body_keywords.split(",") if s.strip()]
            
            mail = EmailAuthorizationService._connect_imap_with_proxy(
                auth.imap_server, auth.imap_port, auth.proxy_host, auth.proxy_port, auth.proxy_username, auth.proxy_password
            )
            try:
                mail.login(auth.email, auth.authorization_code)
                mail.select("INBOX")
                
                target = recipient
                # 注意：IMAP search 语法对引号很敏感，确保 target 没有特殊字符破坏命令
                query = f'(HEADER To "{target}")' 
                res, data = mail.uid("search", None, query)
                if res != "OK": return None
                
                uids = data[0].split()
                msgs = []
                for uid in uids:
                    res, msg_data = mail.uid("fetch", uid, "(RFC822)")
                    if res != "OK": continue
                    msg = email.message_from_bytes(msg_data[0][1])
                    # 处理 Date 头可能缺失的情况
                    date_str = msg.get("Date")
                    if date_str:
                        date_ = email.utils.parsedate_to_datetime(date_str)
                        msgs.append((date_, msg))
                
                msgs.sort(key=lambda x: x[0], reverse=True)

                for _, msg in msgs:
                    msg_from = msg.get("From", "")
                    if sender.lower() not in msg_from.lower(): continue
                    
                    subject_raw = msg.get("Subject", "")
                    subject = ""
                    d = decode_header(subject_raw)[0]
                    subject = d[0].decode(d[1] or "utf-8", errors="ignore") if isinstance(d[0], bytes) else str(d[0])
                    
                    if subject_keys and not any(k.lower() in subject.lower() for k in subject_keys): continue
                    
                    body = EmailAuthorizationService._extract_body(msg, True)
                    if body_keys and not any(k.lower() in body.lower() for k in body_keys): continue

                    # 准备转发
                    # 注意：直接修改 Header 转发可能会破坏 DKIM 签名，更好的方式是作为附件转发
                    # 但这里保持原逻辑
                    del msg['From']
                    del msg['To']
                    del msg['Subject']
                    msg['From'] = auth.email
                    msg['To'] = forward_to
                    msg['Subject'] = f"FWD: {subject}"
                    
                    # 调用同步的 send 方法
                    EmailAuthorizationService.send_email_smtp(auth, msg)
                    
                    return f"邮件 '{subject}' 已成功转发至: {forward_to}"
                return None
            finally:
                try:
                    mail.logout()
                except: pass

        return await run_in_threadpool(_worker)
    
    @staticmethod
    async def forward_first_matching_email2(
            db: Session,
            auth,
            forward_to: str,
            sender: str,
            recipient: str,
            subject_keywords: str,
            body_keywords: str
    ):
        # =========================================================
        # 第一步：在数据库中查找最新的 UID (主线程/DB线程执行)
        # =========================================================
        
        # 1. 构建动态 SQL
        # 假设表名为 emails，字段为 uid, sender, recipient, subject, body_text
        sql = "SELECT uid, subject FROM emails WHERE 1=1"
        params = {}

        # 2. 处理发件人 (模糊匹配)
        if sender.strip():
            sql += " AND sender LIKE :sender"
            params['sender'] = f"%{sender.strip()}%"

        # 3. 处理收件人 (模糊匹配)
        if recipient.strip():
            sql += " AND recipient LIKE :recipient"
            params['recipient'] = f"%{recipient.strip()}%"

        # 4. 处理主题关键词 (OR 关系)
        subj_keys = [k.strip() for k in subject_keywords.split(',') if k.strip()]
        if subj_keys:
            for i, k in enumerate(subj_keys):
                key_name = f"subj_{i}"
                # 直接拼接到主 SQL 中，要求同时满足
                sql += f" AND subject LIKE :{key_name}"
                params[key_name] = f"%{k}%"

        # 5. 处理内容关键词 (OR 关系)
        body_keys = [k.strip() for k in body_keywords.split(',') if k.strip()]
        if body_keys:
            for i, k in enumerate(body_keys):
                key_name = f"body_{i}"
                # 直接拼接到主 SQL 中，要求同时满足
                sql += f" AND body_text LIKE :{key_name}"
                params[key_name] = f"%{k}%"

        # 6. 获取最新的一条
        sql += " ORDER BY uid DESC LIMIT 1"

        try:
            # 执行查询
            result_proxy = await db.execute(text(sql), params)
            result = result_proxy.fetchone()
            
            if not result:
                logger.info(f"DB Search: No email found for {sender} -> {recipient}")
                return None
                
            target_uid = result.uid
            target_subject = result.subject
            logger.info(f"DB Search: Found UID {target_uid} matching criteria. Subject: {target_subject}")
            
        except Exception as e:
            logger.error(f"DB Search Error: {e}")
            return f"数据库查询失败: {str(e)}"

        # =========================================================
        # 第二步：去 IMAP 拉取原始内容并转发 (放入线程池执行 IO 操作)
        # =========================================================
        
        def _worker():
            mail = None
            try:
                # 1. 连接 IMAP
                mail = EmailAuthorizationService._connect_imap_with_proxy(
                    auth.imap_server, auth.imap_port, 
                    auth.proxy_host, auth.proxy_port, 
                    auth.proxy_username, auth.proxy_password
                )
                mail.login(auth.email, auth.authorization_code)
                mail.select("INBOX")
                
                # 2. 根据 UID 精准拉取 (使用 fetch)
                # 注意：IMAPClient 的 fetch 方法
                # UID 必须转为 int 或者 sequence set 字符串
                res, data = mail.uid('fetch', str(target_uid), '(RFC822)')
                
                # 🔴 修正点：不要写 if target_uid in res
                # res 是状态字符串 "OK"，data 是包含邮件内容的列表
                if res != 'OK':
                    return f"IMAP Fetch 失败，状态码: {res}"
                
                if not data or not data[0]:
                    return f"未找到 UID {target_uid} 的邮件内容 (可能已被物理删除)"

                # data[0] 通常是 tuple (byte_header, byte_content)，但也可能是 None
                if isinstance(data[0], tuple):
                    raw_email_bytes = data[0][1]
                else:
                    # 如果 data[0] 只是 bytes (例如 b')')，说明没拿到邮件体
                    return f"邮件数据格式异常，无法解析: {str(data)}"

                msg = email.message_from_bytes(raw_email_bytes)

                # 3. 处理转发逻辑
                # 使用数据库中查到的标题，确保标题准确
                subject = target_subject 

                # 修改 Header 进行转发
                del msg['From']
                del msg['To']
                del msg['Cc']
                del msg['Subject']
                
                msg['From'] = auth.email  # 发件人覆写为当前授权账号
                msg['To'] = forward_to
                msg['Subject'] = f"FWD: {subject}"
                
                # 4. 发送邮件 (SMTP)
                EmailAuthorizationService.send_email_smtp(auth, msg)
                
                return f"邮件 '{subject}' (UID: {target_uid}) 已成功转发至: {forward_to}"
                
            except Exception as e:
                logger.error(f"IMAP Forward Error: {e}")
                return f"邮件转发过程出错: {str(e)}"
            finally:
                if mail:
                    try:
                        mail.logout()
                    except: pass

        # 在线程池中运行耗时 IO 操作
        return await run_in_threadpool(_worker)

    @staticmethod
    async def send_email(
            auth,
            send_to: str,
            subject: str,
            content_type: str,
            content: str
    ):
        def _worker():
            msg = EmailMessage()
            msg["From"] = auth.email
            msg["To"] = send_to
            msg["Subject"] = subject
            
            if content_type.lower() == "html":
                msg.set_content("") # 占位
                msg.add_alternative(content, subtype="html")
            else:
                msg.set_content(content)

            EmailAuthorizationService.send_email_smtp(auth, msg)
            return f"邮件 '{subject}' 成功发送至: {send_to}"

        return await run_in_threadpool(_worker)

    @staticmethod
    async def send_email_bulk(
            auth,
            send_to: str,
            subject: str,
            content_type: str,
            content: str
    ):
        def _worker():
            bcc_list = [s.strip() for s in send_to.split(",") if s.strip()]
            msg = EmailMessage()
            msg["From"] = auth.email
            msg["To"] = bcc_list[0] if bcc_list else auth.email # Fallback
            msg["Subject"] = subject
            
            if content_type.lower() == "html":
                msg.set_content("")
                msg.add_alternative(content, subtype="html")
            else:
                msg.set_content(content)
            
            EmailAuthorizationService.send_email_smtp(auth, msg, bcc_list=bcc_list)
            return f"邮件 '{subject}' 成功发送至: {send_to}"

        return await run_in_threadpool(_worker)

    # ----------------------------------------------------------------------
    # 底层 SMTP 发送 (保持同步，供 Worker 调用)
    # ----------------------------------------------------------------------
    @staticmethod
    def send_email_smtp(auth, msg, bcc_list=None):
        if bcc_list is None:
            bcc_list = []
        
        # 这里的 connect 内部已经加了锁，是安全的
        mail = EmailAuthorizationService._connect_smtp_with_proxy(
            auth.smtp_server,
            auth.smtp_port,
            auth.proxy_host,
            auth.proxy_port,
            auth.proxy_username,
            auth.proxy_password,
        )
        try:
            mail.login(auth.email, auth.authorization_code)
            if bcc_list:
                mail.send_message(msg, to_addrs=bcc_list)
            else:
                mail.send_message(msg)
        finally:
            mail.quit()

    @staticmethod
    def _extract_body(msg, only_text: bool = True) -> str:
        # 纯 CPU 计算，不需要 async，保留原样
        body_parts = []
        if msg.is_multipart():
            for part in msg.walk():
                ctype = part.get_content_type()
                if only_text and ctype != "text/plain":
                    continue
                if not only_text and ctype not in ["text/plain", "text/html"]:
                    continue
                if part.get("Content-Disposition"):
                    continue
                charset = part.get_content_charset() or "utf-8"
                try:
                    payload = part.get_payload(decode=True)
                    if payload:
                        text = payload.decode(charset, errors="ignore")
                        body_parts.append(text)
                except Exception:
                    continue
        else:
            charset = msg.get_content_charset() or "utf-8"
            payload = msg.get_payload(decode=True)
            if payload:
                body_parts.append(payload.decode(charset, errors="ignore"))
        
        body = "\n".join(body_parts)
        return re.sub(r"\s+", " ", body.strip())