""" 邮件发送入口 调用同目录下的 dealer-email-outreach 脚本 """ import subprocess import sys from pathlib import Path from typing import Dict, Any, Optional def _get_script_dir() -> Path: return Path(__file__).parent def _get_skill_dir() -> Path: return _get_script_dir().parent.parent def _get_asset(name: str) -> Path: return _get_skill_dir() / "assets" / name def prepare_emails( excel_path: str, sheet_name: str, filter_str: str, subject: str, template_name: str = "email_template.md", output_path: str = "outreach_preview.json" ) -> str: """生成邮件预览 JSON""" script = _get_script_dir() / "prepare_outreach_emails.py" template = _get_asset(template_name) if not script.exists(): raise FileNotFoundError(f"未找到脚本: {script}") if not template.exists(): raise FileNotFoundError(f"未找到模板: {template}") cmd = [ sys.executable, str(script), "--excel", excel_path, "--sheet", sheet_name, "--filter", filter_str, "--template", str(template), "--subject", subject, "--output", output_path, ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: raise RuntimeError(f"生成邮件预览失败: {result.stderr}") return output_path def send_emails( preview_path: str, sender: str, auth_code: str, smtp_host: str = "smtp.qq.com", smtp_port: int = 465 ) -> Dict[str, Any]: """发送邮件""" script = _get_script_dir() / "send_outreach_emails.py" if not script.exists(): raise FileNotFoundError(f"未找到脚本: {script}") cmd = [ sys.executable, str(script), "--input", preview_path, "--smtp-host", smtp_host, "--smtp-port", str(smtp_port), "--sender", sender, "--auth-code", auth_code, "--confirm-send", ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: raise RuntimeError(f"发送邮件失败: {result.stderr}") return { "stdout": result.stdout, "stderr": result.stderr, } if __name__ == "__main__": print("邮件模块脚本路径:", _get_script_dir()) print("模板路径:", _get_asset("email_template.md"))