webapp/app/services/email_authorizations_service.py

import re
import time
import socket
import smtplib
import socks
import imaplib
import email
from datetime import datetime, timedelta, timezone
from email.message import EmailMessage
from email.header import decode_header
from sqlalchemy.orm import Session
from typing import List, Optional
from app.core.logger import logger
from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError
from app.models.email_authorizations import EmailAuthorization
from app.schemas.email_authorizations import EmailAuthorizationCreate, EmailAuthorizationUpdate


class EmailAuthorizationService:
    
    DEFAULT_READ_TOP_N_EMAIL = 10       # 最多读取最近多少封邮件
    RETRY_DELAY_SECONDS = 5             # 每次轮询间隔
    
    @staticmethod
    def get_all(db: Session) -> List[EmailAuthorization]:
        return db.query(EmailAuthorization).order_by(EmailAuthorization.id.desc()).all()

    @staticmethod
    def get_by_id(db: Session, id: int) -> Optional[EmailAuthorization]:
        obj = db.query(EmailAuthorization).filter(EmailAuthorization.id == id).first()
        if not obj:
            raise NotFoundError("Email authorization not found")
        return obj
        
    @staticmethod
    def get_by_email(db: Session, email: str) -> Optional[EmailAuthorization]:
        obj = db.query(EmailAuthorization).filter(EmailAuthorization.email == email).first()
        if not obj:
            raise NotFoundError("Email authorization not found")
        return obj
        
    @staticmethod
    def create(db: Session, obj_in: EmailAuthorizationCreate) -> EmailAuthorization:
        existing = EmailAuthorizationService.get_by_email(db, obj_in.email)
        if existing:
            raise BizLogicError(f"Email {obj_in.email} already exist")
        db_obj = EmailAuthorization(**obj_in.dict(exclude_unset=True))
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    @staticmethod
    def update(db: Session, id: int, obj_in: EmailAuthorizationUpdate) -> Optional[EmailAuthorization]:
        db_obj = db.query(EmailAuthorization).filter(EmailAuthorization.id == id).first()
        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)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    @staticmethod
    def delete(db: Session, id: int) -> Optional[EmailAuthorization]:
        db_obj = db.query(EmailAuthorization).filter(EmailAuthorization.id == id).first()
        if not db_obj:
            raise NotFoundError("Email authorization not found")
        db.delete(db_obj)
        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:
        """
        创建支持 SOCKS5 / HTTP 代理的 IMAP SSL 连接
        """
        if proxy_host and proxy_port and proxy_port > 0:
            original_socket = socket.socket
            socks.setdefaultproxy(
                proxy_type=socks.SOCKS5,  # 可改为 socks.HTTP
                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  # 恢复原始 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:
        """
        创建支持 SOCKS5 / HTTP 代理的 SMTP SSL 连接
        （逻辑完全与 _connect_imap_with_proxy 保持一致，且强制 SSL）
        """
        if proxy_host and proxy_port and proxy_port > 0:
            original_socket = socket.socket
            socks.setdefaultproxy(
                proxy_type=socks.SOCKS5,  # 如需改 HTTP，这里改
                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  # 恢复原始 socket
        else:
            # 无代理
            smtp = smtplib.SMTP_SSL(host, port)

        return smtp


    @staticmethod
    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 头作为收件时间）
        """

        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")

        

    @staticmethod
    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]:
        """
        在有效期内循环读取邮箱，找到符合条件的邮件（使用最后一条 Received 头作为收件时间）
        """

        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()]

        try:
            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")

            _, data = mail.search(None, "ALL")
            mail_ids = data[0].split()
            if not mail_ids:
                return None

            recent_ids = mail_ids[-top:]
            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)
                messages.append(msg)

            if debug:
                logger.debug(f"[DEBUG] 成功解析邮件数: {len(messages)}")

            for msg in messages:
          
                # 匹配发件人/收件人
                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()

            mail.close()
            mail.logout()
            return None  # 超时未找到

        except Exception as e:
            raise e
        
    # ----------------------------------------------------------------------
    # 主方法：模仿 Java forwardFirstMatchingEmail
    # ----------------------------------------------------------------------
    def forward_first_matching_email(
            auth,
            forward_to: str,
            sender: str,
            recipient: str,
            subject_keywords: str,
            body_keywords: str
    ):
        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()]

        try:
            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")

            target = recipient
            query = f'(HEADER To "{target}")'
            res, data = mail.uid("search", None, query)
            if res != "OK":
                return None

            uids = data[0].split()

            # 获取邮件，按 sentDate 排序
            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_ = email.utils.parsedate_to_datetime(msg.get("Date"))
                msgs.append((date_, msg))

            msgs.sort(key=lambda x: x[0], reverse=True)

            # -----------------------------------------------------------
            # 搜索匹配邮件
            # -----------------------------------------------------------
            for _, msg in msgs:
                # 匹配发件人/收件人
                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, True)
                if body_keys and not any(k.lower() in body.lower() for k in body_keys):
                    continue
                
                msg.replace_header("From", EMAIL_ACCOUNT)
                msg.replace_header("To", forward_to)

                # 可选：修改 Subject
                subject_raw = msg.get("Subject", "")
                msg.replace_header("Subject", f"FWD: {subject_raw}")

                # 发送
                EmailAuthorizationService.send_email_smtp(
                    auth,
                    msg
                )

                return f"邮件 '{subject}' 已成功转发至: {forward_to}"
            return None
        except Exception as e:
            raise e
        
    # ----------------------------------------------------------------------
    # 主方法：模仿 Java send_email
    # ----------------------------------------------------------------------
    def send_email(
            auth,
            send_to: str,
            subject: str,
            content_type: str,
            content: str
    ):
        try:
            EMAIL_ACCOUNT = auth.email
            msg = EmailMessage()
            msg["From"] = EMAIL_ACCOUNT
            msg["To"] = send_to
            msg["Subject"] = subject
            
            html_content = None;
            text_content = None;
            if ("html" == content_type.lower()):
                html_content = content
            else:
                text_content = content

            # 设置正文内容
            if html_content and text_content:
                # multipart/alternative
                msg.set_content(text_content)
                msg.add_alternative(html_content, subtype="html")
            elif html_content:
                msg.add_alternative(html_content, subtype="html")
            elif text_content:
                msg.set_content(text_content)
            else:
                msg.set_content("")  # 空邮件
            
            # 发送
            EmailAuthorizationService.send_email_smtp(
                auth,
                msg
            )

            return f"邮件 '{subject}' 成功发送至: {send_to}"
        except Exception as e:
            raise e
        
        
    # ----------------------------------------------------------------------
    # 主方法：模仿 Java send_email
    # ----------------------------------------------------------------------
    def send_email_bulk(
            auth,
            send_to: str,
            subject: str,
            content_type: str,
            content: str
    ):
        bcc_list = [s.strip() for s in send_to.split(",") if s.strip()]
        EMAIL_ACCOUNT = auth.email
        msg = EmailMessage()
        msg["From"] = EMAIL_ACCOUNT
        # TO 可以留空或放一个默认收件人
        msg["To"] = bcc_list[0]  # 或者固定一个自己邮箱作为 TO
        # BCC 添加所有收件人
        bcc_list = bcc_list
        msg["Subject"] = subject
        
        html_content = None;
        text_content = None;
        if ("html" == content_type.lower()):
            html_content = content
        else:
            text_content = content

        # 设置正文内容
        if html_content and text_content:
            # multipart/alternative
            msg.set_content(text_content)
            msg.add_alternative(html_content, subtype="html")
        elif html_content:
            msg.add_alternative(html_content, subtype="html")
        elif text_content:
            msg.set_content(text_content)
        else:
            msg.set_content("")  # 空邮件
        
        # 发送
        EmailAuthorizationService.send_email_smtp(
            auth,
            msg,
            bcc_list = bcc_list
        )

        return f"邮件 '{subject}' 成功发送至: {send_to}"


    # ----------------------------------------------------------------------
    # SMTP 发送邮件
    # ----------------------------------------------------------------------
    @staticmethod
    def send_email_smtp(auth, msg, bcc_list=[]):
        EMAIL_ACCOUNT = auth.email
        EMAIL_PASSWORD = auth.authorization_code
        SMTP_SERVER = auth.smtp_server
        SMTP_PORT = auth.smtp_port
        mail = EmailAuthorizationService._connect_smtp_with_proxy(
            SMTP_SERVER,
            SMTP_PORT,
            auth.proxy_host,
            auth.proxy_port,
            auth.proxy_username,
            auth.proxy_password,
        )
        mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
        if bcc_list:
            mail.send_message(msg, to_addrs=bcc_list)
        else:
            mail.send_message(msg)

    # ==============================================================
    # 辅助函数：提取邮件正文
    # ==============================================================
    @staticmethod
    def _extract_body(msg, only_text: bool = True) -> str:
        """根据 only_text 参数提取邮件内容"""
        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:
                    text = part.get_payload(decode=True).decode(charset, errors="ignore")
                    body_parts.append(text)
                except Exception:
                    continue
        else:
            charset = msg.get_content_charset() or "utf-8"
            body_parts.append(msg.get_payload(decode=True).decode(charset, errors="ignore"))
        body = "\n".join(body_parts)
        return re.sub(r"\s+", " ", body.strip())