Files
dc-docs/scripts/daily_feishu_collector.py
T
Evilom fe5505343e feat: 钉钉群聊飞书文档收集与知识图谱系统初始提交
- 通过悟空(dws CLI)拉取dc战略问题研究院+创新组两个群的消息(377条)
- 提取190个飞书链接、18个文件附件
- 下载HTML/MD/XLSX等报告文件到output/downloaded-files/
- 构建知识图谱(JSON+HTML可视化)
- 生成Obsidian知识库(28个页面,7大主题)
- 生成花园世界全量汇总报告
- 所有脚本路径改为相对路径,便于迁移
2026-06-02 20:24:25 +08:00

61 lines
2.0 KiB
Python

#!/usr/bin/env python3
"""每日收集钉钉群飞书链接 - 简化版"""
import subprocess
import json
import re
import os
from datetime import datetime, timedelta
DWS = r"C:\Users\admin\.real\.bin\dws\bin\dws.exe"
GROUPS = {
"创新组": "cidMuM+itt5PeY7xNSWsv3M0g==",
"dc战略问题研究院": "cidoUneRB4Db8TAXaTrKxkQAw=="
}
def run_dws(args):
try:
result = subprocess.run([DWS] + args, capture_output=True, text=True, encoding='utf-8', timeout=20)
return json.loads(result.stdout) if result.returncode == 0 else None
except:
return None
def get_messages(group_id, since_time):
data = run_dws(["chat", "message", "list", "--group", group_id, "--time", since_time, "--limit", "200", "--format", "json"])
return data.get("result", {}).get("messages", []) if data else []
def extract_feishu_links(messages):
links = []
pattern = r'https?://[a-zA-Z0-9-]+\.feishu\.cn/\S+'
seen = set()
for msg in messages:
content = msg.get("content", "")
found = re.findall(pattern, content)
for url in found:
clean_url = url.split("?")[0]
if clean_url not in seen:
seen.add(clean_url)
links.append({"url": clean_url, "sender": msg.get("sender", "unknown"), "time": msg.get("createTime", "")})
return links
def main():
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d 00:00:00")
all_links = []
for group_name, group_id in GROUPS.items():
messages = get_messages(group_id, yesterday)
links = extract_feishu_links(messages)
for link in links:
link["group"] = group_name
all_links.extend(links)
# Save
output_file = f"feishu_links_{datetime.now().strftime('%Y%m%d')}.json"
with open(output_file, "w", encoding="utf-8") as f:
json.dump({"date": datetime.now().strftime("%Y-%m-%d"), "total": len(all_links), "links": all_links}, f, ensure_ascii=False, indent=2)
print(f"Found {len(all_links)} Feishu links")
if __name__ == "__main__":
main()