import tkinter as tk from tkinter import ttk, filedialog, messagebox, scrolledtext import pandas as pd import requests import json import threading import time import random import re from datetime import datetime import os import shutil from openpyxl import load_workbook from openpyxl.utils import get_column_letter from PIL import Image import io import base64 import zlib class FeishuSubmitApp: def __init__(self, root): self.root = root self.root.title("飞书表格数据提交工具 - 自动提取图片") self.root.geometry("1100x800") # 状态变量 self.is_running = False self.stop_flag = False self.excel_file = "" self.cookie_string = "" # 图片缓存 self.image_cache = {} # {行号: [图片数据列表]} self.image_extract_folder = None self.uploaded_cache = {} # 缓存已上传的图片token {行号: attachment_info} # 从curl提取的默认配置 self.default_headers = { "accept": "application/json, text/plain, */*", "accept-language": "zh-CN,zh;q=0.9", "content-type": "application/json", "f-version": "docs-6-17-1781628518399", "origin": "https://xhd.feishu.cn", "priority": "u=1, i", "referer": "https://xhd.feishu.cn/share/base/form/shrcnJqFUFF0sqlmrfsWpDNf8df", "sec-ch-ua": '"Chromium";v="148", "Google Chrome";v="148", "Not/A)Brand";v="99"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Windows"', "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36", "x-csrftoken": "83aae12269bced10680d3ccf817f82262305e7d0-1781579108", "x-tt-logid": "02178275091573600000000000000000000ffff5bd640caae0cb8", "x-tt-trace": "1" } self.url = "https://xhd.feishu.cn/space/api/bitable/share/content" self.share_token = "shrcnJqFUFF0sqlmrfsWpDNf8df" self.setup_ui() self.load_config() def setup_ui(self): """构建界面""" # 主容器 main_frame = ttk.Frame(self.root, padding="10") main_frame.pack(fill=tk.BOTH, expand=True) # ========== 第一行：文件选择 ========== file_frame = ttk.LabelFrame(main_frame, text="📁 数据文件", padding="10") file_frame.pack(fill=tk.X, pady=(0, 10)) ttk.Label(file_frame, text="Excel文件:").grid(row=0, column=0, sticky=tk.W) self.file_path_var = tk.StringVar() ttk.Entry(file_frame, textvariable=self.file_path_var, width=60).grid(row=0, column=1, padx=(10, 10)) ttk.Button(file_frame, text="浏览...", command=self.select_file).grid(row=0, column=2) ttk.Label(file_frame, text="工作表:").grid(row=1, column=0, sticky=tk.W, pady=(10, 0)) self.sheet_var = tk.StringVar(value="Sheet1") ttk.Entry(file_frame, textvariable=self.sheet_var, width=20).grid(row=1, column=1, sticky=tk.W, padx=(10, 10), pady=(10, 0)) ttk.Label(file_frame, text="(默认: Sheet1)").grid(row=1, column=2, sticky=tk.W, pady=(10, 0)) # ========== 第二行：Cookie配置 ========== cookie_frame = ttk.LabelFrame(main_frame, text="🍪 Cookie 配置", padding="10") cookie_frame.pack(fill=tk.X, pady=(0, 10)) ttk.Label(cookie_frame, text="Cookie:").grid(row=0, column=0, sticky=tk.W) self.cookie_text = tk.Text(cookie_frame, height=4, width=80, wrap=tk.WORD) self.cookie_text.grid(row=0, column=1, padx=(10, 10), pady=(0, 5), sticky=tk.W) cookie_btn_frame = ttk.Frame(cookie_frame) cookie_btn_frame.grid(row=1, column=1, sticky=tk.W, padx=(10, 0)) ttk.Button(cookie_btn_frame, text="从剪贴板粘贴", command=self.paste_cookie).pack(side=tk.LEFT, padx=(0, 10)) ttk.Button(cookie_btn_frame, text="保存配置", command=self.save_config).pack(side=tk.LEFT, padx=(0, 10)) ttk.Button(cookie_btn_frame, text="加载默认", command=self.load_default_cookie).pack(side=tk.LEFT) # ========== 第三行：发送控制 ========== control_frame = ttk.LabelFrame(main_frame, text="⚙️ 发送控制", padding="10") control_frame.pack(fill=tk.X, pady=(0, 10)) # 间隔配置 ttk.Label(control_frame, text="最小间隔(秒):").grid(row=0, column=0, sticky=tk.W) self.min_interval_var = tk.StringVar(value="20") ttk.Entry(control_frame, textvariable=self.min_interval_var, width=10).grid(row=0, column=1, padx=(5, 15), sticky=tk.W) ttk.Label(control_frame, text="最大间隔(秒):").grid(row=0, column=2, sticky=tk.W) self.max_interval_var = tk.StringVar(value="50") ttk.Entry(control_frame, textvariable=self.max_interval_var, width=10).grid(row=0, column=3, padx=(5, 15), sticky=tk.W) ttk.Label(control_frame, text="起始行:").grid(row=0, column=4, sticky=tk.W) self.start_row_var = tk.StringVar(value="2") ttk.Entry(control_frame, textvariable=self.start_row_var, width=8).grid(row=0, column=5, padx=(5, 15), sticky=tk.W) ttk.Label(control_frame, text="(表头在第1行)").grid(row=0, column=6, sticky=tk.W) # 操作按钮 btn_frame = ttk.Frame(control_frame) btn_frame.grid(row=1, column=0, columnspan=7, pady=(15, 0), sticky=tk.W) self.start_btn = ttk.Button(btn_frame, text="🚀 开始发送", command=self.start_send) self.start_btn.pack(side=tk.LEFT, padx=(0, 10)) self.stop_btn = ttk.Button(btn_frame, text="⏹ 停止", command=self.stop_send, state=tk.DISABLED) self.stop_btn.pack(side=tk.LEFT, padx=(0, 10)) ttk.Button(btn_frame, text="清空日志", command=self.clear_log).pack(side=tk.LEFT, padx=(0, 10)) ttk.Button(btn_frame, text="📸 提取图片", command=self.extract_images).pack(side=tk.LEFT) # 状态显示 self.status_var = tk.StringVar(value="就绪") ttk.Label(control_frame, textvariable=self.status_var, foreground="blue").grid(row=2, column=0, columnspan=7, pady=(10, 0), sticky=tk.W) # ========== 日志显示 ========== log_frame = ttk.LabelFrame(main_frame, text="📋 发送日志", padding="10") log_frame.pack(fill=tk.BOTH, expand=True) self.log_text = scrolledtext.ScrolledText(log_frame, height=20, wrap=tk.WORD, font=("Consolas", 9)) self.log_text.pack(fill=tk.BOTH, expand=True) # 日志颜色标签 self.log_text.tag_config("success", foreground="green") self.log_text.tag_config("error", foreground="red") self.log_text.tag_config("info", foreground="blue") self.log_text.tag_config("warning", foreground="orange") self.log_text.tag_config("debug", foreground="purple") # ========== 底部状态栏 ========== status_bar = ttk.Frame(self.root) status_bar.pack(fill=tk.X, side=tk.BOTTOM) ttk.Label(status_bar, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W).pack(fill=tk.X, padx=5, pady=2) def select_file(self): """选择Excel文件""" file_path = filedialog.askopenfilename( title="选择Excel文件", filetypes=[("Excel files", "*.xlsx *.xls"), ("All files", "*.*")] ) if file_path: self.file_path_var.set(file_path) self.excel_file = file_path self.log(f"已选择文件: {file_path}", "info") # 尝试自动识别工作表 try: xl = pd.ExcelFile(file_path) sheets = xl.sheet_names if sheets: self.sheet_var.set(sheets[0]) self.log(f"自动选择工作表: {sheets[0]}", "info") except Exception as e: self.log(f"读取工作表失败: {str(e)}", "warning") def paste_cookie(self): """从剪贴板粘贴Cookie""" try: clipboard_text = self.root.clipboard_get() if "cookie:" in clipboard_text.lower() or "-H" in clipboard_text: cookie_match = re.search(r'-b\s+"([^"]+)"', clipboard_text) if not cookie_match: cookie_match = re.search(r'-H\s+"cookie:\s*([^"]+)"', clipboard_text, re.IGNORECASE) if cookie_match: self.cookie_text.delete(1.0, tk.END) self.cookie_text.insert(1.0, cookie_match.group(1)) self.log("已从剪贴板提取Cookie", "success") return self.cookie_text.delete(1.0, tk.END) self.cookie_text.insert(1.0, clipboard_text) self.log("已粘贴Cookie", "info") except: messagebox.showwarning("提示", "无法读取剪贴板内容") def load_default_cookie(self): """加载默认Cookie""" default_cookie = "_uuid_hera_ab_path_1=7651649498397052109; __tea__ug__uid=4144561781538482924; _gcl_au=1.1.775810441.1781538483; lgw_csrf_token=9dcbe9c422a1a2df5727bebebf4136760e1ad6dc-1781538485; passport_web_did=7651823325329411260; passport_trace_id=7651823325353757894; QXV0aHpDb250ZXh0=ba539bcfe4a042c8bc1e99d4d03add38; s_v_web_id=verify_mqg251ox_Tp4ofGaR_NCf4_4zII_8Ad3_c3HxGb4PXIuN; session=XN0YXJ0-5b9t9966-1122-4520-890c-e9313b1f818a-WVuZA; session_list=XN0YXJ0-5b9t9966-1122-4520-890c-e9313b1f818a-WVuZA; login_recently=1; is_anonymous_session=; _csrf_token=83aae12269bced10680d3ccf817f82262305e7d0-1781579108; lang=zh; et=c0c63f4480fdf14d7e6e9a21bab2b7e1; ccm_cdn_host=//lf-package-cn.feishucdn.com/obj/feishu-static; locale=zh-CN; landing_url=https://www.feishu.cn/; Hm_lvt_a79616d9322d81f12a92402ac6ae32ea=1781538483,1781579751; Hm_lpvt_a79616d9322d81f12a92402ac6ae32ea=1781579751; HMACCOUNT=5CA5EB76287246C7; _ga_VPYRHN104D=GS2.1.s1781579004$o2$g1$t1781579751$j60$l0$h0; _uetvid=9877555068d111f1992d03b61d301033; _ga=GA1.2.692261332.1781538484; i18n_locale=zh-CN; i18n_locale=zh-CN; _gid=GA1.2.1918132955.1781749497; site_env=pre=0; sl_session=eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3ODE3OTMyNTgsInVuaXQiOiJldV9uYyIsInJhdyI6eyJtZXRhIjoiQVdvYzIrdU1BNHZSYUhkQjl4N2lRQUpxTUx6RjlrN012R293dk1YMlRzeThhakM5WXdRQURMSUNLZ0VBUVVGQlFVRkJRVUZCUVVKeFRVd3hha2h6UkUxNFVUMDkiLCJzdW0iOiI1MWU3ZWM3YzU0NTAzNTljMzRhMjE3ODdlM2U1OGNmZjQzNzZmM2IzYjJjNzU4NWVjMGE3MjQyZDZjNTBiNmNkIiwibG9jIjoiemhfY24iLCJhcGMiOiJSZWxlYXNlIiwiaWF0IjoxNzgxNzUwMDU4LCJzYWMiOnsiVXNlclN0YWZmU3RhdHVzIjoiMSIsIlVzZXJUeXBlIjoiNDIifSwibG9kIjpudWxsLCJjbmYiOnsiamt0Ijoic3VWaTdnNDNJX0lUb0w2ZGo5QXhEWnBacUhkYldzNWJaaVhvV0s4akZrdyJ9LCJucyI6ImxhcmsiLCJuc191aWQiOiI3NjQ2MjI4MDcyMDgxMDk5NzI5IiwibnNfdGlkIjoiNzUyNzU1NzgzMTgwMzk0NDk2MiIsIm90IjoxLCJjdCI6MTc4MTU3OTEwNywicnQiOjE3ODE2Njg4OTMsInRjcyI6MiwidGJ0IjowfX0.6MrIkLX5grSs5PUBr3U-TC5keonxZZDADzbJ0c00y9HsalJhhQ0gbmfk2u2w5RWhMkXt46a50yMc9E7Po6kymg; msToken=Oyg-XUAwfnJlSSSvRBdj1W_LxhRCXLlWMx9mk0Ryxvku6FZLH0ysyeYnmGrIG_yvUClyZT2cLYe5rhC1PtVygRRXOhYTvrDcL8Oidp-ZHHI-LNFKPJk-She-3kUxNTnwb3fYj5FfI36vX-1eFZ_neFVmoTtgD1GhK3M3oNjh5H-Y; swp_csrf_token=6fe8dbf3-4ab2-4f46-b4f3-5c9cba7364cd; t_beda37=a427edf4f48d9fa049a454a58600fa2f188ffacaf053a85e3782d6615e737b03; passport_app_access_token=eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3ODE3OTU2NDcsInVuaXQiOiJldV9uYyIsInJhdyI6eyJtX2FjY2Vzc19pbmZvIjp7IjE0MyI6eyJpYXQiOjE3ODE3NDk1MTksImFjY2VzcyI6dHJ1ZX0sIjIiOnsiaWF0IjoxNzgxNzUyMTY1LCJhY2Nlc3MiOnRydWV9LCIxMDAiOnsiaWF0IjoxNzgxNzUyMjk1LCJhY2Nlc3MiOnRydWV9LCI0Ijp7ImlhdCI6MTc4MTc1MjQ0NywiYWNjZXNzIjp0cnVlfX0sInN1bSI6IjUxZTdlYzdjNTQ1MDM1OWMzNGEyMTc4N2UzZTU4Y2ZmNDM3NmYzYjNiMmM3NTg1ZWMwYTcyNDJkNmM1MGI2Y2QifX0.6o43emSe8iOBsXTPUbx4cShf-DdIpGuTTYDR5NDAAvA6ZByV_biiboF32sk5SjLAuTfEEy1xWpZ_35Fb-nxvJg" self.cookie_text.delete(1.0, tk.END) self.cookie_text.insert(1.0, default_cookie) self.log("已加载默认Cookie", "info") def save_config(self): """保存Cookie配置到文件""" cookie = self.cookie_text.get(1.0, tk.END).strip() if not cookie: messagebox.showwarning("提示", "Cookie不能为空") return try: with open("cookie_config.txt", "w", encoding="utf-8") as f: f.write(cookie) self.log("Cookie配置已保存", "success") messagebox.showinfo("成功", "Cookie配置已保存到 cookie_config.txt") except Exception as e: self.log(f"保存失败: {str(e)}", "error") def load_config(self): """加载保存的Cookie配置""" try: with open("cookie_config.txt", "r", encoding="utf-8") as f: cookie = f.read().strip() if cookie: self.cookie_text.delete(1.0, tk.END) self.cookie_text.insert(1.0, cookie) self.log("已加载保存的Cookie配置", "info") except FileNotFoundError: pass except Exception as e: self.log(f"加载配置失败: {str(e)}", "warning") def log(self, message, tag="info"): """添加日志""" timestamp = datetime.now().strftime("%H:%M:%S") self.log_text.insert(tk.END, f"[{timestamp}] {message}\n", tag) self.log_text.see(tk.END) self.root.update_idletasks() def clear_log(self): """清空日志""" self.log_text.delete(1.0, tk.END) def parse_cookie(self, cookie_string): """解析Cookie字符串为字典""" cookies = {} for item in cookie_string.split(';'): item = item.strip() if '=' in item: key, value = item.split('=', 1) cookies[key.strip()] = value.strip() return cookies def _get_csrf_token(self): """从cookie中提取csrf_token""" cookie_text = self.cookie_text.get(1.0, tk.END).strip() # 匹配 _csrf_token=xxx 或 csrf_token=xxx match = re.search(r'_csrf_token=([^;]+)', cookie_text) if match: return match.group(1) # 也尝试匹配 x-csrftoken match = re.search(r'x-csrftoken[=:]\s*([^;\s]+)', cookie_text, re.IGNORECASE) if match: return match.group(1) # 返回默认值 return "83aae12269bced10680d3ccf817f82262305e7d0-1781579108" def _get_mime_type(self, filename): """获取文件MIME类型""" ext = os.path.splitext(filename)[1].lower() mime_map = { '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.png': 'image/png', '.gif': 'image/gif', '.bmp': 'image/bmp', '.webp': 'image/webp' } return mime_map.get(ext, 'image/jpeg') def upload_image(self, image_data, filename="image.jpg"): """ 完整上传图片到飞书，返回file_token信息 """ try: cookie_string = self.cookie_text.get(1.0, tk.END).strip() if not cookie_string: self.log(" ❌ Cookie为空，无法上传图片", "error") return None cookies = self.parse_cookie(cookie_string) csrf_token = self._get_csrf_token() self.log(f" 🔍 使用的CSRF token: {csrf_token[:30] if csrf_token else 'None'}...", "debug") self.log(f" 🔍 图片大小: {len(image_data)} bytes", "debug") # 构建通用请求头 base_headers = { "accept": "application/json, text/plain, */*", "accept-language": "zh-CN,zh;q=0.9", "f-version": "docs-6-17-1781628518399", "x-csrftoken": csrf_token, "x-tt-logid": "02178275091573600000000000000000000ffff5bd640caae0cb8", "x-tt-trace": "1", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } # ========== 步骤1: 获取uploadCode ========== self.log(f" 📤 [1/4] 获取上传code...", "info") code_url = "https://xhd.feishu.cn/space/api/bitable/external/share/uploadCode" code_params = { "shareToken": self.share_token, "fileName": filename, "size": len(image_data), "mountPoint": "bitable_tmp_point" } self.log(f" 📋 请求URL: {code_url}", "debug") self.log(f" 📋 请求参数: {json.dumps(code_params, ensure_ascii=False)}", "debug") code_response = requests.get(code_url, params=code_params, headers=base_headers, cookies=cookies, timeout=30) self.log(f" 📡 响应状态码: {code_response.status_code}", "debug") if code_response.status_code != 200: self.log(f" ❌ 获取code失败: HTTP {code_response.status_code}", "error") self.log(f" ❌ 响应内容: {code_response.text}", "error") return None try: code_data = code_response.json() except Exception as e: self.log(f" ❌ 解析JSON失败: {str(e)}", "error") return None upload_code = code_data.get('data', {}).get('uploadCode') if not upload_code: self.log(f" ❌ 响应中没有uploadCode字段", "error") self.log(f" ❌ data内容: {code_data.get('data', {})}", "error") return None self.log(f" ✅ 获取uploadCode成功: {upload_code[:30]}...", "success") # ========== 步骤2: 准备上传 ========== self.log(f" 📤 [2/4] 准备上传，获取upload_id...", "info") prepare_url = "https://xhd.feishu.cn/space/api/box/upload/prepare/authcode/" prepare_headers = base_headers.copy() prepare_headers["content-type"] = "application/json" prepare_headers["x-command"] = "space.api.box.upload.prepare.authcode" prepare_body = { "code": upload_code, "mount_point": "bitable_tmp_point", "mount_node_token": self.share_token } self.log(f" 📋 请求体: {json.dumps(prepare_body, ensure_ascii=False)}", "debug") prepare_response = requests.post(prepare_url, headers=prepare_headers, cookies=cookies, json=prepare_body, timeout=30) self.log(f" 📡 响应状态码: {prepare_response.status_code}", "debug") if prepare_response.status_code != 200: self.log(f" ❌ 准备上传失败: HTTP {prepare_response.status_code}", "error") self.log(f" ❌ 响应内容: {prepare_response.text}", "error") return None try: prepare_data = prepare_response.json() except Exception as e: self.log(f" ❌ 解析JSON失败: {str(e)}", "error") return None if prepare_data.get('code') != 0: self.log(f" ❌ 准备上传失败: code={prepare_data.get('code')}, msg={prepare_data.get('msg', '未知错误')}", "error") return None upload_id = prepare_data.get('data', {}).get('upload_id') if not upload_id: self.log(f" ❌ 响应中没有upload_id字段", "error") self.log(f" ❌ data内容: {prepare_data.get('data', {})}", "error") return None self.log(f" ✅ 获取upload_id成功: {upload_id}", "success") self.log(f" 📤 [3/4] 上传文件内容 ({len(image_data)} bytes)...", "info") # 使用 Adler-32 计算 checksum（飞书使用的算法） import zlib checksum = zlib.adler32(image_data) & 0xffffffff upload_url = f"https://internal-api-drive-stream.feishu.cn/space/api/box/stream/upload/merge_block/?upload_id={upload_id}" upload_headers = base_headers.copy() upload_headers["content-type"] = "application/octet-stream" upload_headers["x-command"] = "space.api.box.stream.upload.merge_block" upload_headers["x-seq-list"] = "0" upload_headers["x-block-list-checksum"] = str(checksum) # 注意：是数字字符串，不是十六进制 upload_headers["x-block-origin-size"] = str(len(image_data)) self.log(f" 📋 文件大小: {len(image_data)} bytes, checksum(Adler-32): {checksum}", "debug") upload_response = requests.post(upload_url, headers=upload_headers, cookies=cookies, data=image_data, timeout=60) self.log(f" 📡 响应状态码: {upload_response.status_code}", "debug") self.log(f" 📡 响应内容: {upload_response.text[:500]}", "debug") if upload_response.status_code != 200: self.log(f" ❌ 上传文件失败: HTTP {upload_response.status_code}", "error") self.log(f" ❌ 响应内容: {upload_response.text}", "error") return None # 检查上传是否真正成功 try: upload_result = upload_response.json() if upload_result.get('code') != 0: self.log(f" ❌ 上传文件失败: code={upload_result.get('code')}, msg={upload_result.get('message', '未知错误')}", "error") return None except: pass self.log(f" ✅ 文件上传成功", "success") # ========== 步骤4: 完成上传 ========== self.log(f" 📤 [4/4] 完成上传，获取file_token...", "info") finish_url = "https://xhd.feishu.cn/space/api/box/upload/finish/" finish_headers = base_headers.copy() finish_headers["content-type"] = "application/json" finish_headers["x-command"] = "space.api.box.upload.finish" finish_body = { "upload_id": upload_id, "num_blocks": 1, "push_open_history_record": 0 } self.log(f" 📋 请求体: {json.dumps(finish_body, ensure_ascii=False)}", "debug") finish_response = requests.post(finish_url, headers=finish_headers, cookies=cookies, json=finish_body, timeout=30) self.log(f" 📡 响应状态码: {finish_response.status_code}", "debug") self.log(f" 📡 响应内容: {finish_response.text[:500]}", "debug") if finish_response.status_code != 200: self.log(f" ❌ 完成上传失败: HTTP {finish_response.status_code}", "error") self.log(f" ❌ 响应内容: {finish_response.text}", "error") return None try: finish_data = finish_response.json() except Exception as e: self.log(f" ❌ 解析JSON失败: {str(e)}", "error") return None if finish_data.get('code') != 0: self.log(f" ❌ 完成上传失败: code={finish_data.get('code')}, msg={finish_data.get('message', '未知错误')}", "error") self.log(f" ❌ 完整响应: {json.dumps(finish_data, ensure_ascii=False, indent=2)}", "error") return None file_token = finish_data.get('data', {}).get('file_token') if not file_token: self.log(f" ❌ 响应中没有file_token字段", "error") self.log(f" ❌ data内容: {finish_data.get('data', {})}", "error") return None # 获取图片信息 width, height = 0, 0 try: from PIL import Image import io pil_img = Image.open(io.BytesIO(image_data)) width, height = pil_img.size except: pass result = { 'attachmentToken': file_token, 'id': file_token, 'mimeType': self._get_mime_type(filename), 'name': filename, 'size': len(image_data), 'timeStamp': int(time.time() * 1000), 'height': height, 'width': width } self.log(f" ✅ 图片上传完成! file_token: {file_token}", "success") return result except requests.exceptions.Timeout: self.log(f" ❌ 上传超时", "error") return None except Exception as e: self.log(f" ❌ 上传图片异常: {str(e)}", "error") import traceback self.log(f" 详细: {traceback.format_exc()}", "debug") return None def extract_images_to_cache(self, file_path, sheet_name): """ 提取Excel中的所有图片到缓存 返回: dict {行号: [图片数据列表]} """ image_cache = {} try: wb = load_workbook(file_path, data_only=True) ws = wb[sheet_name] if hasattr(ws, '_images') and ws._images: self.log(f" 📸 开始提取 {len(ws._images)} 张图片到缓存...", "info") for idx, img in enumerate(ws._images): try: # 获取位置信息 if hasattr(img, 'anchor') and hasattr(img.anchor, '_from'): anchor_from = img.anchor._from row = anchor_from.row + 1 col = anchor_from.col + 1 else: row = idx + 1 col = 0 # 获取图片数据 image_data = None img_format = 'jpg' filename = f"image_{row}_{idx}.jpg" if hasattr(img, '_data') and callable(img._data): image_data = img._data() elif hasattr(img, '_data') and not callable(img._data): image_data = img._data elif hasattr(img, 'image'): if hasattr(img.image, '_data') and callable(img.image._data): image_data = img.image._data() elif hasattr(img.image, '_data'): image_data = img.image._data if image_data is None: continue # 检测图片格式 try: from PIL import Image import io pil_img = Image.open(io.BytesIO(image_data)) img_format = pil_img.format if pil_img.format else 'jpg' filename = f"image_{row}_{idx}.{img_format.lower()}" except: pass if row not in image_cache: image_cache[row] = [] image_cache[row].append({ 'data': image_data, 'col': col, 'format': img_format, 'filename': filename }) except Exception as e: self.log(f" ⚠️ 提取图片失败: {str(e)}", "warning") wb.close() total = sum(len(imgs) for imgs in image_cache.values()) self.log(f" ✅ 成功提取 {total} 张图片到缓存，分布在 {len(image_cache)} 行", "success") except Exception as e: self.log(f" ❌ 提取图片失败: {str(e)}", "error") return image_cache def build_request_body(self, row_dict, row_index): """根据Excel行数据构建请求体""" try: # 获取各列数据 name = str(row_dict.get("姓名", "")) if pd.notna(row_dict.get("姓名")) else "" gender = str(row_dict.get("性别", "")) if pd.notna(row_dict.get("性别")) else "" phone = str(row_dict.get("手机号码", "")) if pd.notna(row_dict.get("手机号码")) else "" remark = str(row_dict.get("备注", "")) if pd.notna(row_dict.get("备注")) else "" institution = str(row_dict.get("机构", "")) if pd.notna(row_dict.get("机构")) else "" target_school_text = str(row_dict.get("意向学校", "")) if pd.notna(row_dict.get("意向学校")) else "" consulting_text = str(row_dict.get("咨询项目", "")) if pd.notna(row_dict.get("咨询项目")) else "" product_text = str(row_dict.get("产品", "")) if pd.notna(row_dict.get("产品")) else "" source = str(row_dict.get("线索来源", "")) if pd.notna(row_dict.get("线索来源")) else "" market_person = str(row_dict.get("市场人员", "")) if pd.notna(row_dict.get("市场人员")) else "" # 去掉数字的小数点 phone = phone.split('.')[0] self.log(f" 📝 提取数据: 姓名={name}, 手机={phone}, 意向学校={target_school_text}, 咨询项目={consulting_text}, 产品={product_text}", "debug") # 意向校区映射字典 school_mapping = { "上海": "optqw2jaJ5", "徐汇": "optTD4UQsM", "人广": "opt7hQjSZQ", "长宁": "opt7A8WBtE", "浦东": "optnyoxnTK", "杨浦": "optgArj17r", "闵行": "optCIwo1J1", "松江": "optuZzXsYm", "锦秋个人学校": "optLQYDW3S", "封闭学院": "optitJpd5G", "大学生-市区": "optHlvKsKy", "大学生-曹崇杨": "optdcoQXBt", "大学生-临港": "opt3i4fOKJ", "大学生-松江": "optG7eGFUg", "大学生-奉贤": "optJJqnLgY", "常州": "opt5wuiYjW" } # 咨询项目映射字典 consulting_mapping = { "语培": "opt96MRele", "留学": "optQFQsXIe", "考研": "opti3wbRF7", "AF艺术": "optaZC2VCL", "美行音乐": "optzSqwgLp", "锦秋个人学校": "opt9N4AGbJ", "研学考团": "optdYBLknN", "尚思": "optf8lTDcV" } # 语培产品映射字典（完整版） product_mapping = { "ACT": "optlGkVm5n", "AEAS": "optqX3b1rJ", "ALEVEL": "optgMNcV8D", "AP": "optWKWYQgo", "BC": "opt8huX4Nv", "DSE": "optxFBYmUs", "ESL": "optaYj9RZM", "GMAT": "optgwu03Nf", "GRE": "optEOyf5TR", "IB": "optBhWIw7x", "IGCSE": "optIuoUFpo", "KET": "optjcLdrsB", "PET": "optY42DNFo", "PTE": "optOrHJvTX", "SAT": "optTxT8NEu", "SSAT": "optJ2a0tkB", "Toefl Junior": "optEGInE0d", "国际备考": "optgwn1azI", "国际研学及考团": "optAmxtqGN", "国家地理": "opttXZvoj3", "剑桥英语": "optQkkhzD5", "锦秋竞赛": "optUgXXmpF", "考研": "optgQrGHTS", "留学预备": "optWxtOU8b", "美高": "opt36E5u0t", "硕果": "optNKeSnho", "硕果竞赛": "optMzk3TpL", "腾飞计划": "optYTD9h04", "团企培": "optW5UDF6G", "托福": "optKuZ2zCq", "雅思": "optWsal1MB", "用英语讲中国故事": "optinda27s", "在线": "opt2X07PY4", "浙大国际本科": "optZFmVTTC", "语培待定/未知": "optTNitLA5", "大学四级": "optAqYPFmN", "大学六级": "optpdfKgdl", "菁英树": "optBFhBHFX", "其它": "opt698ww1H" } # 处理意向学校：根据文本查找对应的ID target_school_id = None if target_school_text: # 精确匹配 if target_school_text in school_mapping: target_school_id = school_mapping[target_school_text] self.log(f" 🏫 意向学校: {target_school_text} -> {target_school_id}", "debug") else: # 尝试模糊匹配（包含关系） matched = False for school_name, school_id in school_mapping.items(): if school_name in target_school_text or target_school_text in school_name: target_school_id = school_id matched = True self.log(f" 🏫 意向学校模糊匹配: {target_school_text} -> {school_name} ({target_school_id})", "debug") break if not matched: self.log(f" ⚠️ 未找到匹配的意向学校: {target_school_text}", "warning") # 处理咨询项目：根据文本查找对应的ID（支持多选） consulting_ids = [] if consulting_text: # 如果是多个项目，用分隔符分割（如逗号、顿号等） consulting_items = re.split(r'[,，、/ ]+', consulting_text) for item in consulting_items: item = item.strip() if item: if item in consulting_mapping: consulting_ids.append(consulting_mapping[item]) self.log(f" 📚 咨询项目: {item} -> {consulting_mapping[item]}", "debug") else: # 尝试模糊匹配 matched = False for consulting_name, consulting_id in consulting_mapping.items(): if consulting_name in item or item in consulting_name: consulting_ids.append(consulting_id) matched = True self.log(f" 📚 咨询项目模糊匹配: {item} -> {consulting_name} ({consulting_id})", "debug") break if not matched: self.log(f" ⚠️ 未找到匹配的咨询项目: {item}", "warning") # 处理产品：根据文本查找对应的ID（支持多选） product_ids = [] if product_text: product_items = re.split(r'[,，、/ ]+', product_text) for item in product_items: item = item.strip() if item: if item in product_mapping: product_ids.append(product_mapping[item]) self.log(f" 📦 产品: {item} -> {product_mapping[item]}", "debug") else: # 尝试模糊匹配（支持大小写不敏感） matched = False item_lower = item.lower() for product_name, product_id in product_mapping.items(): # 支持大小写不敏感的匹配 if product_name.lower() == item_lower or product_name in item or item in product_name: product_ids.append(product_id) matched = True self.log(f" 📦 产品模糊匹配: {item} -> {product_name} ({product_id})", "debug") break if not matched: self.log(f" ⚠️ 未找到匹配的产品: {item}", "warning") # 构建备注文本 remark_text = f"姓名:{name}，性别:{gender}，手机号码:{phone}" if remark: remark_text += f"，{remark}" # 构建data JSON data_obj = { "fld43Khnuw": { "type": 1, "value": [{"type": "text", "text": remark_text}] }, "fldexDPwcE": {"type": 3, "value": "optmlZt00z"} if institution else {"type": 3}, # 意向学校 - 使用映射后的ID "fldWt6JrgK": {"type": 3, "value": target_school_id} if target_school_id else {"type": 3}, # 咨询项目 - 使用映射后的ID列表（多选） "fldirDDoLV": {"type": 4, "value": consulting_ids} if consulting_ids else {"type": 4}, # 语培产品 - 使用映射后的ID列表（多选） "fldudfcNVO": {"type": 4, "value": product_ids} if product_ids else {"type": 4}, "fldS90QCXo": {"type": 1, "value": [{"type": "text", "text": name}]} if name else {"type": 1}, "fldsA4YdS6": {"type": 3}, "fldSSElgkA": {"type": 13, "value": {"fullPhoneNum": phone}} if phone else {"type": 13}, "fldZPL52gS": {"type": 1}, "fldPIaD4UH": {"type": 13}, "fldUGeDVDF": {"type": 13}, "fldKZ5WKyN": {"type": 1}, "fld0MHPEaW": {"type": 3}, "fldOLvk3zZ": {"type": 3}, "fldkSVVnuu": {"type": 1}, "fldCfRL3qM": {"type": 3, "value": "optTWIvAFb"} if source else {"type": 3}, "fldodpd0fR": { "type": 11, "value": { "users": [{ "userId": "7646228072081099729", "name": "俞玥戎", "avatarUrl": "https://s3-imfile.feishucdn.com/static-resource/v1/v3_00128_e0f4db42-191a-4878-8546-67a4dcb9005g~?image_size=72x72&cut_type=default-face&quality=&format=jpeg&sticker_format=.webp", "enName": "俞玥戎", "notify": False }] } } if market_person else {"type": 11}, "flde7thxT7": {"type": 3, "value": "optEkk65hv"}, "fldaxavu52": {"type": 3, "value": "optHsn4zKU"}, "fldMpwPaI1": {"type": 17} # 图片字段，默认空 } # ========== 检查并处理图片 ========== if self.image_cache and row_index in self.image_cache: images = self.image_cache[row_index] self.log(f" 🖼️ 检测到 {len(images)} 个嵌入图片在第 {row_index} 行", "info") # 上传第一张图片 if images: img_info = images[0] # 检查是否已经上传过 if row_index in self.uploaded_cache: img_value = self.uploaded_cache[row_index] self.log(f" ♻️ 使用已上传的图片: {img_value.get('attachmentToken')}", "debug") else: # 上传图片 self.log(f" 📤 开始上传图片...", "info") uploaded = self.upload_image(img_info['data'], img_info['filename']) if uploaded: self.uploaded_cache[row_index] = uploaded img_value = uploaded self.log(f" ✅ 图片上传成功，token: {uploaded.get('attachmentToken')}", "success") else: self.log(f" ⚠️ 图片上传失败，跳过该图片", "warning") img_value = None # 如果上传成功，添加到data_obj if img_value: data_obj["fldMpwPaI1"] = { "type": 17, "value": [img_value] } self.log(f" 📎 图片已添加到请求中", "debug") else: # 没有图片，确保图片字段为空 data_obj["fldMpwPaI1"] = {"type": 17} # 构建完整的请求体 body = { "shareToken": self.share_token, "data": json.dumps(data_obj, ensure_ascii=False), "preUploadEnable": True } return body except Exception as e: self.log(f"构建请求体失败: {str(e)}", "error") import traceback self.log(f"详细错误: {traceback.format_exc()}", "debug") return None def send_request(self, row_data, row_index): """发送单条请求""" try: self.log(f"\n{'='*50}", "info") self.log(f"📤 准备发送第 {row_index} 行数据", "info") cookie_string = self.cookie_text.get(1.0, tk.END).strip() if not cookie_string: self.log("❌ Cookie为空，请先配置Cookie", "error") return False, "Cookie为空" # 构建请求体 body = self.build_request_body(row_data, row_index) if body is None: self.log("❌ 构建请求体失败", "error") return False, "构建请求体失败" # 记录请求体大小 body_str = json.dumps(body, ensure_ascii=False) self.log(f"📦 请求体大小: {len(body_str)} 字节", "debug") # 准备请求 headers = self.default_headers.copy() cookies = self.parse_cookie(cookie_string) # 更新动态头 headers["x-csrftoken"] = self._get_csrf_token() headers["f-version"] = "docs-6-17-1781628518399" self.log(f"🌐 发送请求到: {self.url}", "debug") # 发送请求 response = requests.post( self.url, headers=headers, cookies=cookies, json=body, timeout=30 ) self.log(f"📡 响应状态码: {response.status_code}", "debug") # 解析响应 try: resp_data = response.json() resp_text = json.dumps(resp_data, ensure_ascii=False, indent=2) if isinstance(resp_data, dict): if resp_data.get('code') != 0 and resp_data.get('code') is not None: self.log(f"⚠️ 服务器返回错误码: {resp_data.get('code')}", "warning") self.log(f" 错误信息: {resp_data.get('msg', '未知错误')}", "error") if 'data' in resp_data: self.log(f" 📊 响应数据: {json.dumps(resp_data['data'], ensure_ascii=False)[:200]}...", "debug") except: resp_text = response.text self.log("⚠️ 响应不是有效的JSON格式", "warning") if response.status_code == 200: self.log(f"✅ 第{row_index}行 发送成功", "success") return True, resp_text else: self.log(f"❌ 第{row_index}行 发送失败 (HTTP {response.status_code})", "error") self.log(f" 响应内容: {resp_text[:500]}...", "error") return False, resp_text except requests.exceptions.Timeout: self.log(f"⏱️ 第{row_index}行 请求超时", "error") return False, "请求超时" except requests.exceptions.ConnectionError: self.log(f"🔌 第{row_index}行 连接失败，请检查网络", "error") return False, "连接失败" except Exception as e: self.log(f"💥 第{row_index}行 异常: {str(e)}", "error") import traceback self.log(f"详细错误: {traceback.format_exc()}", "debug") return False, str(e) finally: self.log(f"{'='*50}\n", "info") def send_all(self): """发送所有数据的主循环""" try: excel_file = self.file_path_var.get() sheet_name = self.sheet_var.get() min_interval = float(self.min_interval_var.get()) max_interval = float(self.max_interval_var.get()) start_row = int(self.start_row_var.get()) - 1 if not excel_file: self.log("❌ 请先选择Excel文件", "error") self.status_var.set("❌ 请先选择Excel文件") return # 清空上次上传缓存 self.uploaded_cache = {} # ========== 自动提取所有图片到缓存 ========== self.log(f"🔍 正在自动提取Excel中的嵌入图片...", "info") self.image_cache = self.extract_images_to_cache(excel_file, sheet_name) if self.image_cache: total_images = sum(len(imgs) for imgs in self.image_cache.values()) self.log(f"📊 共提取 {total_images} 张图片，分布在 {len(self.image_cache)} 行", "info") rows_with_images = sorted(self.image_cache.keys()) if len(rows_with_images) <= 20: self.log(f"📋 有图片的行: {rows_with_images}", "debug") else: self.log(f"📋 有图片的行: {rows_with_images[:20]} ... (共{len(rows_with_images)}行)", "debug") else: self.log(f"📷 未检测到任何嵌入图片", "info") # ========== 读取Excel数据 ========== try: df = pd.read_excel(excel_file, sheet_name=sheet_name) self.log(f"📊 成功读取Excel: {len(df)} 行数据", "info") self.log(f"📋 列名: {', '.join(df.columns.tolist())}", "info") except Exception as e: self.log(f"❌ 读取Excel失败: {str(e)}", "error") self.status_var.set(f"❌ 读取失败: {str(e)}") return if len(df) == 0: self.log("⚠️ Excel中没有数据", "warning") self.status_var.set("⚠️ 没有数据") return df = df.iloc[start_row:] total = len(df) success_count = 0 fail_count = 0 self.log(f"🚀 开始发送，共 {total} 条数据", "info") self.status_var.set(f"🔄 发送中... 0/{total}") for idx, (_, row) in enumerate(df.iterrows()): if self.stop_flag: self.log("⏹️ 已停止发送", "warning") self.status_var.set("⏹️ 已停止") break # ===== 关键修复：计算Excel实际行号 ===== # DataFrame的索引 + 2（因为第1行是表头，DataFrame从0开始） # 再加上start_row（起始行的偏移） excel_row_number = idx + 2 + start_row self.log(f" 📍 DataFrame索引: {idx}, Excel行号: {excel_row_number}", "debug") # 检查该行是否有图片（使用Excel实际行号） if self.image_cache and excel_row_number in self.image_cache: img_count = len(self.image_cache[excel_row_number]) self.log(f" 🖼️ 第{excel_row_number}行有 {img_count} 个嵌入图片，将自动上传并添加", "info") success, resp_text = self.send_request(row.to_dict(), excel_row_number) if success: success_count += 1 else: fail_count += 1 self.status_var.set(f"🔄 发送中... {idx+1}/{total} (成功:{success_count} 失败:{fail_count})") if idx < total - 1 and not self.stop_flag: interval = random.uniform(min_interval, max_interval) self.log(f"⏳ 等待 {interval:.1f} 秒...", "info") time.sleep(interval) self.log(f"🏁 发送完成！成功: {success_count}, 失败: {fail_count}", "success" if fail_count == 0 else "warning") self.status_var.set(f"✅ 完成 成功:{success_count} 失败:{fail_count}") except Exception as e: self.log(f"💥 程序异常: {str(e)}", "error") import traceback self.log(f"详细错误: {traceback.format_exc()}", "debug") self.status_var.set(f"❌ 异常: {str(e)}") finally: self.is_running = False self.start_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.DISABLED) def start_send(self): """启动发送线程""" if self.is_running: return if not self.file_path_var.get(): messagebox.showwarning("提示", "请先选择Excel文件") return cookie = self.cookie_text.get(1.0, tk.END).strip() if not cookie: messagebox.showwarning("提示", "请先配置Cookie") return try: float(self.min_interval_var.get()) float(self.max_interval_var.get()) int(self.start_row_var.get()) except ValueError: messagebox.showerror("错误", "间隔或行号请输入有效数字") return self.is_running = True self.stop_flag = False self.start_btn.config(state=tk.DISABLED) self.stop_btn.config(state=tk.NORMAL) self.status_var.set("🔄 正在发送...") thread = threading.Thread(target=self.send_all, daemon=True) thread.start() def stop_send(self): """停止发送""" self.stop_flag = True self.log("⏹️ 正在停止...", "warning") self.stop_btn.config(state=tk.DISABLED) # ========== 提取图片功能（手动） ========== def extract_images_to_folder(self, file_path, sheet_name, output_folder=None): """提取Excel中的所有图片并保存到文件夹""" try: if output_folder is None: base_dir = os.path.dirname(file_path) base_name = os.path.splitext(os.path.basename(file_path))[0] output_folder = os.path.join(base_dir, f"{base_name}_提取的图片") if not os.path.exists(output_folder): os.makedirs(output_folder) mapping_file = os.path.join(output_folder, "图片映射表.txt") wb = load_workbook(file_path, data_only=True) ws = wb[sheet_name] image_count = 0 error_count = 0 with open(mapping_file, 'w', encoding='utf-8') as f: f.write("="*60 + "\n") f.write("Excel图片提取映射表\n") f.write(f"源文件: {file_path}\n") f.write(f"工作表: {sheet_name}\n") f.write(f"提取时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write("="*60 + "\n\n") f.write("行号\t列号\t列字母\t文件名\t尺寸\t格式\n") f.write("-"*60 + "\n") if hasattr(ws, '_images') and ws._images: self.log(f" 📸 开始提取 {len(ws._images)} 张图片...", "debug") for idx, img in enumerate(ws._images): try: if hasattr(img, 'anchor') and hasattr(img.anchor, '_from'): anchor_from = img.anchor._from row = anchor_from.row + 1 col = anchor_from.col + 1 col_letter = get_column_letter(col) else: row = idx + 1 col = 0 col_letter = '?' image_data = None if hasattr(img, '_data') and callable(img._data): image_data = img._data() elif hasattr(img, '_data') and not callable(img._data): image_data = img._data elif hasattr(img, 'image'): if hasattr(img.image, '_data') and callable(img.image._data): image_data = img.image._data() elif hasattr(img.image, '_data'): image_data = img.image._data if image_data is None: error_count += 1 continue # 检测格式并保存 img_format = 'jpg' width, height = 0, 0 try: from PIL import Image import io pil_img = Image.open(io.BytesIO(image_data)) img_format = pil_img.format if pil_img.format else 'jpg' width, height = pil_img.size filename = f"第{row}行_列{col_letter}_{idx+1}.{img_format.lower()}" filepath = os.path.join(output_folder, filename) pil_img.save(filepath) except: filename = f"第{row}行_列{col_letter}_{idx+1}.jpg" filepath = os.path.join(output_folder, filename) with open(filepath, 'wb') as f_out: f_out.write(image_data) f.write(f"{row}\t{col}\t{col_letter}\t{filename}\t{width}x{height}\t{img_format}\n") image_count += 1 if image_count % 50 == 0: self.log(f" 📸 已提取 {image_count} 张图片...", "debug") except Exception as e: error_count += 1 self.log(f" ⚠️ 提取图片失败: {str(e)}", "warning") wb.close() with open(mapping_file, 'a', encoding='utf-8') as f: f.write("\n" + "="*60 + "\n") f.write(f"提取完成: 成功 {image_count} 张, 失败 {error_count} 张\n") f.write("="*60 + "\n") self.log(f"✅ 成功提取 {image_count} 张图片到: {output_folder}", "success") if error_count > 0: self.log(f"⚠️ 有 {error_count} 张图片提取失败", "warning") self.log(f"📋 映射文件: {mapping_file}", "info") return output_folder except Exception as e: self.log(f"❌ 提取图片失败: {str(e)}", "error") return None def extract_images(self): """手动提取图片的入口方法""" excel_file = self.file_path_var.get() if not excel_file: messagebox.showwarning("提示", "请先选择Excel文件") return sheet_name = self.sheet_var.get() if not messagebox.askyesno("确认", f"即将提取Excel中的所有图片\n\n" f"文件: {os.path.basename(excel_file)}\n" f"工作表: {sheet_name}\n\n" f"是否继续？"): return def extract_thread(): self.log("📸 开始提取图片...", "info") self.status_var.set("🔄 正在提取图片...") self.start_btn.config(state=tk.DISABLED) output_folder = self.extract_images_to_folder(excel_file, sheet_name) if output_folder: self.status_var.set(f"✅ 图片已提取到: {output_folder}") messagebox.showinfo("提取成功", f"图片已成功提取！\n\n" f"保存位置: {output_folder}\n\n" f"请查看文件夹中的 '图片映射表.txt' 文件") else: self.status_var.set("❌ 提取失败") messagebox.showerror("提取失败", "图片提取失败，请查看日志了解详情。") self.start_btn.config(state=tk.NORMAL) thread = threading.Thread(target=extract_thread, daemon=True) thread.start() if __name__ == "__main__": root = tk.Tk() app = FeishuSubmitApp(root) root.mainloop()