From dc2a55bbf4c7a0efe919276f018e788e78f194e5 Mon Sep 17 00:00:00 2001 From: "qichi.liang" Date: Wed, 31 Dec 2025 00:03:34 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E9=A3=9E=E4=B9=A6?= =?UTF-8?q?=E8=A1=A8=E6=A0=BC=E6=A8=A1=E5=9D=97=E6=94=AF=E6=8C=81=E6=8E=92?= =?UTF-8?q?=E7=8F=AD=E4=BA=BA=E5=91=98=E4=BF=A1=E6=81=AF=E8=8E=B7=E5=8F=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 src/feishu_v2.py: 飞书表格API客户端，支持数据库存储和2026年全年排班表 - 新增 src/schedule_database.py: 排班信息数据库模块，用于缓存排班数据 - 新增 docs/feishu_data_flow.md: 飞书数据流文档 - 新增 plans/feishu_scheduling_plan.md: 飞书排班表模块设计文档 - 更新 src/report.py: 使用新的飞书模块获取排班人员信息 - 更新 src/gui.py: 启动时自动获取新数据，添加auto_fetch_data方法 - 更新 .env.example: 添加飞书配置示例 - 更新 AGENTS.md: 更新项目文档 - 更新 main.py: 集成飞书模块 功能特性: 1. 支持从飞书表格获取排班人员信息 2. 支持2025年月度表格和2026年全年排班表 3. 使用SQLite数据库缓存，减少API调用 4. 自动检测表格更新 5. GUI启动时自动获取最新数据 6. 日报中正确显示次日班次人员信息 --- .env.example | 5 + AGENTS.md | 23 +- docs/feishu_data_flow.md | 179 +++++++++ main.py | 5 + plans/feishu_scheduling_plan.md | 131 +++++++ src/feishu.py | 467 +++++++++++++++++++++++ src/feishu_v2.py | 642 ++++++++++++++++++++++++++++++++ src/gui.py | 91 ++++- src/report.py | 55 ++- src/schedule_database.py | 323 ++++++++++++++++ 10 files changed, 1908 insertions(+), 13 deletions(-) create mode 100644 docs/feishu_data_flow.md create mode 100644 plans/feishu_scheduling_plan.md create mode 100644 src/feishu.py create mode 100644 src/feishu_v2.py create mode 100644 src/schedule_database.py diff --git a/.env.example b/.env.example index e75b221..da185a3 100644 --- a/.env.example +++ b/.env.example @@ -2,3 +2,8 @@ CONFLUENCE_BASE_URL=https://confluence.westwell-lab.com/rest/api CONFLUENCE_TOKEN=your-token-here 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 diff --git a/AGENTS.md b/AGENTS.md index bf89daf..22558fd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -22,7 +22,10 @@ OrbitIn/ ├── debug/ # 调试输出目录 │ └── layout_output_*.txt # 带时间戳的调试文件 ├── data/ # 数据目录 -│ └── daily_logs.db # SQLite3 数据库 +│ ├── daily_logs.db # SQLite3 数据库 +│ └── schedule_cache.json # 排班数据缓存 +├── plans/ # 设计文档目录 +│ └── feishu_scheduling_plan.md # 飞书排班表模块设计 └── src/ # 代码模块 ├── __init__.py ├── confluence.py # Confluence API 客户端 @@ -30,7 +33,8 @@ OrbitIn/ ├── parser.py # 日志解析器 ├── database.py # SQLite3 数据库操作 ├── report.py # 日报生成器 - └── gui.py # GUI 图形界面 + ├── gui.py # GUI 图形界面 + └── feishu.py # 飞书表格 API 客户端（新增） ``` ## 核心模块 @@ -69,6 +73,7 @@ OrbitIn/ - `generate_report(date)` - 生成日报 - `print_report(date)` - 打印日报 +- `get_shift_personnel(date)` - 获取班次人员（从飞书排班表获取） ### OrbitInGUI (src/gui.py:22) @@ -76,6 +81,12 @@ OrbitIn/ - 支持获取数据、生成日报、添加未统计数据 - 日报内容可复制 +### FeishuScheduleManager (src/feishu.py:150) + +- `get_schedule_for_date(date)` - 获取指定日期的排班信息 +- `get_schedule_for_today()` - 获取今天的排班信息 +- `get_schedule_for_tomorrow()` - 获取明天的排班信息 + ## 文本格式约定 - 列表前缀：`•` 用于 `ul`，数字+点用于 `ol` @@ -113,12 +124,18 @@ python3 src/gui.py ## 配置 -在 `.env` 文件中配置 Confluence 连接信息： +在 `.env` 文件中配置连接信息： ```bash +# Confluence 配置 CONFLUENCE_BASE_URL=https://confluence.westwell-lab.com/rest/api CONFLUENCE_TOKEN=your-api-token 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 ``` ## 测试模式 diff --git a/docs/feishu_data_flow.md b/docs/feishu_data_flow.md new file mode 100644 index 0000000..edcf0e8 --- /dev/null +++ b/docs/feishu_data_flow.md @@ -0,0 +1,179 @@ +# 飞书数据获取流程 + +## 整体流程 + +```mermaid +flowchart TD + A[开始: 生成日报] --> B[调用 get_shift_personnel] + B --> C[创建 FeishuScheduleManager] + C --> D[调用 get_schedule_for_date] + + D --> E[解析日期: 2025-12-30 → 12/30] + E --> F[检查缓存: data/schedule_cache.json] + + F --> G{缓存是否存在?} + G -->|是| H[直接返回缓存数据] + G -->|否| I[调用 API 获取数据] + + I --> J[调用 get_sheets_info] + J --> K[GET /spreadsheets/{token}/sheets/query] + K --> L[返回表格列表: 8月, 9月, 10月, 11月, 12月, 2026年...] + + L --> M[根据月份选择表格: 12月 → sheet_id='zcYLIk'] + M --> N[调用 get_sheet_data] + N --> O[GET /spreadsheets/{token}/values/zcYLIk!A:AF] + O --> P[返回表格数据: 姓名, 12月1日, 12月2日...] + + P --> Q[调用 ScheduleDataParser.parse] + Q --> R[解析日期列: 查找12月30日对应的列索引] + R --> S[筛选班次人员: 白班='白', 夜班='夜'] + + S --> T[返回结果: 白班人员列表, 夜班人员列表] + T --> U[保存到缓存] + U --> V[返回给日报模块] + V --> W[填充到日报中] +``` + +## API调用详情 + +### 1. 获取表格列表 + +**请求**: +``` +GET https://open.feishu.cn/open-apis/sheets/v3/spreadsheets/EgNPssi2ghZ7BLtGiTxcIBUmnVh/sheets/query +Authorization: Bearer u-dbctiP9qx1wF.wfoMV2ZHGkh1DNl14oriM8aZMI0026k +``` + +**响应**: +```json +{ + "code": 0, + "data": { + "sheets": [ + {"sheet_id": "904236", "title": "8月"}, + {"sheet_id": "ATgwLm", "title": "9月"}, + {"sheet_id": "2ml4B0", "title": "10月"}, + {"sheet_id": "y5xv1D", "title": "11月"}, + {"sheet_id": "zcYLIk", "title": "12月"}, + {"sheet_id": "R35cIj", "title": "2026年排班表"}, + {"sheet_id": "wMXHQg", "title": "12月（副本）"} + ] + } +} +``` + +### 2. 获取表格数据 + +**请求**: +``` +GET https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/EgNPssi2ghZ7BLtGiTxcIBUmnVh/values/zcYLIk!A:AF +Authorization: Bearer u-dbctiP9qx1wF.wfoMV2ZHGkh1DNl14oriM8aZMI0026k +params: { + valueRenderOption: "ToString", + dateTimeRenderOption: "FormattedString" +} +``` + +**响应**: +```json +{ + "code": 0, + "data": { + "valueRange": { + "range": "zcYLIk!A1:AF11", + "values": [ + ["姓名", "12月1日", "12月2日", "12月3日", "12月4日", ...], + ["张勤", "白", "白", "白", "白", ...], + ["刘炜彬", "白", null, "夜", "夜", ...], + ["杨俊豪", "白", "白", "白", "白", ...], + ["梁启迟", "夜", "夜", "夜", "夜", ...], + ... + ] + } + } +} +``` + +## 数据解析流程 + +### 1. 查找日期列索引 + +```python +# 查找 "12月30日" 在表头中的位置 +headers = ["姓名", "12月1日", "12月2日", ..., "12月30日", ...] +target = "12/30" # 从 "2025-12-30" 转换而来 + +# 遍历表头找到匹配的日期 +for i, header in enumerate(headers): + if header == "12月30日": + column_index = i + break +# 结果: column_index = 31 (第32列) +``` + +### 2. 筛选班次人员 + +```python +# 遍历所有人员行 +for row in values[1:]: # 跳过表头 + name = row[0] # 姓名 + shift = row[31] # 12月30日的班次 + + if shift == "白": + day_shift_list.append(name) + elif shift == "夜": + night_shift_list.append(name) + +# 结果 +# day_shift_list = ["张勤", "杨俊豪", "冯栋", "汪钦良"] +# night_shift_list = ["刘炜彬", "梁启迟"] +``` + +### 3. 生成日报输出 + +```python +day_shift_str = "、".join(day_shift_list) # "张勤、杨俊豪、冯栋、汪钦良" +night_shift_str = "、".join(night_shift_list) # "刘炜彬、梁启迟" + +# 日报中的格式 +lines.append(f"12/31 白班人员：{day_shift_str}") +lines.append(f"12/31 夜班人员：{night_shift_str}") +``` + +## 缓存机制 + +```mermaid +flowchart LR + A[首次请求] --> B[调用API] + B --> C[保存缓存: data/schedule_cache.json] + C --> D{"1小时内再次请求"} + D -->|是| E[直接读取缓存] + D -->|否| F[重新调用API] +``` + +缓存文件格式: +```json +{ + "last_update": "2025-12-30T15:00:00", + "data": { + "2025-12-12/30": { + "day_shift": "张勤、杨俊豪、冯栋、汪钦良", + "night_shift": "刘炜彬、梁启迟", + "day_shift_list": ["张勤", "杨俊豪", "冯栋", "汪钦良"], + "night_shift_list": ["刘炜彬", "梁启迟"] + } + } +} +``` + +## 关键代码位置 + +| 功能 | 文件 | 行号 | +|------|------|------| +| 飞书API客户端 | [`src/feishu.py`](src/feishu.py:10) | 10 | +| 获取表格列表 | [`src/feishu.py`](src/feishu.py:28) | 28 | +| 获取表格数据 | [`src/feishu.py`](src/feishu.py:42) | 42 | +| 数据解析器 | [`src/feishu.py`](src/feishu.py:58) | 58 | +| 缓存管理 | [`src/feishu.py`](src/feishu.py:150) | 150 | +| 主管理器 | [`src/feishu.py`](src/feishu.py:190) | 190 | +| 日报集成 | [`src/report.py`](src/report.py:98) | 98 | diff --git a/main.py b/main.py index 3056934..6afbd05 100644 --- a/main.py +++ b/main.py @@ -23,6 +23,11 @@ CONF_BASE_URL = os.getenv('CONFLUENCE_BASE_URL') CONF_TOKEN = os.getenv('CONFLUENCE_TOKEN') CONF_CONTENT_ID = os.getenv('CONFLUENCE_CONTENT_ID') +# 飞书配置（可选） +FEISHU_BASE_URL = os.getenv('FEISHU_BASE_URL') +FEISHU_TOKEN = os.getenv('FEISHU_TOKEN') +FEISHU_SPREADSHEET_TOKEN = os.getenv('FEISHU_SPREADSHEET_TOKEN') + DEBUG_DIR = 'debug' diff --git a/plans/feishu_scheduling_plan.md b/plans/feishu_scheduling_plan.md new file mode 100644 index 0000000..b44812a --- /dev/null +++ b/plans/feishu_scheduling_plan.md @@ -0,0 +1,131 @@ +# 飞书排班表模块实施计划 + +## 概述 + +添加飞书表格模块，用于获取码头作业人员的排班信息，并将人员姓名填充到日报中。 + +## 数据结构 + +### 飞书排班表格式 +``` +列A: 姓名 +列B-列AF: 12月1日-12月31日的班次（白/夜/null） +``` + +### 日报表填充逻辑 +- 根据日报中的日期（如 12/30）对应排班表中的日期列 +- 筛选出当天的白班人员名单和夜班人员名单 +- 格式：`人员1、人员2、人员3...` + +## 实现步骤 + +### 1. 创建 `src/feishu.py` 模块 + +#### FeishuSheetsClient 类 + +```python +class FeishuSheetsClient: + """飞书表格 API 客户端""" + + def __init__(self, base_url: str, token: str, spreadsheet_token: str): + # 初始化配置 + + def get_sheets_info(self) -> List[Dict]: + """获取所有表格信息（sheet_id 和 title）""" + # GET /spreadsheets/{spreadsheet_token}/sheets/query + + def get_sheet_data(self, sheet_id: str, range_: str = 'A:AF') -> Dict: + """获取指定表格的数据""" + # GET /spreadsheets/{spreadsheet_token}/values/{sheet_id}?range={range_} +``` + +#### ScheduleDataParser 类 + +```python +class ScheduleDataParser: + """排班数据解析器""" + + def parse(self, values: List[List[str]], target_date: str) -> Dict: + """ + 解析排班数据，获取指定日期的班次人员 + + 参数: + values: 飞书表格返回的二维数组 + target_date: 目标日期（格式: 12月30日 或 12/30） + + 返回: + { + 'day_shift': '张勤、刘炜彬、杨俊豪', + 'night_shift': '梁启迟、江唯、汪钦良' + } + """ +``` + +### 2. 修改 `src/report.py` + +在 `DailyReportGenerator` 类中： + +```python +def get_shift_personnel(self, date: str) -> Dict: + """获取班次人员（从飞书排班表获取）""" + # 使用 FeishuSheetsClient 获取排班数据 + # 使用 ScheduleDataParser 解析人员名单 + # 返回 day_shift 和 night_shift +``` + +### 3. 更新 `.env.example` + +```bash +# 飞书表格配置 +FEISHU_BASE_URL=https://open.feishu.cn/open-apis/sheets/v3 +FEISHU_TOKEN=your-feishu-api-token +FEISHU_SPREADSHEET_TOKEN=EgNPssi2ghZ7BLtGiTxcIBUmnVh +``` + +### 4. 添加缓存机制 + +为减少 API 调用，实现简单的缓存： + +- 缓存有效期：1小时 +- 缓存位置：`data/schedule_cache.json` +- 缓存格式： + ```json + { + "last_update": "2025-12-30T10:00:00", + "data": {...} + } + ``` + +## 技术决策 + +### 日期匹配逻辑 + +1. **日报日期到排班日期的映射**： + - 日报中的 `12/30` 对应排班表中的 `12月30日` + - 班次人员需要显示在对应日期的下一班（白班→夜班→次日白班） + +2. **sheet_id 选择策略**： + - 2025年：每月一张表，需要根据年月选择对应的 sheet + - 2026年：12个月整合在一块，使用单一 sheet + +### 错误处理 + +- API 调用失败时降级使用默认配置（如返回空的值班手机） +- 缓存过期时重新获取 +- 日期匹配失败时返回空列表 + +## 文件变更 + +| 文件 | 操作 | 说明 | +|------|------|------| +| `src/feishu.py` | 新建 | 飞书表格客户端模块 | +| `src/report.py` | 修改 | 集成人员信息获取 | +| `.env.example` | 修改 | 添加飞书配置 | +| `main.py` | 修改 | 添加飞书相关配置加载 | + +## 测试用例 + +1. **正常流程**：获取 12 月 30 日的班次人员 +2. **边界情况**：跨月日期匹配 +3. **错误处理**：API 调用失败时的降级策略 +4. **缓存测试**：验证缓存生效和过期逻辑 diff --git a/src/feishu.py b/src/feishu.py new file mode 100644 index 0000000..53f5f7f --- /dev/null +++ b/src/feishu.py @@ -0,0 +1,467 @@ +#!/usr/bin/env python3 +""" +飞书表格 API 客户端模块 +用于获取码头作业人员排班信息 +""" +import requests +import json +import os +import time +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Tuple +import logging + +logger = logging.getLogger(__name__) + + +class FeishuSheetsClient: + """飞书表格 API 客户端""" + + def __init__(self, base_url: str, token: str, spreadsheet_token: str): + """ + 初始化客户端 + + 参数: + base_url: 飞书 API 基础URL + token: Bearer 认证令牌 + spreadsheet_token: 表格 token + """ + self.base_url = base_url.rstrip('/') + self.spreadsheet_token = spreadsheet_token + self.headers = { + 'Authorization': f'Bearer {token}', + 'Content-Type': 'application/json', + 'Accept': 'application/json' + } + + def get_sheets_info(self) -> List[Dict]: + """ + 获取所有表格信息（sheet_id 和 title） + + 返回: + 表格信息列表 [{'sheet_id': 'xxx', 'title': 'xxx'}, ...] + """ + url = f'{self.base_url}/spreadsheets/{self.spreadsheet_token}/sheets/query' + + try: + response = requests.get(url, headers=self.headers, timeout=30) + response.raise_for_status() + data = response.json() + + if data.get('code') != 0: + logger.error(f"飞书API错误: {data.get('msg')}") + return [] + + sheets = data.get('data', {}).get('sheets', []) + result = [] + for sheet in sheets: + result.append({ + 'sheet_id': sheet.get('sheet_id'), + 'title': sheet.get('title') + }) + + logger.info(f"获取到 {len(result)} 个表格") + return result + + except requests.exceptions.RequestException as e: + logger.error(f"获取表格信息失败: {e}") + return [] + except Exception as e: + logger.error(f"解析表格信息失败: {e}") + return [] + + def get_sheet_data(self, sheet_id: str, range_: str = 'A:AF') -> Dict: + """ + 获取指定表格的数据 + + 参数: + sheet_id: 表格ID + range_: 数据范围，默认 A:AF (31列) + + 返回: + 飞书API返回的原始数据 + """ + # 注意：获取表格数据使用 v2 API，而不是 v3 + # 根据你提供的示例：GET 'https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{token}/values/{sheet_id}!A:AF' + url = f'{self.base_url.replace("/v3", "/v2")}/spreadsheets/{self.spreadsheet_token}/values/{sheet_id}!{range_}' + params = { + 'valueRenderOption': 'ToString', + 'dateTimeRenderOption': 'FormattedString' + } + + try: + response = requests.get(url, headers=self.headers, params=params, timeout=30) + response.raise_for_status() + data = response.json() + + if data.get('code') != 0: + logger.error(f"飞书API错误: {data.get('msg')}") + return {} + + return data.get('data', {}) + + except requests.exceptions.RequestException as e: + logger.error(f"获取表格数据失败: {e}") + return {} + except Exception as e: + logger.error(f"解析表格数据失败: {e}") + return {} + + +class ScheduleDataParser: + """排班数据解析器""" + + @staticmethod + def _parse_chinese_date(date_str: str) -> Optional[str]: + """ + 解析中文日期格式 + + 参数: + date_str: 中文日期，如 "12月30日" 或 "12/30" 或 "12月1日" + + 返回: + 标准化日期字符串 "MM月DD日" + """ + if not date_str: + return None + + # 如果是 "12/30" 格式 + if '/' in date_str: + try: + month, day = date_str.split('/') + # 移除可能的空格 + month = month.strip() + day = day.strip() + return f"{int(month)}月{int(day)}日" + except: + return None + + # 如果是 "12月30日" 格式 + if '月' in date_str and '日' in date_str: + return date_str + + # 如果是 "12月1日" 格式（已经包含"日"字） + if '月' in date_str: + # 检查是否已经有"日"字 + if '日' not in date_str: + return f"{date_str}日" + return date_str + + return None + + @staticmethod + def _find_date_column_index(headers: List[str], target_date: str) -> Optional[int]: + """ + 在表头中查找目标日期对应的列索引 + + 参数: + headers: 表头行 ["姓名", "12月1日", "12月2日", ...] + target_date: 目标日期 "12月30日" + + 返回: + 列索引（从0开始），未找到返回None + """ + if not headers or not target_date: + return None + + # 标准化目标日期 + target_std = ScheduleDataParser._parse_chinese_date(target_date) + if not target_std: + return None + + # 遍历表头查找匹配的日期 + for i, header in enumerate(headers): + header_std = ScheduleDataParser._parse_chinese_date(header) + if header_std == target_std: + return i + + return None + + def parse(self, values: List[List[str]], target_date: str) -> Dict: + """ + 解析排班数据，获取指定日期的班次人员 + + 参数: + values: 飞书表格返回的二维数组 + target_date: 目标日期（格式: "12月30日" 或 "12/30"） + + 返回: + { + 'day_shift': '张勤、刘炜彬、杨俊豪', + 'night_shift': '梁启迟、江唯、汪钦良', + 'day_shift_list': ['张勤', '刘炜彬', '杨俊豪'], + 'night_shift_list': ['梁启迟', '江唯', '汪钦良'] + } + """ + if not values or len(values) < 2: + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 第一行是表头 + headers = values[0] + date_column_index = self._find_date_column_index(headers, target_date) + + if date_column_index is None: + logger.warning(f"未找到日期列: {target_date}") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 收集白班和夜班人员 + day_shift_names = [] + night_shift_names = [] + + # 从第二行开始是人员数据 + for row in values[1:]: + if len(row) <= date_column_index: + continue + + name = row[0] if row else '' + shift = row[date_column_index] if date_column_index < len(row) else '' + + if not name or not shift: + continue + + if shift == '白': + day_shift_names.append(name) + elif shift == '夜': + night_shift_names.append(name) + + # 格式化输出 + day_shift_str = '、'.join(day_shift_names) if day_shift_names else '' + night_shift_str = '、'.join(night_shift_names) if night_shift_names else '' + + return { + 'day_shift': day_shift_str, + 'night_shift': night_shift_str, + 'day_shift_list': day_shift_names, + 'night_shift_list': night_shift_names + } + + +class ScheduleCache: + """排班数据缓存""" + + def __init__(self, cache_file: str = 'data/schedule_cache.json'): + self.cache_file = cache_file + self.cache_ttl = 3600 # 1小时 + + def load(self) -> Optional[Dict]: + """加载缓存""" + try: + if not os.path.exists(self.cache_file): + return None + + with open(self.cache_file, 'r', encoding='utf-8') as f: + cache_data = json.load(f) + + # 检查缓存是否过期 + last_update = cache_data.get('last_update') + if last_update: + last_time = datetime.fromisoformat(last_update) + if (datetime.now() - last_time).total_seconds() < self.cache_ttl: + return cache_data.get('data') + + return None + + except Exception as e: + logger.warning(f"加载缓存失败: {e}") + return None + + def save(self, data: Dict): + """保存缓存""" + try: + # 确保目录存在 + os.makedirs(os.path.dirname(self.cache_file), exist_ok=True) + + cache_data = { + 'last_update': datetime.now().isoformat(), + 'data': data + } + + with open(self.cache_file, 'w', encoding='utf-8') as f: + json.dump(cache_data, f, ensure_ascii=False, indent=2) + + logger.info(f"缓存已保存到 {self.cache_file}") + + except Exception as e: + logger.error(f"保存缓存失败: {e}") + + +class FeishuScheduleManager: + """飞书排班管理器（主入口）""" + + def __init__(self, base_url: str = None, token: str = None, + spreadsheet_token: str = None): + """ + 初始化管理器 + + 参数: + base_url: 飞书API基础URL，从环境变量读取 + token: 飞书API令牌，从环境变量读取 + spreadsheet_token: 表格token，从环境变量读取 + """ + # 从环境变量读取配置 + self.base_url = base_url or os.getenv('FEISHU_BASE_URL', 'https://open.feishu.cn/open-apis/sheets/v3') + self.token = token or os.getenv('FEISHU_TOKEN', '') + self.spreadsheet_token = spreadsheet_token or os.getenv('FEISHU_SPREADSHEET_TOKEN', '') + + if not self.token or not self.spreadsheet_token: + logger.warning("飞书配置不完整，请检查环境变量") + + self.client = FeishuSheetsClient(self.base_url, self.token, self.spreadsheet_token) + self.parser = ScheduleDataParser() + self.cache = ScheduleCache() + + def get_schedule_for_date(self, date_str: str, use_cache: bool = True) -> Dict: + """ + 获取指定日期的排班信息 + + 参数: + date_str: 日期字符串，格式 "2025-12-30" 或 "12/30" + use_cache: 是否使用缓存 + + 返回: + 排班信息字典 + """ + # 转换日期格式 + try: + if '-' in date_str: + # "2025-12-30" -> "12/30" + dt = datetime.strptime(date_str, '%Y-%m-%d') + target_date = dt.strftime('%m/%d') + year_month = dt.strftime('%Y-%m') + month_name = dt.strftime('%m月') # "12月" + else: + # "12/30" -> "12/30" + target_date = date_str + # 假设当前年份 + current_year = datetime.now().year + month = int(date_str.split('/')[0]) + year_month = f"{current_year}-{month:02d}" + month_name = f"{month}月" + except: + target_date = date_str + year_month = datetime.now().strftime('%Y-%m') + month_name = datetime.now().strftime('%m月') + + # 尝试从缓存获取 + if use_cache: + cached_data = self.cache.load() + cache_key = f"{year_month}_{target_date}" + if cached_data and cache_key in cached_data: + logger.info(f"从缓存获取 {cache_key} 的排班信息") + return cached_data[cache_key] + + # 获取表格信息 + sheets = self.client.get_sheets_info() + if not sheets: + logger.error("未获取到表格信息") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 根据月份选择对应的表格 + sheet_id = None + sheet_title = None + + # 优先查找月份表格，如 "12月" + for sheet in sheets: + title = sheet.get('title', '') + if month_name in title: + sheet_id = sheet['sheet_id'] + sheet_title = title + logger.info(f"找到月份表格: {title} (ID: {sheet_id})") + break + + # 如果没有找到月份表格，使用第一个表格 + if not sheet_id and sheets: + sheet_id = sheets[0]['sheet_id'] + sheet_title = sheets[0]['title'] + logger.warning(f"未找到 {month_name} 表格，使用第一个表格: {sheet_title}") + + if not sheet_id: + logger.error("未找到可用的表格") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 获取表格数据 + sheet_data = self.client.get_sheet_data(sheet_id) + if not sheet_data: + logger.error("未获取到表格数据") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + values = sheet_data.get('valueRange', {}).get('values', []) + if not values: + logger.error("表格数据为空") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 解析数据 + result = self.parser.parse(values, target_date) + + # 更新缓存 + if use_cache: + cached_data = self.cache.load() or {} + cache_key = f"{year_month}_{target_date}" + cached_data[cache_key] = result + self.cache.save(cached_data) + + return result + + def get_schedule_for_today(self) -> Dict: + """获取今天的排班信息""" + today = datetime.now().strftime('%Y-%m-%d') + return self.get_schedule_for_date(today) + + def get_schedule_for_tomorrow(self) -> Dict: + """获取明天的排班信息""" + tomorrow = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d') + return self.get_schedule_for_date(tomorrow) + + +if __name__ == '__main__': + # 测试代码 + import sys + + # 设置日志 + logging.basicConfig(level=logging.INFO) + + # 从环境变量读取配置 + manager = FeishuScheduleManager() + + if len(sys.argv) > 1: + date_str = sys.argv[1] + else: + date_str = datetime.now().strftime('%Y-%m-%d') + + print(f"获取 {date_str} 的排班信息...") + schedule = manager.get_schedule_for_date(date_str) + + print(f"白班人员: {schedule['day_shift']}") + print(f"夜班人员: {schedule['night_shift']}") + print(f"白班列表: {schedule['day_shift_list']}") + print(f"夜班列表: {schedule['night_shift_list']}") \ No newline at end of file diff --git a/src/feishu_v2.py b/src/feishu_v2.py new file mode 100644 index 0000000..6ad47e6 --- /dev/null +++ b/src/feishu_v2.py @@ -0,0 +1,642 @@ +#!/usr/bin/env python3 +""" +飞书表格 API 客户端模块 v2 +支持数据库存储和2026年全年排班表 +""" +import requests +import json +import os +import time +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Tuple +import logging +import hashlib + +from src.schedule_database import ScheduleDatabase + +logger = logging.getLogger(__name__) + + +class FeishuSheetsClient: + """飞书表格 API 客户端""" + + def __init__(self, base_url: str, token: str, spreadsheet_token: str): + """ + 初始化客户端 + + 参数: + base_url: 飞书 API 基础URL + token: Bearer 认证令牌 + spreadsheet_token: 表格 token + """ + self.base_url = base_url.rstrip('/') + self.spreadsheet_token = spreadsheet_token + self.headers = { + 'Authorization': f'Bearer {token}', + 'Content-Type': 'application/json', + 'Accept': 'application/json' + } + + def get_sheets_info(self) -> List[Dict]: + """ + 获取所有表格信息（sheet_id 和 title） + + 返回: + 表格信息列表 [{'sheet_id': 'xxx', 'title': 'xxx'}, ...] + """ + url = f'{self.base_url}/spreadsheets/{self.spreadsheet_token}/sheets/query' + + try: + response = requests.get(url, headers=self.headers, timeout=30) + response.raise_for_status() + data = response.json() + + if data.get('code') != 0: + logger.error(f"飞书API错误: {data.get('msg')}") + return [] + + sheets = data.get('data', {}).get('sheets', []) + result = [] + for sheet in sheets: + result.append({ + 'sheet_id': sheet.get('sheet_id'), + 'title': sheet.get('title') + }) + + logger.info(f"获取到 {len(result)} 个表格") + return result + + except requests.exceptions.RequestException as e: + logger.error(f"获取表格信息失败: {e}") + return [] + except Exception as e: + logger.error(f"解析表格信息失败: {e}") + return [] + + def get_sheet_data(self, sheet_id: str, range_: str = 'A:AF') -> Dict: + """ + 获取指定表格的数据 + + 参数: + sheet_id: 表格ID + range_: 数据范围，默认 A:AF (31列) + + 返回: + 飞书API返回的原始数据，包含revision版本号 + """ + # 注意：获取表格数据使用 v2 API，而不是 v3 + url = f'{self.base_url.replace("/v3", "/v2")}/spreadsheets/{self.spreadsheet_token}/values/{sheet_id}!{range_}' + params = { + 'valueRenderOption': 'ToString', + 'dateTimeRenderOption': 'FormattedString' + } + + try: + response = requests.get(url, headers=self.headers, params=params, timeout=30) + response.raise_for_status() + data = response.json() + + if data.get('code') != 0: + logger.error(f"飞书API错误: {data.get('msg')}") + return {} + + return data.get('data', {}) + + except requests.exceptions.RequestException as e: + logger.error(f"获取表格数据失败: {e}") + return {} + except Exception as e: + logger.error(f"解析表格数据失败: {e}") + return {} + + +class ScheduleDataParser: + """排班数据解析器（支持2026年全年排班表）""" + + @staticmethod + def _parse_chinese_date(date_str: str) -> Optional[str]: + """ + 解析中文日期格式 + + 参数: + date_str: 中文日期，如 "12月30日" 或 "12/30" 或 "12月1日" 或 "1月1日" + + 返回: + 标准化日期字符串 "M月D日" (不补零) + """ + if not date_str: + return None + + # 如果是 "12/30" 格式 + if '/' in date_str: + try: + month, day = date_str.split('/') + # 移除可能的空格和前导零 + month = month.strip().lstrip('0') + day = day.strip().lstrip('0') + return f"{int(month)}月{int(day)}日" + except: + return None + + # 如果是 "12月30日" 或 "1月1日" 格式 + if '月' in date_str and '日' in date_str: + # 移除前导零，如 "01月01日" -> "1月1日" + parts = date_str.split('月') + if len(parts) == 2: + month_part = parts[0].lstrip('0') + day_part = parts[1].rstrip('日').lstrip('0') + return f"{month_part}月{day_part}日" + return date_str + + # 如果是 "12月1日" 格式（已经包含"日"字） + if '月' in date_str: + # 检查是否已经有"日"字 + if '日' not in date_str: + return f"{date_str}日" + return date_str + + return None + + @staticmethod + def _find_date_column_index(headers: List[str], target_date: str) -> Optional[int]: + """ + 在表头中查找目标日期对应的列索引 + + 参数: + headers: 表头行 ["姓名", "12月1日", "12月2日", ...] + target_date: 目标日期 "12月30日" + + 返回: + 列索引（从0开始），未找到返回None + """ + if not headers or not target_date: + return None + + # 标准化目标日期 + target_std = ScheduleDataParser._parse_chinese_date(target_date) + if not target_std: + return None + + # 遍历表头查找匹配的日期 + for i, header in enumerate(headers): + header_std = ScheduleDataParser._parse_chinese_date(header) + if header_std == target_std: + return i + + return None + + def parse_monthly_sheet(self, values: List[List[str]], target_date: str) -> Dict: + """ + 解析月度表格数据（如12月表格） + + 参数: + values: 飞书表格返回的二维数组 + target_date: 目标日期（格式: "12月30日" 或 "12/30"） + + 返回: + 排班信息字典 + """ + if not values or len(values) < 2: + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 第一行是表头 + headers = values[0] + date_column_index = self._find_date_column_index(headers, target_date) + + if date_column_index is None: + logger.warning(f"未找到日期列: {target_date}") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 收集白班和夜班人员 + day_shift_names = [] + night_shift_names = [] + + # 从第二行开始是人员数据 + for row in values[1:]: + if len(row) <= date_column_index: + continue + + name = row[0] if row else '' + shift = row[date_column_index] if date_column_index < len(row) else '' + + if not name or not shift: + continue + + if shift == '白': + day_shift_names.append(name) + elif shift == '夜': + night_shift_names.append(name) + + # 格式化输出 + day_shift_str = '、'.join(day_shift_names) if day_shift_names else '' + night_shift_str = '、'.join(night_shift_names) if night_shift_names else '' + + return { + 'day_shift': day_shift_str, + 'night_shift': night_shift_str, + 'day_shift_list': day_shift_names, + 'night_shift_list': night_shift_names + } + + def parse_yearly_sheet(self, values: List[List[str]], target_date: str) -> Dict: + """ + 解析年度表格数据（如2026年排班表） + + 参数: + values: 飞书表格返回的二维数组 + target_date: 目标日期（格式: "12月30日" 或 "12/30"） + + 返回: + 排班信息字典 + """ + if not values: + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 查找目标月份的数据块 + target_month = target_date.split('月')[0] if '月' in target_date else '' + if not target_month: + logger.warning(f"无法从 {target_date} 提取月份") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 在年度表格中查找对应的月份块 + current_block_start = -1 + current_month = '' + + for i, row in enumerate(values): + if not row: + continue + + first_cell = str(row[0]) if row else '' + + # 检查是否是月份标题行，如 "福州港1月排班表" + if '排班表' in first_cell and '月' in first_cell: + # 提取月份数字 + for char in first_cell: + if char.isdigit(): + month_str = '' + j = first_cell.index(char) + while j < len(first_cell) and first_cell[j].isdigit(): + month_str += first_cell[j] + j += 1 + if month_str: + current_month = month_str + current_block_start = i + break + + # 如果找到目标月份，检查下一行是否是表头行 + if current_month == target_month and i == current_block_start + 1: + # 当前行是表头行 + headers = row + date_column_index = self._find_date_column_index(headers, target_date) + + if date_column_index is None: + logger.warning(f"在年度表格中未找到日期列: {target_date}") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 收集人员数据（从表头行的下一行开始） + day_shift_names = [] + night_shift_names = [] + + for j in range(i + 1, len(values)): + person_row = values[j] + if not person_row: + # 遇到空行，继续检查下一行 + continue + + # 检查是否是下一个月份块的开始 + if person_row[0] and isinstance(person_row[0], str) and '排班表' in person_row[0] and '月' in person_row[0]: + break + + # 跳过星期行（第一列为空的行） + if not person_row[0]: + continue + + if len(person_row) <= date_column_index: + continue + + name = person_row[0] if person_row else '' + shift = person_row[date_column_index] if date_column_index < len(person_row) else '' + + if not name or not shift: + continue + + if shift == '白': + day_shift_names.append(name) + elif shift == '夜': + night_shift_names.append(name) + + # 格式化输出 + day_shift_str = '、'.join(day_shift_names) if day_shift_names else '' + night_shift_str = '、'.join(night_shift_names) if night_shift_names else '' + + return { + 'day_shift': day_shift_str, + 'night_shift': night_shift_str, + 'day_shift_list': day_shift_names, + 'night_shift_list': night_shift_names + } + + logger.warning(f"在年度表格中未找到 {target_month}月 的数据块") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + def parse(self, values: List[List[str]], target_date: str, sheet_title: str = '') -> Dict: + """ + 解析排班数据，自动判断表格类型 + + 参数: + values: 飞书表格返回的二维数组 + target_date: 目标日期（格式: "12月30日" 或 "12/30"） + sheet_title: 表格标题，用于判断表格类型 + + 返回: + 排班信息字典 + """ + # 根据表格标题判断表格类型 + if '年' in sheet_title and '排班表' in sheet_title: + # 年度表格 + logger.info(f"使用年度表格解析器: {sheet_title}") + return self.parse_yearly_sheet(values, target_date) + else: + # 月度表格 + logger.info(f"使用月度表格解析器: {sheet_title}") + return self.parse_monthly_sheet(values, target_date) + + +class FeishuScheduleManagerV2: + """飞书排班管理器 v2（使用数据库存储）""" + + def __init__(self, base_url: str = None, token: str = None, + spreadsheet_token: str = None): + """ + 初始化管理器 + + 参数: + base_url: 飞书API基础URL，从环境变量读取 + token: 飞书API令牌，从环境变量读取 + spreadsheet_token: 表格token，从环境变量读取 + """ + # 从环境变量读取配置 + self.base_url = base_url or os.getenv('FEISHU_BASE_URL', 'https://open.feishu.cn/open-apis/sheets/v3') + self.token = token or os.getenv('FEISHU_TOKEN', '') + self.spreadsheet_token = spreadsheet_token or os.getenv('FEISHU_SPREADSHEET_TOKEN', '') + + if not self.token or not self.spreadsheet_token: + logger.warning("飞书配置不完整，请检查环境变量") + + self.client = FeishuSheetsClient(self.base_url, self.token, self.spreadsheet_token) + self.parser = ScheduleDataParser() + self.db = ScheduleDatabase() + + def _select_sheet_for_date(self, sheets: List[Dict], target_year_month: str) -> Optional[Dict]: + """ + 为指定日期选择最合适的表格 + + 参数: + sheets: 表格列表 + target_year_month: 目标年月，格式 "2025-12" + + 返回: + 选中的表格信息，未找到返回None + """ + if not sheets: + return None + + # 提取年份和月份 + year = target_year_month[:4] + month = target_year_month[5:7] + + # 对于2026年，优先使用年度表格 + if year == '2026': + # 查找年度表格，如 "2026年排班表" + year_name = f"{year}年" + for sheet in sheets: + title = sheet.get('title', '') + if year_name in title and '排班表' in title: + logger.info(f"找到2026年年度表格: {title}") + return sheet + + # 优先查找月份表格，如 "12月" + month_name = f"{int(month)}月" + for sheet in sheets: + title = sheet.get('title', '') + if month_name in title: + logger.info(f"找到月份表格: {title}") + return sheet + + # 查找年度表格，如 "2026年排班表" + year_name = f"{year}年" + for sheet in sheets: + title = sheet.get('title', '') + if year_name in title and '排班表' in title: + logger.info(f"找到年度表格: {title}") + return sheet + + # 如果没有找到匹配的表格，使用第一个表格 + logger.warning(f"未找到 {target_year_month} 的匹配表格，使用第一个表格: {sheets[0]['title']}") + return sheets[0] + + def get_schedule_for_date(self, date_str: str) -> Dict: + """ + 获取指定日期的排班信息 + + 参数: + date_str: 日期字符串，格式 "2025-12-30" + + 返回: + 排班信息字典 + """ + try: + # 解析日期 + dt = datetime.strptime(date_str, '%Y-%m-%d') + # 生成两种格式的日期字符串，用于匹配不同表格 + target_date_mm_dd = dt.strftime('%m/%d') # "01/01" 用于月度表格 + target_date_chinese = f"{dt.month}月{dt.day}日" # "1月1日" 用于年度表格 + target_year_month = dt.strftime('%Y-%m') # "2025-12" + + logger.info(f"获取 {date_str} 的排班信息 (格式: {target_date_mm_dd}/{target_date_chinese})") + + # 1. 首先尝试从数据库获取 + cached_schedule = self.db.get_schedule(date_str) + if cached_schedule: + logger.info(f"从数据库获取 {date_str} 的排班信息") + return { + 'day_shift': cached_schedule['day_shift'], + 'night_shift': cached_schedule['night_shift'], + 'day_shift_list': cached_schedule['day_shift_list'], + 'night_shift_list': cached_schedule['night_shift_list'] + } + + # 2. 数据库中没有，需要从飞书获取 + logger.info(f"数据库中没有 {date_str} 的排班信息，从飞书获取") + + # 获取表格信息 + sheets = self.client.get_sheets_info() + if not sheets: + logger.error("未获取到表格信息") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 选择最合适的表格 + selected_sheet = self._select_sheet_for_date(sheets, target_year_month) + if not selected_sheet: + logger.error("未找到合适的表格") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + sheet_id = selected_sheet['sheet_id'] + sheet_title = selected_sheet['title'] + + # 3. 获取表格数据 + sheet_data = self.client.get_sheet_data(sheet_id) + if not sheet_data: + logger.error("未获取到表格数据") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + values = sheet_data.get('valueRange', {}).get('values', []) + revision = sheet_data.get('revision', 0) + + if not values: + logger.error("表格数据为空") + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + # 4. 检查表格是否有更新 + need_update = self.db.check_sheet_update( + sheet_id, sheet_title, revision, {'values': values} + ) + + if not need_update and cached_schedule: + # 表格无更新，且数据库中有缓存，直接返回 + logger.info(f"表格无更新，使用数据库缓存") + return { + 'day_shift': cached_schedule['day_shift'], + 'night_shift': cached_schedule['night_shift'], + 'day_shift_list': cached_schedule['day_shift_list'], + 'night_shift_list': cached_schedule['night_shift_list'] + } + + # 5. 解析数据 - 根据表格类型选择合适的日期格式 + # 如果是年度表格，使用中文日期格式；否则使用mm/dd格式 + if '年' in sheet_title and '排班表' in sheet_title: + target_date = target_date_chinese # "1月1日" + else: + target_date = target_date_mm_dd # "01/01" + + logger.info(f"使用日期格式: {target_date} 解析表格: {sheet_title}") + result = self.parser.parse(values, target_date, sheet_title) + + # 6. 保存到数据库 + if result['day_shift'] or result['night_shift']: + self.db.save_schedule(date_str, result, sheet_id, sheet_title) + logger.info(f"已保存 {date_str} 的排班信息到数据库") + + return result + + except Exception as e: + logger.error(f"获取排班信息失败: {e}") + # 降级处理：返回空值 + return { + 'day_shift': '', + 'night_shift': '', + 'day_shift_list': [], + 'night_shift_list': [] + } + + def get_schedule_for_today(self) -> Dict: + """获取今天的排班信息""" + today = datetime.now().strftime('%Y-%m-%d') + return self.get_schedule_for_date(today) + + def get_schedule_for_tomorrow(self) -> Dict: + """获取明天的排班信息""" + tomorrow = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d') + return self.get_schedule_for_date(tomorrow) + + def refresh_all_schedules(self, days: int = 30): + """ + 刷新未来指定天数的排班信息 + + 参数: + days: 刷新未来多少天的排班信息 + """ + logger.info(f"开始刷新未来 {days} 天的排班信息") + + today = datetime.now() + for i in range(days): + date = (today + timedelta(days=i)).strftime('%Y-%m-%d') + logger.info(f"刷新 {date} 的排班信息...") + self.get_schedule_for_date(date) + + logger.info(f"排班信息刷新完成") + + +if __name__ == '__main__': + # 测试代码 + import sys + + # 设置日志 + logging.basicConfig(level=logging.INFO) + + # 从环境变量读取配置 + manager = FeishuScheduleManagerV2() + + if len(sys.argv) > 1: + date_str = sys.argv[1] + else: + date_str = datetime.now().strftime('%Y-%m-%d') + + print(f"获取 {date_str} 的排班信息...") + schedule = manager.get_schedule_for_date(date_str) + + print(f"白班人员: {schedule['day_shift']}") + print(f"夜班人员: {schedule['night_shift']}") + print(f"白班列表: {schedule['day_shift_list']}") + print(f"夜班列表: {schedule['night_shift_list']}") + diff --git a/src/gui.py b/src/gui.py index 51a8791..9ee7fb6 100644 --- a/src/gui.py +++ b/src/gui.py @@ -187,8 +187,8 @@ class OrbitInGUI: self.log_message("=" * 50) self.log_message("按 Ctrl+Enter 快速获取数据") - # 默认显示今日日报 - self.generate_today_report() + # 启动时自动获取新数据 + self.root.after(100, self.auto_fetch_data) def log_message(self, message, is_error=False): """输出日志消息""" @@ -413,6 +413,93 @@ class OrbitInGUI: self.log_message(f"错误: {e}", is_error=True) self.set_status("错误") + def auto_fetch_data(self): + """自动获取新数据（GUI启动时调用）""" + self.set_status("正在自动获取新数据...") + self.log_message("GUI启动，开始自动获取新数据...") + + try: + # 1. 检查飞书配置，如果配置完整则刷新排班信息 + from dotenv import load_dotenv + load_dotenv() + + feishu_token = os.getenv('FEISHU_TOKEN') + feishu_spreadsheet_token = os.getenv('FEISHU_SPREADSHEET_TOKEN') + + if feishu_token and feishu_spreadsheet_token: + try: + self.log_message("正在刷新排班信息...") + from src.feishu_v2 import FeishuScheduleManagerV2 + feishu_manager = FeishuScheduleManagerV2() + # 只刷新未来7天的排班，减少API调用 + feishu_manager.refresh_all_schedules(days=7) + self.log_message("排班信息刷新完成") + except Exception as e: + self.log_message(f"刷新排班信息时出错: {e}", is_error=True) + self.log_message("将继续处理其他任务...") + else: + self.log_message("飞书配置不完整，跳过排班信息刷新") + + # 2. 尝试获取最新的作业数据 + self.log_message("正在尝试获取最新作业数据...") + + base_url = os.getenv('CONFLUENCE_BASE_URL') + token = os.getenv('CONFLUENCE_TOKEN') + content_id = os.getenv('CONFLUENCE_CONTENT_ID') + + if base_url and token and content_id: + try: + # 获取 HTML + self.log_message("正在从 Confluence 获取 HTML...") + from src.confluence import ConfluenceClient + client = ConfluenceClient(base_url, token) + html = client.get_html(content_id) + + if html: + self.log_message(f"获取成功，共 {len(html)} 字符") + + # 提取文本 + self.log_message("正在提取布局文本...") + from src.extractor import HTMLTextExtractor + extractor = HTMLTextExtractor() + layout_text = extractor.extract(html) + + # 解析数据 + self.log_message("正在解析日志数据...") + from src.parser import HandoverLogParser + parser = HandoverLogParser() + logs = parser.parse(layout_text) + + if logs: + # 保存到数据库 + self.log_message("正在保存到数据库...") + db = DailyLogsDatabase() + count = db.insert_many([log.to_dict() for log in logs]) + db.close() + self.log_message(f"已保存 {count} 条新记录") + else: + self.log_message("未解析到新记录") + else: + self.log_message("未获取到 HTML 内容，跳过数据获取") + except Exception as e: + self.log_message(f"获取作业数据时出错: {e}", is_error=True) + else: + self.log_message("Confluence 配置不完整，跳过数据获取") + + # 3. 显示今日日报 + self.log_message("正在生成今日日报...") + self.generate_today_report() + + self.set_status("就绪") + self.log_message("自动获取完成，GUI已就绪") + + except Exception as e: + self.log_message(f"自动获取过程中出现错误: {e}", is_error=True) + self.log_message("将继续显示GUI界面...") + self.set_status("就绪") + # 即使出错也显示今日日报 + self.generate_today_report() + def show_stats(self): """显示数据库统计""" self.set_status("正在统计...") diff --git a/src/report.py b/src/report.py index ffe1eff..56eccb7 100644 --- a/src/report.py +++ b/src/report.py @@ -6,11 +6,15 @@ from datetime import datetime, timedelta from typing import Dict, List, Optional import sys import os +import logging # 添加项目根目录到 Python 路径 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.database import DailyLogsDatabase +from src.feishu_v2 import FeishuScheduleManagerV2 as FeishuScheduleManager + +logger = logging.getLogger(__name__) class DailyReportGenerator: @@ -96,14 +100,49 @@ class DailyReportGenerator: } def get_shift_personnel(self, date: str) -> Dict: - """获取班次人员（从日志文本中解析，需要配合 parser 使用）""" - # 目前数据库中没有人员信息，返回空 - # 可以后续扩展添加人员追踪功能 - return { - 'day_shift': '', - 'night_shift': '', - 'duty_phone': '13107662315' - } + """ + 获取班次人员（从飞书排班表获取） + + 注意：日报中显示的是次日的班次人员，所以需要获取 date+1 的排班 + 例如：生成 12/29 的日报，显示的是 12/30 的人员 + """ + try: + # 初始化飞书排班管理器 + manager = FeishuScheduleManager() + + # 计算次日日期（日报中显示的是次日班次） + parsed_date = datetime.strptime(date, '%Y-%m-%d') + tomorrow = (parsed_date + timedelta(days=1)).strftime('%Y-%m-%d') + + logger.info(f"获取 {date} 日报的班次人员，对应排班表日期: {tomorrow}") + + # 获取次日的排班信息（使用缓存） + schedule = manager.get_schedule_for_date(tomorrow) + + # 如果从飞书获取到数据，使用飞书数据 + if schedule.get('day_shift') or schedule.get('night_shift'): + return { + 'day_shift': schedule.get('day_shift', ''), + 'night_shift': schedule.get('night_shift', ''), + 'duty_phone': '13107662315' + } + + # 如果飞书数据为空，返回空值 + logger.warning(f"无法从飞书获取 {tomorrow} 的排班信息") + return { + 'day_shift': '', + 'night_shift': '', + 'duty_phone': '13107662315' + } + + except Exception as e: + logger.error(f"获取排班信息失败: {e}") + # 降级处理：返回空值 + return { + 'day_shift': '', + 'night_shift': '', + 'duty_phone': '13107662315' + } def generate_report(self, date: str = None) -> str: """生成日报""" diff --git a/src/schedule_database.py b/src/schedule_database.py new file mode 100644 index 0000000..94ef951 --- /dev/null +++ b/src/schedule_database.py @@ -0,0 +1,323 @@ +#!/usr/bin/env python3 +""" +排班人员数据库模块 +""" +import sqlite3 +import os +import json +from datetime import datetime +from typing import List, Dict, Optional, Tuple +import hashlib + + +class ScheduleDatabase: + """排班人员数据库""" + + def __init__(self, db_path: str = 'data/daily_logs.db'): + """ + 初始化数据库 + + 参数: + db_path: 数据库文件路径 + """ + self.db_path = db_path + self._ensure_directory() + self.conn = self._connect() + self._init_schema() + + def _ensure_directory(self): + """确保数据目录存在""" + data_dir = os.path.dirname(self.db_path) + if data_dir and not os.path.exists(data_dir): + os.makedirs(data_dir) + + def _connect(self) -> sqlite3.Connection: + """连接数据库""" + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + return conn + + def _init_schema(self): + """初始化表结构""" + cursor = self.conn.cursor() + + # 创建排班人员表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS schedule_personnel ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + date TEXT NOT NULL, + day_shift TEXT, + night_shift TEXT, + day_shift_list TEXT, -- JSON数组 + night_shift_list TEXT, -- JSON数组 + sheet_id TEXT, + sheet_title TEXT, + data_hash TEXT, -- 数据哈希，用于检测更新 + created_at TEXT DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT DEFAULT CURRENT_TIMESTAMP, + UNIQUE(date) + ) + ''') + + # 创建表格版本表（用于检测表格是否有更新） + cursor.execute(''' + CREATE TABLE IF NOT EXISTS sheet_versions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sheet_id TEXT NOT NULL, + sheet_title TEXT NOT NULL, + revision INTEGER NOT NULL, + data_hash TEXT, + last_checked_at TEXT DEFAULT CURRENT_TIMESTAMP, + UNIQUE(sheet_id) + ) + ''') + + # 索引 + cursor.execute('CREATE INDEX IF NOT EXISTS idx_schedule_date ON schedule_personnel(date)') + cursor.execute('CREATE INDEX IF NOT EXISTS idx_schedule_sheet ON schedule_personnel(sheet_id)') + cursor.execute('CREATE INDEX IF NOT EXISTS idx_sheet_versions ON sheet_versions(sheet_id)') + + self.conn.commit() + + def _calculate_hash(self, data: Dict) -> str: + """计算数据哈希值""" + data_str = json.dumps(data, sort_keys=True, ensure_ascii=False) + return hashlib.md5(data_str.encode('utf-8')).hexdigest() + + def check_sheet_update(self, sheet_id: str, sheet_title: str, revision: int, data: Dict) -> bool: + """ + 检查表格是否有更新 + + 参数: + sheet_id: 表格ID + sheet_title: 表格标题 + revision: 表格版本号 + data: 表格数据 + + 返回: + True: 有更新，需要重新获取 + False: 无更新，可以使用缓存 + """ + cursor = self.conn.cursor() + + # 查询当前版本 + cursor.execute( + 'SELECT revision, data_hash FROM sheet_versions WHERE sheet_id = ?', + (sheet_id,) + ) + result = cursor.fetchone() + + if not result: + # 第一次获取，记录版本 + data_hash = self._calculate_hash(data) + cursor.execute(''' + INSERT INTO sheet_versions (sheet_id, sheet_title, revision, data_hash, last_checked_at) + VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) + ''', (sheet_id, sheet_title, revision, data_hash)) + self.conn.commit() + return True + + # 检查版本号或数据是否有变化 + old_revision = result['revision'] + old_hash = result['data_hash'] + new_hash = self._calculate_hash(data) + + if old_revision != revision or old_hash != new_hash: + # 有更新，更新版本信息 + cursor.execute(''' + UPDATE sheet_versions + SET revision = ?, data_hash = ?, last_checked_at = CURRENT_TIMESTAMP + WHERE sheet_id = ? + ''', (revision, new_hash, sheet_id)) + self.conn.commit() + return True + + # 无更新，更新检查时间 + cursor.execute(''' + UPDATE sheet_versions + SET last_checked_at = CURRENT_TIMESTAMP + WHERE sheet_id = ? + ''', (sheet_id,)) + self.conn.commit() + return False + + def save_schedule(self, date: str, schedule_data: Dict, sheet_id: str = None, sheet_title: str = None) -> bool: + """ + 保存排班信息到数据库 + + 参数: + date: 日期 (YYYY-MM-DD) + schedule_data: 排班数据 + sheet_id: 表格ID + sheet_title: 表格标题 + + 返回: + 是否成功 + """ + try: + cursor = self.conn.cursor() + + # 准备数据 + day_shift = schedule_data.get('day_shift', '') + night_shift = schedule_data.get('night_shift', '') + day_shift_list = json.dumps(schedule_data.get('day_shift_list', []), ensure_ascii=False) + night_shift_list = json.dumps(schedule_data.get('night_shift_list', []), ensure_ascii=False) + data_hash = self._calculate_hash(schedule_data) + + # 使用 INSERT OR REPLACE 来更新已存在的记录 + cursor.execute(''' + INSERT OR REPLACE INTO schedule_personnel + (date, day_shift, night_shift, day_shift_list, night_shift_list, + sheet_id, sheet_title, data_hash, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) + ''', ( + date, day_shift, night_shift, day_shift_list, night_shift_list, + sheet_id, sheet_title, data_hash + )) + + self.conn.commit() + return True + + except sqlite3.Error as e: + print(f"数据库错误: {e}") + return False + + def get_schedule(self, date: str) -> Optional[Dict]: + """ + 获取指定日期的排班信息 + + 参数: + date: 日期 (YYYY-MM-DD) + + 返回: + 排班信息字典，未找到返回None + """ + cursor = self.conn.cursor() + cursor.execute( + 'SELECT * FROM schedule_personnel WHERE date = ?', + (date,) + ) + result = cursor.fetchone() + + if not result: + return None + + # 解析JSON数组 + day_shift_list = json.loads(result['day_shift_list']) if result['day_shift_list'] else [] + night_shift_list = json.loads(result['night_shift_list']) if result['night_shift_list'] else [] + + return { + 'date': result['date'], + 'day_shift': result['day_shift'], + 'night_shift': result['night_shift'], + 'day_shift_list': day_shift_list, + 'night_shift_list': night_shift_list, + 'sheet_id': result['sheet_id'], + 'sheet_title': result['sheet_title'], + 'updated_at': result['updated_at'] + } + + def get_schedule_by_range(self, start_date: str, end_date: str) -> List[Dict]: + """ + 获取日期范围内的排班信息 + + 参数: + start_date: 开始日期 (YYYY-MM-DD) + end_date: 结束日期 (YYYY-MM-DD) + + 返回: + 排班信息列表 + """ + cursor = self.conn.cursor() + cursor.execute(''' + SELECT * FROM schedule_personnel + WHERE date >= ? AND date <= ? + ORDER BY date + ''', (start_date, end_date)) + + results = [] + for row in cursor.fetchall(): + day_shift_list = json.loads(row['day_shift_list']) if row['day_shift_list'] else [] + night_shift_list = json.loads(row['night_shift_list']) if row['night_shift_list'] else [] + + results.append({ + 'date': row['date'], + 'day_shift': row['day_shift'], + 'night_shift': row['night_shift'], + 'day_shift_list': day_shift_list, + 'night_shift_list': night_shift_list, + 'sheet_id': row['sheet_id'], + 'sheet_title': row['sheet_title'], + 'updated_at': row['updated_at'] + }) + + return results + + def delete_old_schedules(self, before_date: str) -> int: + """ + 删除指定日期之前的排班记录 + + 参数: + before_date: 日期 (YYYY-MM-DD) + + 返回: + 删除的记录数 + """ + cursor = self.conn.cursor() + cursor.execute( + 'DELETE FROM schedule_personnel WHERE date < ?', + (before_date,) + ) + deleted_count = cursor.rowcount + self.conn.commit() + return deleted_count + + def get_stats(self) -> Dict: + """获取统计信息""" + cursor = self.conn.cursor() + + cursor.execute('SELECT COUNT(*) FROM schedule_personnel') + total = cursor.fetchone()[0] + + cursor.execute('SELECT MIN(date), MAX(date) FROM schedule_personnel') + date_range = cursor.fetchone() + + cursor.execute('SELECT COUNT(DISTINCT sheet_id) FROM schedule_personnel') + sheet_count = cursor.fetchone()[0] + + return { + 'total': total, + 'date_range': {'start': date_range[0], 'end': date_range[1]}, + 'sheet_count': sheet_count + } + + def close(self): + """关闭连接""" + if self.conn: + self.conn.close() + + +if __name__ == '__main__': + # 测试代码 + db = ScheduleDatabase() + + # 测试保存 + test_schedule = { + 'day_shift': '张勤、杨俊豪', + 'night_shift': '刘炜彬、梁启迟', + 'day_shift_list': ['张勤', '杨俊豪'], + 'night_shift_list': ['刘炜彬', '梁启迟'] + } + + success = db.save_schedule('2025-12-31', test_schedule, 'zcYLIk', '12月') + print(f"保存成功: {success}") + + # 测试获取 + schedule = db.get_schedule('2025-12-31') + print(f"获取结果: {schedule}") + + # 测试统计 + stats = db.get_stats() + print(f"统计: {stats}") + + db.close() \ No newline at end of file