""" 邮件解析 & AI 分类查看器 - 增强版 - 上传/拖拽 .msg 文件(支持多文件) - AI 自动提取:公司名、邮件类型、正文内容 - 新增:邮件正文翻译(并排对照展示,无复制按钮) - 新增:AI 深度解析(智能分析邮件内容) - 公司名/类型/正文支持一键复制 - 部署:Hugging Face Spaces (Gradio SDK) """ import asyncio import json import logging import os import sys from pathlib import Path from typing import Optional import gradio as gr # 兼容补丁:Gradio 4.44.0 的 gradio_client/utils.py 在 JSON schema 中 # additionalProperties=True(bool)时会崩溃:`if "const" in schema` 不适用于 bool。 # Hugging Face Spaces 强制安装 gradio[oauth]==4.44.0,无法升级,故运行时打补丁。 try: import gradio_client.utils as _client_utils # 补丁 1:get_type 遇到 bool schema 时返回 Any,避免 # `if "const" in schema` 对 bool 执行 `in` 操作而崩溃。 _orig_get_type = _client_utils.get_type def _patched_get_type(schema): if isinstance(schema, bool): return "Any" return _orig_get_type(schema) _client_utils.get_type = _patched_get_type # 补丁 2:_json_schema_to_python_type 递归遇到 bool schema 时直接返回 "Any", # 避免走到末尾抛出 APIInfoParseError: Cannot parse schema True _orig_json_schema_to_python_type = _client_utils._json_schema_to_python_type def _patched_json_schema_to_python_type(schema, defs): if isinstance(schema, bool): return "Any" return _orig_json_schema_to_python_type(schema, defs) _client_utils._json_schema_to_python_type = _patched_json_schema_to_python_type except Exception: pass # 补丁 3:禁用 Jinja2 模板缓存,避免 cache key 类型错误 # Jinja2 3.1.5+ 修改了缓存 key 的生成方式,与 Gradio 4.44.0 的 Starlette # 模板渲染不兼容,导致 TypeError: unhashable type: 'dict'。 # 禁用缓存(cache=None)可彻底规避此问题,对性能影响极小。 try: from starlette.templating import Jinja2Templates as _J2T _orig_j2t_init = _J2T.__init__ def _patched_j2t_init(self, *args, **kwargs): _orig_j2t_init(self, *args, **kwargs) self.env.cache = None # 禁用模板缓存 _J2T.__init__ = _patched_j2t_init except Exception: pass sys.path.insert(0, str(Path(__file__).parent)) from src import config as cfg from src.email_parser import parse_msg_file, ParsedEmail from src.ai_classifier import classify_email_sync, ClassificationResult from src.translator import translate_email_sync, TranslationResult logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger(__name__) # ──────────────────────────────────────────── # CSS & JS:事件委托复制(无内联 onclick) # ──────────────────────────────────────────── CUSTOM_CSS = """ .gradio-container { font-family: 'Inter', -apple-system, BlinkMacSystemFont, sans-serif; } .header-banner { background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); border-radius: 16px; padding: 28px 32px; margin-bottom: 8px; color: white; } .header-banner h1 { margin: 0 0 6px 0; font-size: 1.7rem; font-weight: 700; } .header-banner p { margin: 0; opacity: 0.85; font-size: 0.95rem; } .config-panel { border: 1px solid #e2e8f0 !important; border-radius: 12px !important; } .email-card { background: #fff; border: 1px solid #e2e8f0; border-radius: 14px; padding: 20px 24px; margin-bottom: 14px; box-shadow: 0 1px 4px rgba(0,0,0,0.06); transition: box-shadow 0.2s; } .email-card:hover { box-shadow: 0 4px 16px rgba(0,0,0,0.10); } .email-card-header { display: flex; align-items: center; gap: 10px; margin-bottom: 14px; flex-wrap: wrap; } .email-filename { font-size: 0.78rem; color: #94a3b8; margin-left: auto; font-family: monospace; } .category-badge { display: inline-flex; align-items: center; gap: 5px; padding: 4px 12px; border-radius: 20px; font-size: 0.82rem; font-weight: 600; color: white; white-space: nowrap; } .field-row { display: flex; align-items: flex-start; gap: 10px; margin-bottom: 10px; padding: 10px 14px; background: #f8fafc; border-radius: 8px; border: 1px solid #f1f5f9; } .field-label { font-size: 0.72rem; font-weight: 700; color: #64748b; text-transform: uppercase; letter-spacing: 0.05em; min-width: 64px; padding-top: 2px; white-space: nowrap; } .field-value { flex: 1; font-size: 0.88rem; color: #1e293b; line-height: 1.5; word-break: break-word; max-height: 120px; overflow-y: auto; } .field-value.company { font-weight: 700; font-size: 1rem; color: #0f172a; } .copy-btn { background: #f1f5f9; border: 1px solid #e2e8f0; border-radius: 6px; padding: 4px 10px; font-size: 0.75rem; color: #475569; cursor: pointer; white-space: nowrap; transition: all 0.15s; flex-shrink: 0; margin-top: 1px; } .copy-btn:hover { background: #e2e8f0; color: #0f172a; } .copy-btn.copied { background: #dcfce7; border-color: #86efac; color: #166534; } .summary-text { font-size: 0.82rem; color: #64748b; font-style: italic; padding: 6px 14px; border-left: 3px solid #e2e8f0; margin-bottom: 4px; } .meta-row { display: flex; gap: 16px; font-size: 0.78rem; color: #94a3b8; flex-wrap: wrap; margin-top: 6px; } .meta-row span { display: flex; align-items: center; gap: 4px; } /* 翻译对照区域 */ .translation-section { margin-top: 14px; border: 1px solid #e2e8f0; border-radius: 10px; overflow: hidden; background: #f8fafc; } .translation-header { background: #f1f5f9; padding: 8px 14px; font-size: 0.78rem; font-weight: 600; color: #475569; border-bottom: 1px solid #e2e8f0; display: flex; align-items: center; gap: 8px; } .translation-columns { display: flex; min-height: 80px; } .translation-original, .translation-result { flex: 1; padding: 12px 14px; font-size: 0.85rem; line-height: 1.6; color: #334155; white-space: pre-wrap; max-height: 300px; overflow-y: auto; } .translation-original { border-right: 1px solid #e2e8f0; background: #fff; } .translation-result { background: #fefce8; } .translation-lang-label { font-size: 0.7rem; color: #94a3b8; font-weight: 600; text-transform: uppercase; letter-spacing: 0.05em; margin-bottom: 6px; } /* AI 解析区域 */ .ai-analysis-section { margin-top: 14px; border: 1px solid #c4b5fd; border-radius: 10px; overflow: hidden; background: #faf5ff; } .ai-analysis-header { background: #ede9fe; padding: 8px 14px; font-size: 0.78rem; font-weight: 600; color: #6d28d9; border-bottom: 1px solid #c4b5fd; display: flex; align-items: center; gap: 8px; } .ai-analysis-content { padding: 14px; font-size: 0.85rem; line-height: 1.7; color: #334155; } .ai-analysis-content h4 { margin: 12px 0 6px 0; font-size: 0.82rem; font-weight: 700; color: #6d28d9; } .ai-analysis-content ul { margin: 6px 0; padding-left: 20px; } .ai-analysis-content li { margin-bottom: 4px; } .status-done { color: #16a34a; font-weight: 600; } .status-error { color: #dc2626; font-weight: 600; } .progress-bar-wrap { background: #f1f5f9; border-radius: 8px; height: 8px; overflow: hidden; margin: 8px 0; } .progress-bar-fill { background: linear-gradient(90deg, #667eea, #764ba2); height: 100%; transition: width 0.3s ease; border-radius: 8px; } .empty-state { text-align: center; padding: 48px 24px; color: #94a3b8; } .empty-state .icon { font-size: 3rem; margin-bottom: 12px; } .stats-bar { display: flex; gap: 20px; padding: 12px 20px; background: #f8fafc; border-radius: 10px; border: 1px solid #e2e8f0; margin-bottom: 16px; flex-wrap: wrap; } .stat-item { display: flex; flex-direction: column; align-items: center; } .stat-num { font-size: 1.4rem; font-weight: 700; color: #0f172a; } .stat-lbl { font-size: 0.7rem; color: #94a3b8; text-transform: uppercase; letter-spacing: 0.05em; } .test-result-ok { color: #16a34a; font-weight: 600; padding: 6px 0; } .test-result-err { color: #dc2626; font-weight: 600; padding: 6px 0; } """ # JavaScript:事件委托,从 data-copy-text 属性读取复制内容 COPY_JS = r""" document.addEventListener('click', function(e) { var btn = e.target.closest('.copy-btn'); if (!btn) return; var text = btn.getAttribute('data-copy-text') || ''; if (!text) return; navigator.clipboard.writeText(text).then(function() { var orig = btn.innerText; btn.innerText = '\u2705 \u5df2\u590d\u5236'; btn.classList.add('copied'); setTimeout(function() { btn.innerText = orig; btn.classList.remove('copied'); }, 2000); }).catch(function() { var ta = document.createElement('textarea'); ta.value = text; ta.style.position = 'fixed'; ta.style.opacity = '0'; document.body.appendChild(ta); ta.focus(); ta.select(); document.execCommand('copy'); document.body.removeChild(ta); var orig = btn.innerText; btn.innerText = '\u2705 \u5df2\u590d\u5236'; btn.classList.add('copied'); setTimeout(function() { btn.innerText = orig; btn.classList.remove('copied'); }, 2000); }); }); """ # ───────────────────────────────────────────── # 工具函数 # ───────────────────────────────────────────── CATEGORY_COLORS = cfg.CATEGORY_COLORS CATEGORY_EMOJI = cfg.CATEGORY_EMOJI def _esc_html(s: str) -> str: """HTML 转义(用于 data 属性值)""" return s.replace("&", "&").replace('"', """).replace("<", "<").replace(">", ">") def _badge_html(category: str) -> str: color = CATEGORY_COLORS.get(category, "#94a3b8") emoji = CATEGORY_EMOJI.get(category, "[邮件]") return f'{emoji} {category}' def _copy_btn_html(text: str, btn_id: str, label: str = "[复制]") -> str: """生成复制按钮(使用 data-copy-text 属性,无内联 JS)""" safe = _esc_html(text) return f'' def _render_translation_section(idx: int, translation) -> str: """渲染翻译对照区域(并排),无复制按钮""" if not translation or not getattr(translation, "translated_text", ""): return "" src_lang = getattr(translation, "source_language", "Unknown") translated = getattr(translation, "translated_text", "") is_ok = getattr(translation, "is_already_target", False) orig = getattr(translation, "original_text", "(无正文)") if is_ok: note = '(原文已是目标语言)' else: note = f'检测语言:{src_lang}' return ( '
' '
' f' [翻译] 邮件翻译对照 {note}' '
' '
' '
' f'
原文({src_lang})
' f'
{orig}
' '
' '
' '
译文
' f'
{translated}
' '
' '
' '
' ) def _render_ai_analysis_section(idx: int, analysis: str) -> str: """渲染 AI 深度解析区域""" if not analysis: return "" import re html = analysis html = re.sub(r"\*\*(.+?)\*\*", r"\1", html) html = re.sub(r"\*(.+?)\*", r"\1", html) html = html.replace("\n", "
") return ( '
' '
' ' [AI] AI 深度解析结果' '
' '
' f' {html}' '
' '
' ) def _render_email_card( idx: int, parsed, result, translation=None, ai_analysis=None, ) -> str: """渲染单封邮件 HTML 卡片""" if result is None: status = '[warn] AI 分类跳过(未配置 API Key)' elif getattr(result, "error", ""): status = f'[x] {getattr(result, "error", "")[:80]}' else: status = '[ok] 分类完成' company = getattr(result, "company_name", "") if result and not getattr(result, "error", "") else "—" category = getattr(result, "category", "") if result and not getattr(result, "error", "") else "—" summary = getattr(result, "summary", "") if result and not getattr(result, "error", "") else "" body = getattr(parsed, "body_text", "") if body and len(body) > 800: body_disp = body[:800] + "...(截断,完整内容见翻译对照)" elif not body: body_disp = "(正文为空)" else: body_disp = body cid_c = f"cp-co-{idx}" cid_b = f"cp-bo-{idx}" cid_t = f"cp-ca-{idx}" badge = _badge_html(category) if category != "—" else '' return ( '
' f'
{badge} {status}
' '
' ' 公司名' f' {company}' f' {_copy_btn_html(company, cid_c)}' '
' '
' ' 类型' f' {badge}' f' {_copy_btn_html(category, cid_t)}' '
' '
' ' 正文' f' {body_disp}' f' {_copy_btn_html(body, cid_b)}' '
' f' {("
[tip] " + summary + "
") if summary else ""}' '
' f' [sender] {getattr(parsed, "sender", "") or "—"}' f' [date] {getattr(parsed, "date_str", "") or "—"}' f' [subject] {(getattr(parsed, "subject", "") or "")[:50]}' '
' f' {_render_translation_section(idx, translation)}' f' {_render_ai_analysis_section(idx, ai_analysis)}' '
' ) def _render_results_html( parsed_list: list, results: list, translations: list = None, analyses: list = None, ) -> str: if not parsed_list: return '
[doc]

上传 .msg 文件后,解析结果将在此展示

' translations = translations or [None] * len(parsed_list) analyses = analyses or [None] * len(parsed_list) total = len(parsed_list) done = sum(1 for r in results if r and not getattr(r, "error", "")) err = total - done cats = {} for r in results: if r and getattr(r, "category", ""): c = getattr(r, "category", "") cats[c] = cats.get(c, 0) + 1 cat_html = "" for k, v in sorted(cats.items(), key=lambda x: -x[1]): c = CATEGORY_COLORS.get(k, "#64748b") cat_html += f'
{v}{k}
' stats = ( '
' f'
{total}总封数
' f'
{done}已分类
' f'
{err}失败
' f'{cat_html}' '
' ) cards = "" for i, p in enumerate(parsed_list): cards += _render_email_card(i, p, results[i], translations[i], analyses[i]) return f'{stats}{cards}' # ───────────────────────────────────────────── # 主处理函数 # ───────────────────────────────────────────── def process_files(files, provider: str, api_key: str, model: str, progress=gr.Progress(track_tqdm=False)): """解析 .msg + AI 分类""" if not files: return _render_results_html([], []), [], [], [], [] cfg.apply_runtime_config(provider.lower(), api_key.strip(), model.strip()) paths = [] for f in files: if hasattr(f, "name"): paths.append(f.name) elif isinstance(f, str): paths.append(f) total = len(paths) parsed_list = [] results = [] has_key = bool(cfg.get_llm_api_key()) progress(0, desc=f"[解析] 正在解析 {total} 封邮件…") for i, fp in enumerate(paths): progress((i + 0.5) / (total * 2), desc=f"[解析] 解析 {i+1}/{total}") parsed_list.append(parse_msg_file(fp)) for i, p in enumerate(parsed_list): progress(0.5 + (i + 0.5) / (total * 2), desc=f"[AI] 分类 {i+1}/{total}") if not has_key: results.append(None) elif not getattr(p, "is_valid", True): results.append(ClassificationResult(error=f"解析失败:{getattr(p, 'parse_error', '')}")) else: results.append(classify_email_sync(p)) progress(1.0, desc=f"[ok] 处理完成,共 {total} 封") return _render_results_html(parsed_list, results), parsed_list, results, [], [] def translate_all(parsed_list, results, translations, target_lang: str, progress=gr.Progress(track_tqdm=False)): """为所有已解析邮件翻译正文""" if not parsed_list: return _render_results_html([], []), [], [] new_t = list(translations) if translations else [None] * len(parsed_list) total = len(parsed_list) for i, p in enumerate(parsed_list): progress(i / total, desc=f"[翻译] 翻译 {i+1}/{total}…") body = getattr(p, "body_text", "") if not body or not body.strip(): new_t[i] = None continue tr = translate_email_sync(body, target_lang) tr.original_text = body[:2000] + ("…" if len(body) > 2000 else "") new_t[i] = tr progress(1.0, desc="[ok] 翻译完成") return _render_results_html(parsed_list, results, new_t), new_t def analyze_all(parsed_list, results, analyses, progress=gr.Progress(track_tqdm=False)): """为所有已分类邮件做 AI 深度解析""" if not parsed_list: return _render_results_html([], []), [], [] import json as _json new_a = list(analyses) if analyses else [None] * len(parsed_list) total = len(parsed_list) for i, p in enumerate(parsed_list): progress(i / total, desc=f"[AI] 解析 {i+1}/{total}…") body = getattr(p, "body_text", "") if not body or not cfg.get_llm_api_key(): new_a[i] = None continue company = getattr(results[i], "company_name", "") if i < len(results) and results[i] else "" category = getattr(results[i], "category", "") if i < len(results) and results[i] else "" system_prompt = """You are a senior recruitment strategist and email analyst. Analyze the recruitment-related email deeply and provide actionable insights. You MUST respond with valid JSON only: { "intent_analysis": "", "suggested_action": "", "tone_assessment": "", "key_phrases": ["", ""], "follow_up_template": "" }""" user_prompt = ( f"Company: {company}\n" f"Category: {category}\n" f"From: {getattr(p, 'sender', '')}\n" f"Subject: {getattr(p, 'subject', '')}\n" f"\nBody:\n{body[:3000]}\n" f"\nRespond in JSON format as specified." ) try: from openai import AsyncOpenAI import asyncio as _a api_key = cfg.get_llm_api_key() model = cfg.get_llm_model() base_url = cfg.get_llm_base_url() ck = {"api_key": api_key} if base_url: ck["base_url"] = base_url client = AsyncOpenAI(**ck) async def _call(): resp = await client.chat.completions.create( model=model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=0.2, max_tokens=1024, response_format={"type": "json_object"}, ) return resp.choices[0].message.content or "" loop = _a.new_event_loop() content = loop.run_until_complete(_call()) loop.close() data = _json.loads(content) parts = [] if data.get("intent_analysis"): parts.append(f'

[意图] 意图分析

{data["intent_analysis"]}

') if data.get("suggested_action"): parts.append(f'

[行动] 建议行动

{data["suggested_action"]}

') if data.get("tone_assessment"): _tm = {"professional": "专业正式", "friendly": "友好亲切", "cold": "冷淡敷衍", "encouraging": "鼓励积极", "standard": "标准常规"} _t = _tm.get(data["tone_assessment"], data["tone_assessment"]) parts.append(f'

[语气] 语气评估

{_t}

') if data.get("key_phrases"): _ph = "".join(f"
  • {x}
  • " for x in data["key_phrases"]) parts.append(f"

    [关键词] 关键短语

      {_ph}
    ") if data.get("follow_up_template"): parts.append( f'

    [模板] 跟进邮件模板

    ' f'
    {data["follow_up_template"]}
    ' ) new_a[i] = "\n".join(parts) if parts else "(无内容)" except Exception as e: logger.error(f"AI 解析失败: {e}") new_a[i] = f"[warn] 解析失败:{str(e)[:100]}" progress(1.0, desc="[ok] AI 解析完成") return _render_results_html(parsed_list, results, None, new_a), new_a # ───────────────────────────────────────────── # 连接测试 # ───────────────────────────────────────────── def test_llm_connection(provider: str, api_key: str, model: str) -> str: if not api_key.strip(): return "[x] 请先填写 API Key" cfg.apply_runtime_config(provider.lower(), api_key.strip(), model.strip()) try: import openai burl = cfg.get_llm_base_url() ck = {"api_key": api_key.strip()} if burl: ck["base_url"] = burl client = openai.OpenAI(**ck) resp = client.chat.completions.create( model=cfg.get_llm_model(), messages=[{"role": "user", "content": "Reply with just: OK"}], max_tokens=10, temperature=0, ) reply = resp.choices[0].message.content.strip() return f'[ok] 连接成功!模型回复:{reply}' except Exception as e: return f'[x] 连接失败:{str(e)[:120]}' # ───────────────────────────────────────────── # Gradio 界面构建 # ───────────────────────────────────────────── def build_app() -> gr.Blocks: curr = cfg.get_current_config() with gr.Blocks( title="邮件分析查看器", css=CUSTOM_CSS, theme=gr.themes.Soft( primary_hue=gr.themes.colors.violet, neutral_hue=gr.themes.colors.slate, ), ) as demo: parsed_st = gr.State([]) results_st = gr.State([]) trans_st = gr.State([]) analyses_st = gr.State([]) gr.HTML( '
    ' '

    [邮件] 邮件解析 & AI 智能分析查看器

    ' '

    上传 .msg 邮件文件,AI 自动提取 / 分类 / 翻译 / 深度解析

    ' '
    ' ) with gr.Accordion("⚙️ API 配置(展开设置 LLM 密钥)", open=not bool(curr["api_key"]), elem_classes="config-panel"): with gr.Row(): prov_dd = gr.Dropdown( choices=["deepseek", "openai"], value=curr["provider"], label="LLM 提供商", scale=1, ) key_box = gr.Textbox( value=curr["api_key"], label="API Key", type="password", scale=3, ) mdl_box = gr.Textbox( value=curr["model"], label="模型名称", scale=2, ) with gr.Row(): tst_btn = gr.Button("🔌 测试连接", variant="secondary", size="sm") tst_res = gr.HTML(label="") tst_btn.click(test_llm_connection, [prov_dd, key_box, mdl_box], tst_res) gr.HTML("
    ") with gr.Row(): with gr.Column(scale=1): upl = gr.File( label="[上传] 拖拽或点击上传 .msg 文件(支持多文件)", file_types=[".msg"], file_count="multiple", height=180, ) go_btn = gr.Button( "[开始] 开始解析并分类", variant="primary", size="lg", ) with gr.Row(): tgt_lang = gr.Dropdown( choices=["Chinese", "English", "Japanese", "German", "French"], value="Chinese", label="[翻译] 翻译目标语言", scale=2, ) tr_btn = gr.Button( "[翻译] 翻译全部邮件正文", variant="secondary", size="sm", ) an_btn = gr.Button( "[AI] AI 深度解析全部邮件", variant="secondary", size="sm", ) gr.HTML( '
    ' "[功能] 公司名 / 类型 / 正文 支持一键复制 · " "翻译并排对照 · AI 深度解析(意图 / 行动 / 语气 / 模板)" '
    ' ) gr.HTML("
    ") res_html = gr.HTML( value=( '
    ' '
    [doc]
    ' '

    上传 .msg 文件并点击「开始解析」后,结果将在此展示

    ' '
    ' ), label="", ) # 按钮回调 go_btn.click( process_files, [upl, prov_dd, key_box, mdl_box], [res_html, parsed_st, results_st, trans_st, analyses_st], ) tr_btn.click( translate_all, [parsed_st, results_st, trans_st, tgt_lang], [res_html, trans_st], ) an_btn.click( analyze_all, [parsed_st, results_st, analyses_st], [res_html, analyses_st], ) gr.HTML( '
    ' '基于 extract-msg · AI 功能基于 OpenAI / DeepSeek' '
    ' ) return demo # ───────────────────────────────────────────── # 启动 # ───────────────────────────────────────────── if __name__ == "__main__": app = build_app() app.launch( server_name="0.0.0.0", server_port=int(os.getenv("PORT", 7860)), show_api=False, share=False, # HF Spaces:不需要 public tunnel debug=False, # 生产环境关闭 debug )