Evilom
64 lines
2.4 KiB
Python
64 lines
2.4 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()
|
|
failures = 0
|
|
|
|
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"]
|
|
failures += 1 if run("step1_collect_messages.py", s1) else 0
|
|
|
|
s2 = [] if args.full else ["--incremental"]
|
|
failures += 1 if run("step2_extract_links.py", s2) else 0
|
|
|
|
s3 = [] if args.full else ["--incremental"]
|
|
failures += 1 if run("step3_fetch_feishu_docs.py", s3) else 0
|
|
|
|
if not args.skip_download:
|
|
s4 = [] if args.full else ["--incremental"]
|
|
failures += 1 if run("step4_download_files.py", s4) else 0
|
|
|
|
s5 = []
|
|
if args.quick: s5 += ["--since", "today", "--brief"]
|
|
elif args.since: s5 += ["--since", args.since]
|
|
elif args.days: s5 += ["--days", str(args.days)]
|
|
failures += 1 if run("step5_generate_summary.py", s5) else 0
|
|
|
|
if not args.skip_kb:
|
|
failures += 1 if run("step6_update_knowledge_base.py") else 0
|
|
|
|
status = "全部完成" if failures == 0 else f"完成，但有 {failures} 步失败"
|
|
print(f"\n{'='*50}\n {status}! ({time.time()-t0:.1f}s)\n{'='*50}")
|
|
return 1 if failures else 0
|
|
|
|
if __name__ == "__main__": sys.exit(main())
|