hlzd-rfp-response

name	hlzd-rfp-response
description	B2B 询盘 / 邮件响应自动化 —— 8 大邮箱配置矩阵（Gmail/网易/QQ/Outlook/企业微信/飞书/阿里云/自建）+ Gmail OAuth + IMAP/IMAPS + 凭证加密 + 飞书卡片配置向导。
license	MIT
metadata	{"author":"HLZD (海联智达 / 海良数科)","version":"0.1.0","industry":"cross-border-b2b","category":"sales-automation","triggered_by":["询盘响应","邮件响应","询盘自动回复","邮件自动回复","邮件配置","Gmail OAuth","网易授权码","IMAP 配置","飞书卡片","邮件模板","邮件群发","inquiry response","email reply","mail automation","IMAP setup","OAuth flow"]}

HLZD 询盘响应 v0.2.1（邮件接入升级版）

v0.2.1 关键升级：基于你"用网易 + Gmail"的真实场景，全面升级邮件接入层

• 借鉴 imap_tools（831 stars）—— Python IMAP 标准库（OAuth2 + IDLE + 无外部依赖）
• 借鉴 google-api-python-client（8871 stars）—— Gmail 官方 OAuth 2.0 流程
• 覆盖 8 大邮箱——网易 163/126/yeah、Gmail、QQ、Outlook、飞书邮箱、企业邮箱

---

🎯 你之前的失真（直接承认）

我之前 SKILL.md 写的"邮件 API 拉取"是抽象假设：

• ❌ 没考虑你用网易个人邮箱（需要授权码，不是密码）
• ❌ 没考虑你用 Gmail（必须 OAuth 2.0，2024 起强制）
• ❌ 没考虑飞书邮箱（OAuth 2.0）
• ❌ 假设"邮件 API"是简单事，实际是5 步 OAuth 流程 + 授权码获取

现在直接修正，基于真实场景设计。

---

📋 8 大邮箱配置矩阵（覆盖你所有真实场景）

邮箱服务商	IMAP 服务器	端口	SSL	认证方式	客户端库
网易 163 ⭐ 你在用	imap.163.com	993	✅	用户名 + 授权码（非登录密码）	imap_tools
网易 126	imap.126.com	993	✅	用户名 + 授权码	imap_tools
网易 yeah.net	imap.yeah.net	993	✅	用户名 + 授权码	imap_tools
Gmail ⭐ 你在用	imap.gmail.com	993	✅	OAuth 2.0 token	imap_tools + google-api-python-client
QQ 邮箱	imap.qq.com	993	✅	用户名 + 授权码	imap_tools
Outlook / Office365	outlook.office365.com	993	✅	OAuth 2.0 token	imap_tools
飞书邮箱	imap.feishu.cn	993	✅	OAuth 2.0 token	imap_tools
企业邮箱（如 HLZD.com）	imap.hlzd.com	993	✅	用户名 + 密码	imap_tools

所有 8 大邮箱都被 imap_tools（831 stars）+ google-api-python-client（8871 stars）覆盖，无需自己实现。

---

🏆 借鉴证据

ikvk/imap_tools（831 stars）—— Python IMAP 标准库

关键特性（直接对 HLZD 有用）：

• ✅ OAuth2 认证（xoauth2 方法）—— Gmail/Outlook 集成
• ✅ MailBox 主类（SSL/TLS 连接）
• ✅ 查询构建系统（Pythonic 语法 → IMAP 搜索命令）
• ✅ IDLE 功能（实时监控新邮件）
• ✅ 无外部依赖

真实代码示例（imap_tools 文档摘录）：

from imap_tools import MailBox

# Gmail 接入（OAuth 2.0 token）
with MailBox('imap.gmail.com').xoauth2(user_email, oauth2_token) as mailbox:
    for msg in mailbox.fetch():
        print(msg.subject, msg.from_, msg.text)

# 网易 163 接入（授权码）
with MailBox('imap.163.com').login(user_email, auth_code) as mailbox:
    for msg in mailbox.fetch():
        print(msg.subject, msg.from_, msg.text)

googleapis/google-api-python-client（8,871 stars）—— Gmail 官方

用途：Gmail OAuth 2.0 完整流程（imap_tools 只处理认证字符串，OAuth 流程本身需要 google-auth-oauthlib）

---

🚀 升级后的核心能力矩阵

能力	v0.1	v0.2.1（升级）
网易 163 接入	抽象"邮件 API"	✅ imap_tools + 授权码流程
Gmail 接入	❌ 不支持	✅ imap_tools + OAuth 2.0（5 步流程）
多邮箱支持	❌ 假设单一邮箱	✅ 同时支持 8 大邮箱
飞书邮箱	❌ 不支持	✅ imap_tools + 飞书 OAuth
IMAP IDLE	❌ 轮询	✅ imap_tools 内置 IDLE
凭证加密	❌ 明文	✅ Panmira secrets
一键配置向导	❌ 无	✅ 飞书卡片引导（8 邮箱分别）

---

🛠️ scripts/ 设计 v0.2.1

scripts/
├── fetch_emails.py                # ⭐ v0.2.1: imap_tools 统一接入（8 邮箱）
├── oauth_gmail.py                 # ⭐ v0.2.1: Gmail OAuth 2.0 完整流程
├── oauth_outlook.py               # ⭐ v0.2.1: Outlook OAuth 2.0
├── oauth_feishu_mail.py           # ⭐ v0.2.1: 飞书邮箱 OAuth
├── netease_auth_code.py           # ⭐ v0.2.1: 网易授权码流程文档化
├── setup_email_wizard.py          # ⭐ v0.2.1: 飞书卡片配置向导
├── classify_by_sender.py          # v0.2: 基于发件人聚合（Inbox Zero）
├── extract_entities_cn.py         # v0.2: HanLP 中文实体抽取
├── extract_entities_langextract.py # v0.2: LangExtract
├── classify_intent.py             # 4 类意图分类
├── structured_output.py           # v0.2: Pydantic 输出验证
├── detect_lang.py                 # 多语种识别
├── translate.py                   # LLM 翻译
├── research_customer.py           # 客户背景调查
├── grade_customer.py              # A/B/C 分级
├── draft_reply.py                 # 分级起草
├── get_reply_context.py           # v0.2: 6 个月历史窗口
├── inquiry_state_machine.py       # v0.2: 6 状态状态机
├── event_triggers.py              # v0.2: 6 种触发事件
├── event_models.py                # 借鉴 Tracardi Event
├── secrets_manager.py             # ⭐ v0.2.1: Panmira secrets 凭证管理
├── idle_listener.py               # ⭐ v0.2.1: IMAP IDLE 实时监听
├── stats.py                       # 今日询盘统计
└── orchestrator.py                # 主调度

---

🔐 scripts/fetch_emails.py 完整设计

"""
HLZD-询盘响应 邮件接入层（v0.2.1）
基于 imap_tools (831 stars) + google-api-python-client (8871 stars)
覆盖 8 大邮箱
"""
from imap_tools import MailBox, AND, OR, NOT
from typing import Literal, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum

# ============== 8 大邮箱 Provider 定义 ==============

class EmailProvider(Enum):
    """8 大邮箱服务商（基于 imap_tools）"""
    NETEASE_163     = ("imap.163.com",                993, "ssl")  # 网易 163 ⭐
    NETEASE_126     = ("imap.126.com",                993, "ssl")  # 网易 126
    NETEASE_YEAH    = ("imap.yeah.net",               993, "ssl")  # 网易 yeah
    GMAIL           = ("imap.gmail.com",              993, "ssl")  # Gmail ⭐
    QQ              = ("imap.qq.com",                 993, "ssl")  # QQ 邮箱
    OUTLOOK         = ("outlook.office365.com",      993, "ssl")  # Outlook
    FEISHU_MAIL     = ("imap.feishu.cn",              993, "ssl")  # 飞书邮箱
    GENERIC_IMAP    = (,              , )  


     ():  .value[]


     ():  .value[]


     ():  .value[]




 :
    
    user_id:                     
    account_id:                  
    provider: EmailProvider         
    email_address:               

    
    auth_code_ref: [] =    

    
    oauth_token_ref: [] =           
    oauth_refresh_token_ref: [] =   
    oauth_expires_at: [datetime] = 

    
    created_at: datetime = 
    last_sync_at: [datetime] = 
    is_active:  = 



 :
    

     ():
        .config = config
        .secrets = secrets_manager  
        .mailbox = 

     ():
        
        host, port, ssl = .config.provider.host, .config.provider.port, .config.provider.ssl

         .config.provider  [
            EmailProvider.GMAIL, EmailProvider.OUTLOOK, EmailProvider.FEISHU_MAIL
        ]:
            
            access_token = ._get_valid_oauth_token()
            .mailbox = MailBox(host, port=port)
            .mailbox.xoauth2(.config.email_address, access_token)
        :
            
            auth_code = .secrets.get(.config.auth_code_ref)
            .mailbox = MailBox(host, port=port)
            .mailbox.login(.config.email_address, auth_code)

     ():
        
        access_token = .secrets.get(.config.oauth_token_ref)
        expires_at = .config.oauth_expires_at

        
         expires_at  expires_at < datetime.now() + timedelta(minutes=):
            refresh_token = .secrets.get(.config.oauth_refresh_token_ref)
            new_token = refresh_oauth_token(.config.provider, refresh_token)
            
            .secrets.(.config.oauth_token_ref, new_token[])
            .config.oauth_expires_at = new_token[]
            access_token = new_token[]

         access_token

     ():
        
         .mailbox  mb:
            criteria = AND(
                date_gte=datetime.now() - timedelta(days=days),
                seen= only_unseen  only_unseen  ,
            )
             msg  mb.fetch(criteria, limit=limit):
                 {
                    : msg.uid,
                    : msg.from_,
                    : msg.to,
                    : msg.subject,
                    : msg.text  msg.html,
                    : msg.date,
                    : [att.filename  att  msg.attachments],
                    : .config.provider.name,
                    : msg,  
                }

     ():
        
         .mailbox  mb:
            
            mb.idle(callback=callback, timeout=timeout)
            
            

     () -> :
        
        :
            .connect()
            
            folders = [f.name  f  .mailbox.folder.()]
            .mailbox.logout()
             {: , : folders, : .config.provider.name}
         Exception  e:
             {: , : (e), : .config.provider.name}

---

🔐 scripts/oauth_gmail.py 完整设计

"""
Gmail OAuth 2.0 完整流程
基于 google-api-python-client (8871 stars) + google-auth-oauthlib
"""
from google_auth_oauthlib.flow import Flow
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
import json

GMAIL_SCOPES = [
    "https://mail.google.com/",          # IMAP/SMTP 访问（Gmail 强制 scope）
    "https://www.googleapis.com/auth/gmail.readonly",
    "openid",
    "https://www.googleapis.com/auth/userinfo.email",
]

def oauth_gmail_flow(client_secret_json: dict, redirect_uri: str = "https://panmira.hlzd.com/oauth/gmail/callback"):
    """Gmail OAuth 2.0 完整 5 步流程"""

    # Step 1: 创建 Flow（客户端密钥 JSON）
    flow = Flow.from_client_config(
        client_secret_json,
        scopes=GMAIL_SCOPES,
        redirect_uri=redirect_uri,
    )

    # Step 2: 生成授权 URL（用户跳转 Google 登录）
    auth_url, state = flow.authorization_url(
        access_type="offline",   # 必须 offline 才返回 refresh_token
        prompt="consent",        # 强制每次同意（确保返回 refresh_token）
        include_granted_scopes="true",
    )

    # Step 3: 用户授权后，Google 回调 redirect_uri?code=xxx&state=xxx
    # Step 4: 我们的后端接收回调，用 code 换 token
    # （这步在 /oauth/gmail/callback 路由中调用 fetch_token）

    # Step 5: 返回 Credentials 对象
    # credentials = flow.credentials
    # return {
    #     "access_token": credentials.token,
    #     "refresh_token": credentials.refresh_token,  # 重要：用于后续自动刷新
    #     "expires_at": credentials.expiry,
    
    

 () -> :
    
    creds = Credentials(
        token=,
        refresh_token=refresh_token,
        token_uri=,
        client_id=client_id,
        client_secret=client_secret,
        scopes=GMAIL_SCOPES,
    )
    creds.refresh(Request())
     {
        : creds.token,
        : creds.expiry,
    }

---

🛡️ scripts/secrets_manager.py 凭证加密存储

"""
Panmira secrets 凭证管理
所有邮箱授权码/OAuth token 加密存储
"""
from pathlib import Path

class SecretsManager:
    """凭证加密存储（Panmira secrets 标准接口）"""

    def __init__(self, master_key: bytes = None):
        self.master_key = master_key or self._load_or_generate_master_key()
        self.store_path = Path("~/.hlzd-inquiry/secrets.json.enc")

    def get(self, secret_ref: str) -> str:
        """获取凭证（自动解密）"""
        encrypted = self._load_store().get(secret_ref)
        return self._decrypt(encrypted)

    def set(self, secret_ref: str, plaintext: str):
        """存储凭证（自动加密）"""
        store = self._load_store()
        store[secret_ref] = self._encrypt(plaintext)
        self._save_store(store)

    def delete(self, secret_ref: str):
        """删除凭证"""
        store = self._load_store()
        if secret_ref in store:
            del store[secret_ref]
            self._save_store(store)

    def () -> :
        
         cryptography.hazmat.primitives.ciphers.aead  AESGCM
         os
        nonce = os.urandom()
        ciphertext = AESGCM(.master_key).encrypt(nonce, plaintext.encode(), )
         (nonce + ciphertext).()

     () -> :
        
         cryptography.hazmat.primitives.ciphers.aead  AESGCM
        data = .fromhex(encrypted_hex)
        nonce = data[:]
        ciphertext = data[:]
         AESGCM(.master_key).decrypt(nonce, ciphertext, ).decode()

---

📱 scripts/setup_email_wizard.py 飞书卡片配置向导

---

更多细节

完整设计文档（v0.1.x 实现细节 + 错误地图 + 性能目标）见 references/deep-dive.md。