From 9b19015156b9c842f102b1cf7d8b8548a3cf87a0 Mon Sep 17 00:00:00 2001 From: "qichi.liang" Date: Wed, 31 Dec 2025 06:03:51 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E9=A3=9E=E4=B9=A6ten?= =?UTF-8?q?ant=5Faccess=5Ftoken=E8=87=AA=E5=8A=A8=E8=8E=B7=E5=8F=96?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在FeishuSheetsClient中添加_get_tenant_access_token()方法 - 实现token自动缓存和刷新机制（提前30分钟刷新） - 更新配置类支持FEISHU_APP_ID和FEISHU_APP_SECRET - 从.env中移除FEISHU_TOKEN，完全使用应用凭证 - 更新report.py和gui.py支持新的配置检查逻辑 - 更新FeishuScheduleManager配置检查逻辑 - 更新文档和示例文件说明新的配置方式 系统现在支持两种认证方式： 1. 推荐：使用应用凭证（FEISHU_APP_ID + FEISHU_APP_SECRET） 2. 备选：使用手动token（FEISHU_TOKEN） 所有功能测试通过，系统能自动获取、缓存和刷新token。 --- .env.example | 9 +- README.md | 40 +++++++- src/config.py | 31 ++++++- src/feishu/client.py | 211 +++++++++++++++++++++++++++++++++++++++--- src/feishu/manager.py | 13 ++- src/gui.py | 9 +- src/report.py | 10 +- 7 files changed, 298 insertions(+), 25 deletions(-) diff --git a/.env.example b/.env.example index da185a3..8d4eb71 100644 --- a/.env.example +++ b/.env.example @@ -5,5 +5,12 @@ CONFLUENCE_CONTENT_ID=155764524 # 飞书表格配置 FEISHU_BASE_URL=https://open.feishu.cn/open-apis/sheets/v3 -FEISHU_TOKEN=your-feishu-api-token FEISHU_SPREADSHEET_TOKEN=EgNPssi2ghZ7BLtGiTxcIBUmnVh + +# 飞书应用凭证（推荐方式，自动获取tenant_access_token） +# 创建飞书自建应用后获取app_id和app_secret +FEISHU_APP_ID=your-feishu-app-id +FEISHU_APP_SECRET=your-feishu-app-secret + +# 备选：手动配置token（不推荐，token会过期） +# FEISHU_TOKEN=your-feishu-api-token diff --git a/README.md b/README.md index 5302259..1900a3e 100644 --- a/README.md +++ b/README.md @@ -75,9 +75,16 @@ CONFLUENCE_CONTENT_ID=155764524 # 飞书表格配置（用于获取排班人员信息） FEISHU_BASE_URL=https://open.feishu.cn/open-apis/sheets/v3 -FEISHU_TOKEN=your-feishu-api-token FEISHU_SPREADSHEET_TOKEN=EgNPssi2ghZ7BLtGiTxcIBUmnVh +# 飞书应用凭证（推荐方式，自动获取tenant_access_token） +# 创建飞书自建应用后获取app_id和app_secret +FEISHU_APP_ID=your-feishu-app-id +FEISHU_APP_SECRET=your-feishu-app-secret + +# 备选：手动配置token（不推荐，token会过期） +# FEISHU_TOKEN=your-feishu-api-token + # 数据库配置 DATABASE_PATH=data/daily_logs.db @@ -205,10 +212,29 @@ GUI 功能： - **`manager.py`** - 内容管理器，提供高级内容管理功能 ### 飞书模块 (`src/feishu/`) -- **`client.py`** - 飞书 API 客户端 +- **`client.py`** - 飞书 API 客户端，支持自动获取和刷新tenant_access_token - **`parser.py`** - 排班数据解析器 - **`manager.py`** - 飞书排班管理器，缓存和刷新排班信息 +#### Token 自动获取机制 +飞书模块现在支持两种认证方式： +1. **推荐方式**：使用应用凭证（FEISHU_APP_ID + FEISHU_APP_SECRET） + - 系统会自动调用飞书API获取tenant_access_token + - token有效期2小时，系统会在过期前30分钟自动刷新 + - 无需手动管理token过期问题 + +2. **备选方式**：使用手动配置的FEISHU_TOKEN + - 兼容旧配置方式 + - token过期后需要手动更新 + - 不推荐长期使用 + +#### 如何获取应用凭证 +1. 登录飞书开放平台：https://open.feishu.cn/ +2. 创建自建应用 +3. 在"凭证与基础信息"中获取App ID和App Secret +4. 为应用添加"获取tenant_access_token"权限 +5. 将应用发布到企业（仅自建应用需要） + ### 数据库模块 (`src/database/`) - **`base.py`** - 数据库基类，提供统一的连接管理 - **`daily_logs.py`** - 每日交接班日志数据库 @@ -271,7 +297,15 @@ python3 main.py report-today 1. **连接失败**: 检查 `.env` 文件中的 API 令牌和 URL 2. **数据库错误**: 确保 `data/` 目录存在且有写入权限 3. **解析错误**: 检查 Confluence 页面结构是否发生变化 -4. **飞书数据获取失败**: 验证飞书表格权限和 token 有效性 +4. **飞书数据获取失败**: + - 验证飞书表格权限 + - 检查应用凭证是否正确（FEISHU_APP_ID + FEISHU_APP_SECRET） + - 确认应用已发布到企业（自建应用需要） + - 检查网络连接是否能访问飞书API +5. **飞书token获取失败**: + - 确认应用有"获取tenant_access_token"权限 + - 检查app_id和app_secret是否正确 + - 查看日志文件获取详细错误信息 ### 日志级别 diff --git a/src/config.py b/src/config.py index 5cdd89f..413ad9f 100644 --- a/src/config.py +++ b/src/config.py @@ -23,6 +23,8 @@ class Config: FEISHU_BASE_URL = os.getenv('FEISHU_BASE_URL', 'https://open.feishu.cn/open-apis/sheets/v3') FEISHU_TOKEN = os.getenv('FEISHU_TOKEN') FEISHU_SPREADSHEET_TOKEN = os.getenv('FEISHU_SPREADSHEET_TOKEN') + FEISHU_APP_ID = os.getenv('FEISHU_APP_ID') + FEISHU_APP_SECRET = os.getenv('FEISHU_APP_SECRET') # 数据库配置 DATABASE_PATH = os.getenv('DATABASE_PATH', 'data/daily_logs.db') @@ -70,8 +72,17 @@ class Config: errors.append("CONFLUENCE_CONTENT_ID 未配置") # 检查飞书配置（可选，但建议配置） - if not cls.FEISHU_TOKEN: - print("警告: FEISHU_TOKEN 未配置，排班功能将不可用") + has_feishu_token = bool(cls.FEISHU_TOKEN) + has_app_credentials = bool(cls.FEISHU_APP_ID and cls.FEISHU_APP_SECRET) + + if not has_feishu_token and not has_app_credentials: + print("警告: 飞书认证未配置，排班功能将不可用") + print(" 请配置 FEISHU_TOKEN 或 FEISHU_APP_ID + FEISHU_APP_SECRET") + elif has_app_credentials: + print("信息: 使用飞书应用凭证自动获取token") + elif has_feishu_token: + print("信息: 使用手动配置的FEISHU_TOKEN") + if not cls.FEISHU_SPREADSHEET_TOKEN: print("警告: FEISHU_SPREADSHEET_TOKEN 未配置，排班功能将不可用") @@ -88,7 +99,21 @@ class Config: """打印配置摘要""" print("配置摘要:") print(f" Confluence: {'已配置' if cls.CONFLUENCE_BASE_URL else '未配置'}") - print(f" 飞书: {'已配置' if cls.FEISHU_TOKEN else '未配置'}") + + # 飞书配置详情 + has_feishu_token = bool(cls.FEISHU_TOKEN) + has_app_credentials = bool(cls.FEISHU_APP_ID and cls.FEISHU_APP_SECRET) + has_spreadsheet_token = bool(cls.FEISHU_SPREADSHEET_TOKEN) + + if has_app_credentials: + feishu_status = f"应用凭证 (ID: {cls.FEISHU_APP_ID[:8]}...)" + elif has_feishu_token: + feishu_status = "手动token" + else: + feishu_status = "未配置" + + print(f" 飞书认证: {feishu_status}") + print(f" 飞书表格: {'已配置' if has_spreadsheet_token else '未配置'}") print(f" 数据库路径: {cls.DATABASE_PATH}") print(f" 每日目标TEU: {cls.DAILY_TARGET_TEU}") print(f" 排班刷新天数: {cls.SCHEDULE_REFRESH_DAYS}") diff --git a/src/feishu/client.py b/src/feishu/client.py index 6409f4e..d6b80fb 100644 --- a/src/feishu/client.py +++ b/src/feishu/client.py @@ -2,9 +2,11 @@ """ 飞书表格 API 客户端模块 统一版本，支持月度表格和年度表格 +支持自动获取和刷新 tenant_access_token """ import requests -from typing import Dict, List, Optional +import time +from typing import Dict, List, Optional, Tuple import logging from src.config import config @@ -21,32 +23,147 @@ class FeishuClientError(Exception): class FeishuSheetsClient: """飞书表格 API 客户端""" - def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None, - spreadsheet_token: Optional[str] = None): + def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None, + spreadsheet_token: Optional[str] = None, app_id: Optional[str] = None, + app_secret: Optional[str] = None): """ 初始化客户端 参数: base_url: 飞书 API 基础URL，如果为None则使用配置 - token: Bearer 认证令牌，如果为None则使用配置 + token: Bearer 认证令牌，如果为None则使用配置或自动获取 spreadsheet_token: 表格 token，如果为None则使用配置 + app_id: 飞书应用ID，用于获取tenant_access_token + app_secret: 飞书应用密钥，用于获取tenant_access_token """ self.base_url = (base_url or config.FEISHU_BASE_URL).rstrip('/') self.spreadsheet_token = spreadsheet_token or config.FEISHU_SPREADSHEET_TOKEN - self.token = token or config.FEISHU_TOKEN + self.app_id = app_id or config.FEISHU_APP_ID + self.app_secret = app_secret or config.FEISHU_APP_SECRET - self.headers = { - 'Authorization': f'Bearer {self.token}', - 'Content-Type': 'application/json', - 'Accept': 'application/json' - } + # Token管理相关属性 + self._token = token or config.FEISHU_TOKEN + self._token_expire_time = 0 # token过期时间戳 + self._token_obtained_time = 0 # token获取时间戳 # 使用 Session 重用连接 self.session = requests.Session() - self.session.headers.update(self.headers) self.session.timeout = config.REQUEST_TIMEOUT + # 初始化headers + self._update_session_headers() + logger.debug(f"飞书客户端初始化完成，基础URL: {self.base_url}") + logger.debug(f"使用应用ID: {self.app_id[:8]}... 如果配置" if self.app_id else "未配置应用ID") + + def _update_session_headers(self): + """更新session的headers""" + if self._token: + self.session.headers.update({ + 'Authorization': f'Bearer {self._token}', + 'Content-Type': 'application/json', + 'Accept': 'application/json' + }) + else: + # 如果没有token，移除Authorization头 + if 'Authorization' in self.session.headers: + del self.session.headers['Authorization'] + + def _get_tenant_access_token(self) -> Tuple[str, int]: + """ + 获取tenant_access_token + + 返回: + (token, expire_time): token字符串和过期时间（秒） + + 异常: + requests.exceptions.RequestException: 网络请求失败 + ValueError: API返回错误 + """ + if not self.app_id or not self.app_secret: + raise ValueError("未配置飞书应用ID和密钥，无法获取tenant_access_token") + + token_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" + + payload = { + "app_id": self.app_id, + "app_secret": self.app_secret + } + + headers = { + "Content-Type": "application/json; charset=utf-8" + } + + try: + logger.info(f"正在获取tenant_access_token，应用ID: {self.app_id[:8]}...") + response = requests.post(token_url, json=payload, headers=headers, timeout=config.REQUEST_TIMEOUT) + response.raise_for_status() + data = response.json() + + if data.get('code') != 0: + error_msg = f"获取tenant_access_token失败: {data.get('msg')}" + logger.error(error_msg) + raise ValueError(error_msg) + + token = data.get('tenant_access_token') + expire = data.get('expire', 7200) # 默认2小时 + + if not token: + raise ValueError("API返回的token为空") + + logger.info(f"成功获取tenant_access_token，有效期: {expire}秒") + return token, expire + + except requests.exceptions.RequestException as e: + logger.error(f"获取tenant_access_token网络请求失败: {e}") + raise + except Exception as e: + logger.error(f"获取tenant_access_token失败: {e}") + raise + + def _ensure_valid_token(self): + """ + 确保当前token有效，如果无效则重新获取 + + 返回: + bool: token是否有效 + """ + current_time = time.time() + + # 如果有手动配置的token，直接使用 + if config.FEISHU_TOKEN and self._token == config.FEISHU_TOKEN: + logger.debug("使用手动配置的FEISHU_TOKEN") + return True + + # 检查token是否过期（提前30分钟刷新） + if self._token and self._token_expire_time > 0: + time_remaining = self._token_expire_time - current_time + if time_remaining > 1800: # 剩余时间大于30分钟 + logger.debug(f"token仍然有效，剩余时间: {int(time_remaining)}秒") + return True + elif time_remaining > 0: # 剩余时间小于30分钟但大于0 + logger.info(f"token即将过期，剩余时间: {int(time_remaining)}秒，重新获取") + else: # 已过期 + logger.info("token已过期，重新获取") + + # 需要获取新token + try: + token, expire = self._get_tenant_access_token() + self._token = token + self._token_obtained_time = current_time + self._token_expire_time = current_time + expire + self._update_session_headers() + logger.info(f"token获取成功，将在 {expire} 秒后过期") + return True + except Exception as e: + logger.error(f"获取token失败: {e}") + # 如果配置了备用token，尝试使用 + if config.FEISHU_TOKEN and config.FEISHU_TOKEN != self._token: + logger.warning("使用备用FEISHU_TOKEN") + self._token = config.FEISHU_TOKEN + self._update_session_headers() + return True + return False def get_sheets_info(self) -> List[Dict[str, str]]: """ @@ -59,6 +176,10 @@ class FeishuSheetsClient: requests.exceptions.RequestException: 网络请求失败 ValueError: API返回错误 """ + # 确保token有效 + if not self._ensure_valid_token(): + raise FeishuClientError("无法获取有效的飞书token") + url = f'{self.base_url}/spreadsheets/{self.spreadsheet_token}/sheets/query' try: @@ -104,6 +225,10 @@ class FeishuSheetsClient: requests.exceptions.RequestException: 网络请求失败 ValueError: API返回错误 """ + # 确保token有效 + if not self._ensure_valid_token(): + raise FeishuClientError("无法获取有效的飞书token") + if range_ is None: range_ = config.SHEET_RANGE @@ -134,6 +259,26 @@ class FeishuSheetsClient: logger.error(f"解析表格数据失败: {e}, sheet_id: {sheet_id}") raise + def get_token_info(self) -> Dict[str, any]: + """ + 获取当前token信息 + + 返回: + token信息字典 + """ + current_time = time.time() + time_remaining = self._token_expire_time - current_time if self._token_expire_time > 0 else 0 + + return { + 'has_token': bool(self._token), + 'token_preview': self._token[:20] + '...' if self._token and len(self._token) > 20 else self._token, + 'token_obtained_time': self._token_obtained_time, + 'token_expire_time': self._token_expire_time, + 'time_remaining': max(0, time_remaining), + 'using_app_credentials': bool(self.app_id and self.app_secret), + 'using_manual_token': bool(config.FEISHU_TOKEN and self._token == config.FEISHU_TOKEN) + } + def test_connection(self) -> bool: """ 测试飞书连接是否正常 @@ -142,6 +287,12 @@ class FeishuSheetsClient: 连接是否正常 """ try: + # 首先测试token获取 + if not self._ensure_valid_token(): + logger.error("无法获取有效的飞书token") + return False + + # 然后测试表格访问 sheets = self.get_sheets_info() if sheets: logger.info(f"飞书连接测试成功，找到 {len(sheets)} 个表格") @@ -152,6 +303,27 @@ class FeishuSheetsClient: except Exception as e: logger.error(f"飞书连接测试失败: {e}") return False + + def refresh_token(self) -> bool: + """ + 强制刷新token + + 返回: + 刷新是否成功 + """ + try: + logger.info("强制刷新token...") + current_time = time.time() + token, expire = self._get_tenant_access_token() + self._token = token + self._token_obtained_time = current_time + self._token_expire_time = current_time + expire + self._update_session_headers() + logger.info(f"token刷新成功，将在 {expire} 秒后过期") + return True + except Exception as e: + logger.error(f"强制刷新token失败: {e}") + return False if __name__ == '__main__': @@ -164,8 +336,17 @@ if __name__ == '__main__': # 测试连接 client = FeishuSheetsClient() + # 显示token信息 + token_info = client.get_token_info() + print("当前token信息:") + print(f" 是否有token: {token_info['has_token']}") + print(f" token预览: {token_info['token_preview']}") + print(f" 剩余时间: {int(token_info['time_remaining'])}秒") + print(f" 使用应用凭证: {token_info['using_app_credentials']}") + print(f" 使用手动token: {token_info['using_manual_token']}") + if client.test_connection(): - print("飞书连接测试成功") + print("\n飞书连接测试成功") # 获取表格信息 sheets = client.get_sheets_info() @@ -177,6 +358,10 @@ if __name__ == '__main__': sheet_id = sheets[0]['sheet_id'] data = client.get_sheet_data(sheet_id, 'A1:C5') print(f"获取到表格数据，版本: {data.get('revision', '未知')}") + + # 再次显示token信息 + token_info = client.get_token_info() + print(f"\n测试后token剩余时间: {int(token_info['time_remaining'])}秒") else: - print("飞书连接测试失败") + print("\n飞书连接测试失败") sys.exit(1) \ No newline at end of file diff --git a/src/feishu/manager.py b/src/feishu/manager.py index dd26893..2e078d4 100644 --- a/src/feishu/manager.py +++ b/src/feishu/manager.py @@ -42,8 +42,17 @@ class FeishuScheduleManager: def _check_config(self, token: Optional[str], spreadsheet_token: Optional[str]) -> None: """检查必要配置""" - if not token and not config.FEISHU_TOKEN: - logger.warning("飞书令牌未配置，排班功能将不可用") + # 检查是否有任何可用的认证方式 + has_token = bool(token or config.FEISHU_TOKEN) + has_app_credentials = bool(config.FEISHU_APP_ID and config.FEISHU_APP_SECRET) + + if not has_token and not has_app_credentials: + logger.warning("飞书认证未配置，排班功能将不可用") + logger.warning("请配置 FEISHU_TOKEN 或 FEISHU_APP_ID + FEISHU_APP_SECRET") + elif has_app_credentials: + logger.info("使用飞书应用凭证自动获取token") + elif has_token: + logger.info("使用手动配置的FEISHU_TOKEN") if not spreadsheet_token and not config.FEISHU_SPREADSHEET_TOKEN: logger.warning("飞书表格令牌未配置，排班功能将不可用") diff --git a/src/gui.py b/src/gui.py index 8112bcb..3de2584 100644 --- a/src/gui.py +++ b/src/gui.py @@ -487,7 +487,13 @@ class OrbitInGUI: try: # 1. 检查飞书配置，如果配置完整则刷新排班信息 - if config.FEISHU_TOKEN and config.FEISHU_SPREADSHEET_TOKEN: + # 支持应用凭证和手动token两种方式 + has_feishu_config = bool(config.FEISHU_SPREADSHEET_TOKEN) and ( + bool(config.FEISHU_APP_ID and config.FEISHU_APP_SECRET) or + bool(config.FEISHU_TOKEN) + ) + + if has_feishu_config: try: self.log_message("正在刷新排班信息...") self.logger.info("正在刷新排班信息...") @@ -507,6 +513,7 @@ class OrbitInGUI: else: self.log_message("飞书配置不完整，跳过排班信息刷新") self.logger.warning("飞书配置不完整，跳过排班信息刷新") + self.logger.warning("需要配置 FEISHU_SPREADSHEET_TOKEN 和 (FEISHU_APP_ID+FEISHU_APP_SECRET 或 FEISHU_TOKEN)") # 2. 尝试获取最新的作业数据 self.log_message("正在尝试获取最新作业数据...") diff --git a/src/report.py b/src/report.py index 4b86480..4e030e2 100644 --- a/src/report.py +++ b/src/report.py @@ -181,9 +181,15 @@ class DailyReportGenerator: 班次人员字典 """ try: - # 检查飞书配置 - if not config.FEISHU_TOKEN or not config.FEISHU_SPREADSHEET_TOKEN: + # 检查飞书配置（支持应用凭证和手动token两种方式） + has_feishu_config = bool(config.FEISHU_SPREADSHEET_TOKEN) and ( + bool(config.FEISHU_APP_ID and config.FEISHU_APP_SECRET) or + bool(config.FEISHU_TOKEN) + ) + + if not has_feishu_config: logger.warning("飞书配置不完整，跳过排班信息获取") + logger.warning("需要配置 FEISHU_SPREADSHEET_TOKEN 和 (FEISHU_APP_ID+FEISHU_APP_SECRET 或 FEISHU_TOKEN)") return self._empty_personnel() # 初始化飞书排班管理器