58 lines
2.4 KiB
Python
58 lines
2.4 KiB
Python
import json
|
||
import csv
|
||
import re
|
||
|
||
import lark_oapi as lark
|
||
from lark_oapi.api.drive.v1 import *
|
||
|
||
|
||
def insert_to_tree(tree: list, path: list, file_item: dict = None):
|
||
current_level = tree
|
||
for folder_name in path:
|
||
existing = next((item for item in current_level if item.get("folder_name") == folder_name), None)
|
||
if not existing:
|
||
existing = {"folder_name": folder_name, "children": []}
|
||
current_level.append(existing)
|
||
current_level = existing["children"]
|
||
if file_item: current_level.append(file_item)
|
||
def extract_folder_token_from_url(url: str) -> str:
|
||
return re.search(r"/folder/([a-zA-Z0-9]+)", url).group(1) if re.search(r"/folder/([a-zA-Z0-9]+)", url) else ""
|
||
|
||
def list_files_recursive(client: lark.Client, folder_token: str, option: lark.RequestOption, tree: list, path: list = ["根目录"]):
|
||
response = client.drive.v1.file.list(ListFileRequest.builder().folder_token(folder_token).page_size(50)
|
||
.order_by("EditedTime").direction("DESC").user_id_type("open_id").build(), option)
|
||
if not response.success():
|
||
insert_to_tree(tree, path, {"file_name": f"[ERROR] 拉取失败 folder_token={folder_token}", "file_type": "error", "file_url": f"code={response.code}, msg={response.msg}"})
|
||
return
|
||
for file in response.data.files:
|
||
if file.type == "folder":
|
||
list_files_recursive(client, file.token, option, tree, path + [file.name])
|
||
else:
|
||
insert_to_tree(tree, path, {"file_name": file.name, "file_type": file.type, "file_url": file.url})
|
||
|
||
|
||
def main():
|
||
# 替换成你自己的
|
||
app_id = "cli_a8180bd0bb6e100d"
|
||
app_secret = "vC0j95daj8X0wNGlNfiFtfyuhrocvfyC"
|
||
folder_token = "Ee11fb0Dll8Zhadni4KcTrVYnvb" # 可以通过 URL 提取最后一段得到
|
||
# 创建 client
|
||
client = lark.Client.builder() \
|
||
.app_id(app_id) \
|
||
.app_secret(app_secret) \
|
||
.log_level(lark.LogLevel.DEBUG) \
|
||
.build()
|
||
|
||
# 创建访问凭证 option（自动带 tenant_access_token）
|
||
option = lark.RequestOption.builder().build()
|
||
|
||
# 写入 CSV 文件
|
||
with open("output/feishu_folder_list.csv", "w", newline="", encoding="utf-8-sig") as f:
|
||
writer = csv.writer(f)
|
||
writer.writerow(["文件类型", "文件名", "文件URL", "上一级目录"])
|
||
list_files_recursive(client, folder_token, option, writer)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|