read_eml.py

下载原文件

"""离线解码邮件全文。分类只匹配本例明确措辞，未识别项回原文核对。
python3 read_eml.py 回执邮件 --json 邮件读取.json
不指定--json时输出完整正文CSV。
"""
import argparse
import csv
import html
import json
import re
import sys
from datetime import timezone, timedelta
from email import policy
from email.parser import BytesParser
from email.utils import parsedate_to_datetime
from pathlib import Path


def classify(subject, body, auto):
    if auto: return '自动回复'
    if '反馈已录系统' in body: return '反馈已录'
    if '补发一面反馈' in body or ('反馈先发邮件' in body and '系统我先不录' in body): return '反馈未录'
    if '能否改到' in body or '想改期' in body: return '改期请求'
    if re.search(r'确认.*参加.*面', body, re.S): return '排期确认'
    if re.search(r'接受\s*offer', body, re.I): return '接受Offer'
    if '请反馈' in subject or '麻烦今天录' in body: return '催办'
    return '未识别，需核原文'


def read_messages(folder):
    records = []
    for path in sorted(Path(folder).glob('*.eml')):
        message = BytesParser(policy=policy.default).parsebytes(path.read_bytes())
        part = message.get_body(preferencelist=('plain', 'html'))
        if part is None: raise ValueError(f'邮件没有可读正文：{path.name}')
        body = part.get_content()
        if part.get_content_subtype() == 'html':
            body = re.sub(r'</?(?:p|tr|td|th|br|div)\b[^>]*>', '\n', body, flags=re.I)
            body = html.unescape(re.sub(r'<[^>]+>', '', body))
        subject = str(message.get('Subject', ''))
        timestamp = parsedate_to_datetime(message['Date'])
        if timestamp.tzinfo is None: raise ValueError(f'邮件时间缺时区：{path.name}')
        timestamp = timestamp.astimezone(timezone(timedelta(hours=8)))
        auto = bool(message.get('Auto-Submitted') and message['Auto-Submitted'] != 'no') or str(message.get('X-Autoreply', '')).lower() == 'yes' or subject.startswith(('自动回复', 'Automatic reply'))
        records.append({'文件': path.name, '发件人': str(message.get('From', '')), '时间': timestamp.isoformat(), '主题': subject, '类型': classify(subject, body, auto), '正文编号': sorted(set(re.findall(r'APP-\d{3}-\d{2}', subject+'\n'+body))), '文件名编号': sorted(set(re.findall(r'APP-\d{3}-\d{2}', path.name))), '号码代号': sorted(set(re.findall(r'\bTEL-\d+\b', body))), '正文': body})
    if not records: raise ValueError('没有找到.eml邮件')
    return records


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('folder'); parser.add_argument('--json', type=Path)
    args = parser.parse_args(); records = read_messages(args.folder)
    if args.json:
        if args.json.exists(): raise ValueError('输出已存在，请换新文件名')
        args.json.write_text(json.dumps(records, ensure_ascii=False, indent=2)+'\n', encoding='utf-8')
    else:
        writer = csv.writer(sys.stdout)
        fields = ['文件','发件人','时间','主题','类型','正文编号','文件名编号','正文']
        writer.writerow(fields)
        for record in records: writer.writerow([record[k] if not isinstance(record[k],list) else ';'.join(record[k]) for k in fields])


if __name__ == '__main__':
    try: main()
    except (ValueError,OSError,TypeError) as exc:
        print(f'邮件读取未完成：{exc}', file=sys.stderr); sys.exit(2)