my-assistant/scripts/email_summary.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
邮件摘要脚本 - 每天早上9点推送前一天的邮件摘要
重点关注 fj@77ircloud.com 的邮件
"""

import imaplib
import ssl
import email
from email.header import decode_header
from datetime import datetime, timedelta
import json
import sys

def decode_str(s):
    if s is None:
        return ''
    parts = decode_header(s)
    result = []
    for part, charset in parts:
        if isinstance(part, bytes):
            charset = charset or 'utf-8'
            try:
                result.append(part.decode(charset, errors='replace'))
            except:
                result.append(part.decode('utf-8', errors='replace'))
        else:
            result.append(part)
    return ''.join(result)

def get_email_body(msg):
    """获取邮件正文"""
    body = ''
    if msg.is_multipart():
        for part in msg.walk():
            content_type = part.get_content_type()
            if content_type == 'text/plain':
                try:
                    charset = part.get_content_charset() or 'utf-8'
                    body = part.get_payload(decode=True).decode(charset, errors='replace')
                    break
                except:
                    pass
            elif content_type == 'text/html':
                try:
                    charset = part.get_content_charset() or 'utf-8'
                    body = part.get_payload(decode=True).decode(charset, errors='replace')
                except:
                    pass
    else:
        try:
            charset = msg.get_content_charset() or 'utf-8'
            body = msg.get_payload(decode=True).decode(charset, errors='replace')
        except:
            pass
    return body[:500] if body else '(无正文)'

def fetch_yesterday_emails():
    """获取昨天的邮件"""
    try:
        context = ssl.create_default_context()
        mail = imaplib.IMAP4_SSL('imap.exmail.qq.com', 993, ssl_context=context)
        mail.login('lgc@77ircloud.com', 'bVPPcg6XgB7qRPw7')
        
        mail.select('INBOX')
        status, messages = mail.search(None, 'ALL')
        email_ids = messages[0].split()
        
        yesterday = (datetime.now() - timedelta(days=1)).date()
        yesterday_start = yesterday.strftime('%d-%b-%Y')
        
        all_emails = []
        important_emails = []
        
        for uid in reversed(email_ids[-50:] if len(email_ids) >= 50 else email_ids):
            status, msg_data = mail.fetch(uid, '(RFC822.HEADER RFC822.TEXT)')
            if not msg_data or not msg_data[0]:
                continue
                
            raw_email = msg_data[0][1] if isinstance(msg_data[0], tuple) else msg_data[0]
            if isinstance(raw_email, tuple):
                raw_email = raw_email[1]
                
            try:
                msg = email.message_from_bytes(raw_email if isinstance(raw_email, bytes) else raw_email.encode('utf-8'))
            except:
                continue
            
            subject = decode_str(msg.get('Subject', '(无主题)'))
            sender = decode_str(msg.get('From', ''))
            date_str = msg.get('Date', '')
            
            # 解析日期
            try:
                email_date = email.utils.parsedate_to_datetime(date_str).date()
            except:
                continue
            
            # 只取昨天的邮件
            if email_date != yesterday:
                continue
            
            email_info = {
                'subject': subject,
                'sender': sender,
                'date': date_str,
                'body': get_email_body(msg)
            }
            
            all_emails.append(email_info)
            
            # 重点关注 fj@77ircloud.com
            if 'fj@77ircloud.com' in sender:
                important_emails.append(email_info)
        
        mail.logout()
        
        return all_emails, important_emails, yesterday.strftime('%Y年%m月%d日')
        
    except Exception as e:
        return [], [], str(e)

def format_summary():
    """格式化邮件摘要"""
    emails, important, date_str = fetch_yesterday_emails()
    
    if isinstance(important, str):
        return f"⚠️ 获取邮件失败: {important}"
    
    if not emails:
        return f"📬 {date_str} 无新邮件"
    
    lines = [f"📬 {date_str} 邮件摘要（共 {len(emails)} 封）\n"]
    
    if important:
        lines.append(f"⭐ 重点关注（fj@77ircloud.com）：{len(important)} 封\n")
        for i, mail in enumerate(important, 1):
            lines.append(f"  {i}. 📧 {mail['subject']}")
            lines.append(f"     发件人: {mail['sender']}")
            lines.append(f"     时间: {mail['date']}")
            if mail['body'] and mail['body'] != '(无正文)':
                lines.append(f"     摘要: {mail['body'][:200]}...")
            lines.append("")
    
    lines.append(f"\n📋 其他邮件（{len(emails) - len(important)} 封）：")
    other = [e for e in emails if e not in important]
    for i, mail in enumerate(other[:5], 1):
        lines.append(f"  {i}. 📧 {mail['subject']} - {mail['sender']} ({mail['date']})")
    
    if len(other) > 5:
        lines.append(f"  ... 还有 {len(other) - 5} 封邮件")
    
    return "\n".join(lines)

if __name__ == '__main__':
    print(format_summary())