Evilom
ab2ca1d836
Reusable 6-step pipeline for collecting DingTalk group messages, extracting Feishu doc links, fetching content, and generating summaries. Skills structure: - SKILL.md: trigger rules, workflow, config reference - config.yaml: group IDs, collection settings - scripts/paths.py: shared path resolution - scripts/step1-6: modular pipeline steps - scripts/run_all.py: one-click runner
65 lines
3.3 KiB
Python
65 lines
3.3 KiB
Python
"""Step 5: 生成结构化总结"""
|
|
import argparse, json, os, sys
|
|
from datetime import datetime, timedelta
|
|
from collections import defaultdict
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from paths import DATA_DIR, LINKS_DIR, DOCS_DIR, REPORTS_DIR
|
|
|
|
def load_json(path):
|
|
if not os.path.isfile(path): return None
|
|
with open(path,"r",encoding="utf-8") as f: return json.load(f)
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("--since",type=str); p.add_argument("--days",type=int)
|
|
p.add_argument("--full",action="store_true"); p.add_argument("--brief",action="store_true")
|
|
args = p.parse_args()
|
|
|
|
messages = load_json(os.path.join(DATA_DIR,"raw-messages","all_messages_combined.json")) or {}
|
|
content = load_json(os.path.join(LINKS_DIR,"all_feishu_content.json")) or {}
|
|
files = load_json(os.path.join(LINKS_DIR,"all_file_attachments.json")) or []
|
|
docs = {}
|
|
for d in content.get("documents",[]):
|
|
did = d.get("doc_id","") or d.get("url","")
|
|
if did: docs[did] = d
|
|
|
|
if args.since:
|
|
since = datetime.now().strftime("%Y-%m-%d 00:00:00") if args.since=="today" else (datetime.now()-timedelta(days=1)).strftime("%Y-%m-%d 00:00:00") if args.since=="yesterday" else args.since
|
|
messages = {g:[m for m in msgs if m.get("createTime","")>=since] for g,msgs in messages.items()}
|
|
elif args.days:
|
|
since = (datetime.now()-timedelta(days=args.days)).strftime("%Y-%m-%d 00:00:00")
|
|
messages = {g:[m for m in msgs if m.get("createTime","")>=since] for g,msgs in messages.items()}
|
|
|
|
lines = [f"# 钉钉群聊飞书文档采集报告", f"> 生成: {datetime.now().strftime('%Y-%m-%d %H:%M')}", ""]
|
|
total_m = sum(len(v) for v in messages.values())
|
|
lines += ["## 数据总览","","| 指标 | 数量 |","|------|------|",
|
|
f"| 消息 | {total_m} |", f"| 文档 | {len(docs)} |", f"| 附件 | {len(files)} |", ""]
|
|
|
|
for grp, msgs in messages.items():
|
|
if not msgs: continue
|
|
times = sorted([m.get("createTime","") for m in msgs])
|
|
gd = [d for d in docs.values() if d.get("group")==grp]
|
|
gf = [f for f in files if f.get("group")==grp]
|
|
lines += [f"### {grp}", f"- 消息: {len(msgs)} ({times[0]} ~ {times[-1]})", f"- 文档: {len(gd)}", f"- 附件: {len(gf)}", ""]
|
|
if args.brief: continue
|
|
senders = defaultdict(lambda: {"m":0,"d":0})
|
|
for m in msgs: senders[m.get("sender","?")]["m"]+=1
|
|
for d in gd: senders[d.get("sender","?")]["d"]+=1
|
|
lines += ["| 发送者 | 消息 | 文档 |","|--------|------|------|"]
|
|
for s,st in sorted(senders.items(),key=lambda x:-(x[1]["m"]+x[1]["d"])):
|
|
lines.append(f"| {s} | {st['m']} | {st['d']} |")
|
|
lines.append("")
|
|
lines.append("**最新消息:**")
|
|
for m in sorted(msgs,key=lambda x:x.get("createTime",""),reverse=True)[:10]:
|
|
c = m.get('content','')[:80].replace('\n',' ')
|
|
lines.append(f"- [{m.get('createTime','')}] **{m.get('sender','?')}**: {c}")
|
|
lines.append("")
|
|
|
|
os.makedirs(REPORTS_DIR, exist_ok=True)
|
|
report = "\n".join(lines)
|
|
for name in ["latest_summary.md", f"summary_{datetime.now().strftime('%Y%m%d')}.md"]:
|
|
with open(os.path.join(REPORTS_DIR,name),"w",encoding="utf-8") as f: f.write(report)
|
|
print(f"报告已生成: {REPORTS_DIR}")
|
|
|
|
if __name__ == "__main__": main()
|