universal-crawler/notify.py

# -*- coding: utf-8 -*-
"""邮件通知 (复用 send_email.py 技能脚本)"""
import os
import subprocess
import sys

SEND_EMAIL = os.path.join(
    os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
    "skills", "send_email.py",
)


def notify_email(task, run):
    """发送完成通知邮件, 返回 (bool, msg)"""
    if not os.path.exists(SEND_EMAIL):
        return False, "send_email.py 不存在"
    cfg = task.get("config", {})
    to = cfg.get("notify_email") or "wlq@tphai.com"
    results = run.get("results", [])
    ok = sum(1 for r in results if r.get("status") == "OK")
    fail = len(results) - ok
    lines = [
        f"项目: {task['name']}",
        f"模式: {task['mode']}",
        f"状态: {run.get('status')}",
        f"成功: {ok}   失败: {fail}   图片: {run.get('stats', {}).get('images', 0)}",
        f"输出目录: {run.get('out_dir', '')}",
        "",
        "明细:",
    ]
    for r in results[:50]:
        lines.append(f"  [{r.get('status')}] {r.get('title', '')[:40]} {r.get('url', '')}")
    if len(results) > 50:
        lines.append(f"  ... 共 {len(results)} 条")
    body = "\n".join(lines)
    return send_attachment(f"[爬虫完成] {task['name']}", body, to)


def send_attachment(subject, body, to, attach_path=None):
    """发送带附件的邮件 (复用 send_email.py), 返回 (bool, msg)
    attach_path: 附件文件路径, 可多个
    """
    if not os.path.exists(SEND_EMAIL):
        return False, "send_email.py 不存在"
    cmd = [sys.executable, SEND_EMAIL, subject, body, "--to", to]
    if attach_path:
        if isinstance(attach_path, str):
            attach_path = [attach_path]
        for p in attach_path:
            if os.path.isfile(p):
                cmd += ["--attach", p]
    try:
        r = subprocess.run(cmd, timeout=600, capture_output=True)
        return r.returncode == 0, (r.stderr or b"").decode(errors="ignore")[:200]
    except Exception as e:
        return False, str(e)