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
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
"""一键执行完整采集流程
|
|
|
|
用法:
|
|
python run_all.py # 增量（最近3天）
|
|
python run_all.py --days 1 # 最近1天
|
|
python run_all.py --full # 全量
|
|
python run_all.py --quick # 只看今天
|
|
python run_all.py --skip-download # 跳过文件下载
|
|
python run_all.py --skip-kb # 跳过知识库更新
|
|
"""
|
|
|
|
import argparse, subprocess, sys, os, time
|
|
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
def run(script, args=None):
|
|
cmd = [sys.executable, os.path.join(SCRIPT_DIR, script)] + (args or [])
|
|
print(f"\n{'='*50}\n {script} {' '.join(args or [])}\n{'='*50}")
|
|
t = time.time()
|
|
r = subprocess.run(cmd)
|
|
print(f" {'OK' if r.returncode==0 else 'WARN'} ({time.time()-t:.1f}s)")
|
|
return r.returncode
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("--full",action="store_true"); p.add_argument("--days",type=int)
|
|
p.add_argument("--since",type=str); p.add_argument("--quick",action="store_true")
|
|
p.add_argument("--skip-download",action="store_true"); p.add_argument("--skip-kb",action="store_true")
|
|
args = p.parse_args()
|
|
t0 = time.time()
|
|
|
|
s1 = []
|
|
if args.full: s1.append("--full")
|
|
elif args.since: s1 += ["--since", args.since]
|
|
elif args.days: s1 += ["--days", str(args.days)]
|
|
elif args.quick: s1 += ["--days", "1"]
|
|
run("step1_collect_messages.py", s1)
|
|
|
|
s2 = [] if args.full else ["--incremental"]
|
|
run("step2_extract_links.py", s2)
|
|
|
|
s3 = [] if args.full else ["--incremental"]
|
|
run("step3_fetch_feishu_docs.py", s3)
|
|
|
|
if not args.skip_download:
|
|
s4 = [] if args.full else ["--incremental"]
|
|
run("step4_download_files.py", s4)
|
|
|
|
s5 = []
|
|
if args.quick: s5 += ["--since", "today", "--brief"]
|
|
elif args.since: s5 += ["--since", args.since]
|
|
elif args.days: s5 += ["--days", str(args.days)]
|
|
run("step5_generate_summary.py", s5)
|
|
|
|
if not args.skip_kb:
|
|
run("step6_update_knowledge_base.py")
|
|
|
|
print(f"\n{'='*50}\n 全部完成! ({time.time()-t0:.1f}s)\n{'='*50}")
|
|
|
|
if __name__ == "__main__": main()
|