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
import email.policy
from email.message import EmailMessage
from email.utils import formatdate, make_msgid
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
    ):
        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")
                
                # 1. 搜索目标邮件
                target = recipient
                query = f'(HEADER To "{target}")' 
                res, data = mail.uid("search", None, query)
                if res != "OK": return None
                
                uids = data[0].split()
                msgs_to_check = []
                for uid in uids:
                    # 使用 RFC822 获取完整内容
                    res, msg_data = mail.uid("fetch", uid, "(RFC822)")
                    if res != "OK" or not msg_data: continue
                    
                    # 临时解析用于排序和初步过滤
                    raw_bytes = msg_data[0][1]
                    tmp_msg = email.message_from_bytes(raw_bytes, policy=email.policy.default)
                    
                    date_str = tmp_msg.get("Date")
                    if date_str:
                        try:
                            date_dt = parsedate_to_datetime(date_str)
                            msgs_to_check.append((date_dt, tmp_msg, raw_bytes))
                        except:
                            continue
                
                # 按时间降序排序（最新的优先）
                msgs_to_check.sort(key=lambda x: x[0], reverse=True)

                for _, orig_msg, raw_bytes in msgs_to_check:
                    # --- 过滤逻辑 ---
                    msg_from = orig_msg.get("From", "")
                    if sender.lower() not in msg_from.lower(): continue
                    
                    subject = orig_msg.get("Subject", "")
                    if subject_keys and not any(k.lower() in subject.lower() for k in subject_keys): continue
                    
                    body_content = EmailAuthorizationService._extract_body(orig_msg, True)
                    if body_keys and not any(k.lower() in body_content.lower() for k in body_keys): continue

                    # --- 匹配成功：开始构造转发邮件 ---
                    
                    # 1. 提取原始信息用于视觉转发头
                    orig_from = orig_msg.get("From", "Unknown")
                    orig_date = orig_msg.get("Date", "Unknown")
                    orig_subject = orig_msg.get("Subject", "No Subject")
                    orig_to = orig_msg.get("To", "Unknown")
                    orig_msg_id = orig_msg.get("Message-ID")

                    fwd_info = (
                        f"\n\n---------- Forwarded message ----------\n"
                        f"From: {orig_from}\n"
                        f"Date: {orig_date}\n"
                        f"Subject: {orig_subject}\n"
                        f"To: {orig_to}\n\n"
                    )

                    # 2. 构造新的邮件对象 (重新基于原始字节解析，确保附件完整)
                    msg = email.message_from_bytes(raw_bytes, policy=email.policy.default)

                    # 3. 清理并重置 Header
                    headers_to_clean = ['From', 'To', 'Cc', 'Bcc', 'Subject', 'Date', 'Message-ID', 'In-Reply-To', 'References']
                    for h in headers_to_clean:
                        del msg[h]
                    
                    msg['From'] = auth.email
                    msg['To'] = forward_to
                    msg['Subject'] = f"Fwd: {orig_subject}"
                    msg['Date'] = formatdate(localtime=True)
                    msg['Message-ID'] = make_msgid(domain=auth.email.split('@')[-1])

                    # 4. 【核心】建立上下文关联 (Threading)
                    if orig_msg_id:
                        msg['In-Reply-To'] = orig_msg_id
                        msg['References'] = orig_msg_id

                    # 5. 【核心】注入视觉转发头 (Visual Prepend)
                    try:
                        if msg.is_multipart():
                            # 遍历部分，找到主要正文并插入
                            for part in msg.walk():
                                ctype = part.get_content_type()
                                if ctype == "text/plain":
                                    part.set_content(fwd_info + part.get_content())
                                    break
                                elif ctype == "text/html":
                                    html_fwd = fwd_info.replace("\n", "<br>")
                                    part.set_content(f"<div>{html_fwd}</div>" + part.get_content(), subtype="html")
                                    break
                        else:
                            msg.set_content(fwd_info + msg.get_content())
                    except Exception as e:
                        logger.warning(f"Prepend visual header failed: {e}")

                    # 6. 发送邮件
                    EmailAuthorizationService.send_email_smtp(auth, msg)
                    
                    # 7. 同步发件记录 (IMAP Sent)
                    EmailAuthorizationService._append_to_sent(auth, msg)
                    
                    return f"邮件 '{orig_subject}' 已成功关联转发至: {forward_to}"
                
                return None
            except Exception as e:
                logger.error(f"Forward matching email error: {e}")
                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)}"

                # 使用 default policy 解析，方便后续修改
                orig_msg = email.message_from_bytes(data[0][1], policy=email.policy.default)

                # --- 1. 提取原始邮件信息用于构造转发头 ---
                orig_from = orig_msg.get("From", "Unknown")
                orig_date = orig_msg.get("Date", "Unknown")
                orig_subject = orig_msg.get("Subject", "No Subject")
                orig_to = orig_msg.get("To", "Unknown")
                orig_msg_id = orig_msg.get("Message-ID")

                # --- 2. 构造视觉上的“转发信息栏” ---
                fwd_header_text = (
                    f"\n\n---------- Forwarded message ----------\n"
                    f"From: {orig_from}\n"
                    f"Date: {orig_date}\n"
                    f"Subject: {orig_subject}\n"
                    f"To: {orig_to}\n\n"
                )

                # --- 3. 构造新的邮件对象 ---
                # 为了保持上下文关联，我们克隆或重新构造，并设置 Threading Headers
                msg = email.message_from_bytes(data[0][1], policy=email.policy.default)

               # 清除旧头
                for h in ['From', 'To', 'Cc', 'Bcc', 'Subject', 'Date', 'Message-ID', 'In-Reply-To', 'References']:
                    del msg[h]
                
                msg['From'] = auth.email
                msg['To'] = forward_to
                msg['Subject'] = f"Fwd: {target_subject}"
                msg['Date'] = formatdate(localtime=True)
                msg['Message-ID'] = make_msgid(domain=auth.email.split('@')[-1])
                
                # --- 4. 关键：建立线索关联 (Threading) ---
                if orig_msg_id:
                    # 这两个头告诉 Gmail 这封信是原邮件的后续
                    msg['In-Reply-To'] = orig_msg_id
                    msg['References'] = orig_msg_id
                    
                # --- 5. 修改正文，注入转发视觉头 ---
                # 处理 Multipart 或简单邮件，将 fwd_header_text 插入到正文最前面
                try:
                    if msg.is_multipart():
                        # 找到第一个文本部分并修改
                        for part in msg.walk():
                            if part.get_content_type() == "text/plain":
                                content = part.get_content()
                                part.set_content(fwd_header_text + content)
                                break
                            elif part.get_content_type() == "text/html":
                                # HTML 转发头稍微复杂点，这里简单处理
                                content = part.get_content()
                                html_fwd = fwd_header_text.replace("\n", "<br>")
                                part.set_content(f"<div>{html_fwd}</div>" + content, subtype="html")
                                break
                    else:
                        content = msg.get_content()
                        msg.set_content(fwd_header_text + content)
                except Exception as e:
                    logger.warning(f"Failed to prepend forward header: {e}")
                
                # 4. 发送邮件 (SMTP)
                EmailAuthorizationService.send_email_smtp(auth, msg)
                
                return f"邮件 '{target_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
    def _append_to_sent(auth, msg: EmailMessage):
        """
        同步发件记录到 IMAP Sent 文件夹
        """
        imap = None
        try:
            # 确保消息包含必要的指纹，否则同步后 Gmail 搜索不到
            if 'Date' not in msg:
                msg["Date"] = formatdate(localtime=True)
            if 'Message-ID' not in msg:
                msg["Message-ID"] = make_msgid(domain=auth.email.split('@')[-1])

            imap = EmailAuthorizationService._connect_imap_with_proxy(
                auth.imap_server, auth.imap_port,
                auth.proxy_host, auth.proxy_port,
                auth.proxy_username, auth.proxy_password,
            )
            imap.login(auth.email, auth.authorization_code)

            # --- 自动探测已发送文件夹 (兼容 Gmail/Outlook/域名邮) ---
            sent_folder = None
            typ, data = imap.list()
            if typ == "OK":
                for entry in data:
                    line = entry.decode()
                    # 寻找包含 \Sent 属性的系统文件夹
                    if '\\Sent' in line:
                        # 兼容各种分隔符，提取最后一个引号内的内容
                        parts = re.findall(r'"([^"]+)"', line)
                        if parts:
                            sent_folder = f'"{parts[-1]}"' # 强制带引号防止空格导致 BAD
                            break
            
            # 兜底逻辑
            if not sent_folder:
                sent_folder = '"[Gmail]/Sent Mail"' if "gmail" in auth.email.lower() else '"Sent"'

            # 执行写入 (使用 \\Seen 标记为已读)
            # imap.append 的参数顺序: 文件夹, 标志, 时间, 内容
            imap.append(
                sent_folder,
                '(\\Seen)',
                imaplib.Time2Internaldate(time.time()),
                msg.as_bytes()
            )
            logger.info(f"Successfully synced to folder: {sent_folder}")

        except Exception as e:
            logger.error(f"Append sent mail failed: {str(e)}")
        finally:
            if imap:
                try: imap.logout()
                except: pass

    @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
            
            msg["Date"] = formatdate(localtime=True)
            msg["Message-ID"] = make_msgid(domain=auth.email.split('@')[-1])
            msg["MIME-Version"] = "1.0"
            msg["X-Mailer"] = "Python-Client-v1.0"
            
            if content_type.lower() == "html":
                msg.set_content("") # 占位
                msg.add_alternative(content, subtype="html")
            else:
                msg.set_content(content)

            # 2. 执行发送
            logger.info(f"[DEBUG] 准备发送邮件: ID={msg['Message-ID']}")
            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)
            recipients = bcc_list if bcc_list else [msg["To"]]
            mail.send_message(
                msg,
                from_addr=auth.email,
                to_addrs=recipients
            )
            
            logger.info(f"[DEBUG] 开始同步到已发送文件夹...")
            EmailAuthorizationService._append_to_sent(auth, 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())