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
32 lines
1.3 KiB
Python
32 lines
1.3 KiB
Python
"""Step 4: 下载钉钉群文件附件"""
|
|
import argparse, json, os, subprocess, sys
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from paths import LINKS_DIR, DOWNLOAD_DIR, find_dws
|
|
|
|
HTML_MD = {".html",".htm",".md",".markdown",".txt"}
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(); p.add_argument("--incremental",action="store_true"); args = p.parse_args()
|
|
path = os.path.join(LINKS_DIR,"all_file_attachments.json")
|
|
if not os.path.isfile(path): print("ERROR: 先运行 step2"); sys.exit(1)
|
|
with open(path,"r",encoding="utf-8") as f: files = json.load(f)
|
|
print(f"附件: {len(files)} 个")
|
|
dws = find_dws()
|
|
new = 0
|
|
for fi in files:
|
|
fid, name = fi.get("fileId",""), fi.get("name","unknown")
|
|
if not fid: continue
|
|
ext = os.path.splitext(name)[1].lower()
|
|
subdir = "html-md" if ext in HTML_MD else "other"
|
|
out = os.path.join(DOWNLOAD_DIR, subdir)
|
|
os.makedirs(out, exist_ok=True)
|
|
print(f" {name} -> {subdir}/")
|
|
try:
|
|
r = subprocess.run([dws,"drive","download","--node",fid,"--output",out],capture_output=True,timeout=120)
|
|
if r.returncode==0: new+=1
|
|
else: print(f" FAIL")
|
|
except Exception as e: print(f" ERR: {e}")
|
|
print(f"\n下载 {new} 个 -> {DOWNLOAD_DIR}")
|
|
|
|
if __name__ == "__main__": main()
|