157 lines
6.2 KiB
Python
157 lines
6.2 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""从钉邮周报 eml 文件批量解析并归档到 weekly-reports 目录与 SQLite 数据库"""
|
||
import glob
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import sys
|
||
from datetime import datetime, date
|
||
from email import policy
|
||
from email.parser import BytesParser
|
||
|
||
BASE = "/home/yangxuan/.openclaw/workspace-resume/weekly-reports"
|
||
INBOUND = "/home/yangxuan/.openclaw/media/inbound"
|
||
DB = os.path.join(BASE, "weekly_reports.db")
|
||
ARCHIVE_DIR = os.path.join(BASE, "2026")
|
||
|
||
# 问题标题之后的签名词，用于截断
|
||
SIGN_TERMS = ("best regards", "杨轩", "地址", "手机", "邮箱", "yangxuan@")
|
||
|
||
def decode_b64_part(part):
|
||
payload = part.get_payload(decode=True)
|
||
if payload is None:
|
||
return ""
|
||
charset = part.get_content_charset() or "utf-8"
|
||
try:
|
||
return payload.decode(charset)
|
||
except Exception:
|
||
return payload.decode("utf-8", errors="replace")
|
||
|
||
def parse_eml(path):
|
||
with open(path, "rb") as f:
|
||
msg = BytesParser(policy=policy.default).parse(f)
|
||
plain = ""
|
||
for part in msg.walk():
|
||
if part.get_content_type() == "text/plain":
|
||
plain = decode_b64_part(part)
|
||
break
|
||
return plain, str(msg["Subject"] or "")
|
||
|
||
def extract_weekly_data(plain):
|
||
"""返回 (project, main_task, days, problems)"""
|
||
days = {} # key: (yyyy,mm,dd) -> list[item]; 按 ymd 排序
|
||
project = ""
|
||
main_task = ""
|
||
problems = []
|
||
cur_ymd = None
|
||
in_problem = False
|
||
for raw in plain.splitlines():
|
||
line = raw.strip()
|
||
if not line:
|
||
continue
|
||
# 项目名称
|
||
if line.startswith("项目名称"):
|
||
project = re.sub(r"^[📅]*\s*项目名称[：:]?\s*", "", line).strip()
|
||
continue
|
||
if line.startswith("主要任务"):
|
||
main_task = re.sub(r"^[📅]*\s*主要任务[：:]?\s*", "", line).strip()
|
||
continue
|
||
# 存在问题 标题
|
||
if "存在问题" in line or re.match(r"^\s*问题", line):
|
||
in_problem = True
|
||
cur_ymd = None
|
||
continue
|
||
# 其他小标题（下周计划 等）退出问题区（避免误匹配正文，如"本周..."）
|
||
if re.match(r"^(下周计划|下周计划:|备注[：:]|说明[：:]|下月计划)", line):
|
||
in_problem = False
|
||
cur_ymd = None
|
||
continue
|
||
# 签名开始，停止一切处理
|
||
low = line.lower()
|
||
if any(s in low for s in SIGN_TERMS):
|
||
break
|
||
# 日期行：可带 emoji、可带/不带年份
|
||
m = re.match(r"^[📅🔹•]?\s*(周一|周二|周三|周四|周五|周六|周日)[（(]\s*(\d{4})-(\d{2})-(\d{2})\s*[）)]", line)
|
||
if not m:
|
||
m = re.match(r"^[📅🔹•]?\s*(周一|周二|周三|周四|周五|周六|周日)[（(]\s*(\d{2})-(\d{2})\s*[）)]", line)
|
||
if m:
|
||
wd = m.group(1)
|
||
if len(m.groups()) == 4:
|
||
y, mm, dd = int(m.group(2)), m.group(3), m.group(4)
|
||
else:
|
||
# 需要年份，用文件主题中的年份，这里用 2026 占位，调用方按主题校正
|
||
y = 2026; mm, dd = m.group(2), m.group(3)
|
||
cur_ymd = (y, mm, dd, wd)
|
||
days.setdefault(cur_ymd, [])
|
||
in_problem = False
|
||
continue
|
||
# 每日明细内容：* 与文字可能被拆成两行，故 cur_ymd 有效时任何非签名行都归入
|
||
if in_problem:
|
||
# 问题区（仅"存在问题"标题之后）
|
||
if line not in ("无", "暂无", "None", ""):
|
||
problems.append(re.sub(r"^[-*·●▪◦]\s*", "", line).strip())
|
||
elif cur_ymd:
|
||
item = re.sub(r"^[-*·●▪◦]\s*", "", line).strip()
|
||
if item:
|
||
days[cur_ymd].append(item)
|
||
return project, main_task, days, problems
|
||
|
||
def main():
|
||
files = sorted(glob.glob(os.path.join(INBOUND, "*.eml")))
|
||
if not files:
|
||
print("未找到 eml 文件")
|
||
return
|
||
results = []
|
||
for path in files:
|
||
fname = os.path.basename(path)
|
||
m = re.search(r"(\d{4}-\d{2}-\d{2})_(\d{4}-\d{2}-\d{2})", fname)
|
||
if not m:
|
||
print(f"⏭ {fname}: 文件名不含日期，跳过")
|
||
continue
|
||
start, end = m.group(1), m.group(2)
|
||
plain, subject = parse_eml(path)
|
||
project, main_task, days, problems = extract_weekly_data(plain)
|
||
y0 = int(start[:4])
|
||
from datetime import date as _d
|
||
iso = _d.fromisoformat(start).isocalendar()
|
||
wnum = iso[1]
|
||
fn = os.path.join(ARCHIVE_DIR, f"2026-W{wnum:02d}-周报.md")
|
||
# 根据项目名确定标题项目代号
|
||
proj_code = "G5"
|
||
if "G6" in project or "G6" in fname:
|
||
proj_code = "G6"
|
||
# 构建 Markdown
|
||
lines = [f"# {proj_code} 开发周报 (2026-W{wnum:02d})", "",
|
||
f"**日期范围：** {start} ~ {end}", "",
|
||
"## 基本信息", f"项目名称： {project or '维云智造G5'}", f"主要任务： {main_task}", "",
|
||
"## 本周工作内容", ""]
|
||
for (y, mm, dd, wd) in sorted(days.keys(), key=lambda x: (x[0], int(x[1]), int(x[2]))):
|
||
ymd = f"{y}-{mm}-{dd}"
|
||
lines.append(f"{wd}（{ymd}）")
|
||
for item in days[(y, mm, dd, wd)]:
|
||
lines.append(f" - {item}")
|
||
lines.append("")
|
||
lines += ["## 存在问题"]
|
||
if problems:
|
||
for p in problems:
|
||
lines.append(f" - {p}")
|
||
else:
|
||
lines.append("无")
|
||
lines += ["", "---", f"*归档时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}*",
|
||
"*来源：解析周报邮件 eml 文件*", ""]
|
||
md = "\n".join(lines)
|
||
os.makedirs(ARCHIVE_DIR, exist_ok=True)
|
||
with open(fn, "w") as f:
|
||
f.write(md)
|
||
# 清理空问题条目（* ）
|
||
print(f"✅ 2026-W{wnum:02d} | {fname[:38]}...")
|
||
print(f" 项目={project} | 任务={main_task[:30]} | 天={len(days)} | 问题={problems}")
|
||
results.append((wnum, start, end, project, main_task, len(days), len(problems)))
|
||
print("\n===== 汇总 =====")
|
||
for r in results:
|
||
print(f" W{r[0]:02d} ({r[1]}~{r[2]}) | {r[3]} | 天{r[5]} 问题{r[6]}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|