#!/usr/bin/env python3 """ Feishu Smart Doc Writer 飞书智能文档写入器 - 自动分段、分批写入 支持首次使用自动引导配置 核心功能： 1. 智能分块写入 - 解决飞书API字数限制导致的空白文档 2. 自动转移所有权 - 创建文档后自动转移给用户 3. 首次使用引导 - 自动询问OpenID并配置 """ import json import os from typing import Dict, Optional from dataclasses import dataclass, asdict # 配置文件路径 CONFIG_PATH = os.path.expanduser("~/.openclaw/workspace/skills/feishu-smart-doc-writer/user_config.json") @dataclass class UserConfig: """用户配置""" owner_openid: str = "" permission_noted: bool = False # 用户是否已确认权限 first_time: bool = True # 是否首次使用 def save(self): """保存配置""" try: os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True) with open(CONFIG_PATH, 'w', encoding='utf-8') as f: json.dump(asdict(self), f, ensure_ascii=False, indent=2) return True except Exception as e: print(f"保存配置失败: {e}") return False @classmethod def load(cls) -> 'UserConfig': """加载配置""" try: if os.path.exists(CONFIG_PATH): with open(CONFIG_PATH, 'r', encoding='utf-8') as f: data = json.load(f) return cls(**data) except Exception as e: print(f"加载配置失败: {e}") return cls() # 引导消息模板 FIRST_TIME_GUIDE = """👋 **欢迎使用 Feishu Smart Doc Writer！** 本 Skill 可以帮助你： ✅ **智能分块写入** - 解决长文档写入时因API限制导致的空白问题 ✅ **自动转移所有权** - 创建文档后自动转移给你，拥有完全控制权 --- ## 🔧 首次使用配置 ### 第1步：获取你的 OpenID **详细步骤（精确路径）：** 1. **登录飞书开放平台** - 网址：https://open.feishu.cn 2. **进入权限管理并前往调试台** - 进入你的**相关应用** - 点击 **"权限管理"** - 搜索权限：`im:message` - 鼠标移动到 **"相关API事件"** - 选择：**【API】发送消息** - 点击右下角：**"前往API调试台"** 3. **找到 "快速复制 open_id"** - 在页面中找到 **蓝色文字** "快速复制 open_id" - 点击这个链接 4. **选择用户并复制** - 在弹出的选择框中，**选择你的账号** - 点击 **"复制"** 按钮 - 得到格式如：`ou_5b921cba0fd6e7c885276a02d730ec19` 💡 **提示**：OpenID 是以 `ou_` 开头的一串字符，不是数字ID --- ### 第2步：开通并发布权限 ⚠️ **重要：需要开通权限并发布应用新版本** **开通权限步骤：** 1. **进入权限管理** - 登录 https://open.feishu.cn - 进入你的应用 - 点击左侧菜单 **"权限管理"** 2. **搜索并开通权限** - 在搜索框输入：`docs:permission.member:transfer` - 找到权限 **"转移云文档的所有权"** - 点击 **"开通"** 按钮 3. **发布新版本（关键！）** - 开通后，点击页面右上角的 **"发布"** 按钮 - 等待发布完成（显示"已发布"状态） - ⚠️ **不发布的话，权限不会生效！** --- ## 💬 请回复配置信息 请按以下格式回复： ``` 配置OpenID：ou_你的OpenID 权限已开通并发布：是 ``` 例如： ``` 配置OpenID：ou_5b921cba0fd6e7c885276a02d730ec19 权限已开通并发布：是 ``` 配置完成后，本 Skill 将自动保存配置，之后创建文档会自动转移所有权给你！ """ async def write_smart(ctx, args: dict) -> dict: """ 智能创建飞书文档（自动分块 + 自动转移所有权） 首次使用时会自动引导配置。 Args: ctx: OpenClaw 上下文 args: 包含 title, content, folder_token, chunk_size, show_progress Returns: {"doc_url": "...", "doc_token": "...", "chunks_count": N, "owner_transferred": True/False} """ # 加载用户配置 config = UserConfig.load() # 首次使用或未完成配置，显示引导 if config.first_time or not config.owner_openid: return { "doc_url": None, "doc_token": None, "chunks_count": 0, "owner_transferred": False, "need_config": True, "message": FIRST_TIME_GUIDE } # 已配置，正常执行 title = args.get("title") content = args.get("content", "") folder_token = args.get("folder_token") chunk_size = args.get("chunk_size", 2000) show_progress = args.get("show_progress", True) if not title: raise ValueError("必须提供 title 参数") from .feishu_smart_doc_writer import FeishuDocWriter, ChunkConfig chunk_config = ChunkConfig( chunk_size=chunk_size, show_progress=show_progress ) writer = FeishuDocWriter(ctx, chunk_config) try: # 使用配置的 owner_openid 自动转移 result = await writer.write_document_with_transfer( title=title, content=content, folder_token=folder_token, owner_openid=config.owner_openid ) transfer_msg = "，所有权已转移" if result.get("owner_transferred") else "" return { "doc_url": result["doc_url"], "doc_token": result["doc_token"], "chunks_count": result["chunks_count"], "owner_transferred": result["owner_transferred"], "need_config": False, "message": f"✅ 文档创建成功，共分 {result['chunks_count']} 块写入{transfer_msg}" } except Exception as e: return { "doc_url": None, "doc_token": None, "chunks_count": 0, "owner_transferred": False, "need_config": False, "message": f"❌ 创建失败: {e}" } async def configure(ctx, args: dict) -> dict: """ 配置 Skill 用户首次使用时，通过此工具配置 OpenID。 Args: ctx: OpenClaw 上下文 args: 包含 openid, permission_checked Returns: {"success": True/False, "message": "..."} """ openid = args.get("openid", "").strip() permission_checked = args.get("permission_checked", False) # 验证 OpenID 格式 if not openid: return { "success": False, "message": "❌ 请提供 OpenID" } if not openid.startswith("ou_"): return { "success": False, "message": "❌ OpenID 格式错误，应以 'ou_' 开头，请检查" } # 保存配置 config = UserConfig() config.owner_openid = openid config.permission_noted = permission_checked config.first_time = False if config.save(): return { "success": True, "openid": openid, "message": f"✅ 配置成功！\n\n你的 OpenID：{openid}\n\n配置已保存，现在可以使用 write_smart 创建文档，所有权会自动转移给你。" } else: return { "success": False, "message": "❌ 配置保存失败" } async def append_smart(ctx, args: dict) -> dict: """ 智能追加内容到飞书文档（自动分块） Args: ctx: OpenClaw 上下文 args: 包含 doc_url, content, chunk_size, show_progress Returns: {"success": True/False, "chunks_count": N} """ doc_url = args.get("doc_url") content = args.get("content", "") chunk_size = args.get("chunk_size", 2000) show_progress = args.get("show_progress", True) if not doc_url: raise ValueError("必须提供 doc_url 参数") from .feishu_smart_doc_writer import FeishuDocWriter, ChunkConfig, ContentChunker config = ChunkConfig( chunk_size=chunk_size, show_progress=show_progress ) writer = FeishuDocWriter(ctx, config) try: success = await writer.append_to_document(doc_url, content) # 计算分块数 chunks = ContentChunker(config).chunk_content(content) return { "success": success, "chunks_count": len(chunks), "message": f"{'✅' if success else '❌'} 追加 {'成功' if success else '失败'}，共分 {len(chunks)} 块" } except Exception as e: return { "success": False, "chunks_count": 0, "message": f"❌ 追加失败: {e}" } async def transfer_ownership(ctx, args: dict) -> dict: """ 转移文档所有权 Args: ctx: OpenClaw 上下文 args: 包含 doc_url, owner_openid Returns: {"success": True/False, "message": "..."} """ doc_url = args.get("doc_url") owner_openid = args.get("owner_openid") if not doc_url or not owner_openid: raise ValueError("必须提供 doc_url 和 owner_openid 参数") from .feishu_smart_doc_writer import FeishuDocWriter, ChunkConfig config = ChunkConfig(show_progress=False) writer = FeishuDocWriter(ctx, config) try: success = await writer.transfer_ownership(doc_url, owner_openid) return { "success": success, "message": f"{'✅' if success else '❌'} 所有权转移{'成功' if success else '失败'}" } except Exception as e: return { "success": False, "message": f"❌ 转移失败: {e}" } async def get_config_status(ctx, args: dict) -> dict: """ 获取当前配置状态 Returns: {"configured": True/False, "openid": "...", "message": "..."} """ config = UserConfig.load() if config.owner_openid: return { "configured": True, "openid": config.owner_openid, "first_time": config.first_time, "message": f"✅ 已配置\nOpenID: {config.owner_openid}" } else: return { "configured": False, "openid": None, "first_time": config.first_time, "message": "⚠️ 未配置\n请使用 configure 工具进行配置" } async def search_docs(ctx, args: dict) -> dict: """ 搜索本地索引中的文档 Args: ctx: OpenClaw 上下文 args: 包含 keyword, search_in(可选) Returns: {"results": [...], "count": N, "message": "..."} """ keyword = args.get("keyword", "").strip() search_in = args.get("search_in", ["name", "summary", "tags"]) if not keyword: return { "results": [], "count": 0, "message": "❌ 请提供搜索关键词" } try: from .index_manager import IndexManager manager = IndexManager() results = manager.search_docs(keyword, search_in) # 格式化结果 formatted_results = [] for doc in results: formatted_results.append({ "name": doc.get("name", ""), "type": doc.get("type", ""), "link": doc.get("link", ""), "summary": doc.get("summary", ""), "status": doc.get("status", ""), "tags": doc.get("tags", ""), "updated": doc.get("updated", "") }) return { "results": formatted_results, "count": len(formatted_results), "message": f"✅ 找到 {len(formatted_results)} 个结果" if formatted_results else f"⚠️ 未找到包含 '{keyword}' 的文档" } except Exception as e: return { "results": [], "count": 0, "message": f"❌ 搜索失败: {e}" } async def list_docs(ctx, args: dict) -> dict: """ 列出所有文档（支持筛选） Args: ctx: OpenClaw 上下文 args: 包含 tag(可选), status(可选), limit(可选) Returns: {"results": [...], "count": N, "message": "..."} """ tag = args.get("tag") status = args.get("status") limit = args.get("limit", 50) try: from .index_manager import IndexManager manager = IndexManager() results = manager.list_docs(tag=tag, status=status, limit=limit) # 格式化结果 formatted_results = [] for doc in results: formatted_results.append({ "name": doc.get("name", ""), "type": doc.get("type", ""), "link": doc.get("link", ""), "summary": doc.get("summary", ""), "status": doc.get("status", ""), "tags": doc.get("tags", ""), "updated": doc.get("updated", "") }) # 构建消息 filter_desc = [] if tag: filter_desc.append(f"标签 '{tag}'") if status: filter_desc.append(f"状态 '{status}'") filter_text = "，".join(filter_desc) if filter_desc else "全部" return { "results": formatted_results, "count": len(formatted_results), "message": f"✅ {filter_text}文档共 {len(formatted_results)} 个" } except Exception as e: return { "results": [], "count": 0, "message": f"❌ 列出文档失败: {e}" } # 版本信息 __version__ = "1.3.0" __all__ = [ "write_smart", "append_smart", "transfer_ownership", "configure", "get_config_status", "search_docs", "list_docs", "UserConfig", "FIRST_TIME_GUIDE" ]