diff --git a/.gitignore b/.gitignore index a6438d2c..7414fd54 100644 --- a/.gitignore +++ b/.gitignore @@ -100,6 +100,7 @@ _大文件外置/财务管理_data/chat.snapshot_收集.db 01_卡资(金)/金仓_存储备份/大文件外置/财务管理_data/chat.snapshot_data.db 01_卡资(金)/金仓_存储备份/大文件外置/财务管理_data/chat.snapshot_收集.db 01_卡资(金)/金仓_存储备份/服务器管理/scripts/.venv_aliyun/lib/python3.14/site-packages/cryptography/hazmat/bindings/_rust.abi3.so +01_卡资(金)/金仓_存储备份/服务器管理/scripts/.venv_tat/lib/python3.14/site-packages/cryptography/hazmat/bindings/_rust.abi3.so 03_卡木(木)/木叶_视频内容/视频切片/切片动效包装/10秒视频/node_modules/.cache/webpack/remotion-production-4.0.427/a233e9cccba253c3b0157f54cad843b8/0.pack 03_卡木(木)/木叶_视频内容/视频切片/切片动效包装/10秒视频/node_modules/.remotion/chrome-headless-shell/mac-x64/chrome-headless-shell-mac-x64/chrome-headless-shell 03_卡木(木)/木叶_视频内容/视频切片/切片动效包装/10秒视频/node_modules/@rspack/binding-darwin-x64/rspack.darwin-x64.node diff --git a/02_卡人(水)/水桥_平台对接/飞书管理/Excel表格与日报_SKILL.md b/02_卡人(水)/水桥_平台对接/飞书管理/Excel表格与日报_SKILL.md new file mode 100644 index 00000000..df78c82b --- /dev/null +++ b/02_卡人(水)/水桥_平台对接/飞书管理/Excel表格与日报_SKILL.md @@ -0,0 +1,73 @@ +--- +name: Excel表格与日报 +description: 本地 Excel/CSV 批量写入飞书表格,并自动生成日报图表(图片)发飞书群。 +triggers: Excel写飞书、Excel导入飞书、批量写飞书表格、飞书表格导入、CSV写飞书、日报图表发飞书、表格日报 +owner: 水桥 +group: 水 +version: "1.0" +updated: "2026-02-24" +--- + +# Excel表格与日报(飞书) + +## 能做什么(Capabilities) + +- 批量读取本地 `.xlsx/.xlsm/.csv` +- 写入飞书电子表格(同一 sheet 内按区块堆叠) +- 自动生成「日报图表」:HTML → PNG(截图) +- 发飞书群:先发文字,再可选发图片 + +## 怎么用(Usage) + +触发词:`Excel写飞书`、`批量写飞书表格`、`表格日报`、`日报图表发飞书` + +## 前置条件(一次性) + +1. **已完成飞书授权**(同目录 `脚本/.feishu_tokens.json` 存在) + - 若没有:先运行 `飞书管理` 的 `auto_log.py` 完成授权/刷新 token。 +2. **脚本运行环境**:使用独立 venv(不污染系统 Python) + - Python:`/Users/karuo/.venvs/karuo-feishu/bin/python` + +## 必填参数(用环境变量最省事) + +- `FEISHU_SPREADSHEET_TOKEN`:目标表格 token +- `FEISHU_SHEET_ID`:目标 sheetId +- `FEISHU_GROUP_WEBHOOK`:飞书群机器人 webhook + +可选(用于脚本内自动 refresh access_token): +- `FEISHU_APP_ID` +- `FEISHU_APP_SECRET` + +## 一键命令(推荐) + +```bash +# 1) 配置目标表格(示例:请替换为你的) +export FEISHU_SPREADSHEET_TOKEN="xxx" +export FEISHU_SHEET_ID="yyy" +export FEISHU_GROUP_WEBHOOK="https://open.feishu.cn/open-apis/bot/v2/hook/zzz" + +# 2) 批量导入并生成日报(默认只发文字) +"/Users/karuo/.venvs/karuo-feishu/bin/python" \ + "/Users/karuo/Documents/个人/卡若AI/02_卡人(水)/水桥_平台对接/飞书管理/脚本/excel_batch_to_feishu_sheet_report.py" \ + --input "/path/to/excel_or_dir" + +# 3) 追加:同时把日报图片发到群里 +"/Users/karuo/.venvs/karuo-feishu/bin/python" \ + "/Users/karuo/Documents/个人/卡若AI/02_卡人(水)/水桥_平台对接/飞书管理/脚本/excel_batch_to_feishu_sheet_report.py" \ + --input "/path/to/excel_or_dir" \ + --send-image +``` + +## 输出位置 + +按卡若AI 输出规范,自动写到: + +`/Users/karuo/Documents/卡若Ai的文件夹/导出/飞书/Excel日报/` + +包含:`report.html`、`report.png`、`charts_flat/`。 + +## 相关文件(Files) + +- 脚本:`飞书管理/脚本/excel_batch_to_feishu_sheet_report.py` +- 发送图片工具(复用):`智能纪要/脚本/send_to_feishu.py` + diff --git a/02_卡人(水)/水桥_平台对接/飞书管理/SKILL.md b/02_卡人(水)/水桥_平台对接/飞书管理/SKILL.md index 9b4d8c37..b06ae1e1 100755 --- a/02_卡人(水)/水桥_平台对接/飞书管理/SKILL.md +++ b/02_卡人(水)/水桥_平台对接/飞书管理/SKILL.md @@ -1,7 +1,7 @@ --- name: 飞书管理 description: 飞书日志/文档自动写入与知识库管理 -triggers: 飞书日志、写入飞书、飞书知识库、飞书运营报表、派对效果数据、104场写入、运营报表填写、派对截图填表发群 +triggers: 飞书日志、写入飞书、飞书知识库、飞书运营报表、派对效果数据、104场写入、运营报表填写、派对截图填表发群、Excel写飞书、批量写飞书表格、表格日报 owner: 水桥 group: 水 version: "1.1" diff --git a/02_卡人(水)/水桥_平台对接/飞书管理/脚本/excel_batch_to_feishu_sheet_report.py b/02_卡人(水)/水桥_平台对接/飞书管理/脚本/excel_batch_to_feishu_sheet_report.py new file mode 100644 index 00000000..1b1b189a --- /dev/null +++ b/02_卡人(水)/水桥_平台对接/飞书管理/脚本/excel_batch_to_feishu_sheet_report.py @@ -0,0 +1,534 @@ +#!/usr/bin/env python3 +""" +批量 Excel/CSV → 飞书电子表格(Sheets)写入,并生成日报图表(图片)+ 发飞书群。 + +设计目标(最小可用): +1) 支持目录/多文件批量导入(xlsx/xlsm/csv) +2) 写入到同一个 spreadsheet 的同一张 sheet(按块堆叠,每个文件一个区块) +3) 生成本地日报:summary + 折线图(最多3列)→ HTML → PNG +4) 通过飞书群机器人 webhook 发送:文字 + 图片 + +注意: +- 写表使用「用户 access_token」(来自同目录 .feishu_tokens.json) +- 若 token 过期:优先提示先跑 auto_log.py(已有静默刷新机制) +- 如需脚本内自动 refresh:请配置环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET +""" + +from __future__ import annotations + +import argparse +import csv +import json +import os +import re +import sys +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any, Iterable, List, Optional, Tuple +from urllib.parse import quote + +import requests + +try: + from openpyxl import load_workbook # type: ignore +except Exception: + load_workbook = None # venv 未安装时给出明确提示 + +try: + import matplotlib + + matplotlib.use("Agg") + import matplotlib.pyplot as plt # type: ignore +except Exception: + plt = None + + +SCRIPT_DIR = Path(__file__).resolve().parent +TOKEN_FILE = SCRIPT_DIR / ".feishu_tokens.json" + +# 输出目录(按卡若AI 输出规范) +OUT_ROOT = Path("/Users/karuo/Documents/卡若Ai的文件夹/导出/飞书/Excel日报") + + +@dataclass +class FeishuAuth: + access_token: str + refresh_token: str | None = None + + +def _now_str() -> str: + return datetime.now().strftime("%Y-%m-%d_%H%M%S") + + +def _to_text(v: Any) -> str: + if v is None: + return "" + if isinstance(v, (datetime,)): + return v.strftime("%Y-%m-%d %H:%M:%S") + return str(v) + + +def _safe_title(name: str) -> str: + base = re.sub(r"\s+", " ", name.strip()) + base = re.sub(r"[^\w\u4e00-\u9fff\-\.\(\) ]+", "_", base) + return base[:80] or "未命名" + + +def load_user_tokens() -> FeishuAuth: + if not TOKEN_FILE.exists(): + raise RuntimeError( + f"未找到飞书 token 文件:{TOKEN_FILE}\n" + f"请先运行:python3 {SCRIPT_DIR}/auto_log.py 完成授权/刷新" + ) + data = json.loads(TOKEN_FILE.read_text(encoding="utf-8")) + access = (data.get("access_token") or "").strip() + refresh = (data.get("refresh_token") or "").strip() or None + if not access: + raise RuntimeError("token 文件存在,但缺少 access_token;请先跑 auto_log.py 刷新。") + return FeishuAuth(access_token=access, refresh_token=refresh) + + +def try_refresh_access_token(auth: FeishuAuth) -> Optional[str]: + """ + 尝试 refresh(需要 FEISHU_APP_ID/FEISHU_APP_SECRET)。 + 不内置默认 app_id/app_secret,避免把密钥扩散到新脚本。 + """ + if not auth.refresh_token: + return None + app_id = (os.environ.get("FEISHU_APP_ID") or "").strip() + app_secret = (os.environ.get("FEISHU_APP_SECRET") or "").strip() + if not app_id or not app_secret: + return None + + r = requests.post( + "https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal", + json={"app_id": app_id, "app_secret": app_secret}, + timeout=15, + ) + app_token = (r.json() or {}).get("app_access_token") + if not app_token: + return None + + r2 = requests.post( + "https://open.feishu.cn/open-apis/authen/v1/oidc/refresh_access_token", + headers={"Authorization": f"Bearer {app_token}", "Content-Type": "application/json"}, + json={"grant_type": "refresh_token", "refresh_token": auth.refresh_token}, + timeout=15, + ) + out = r2.json() or {} + if out.get("code") != 0: + return None + + data = out.get("data") or {} + new_access = (data.get("access_token") or "").strip() + new_refresh = (data.get("refresh_token") or "").strip() or auth.refresh_token + if not new_access: + return None + + # 写回 token 文件(沿用旧结构) + old = json.loads(TOKEN_FILE.read_text(encoding="utf-8")) + old["access_token"] = new_access + old["refresh_token"] = new_refresh + TOKEN_FILE.write_text(json.dumps(old, ensure_ascii=False, indent=2), encoding="utf-8") + return new_access + + +def feishu_put_values( + access_token: str, + spreadsheet_token: str, + range_str: str, + values: List[List[Any]], + value_input_option: str = "RAW", +) -> Tuple[int, dict]: + url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{spreadsheet_token}/values" + params = {"valueInputOption": value_input_option} + payload = {"valueRange": {"range": range_str, "values": values}} + r = requests.put( + url, + params=params, + headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}, + json=payload, + timeout=30, + ) + try: + return r.status_code, r.json() + except Exception: + return r.status_code, {"code": -1, "msg": (r.text or "")[:200]} + + +def feishu_read_values( + access_token: str, + spreadsheet_token: str, + range_str: str, +) -> Tuple[int, dict]: + url = ( + f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{spreadsheet_token}" + f"/values/{quote(range_str, safe='')}" + ) + r = requests.get( + url, + headers={"Authorization": f"Bearer {access_token}"}, + timeout=30, + ) + try: + return r.status_code, r.json() + except Exception: + return r.status_code, {"code": -1, "msg": (r.text or "")[:200]} + + +def _col_letter(n: int) -> str: + # 0->A, 1->B ... 25->Z, 26->AA + s = "" + while True: + s = chr(65 + n % 26) + s + n = n // 26 + if n <= 0: + break + return s + + +def read_csv(path: Path, max_rows: int, max_cols: int) -> List[List[Any]]: + rows: List[List[Any]] = [] + with path.open("r", encoding="utf-8-sig", newline="") as f: + reader = csv.reader(f) + for i, row in enumerate(reader): + if i >= max_rows: + break + rows.append([c for c in row[:max_cols]]) + return rows + + +def read_xlsx(path: Path, worksheet: Optional[str], max_rows: int, max_cols: int) -> List[List[Any]]: + if load_workbook is None: + raise RuntimeError( + "未安装 openpyxl(或依赖异常)。请使用 venv 运行:\n" + f' "/Users/karuo/.venvs/karuo-feishu/bin/python" "{path}" ...' + ) + wb = load_workbook(path, data_only=True, read_only=True) + if worksheet: + ws = wb[worksheet] + else: + ws = wb.active + rows: List[List[Any]] = [] + for i, row in enumerate(ws.iter_rows(values_only=True)): + if i >= max_rows: + break + out_row: List[Any] = [] + for j, v in enumerate(row): + if j >= max_cols: + break + out_row.append("" if v is None else v) + rows.append(out_row) + return rows + + +def normalize_rows(rows: List[List[Any]]) -> List[List[Any]]: + max_len = max((len(r) for r in rows), default=0) + norm: List[List[Any]] = [] + for r in rows: + rr = list(r) + [""] * (max_len - len(r)) + norm.append([_to_text(v) if isinstance(v, (datetime,)) else ("" if v is None else v) for v in rr]) + return norm + + +def detect_numeric_columns(table: List[List[Any]], has_header: bool) -> List[Tuple[int, str, List[float]]]: + if not table: + return [] + start = 1 if has_header else 0 + headers = [f"列{idx+1}" for idx in range(len(table[0]))] + if has_header: + headers = [str(c).strip() if str(c).strip() else f"列{idx+1}" for idx, c in enumerate(table[0])] + + numeric_cols: List[Tuple[int, str, List[float]]] = [] + for ci in range(len(headers)): + vals: List[float] = [] + for r in table[start:]: + if ci >= len(r): + continue + v = r[ci] + if v is None or v == "": + continue + try: + vals.append(float(v)) + except Exception: + # 非数字列跳过 + vals = [] + break + if len(vals) >= 3: + numeric_cols.append((ci, headers[ci], vals)) + return numeric_cols + + +def plot_numeric_columns( + table: List[List[Any]], + has_header: bool, + out_dir: Path, + top_k: int = 3, +) -> List[Path]: + if plt is None: + return [] + cols = detect_numeric_columns(table, has_header) + if not cols: + return [] + + # 简单选择:按有效值数量降序取前 top_k + cols.sort(key=lambda x: len(x[2]), reverse=True) + selected = cols[:top_k] + + img_paths: List[Path] = [] + for idx, (ci, name, vals) in enumerate(selected, start=1): + fig = plt.figure(figsize=(10, 3)) + ax = fig.add_subplot(111) + ax.plot(list(range(1, len(vals) + 1)), vals, linewidth=2) + ax.set_title(name) + ax.set_xlabel("行序号") + ax.set_ylabel("数值") + ax.grid(True, alpha=0.3) + img = out_dir / f"chart_{idx}_{_safe_title(name)}.png" + fig.tight_layout() + fig.savefig(img, dpi=160) + plt.close(fig) + img_paths.append(img) + return img_paths + + +def build_report_html( + title: str, + sections: List[Tuple[str, List[str], List[Path]]], + out_html: Path, +) -> None: + parts: List[str] = [] + parts.append("
") + parts.append("") + + parts.append(f"