images（c2691+c3640+c7200）.zip资源-CSDN下载

将相关产出计划表的逻辑及规则帮忙梳理清楚，整理一份文档，我需要给别人在另外一软件上开发。 import re import pickle import pandas as pd import json import os import configparser import requests from datetime import date, datetime, timedelta import tkinter as tk from tkinter import ttk, messagebox, filedialog from tkcalendar import Calendar, DateEntry from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Date, DateTime, insert, text import socket import sys import pymysql import subprocess import matplotlib.pyplot as plt from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg import time import threading import logging import traceback from sqlalchemy import text, bindparam import concurrent.futures # 修改导入方式 from Login2 import parse_cookies from batch_cycle import BatchCycle # 获取当前脚本的目录 current_dir = os.path.dirname(os.path.abspath(file)) # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', filename=os.path.join(current_dir, 'server_board.log'), filemode='a' ) logger = logging.getLogger(name) # 构建配置文件的完整路径 config_path = os.path.join(current_dir, 'config.json') # 读取配置文件 try: with open(config_path, 'r', encoding='utf-8') as f: config = json.load(f) LABELS = config['labels'] RENAME = config['rename'] except FileNotFoundError: print(f"Error: The file {config_path} does not exist.") # MySQL 数据库连接信息 mysql_host = '10.57.176.92' mysql_user = 'zxr' mysql_password = 'asd123..' mysql_port = '3306' mysql_database = 'bj_database' # 创建数据库引擎 engine = create_engine( f'mysql+pymysql://{mysql_user}:{mysql_password}@{mysql_host}:{mysql_port}/{mysql_database}?charset=utf8mb4', echo=True) # 定义当前脚本版本 CURRENT_VERSION = 1.0 def check_and_update_version(): """检查并更新脚本版本""" global CURRENT_VERSION connection = None try: # 连接数据库 connection = pymysql.connect( host=mysql_host, user=mysql_user, password=mysql_password, database=mysql_database ) with connection.cursor() as cursor: # 查询版本控制表 sql = "SELECT version, file, file_name FROM version_control_sz WHERE CAST(version AS DECIMAL(10, 1)) > %s ORDER BY CAST(version AS DECIMAL(10, 1)) DESC" cursor.execute(sql, (CURRENT_VERSION,)) results = cursor.fetchall() if results: db_version = float(results[0][0]) root = tk.Tk() root.withdraw() # 隐藏主窗口 if messagebox.askyesno("版本更新", f"发现新版本 {db_version}，是否下载？"): file_name = results[0][2] # 在文件名中添加版本号 new_file_name = f"{file_name.rsplit('.', 1)[0]}_v{db_version}.exe" # 使用旧版本文件的实际路径 new_file_path = os.path.join(os.path.dirname(sys.executable), new_file_name) # 下载新版本文件 with open(new_file_path, 'wb') as file: for result in results: if float(result[0]) == db_version: file.write(result[1]) # 检查文件大小 if os.path.getsize(new_file_path) < 1000000: # 假设文件应大于1MB messagebox.showerror("错误", "下载的文件不完整，请检查数据库中的文件数据。") return messagebox.showinfo("版本更新", f"新版本文件已下载: {new_file_path}\n程序将自动重启。") # 创建批处理文件 batch_file = os.path.join(os.path.dirname(sys.executable), "update.bat") with open(batch_file, 'w') as f: f.write(f"@echo off\n") f.write(f"ping 127.0.0.1 -n 5 > nul\n") # 等待5秒 f.write(f"start \"\" \"{new_file_path}\"\n") f.write(f"del \"{sys.executable}\"\n") f.write(f"del \"%~f0\"\n") # 删除批处理文件自身 # 启动批处理文件 subprocess.Popen(batch_file, shell=True) # 退出当前程序 sys.exit() else: print(f"当前版本: {CURRENT_VERSION}, 数据库中无更高版本") except Exception as e: messagebox.showerror("错误", f"检查和更新版本时出错: {e}") finally: if connection: connection.close() print(f"CURRENT_VERSION: {CURRENT_VERSION}") class SimpleLoadingDialog: def init(self, parent): self.top = tk.Toplevel(parent) self.top.title("加载中") self.top.geometry("300x100") self.top.transient(parent) self.top.grab_set() # 获取屏幕宽度和高度 screen_width = self.top.winfo_screenwidth() screen_height = self.top.winfo_screenheight() # 计算窗口位置 x = (screen_width - 300) // 2 y = (screen_height - 100) // 2 # 设置窗口位置 self.top.geometry(f"400x100+{x}+{y}") ttk.Label(self.top, text="数据加载中，请稍候...").pack(expand=True, pady=20) self.top.update_idletasks() def destroy(self): self.top.grab_release() self.top.destroy() class DataProcessor: @staticmethod def fetch_url6_data(factory_id=None): """获取产品信息设置数据""" try: cookies = parse_cookies() if not cookies: print("无法获取cookies，请检查登录状态") return pd.DataFrame() emp_no = cookies.get("ZTEDPGSSOUser") app_access_token = cookies.get("APPACCESSTOKEN") if not emp_no: print("Cookies中未找到工号信息") return pd.DataFrame() url_6 = "https://imes.zte.com.cn/zte-mes-manufactureshare-gateway/zte-mes-manufactureshare-crafttechsys/productInfoSettings/pageList" # 如果未指定工厂ID，则尝试多个工厂 factory_ids = [factory_id] if factory_id else ['53', '55', '58'] # 深圳、河源、南京 factory_names = {'53': '深圳', '55': '河源', '58': '南京'} all_rows = [] for fid in factory_ids: headers = { "Content-Type": "application/json", "User-Agent": "Mozilla/5.0", "x-emp-no": emp_no, "x-auth-value": app_access_token, "x-factory-id": str(fid) } payload = { "order": "desc", "orderField": "lastUpdatedDate", "page": 1, "rows": 10000 } resp = requests.post(url_6, json=payload, cookies=cookies, headers=headers, timeout=30) if resp.status_code == 200: data = resp.json() rows = data.get('bo', {}).get('rows', []) if rows: # 为每条数据添加工厂信息 for row in rows: row['factoryId'] = fid row['factoryName'] = factory_names.get(fid, f'工厂{fid}') all_rows.extend(rows) print(f"工厂ID {fid} 获取到 {len(rows)} 条记录") if all_rows: df6 = pd.DataFrame(all_rows) if not df6.empty and 'RENAME' in globals(): # 只对存在于RENAME中的列进行重命名 rename_cols = {k: v for k, v in RENAME.items() if k in df6.columns} df6.rename(columns=rename_cols, inplace=True) return df6 else: print("未获取到产品信息设置数据") return pd.DataFrame() except Exception as e: print(f"获取产品信息设置数据时出错: {str(e)}") traceback.print_exc() return pd.DataFrame() @staticmethod def process_data(factory_id): try: print(f"\n===== 开始处理工厂数据，当前工厂ID: {factory_id} =====") print(f"当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") cookies = parse_cookies() if not cookies: print("无法获取cookies，请检查登录状态") return pd.DataFrame() # 如果没有提供factory_id，使用默认值 if factory_id is None: factory_id = '53' # 默认深圳工厂 print(f"使用默认工厂ID: {factory_id}") emp_no = cookies.get("ZTEDPGSSOUser") app_access_token = cookies.get("APPACCESSTOKEN") if not emp_no: print("Cookies中未找到工号信息，请检查登录流程") return pd.DataFrame() print(f"获取到工号: {emp_no}") daY = date.today() headers = { "x-factory-id": factory_id, "x-emp-no": emp_no, "x-auth-value": app_access_token } session = requests.Session() # 定义仓库ID映射 warehouse_mapping = { # 深圳工厂 '53': ["1316200666516144130","1324273272430747650","1324273657404940290", "1324274108225511426","1579367656533045249","1753967825651752961", "1873600463992856578","1887371251385540610"], # 河源工厂 '55': ["1308313462591893506","1579283316397719553","1752652434774433794"], # 南京工厂 '58': ["1300275955771727873","1324257039408914433","1324259519807053826", "1324260565228937217","1324265806070505474","1324266346598850562", "1413007623030095873","1468034273549975554","1579369298049413122"] } # 所有仓库ID（合并所有工厂） all_warehouse_ids = list(set( warehouse_mapping['53'] + warehouse_mapping['55'] + warehouse_mapping['58'] )) # 根据工厂ID选择仓库ID列表 if factory_id in warehouse_mapping: warehouse_ids = warehouse_mapping[factory_id] print(f"使用工厂 {factory_id} 的仓库ID列表，共 {len(warehouse_ids)} 个仓库") else: warehouse_ids = all_warehouse_ids print(f"工厂 {factory_id} 未配置，使用全部仓库ID，共 {len(warehouse_ids)} 个仓库") url_1 = "https://imes.zte.com.cn/zte-mes-manufactureshare-gateway/zte-mes-manufactureshare-productionmgmtsys/wipDailyStatistics/query" url_2 = "https://imes.zte.com.cn/zte-mes-manufactureshare-gateway/zte-mes-manufactureshare-planschedulesys/PS/schedulingAdjustPage?workOrderStatusList=%E5%B7%B2%E6%8F%90%E4%BA%A4,%E5%B7%B2%E5%BC%80%E5%B7%A5,%E6%8C%82%E8%B5%B7&workshopCode=&lineCode=&order=asc&sort=scheduleStartDate&page=1&rows=30000&craftSections=%E8%83%8C%E6%9D%BF,%E9%93%A3%E6%9D%BF,DIP,%E6%B6%82%E8%A6%86,Test,SMT-A,SMT-B,%E6%89%8B%E8%B4%B4" url_3 = "https://imes.zte.com.cn/zte-mes-manufactureshare-gateway/zte-mes-manufactureshare-planschedulesys/workOrderPriorityQueue/page" url_4 = "https://imes.zte.com.cn/zte-mes-manufactureshare-gateway/zte-mes-manufactureshare-productionmgmtsys/conciseDaily/conciseDailyQuery" url_5 = "https://imes.zte.com.cn/zte-mes-manufactureshare-gateway/zte-mes-manufactureshare-productiondeliverysys/linesideWarehouseInfoCtrl/getReqHead" print(f"\n===== URL列表 =====") print(f"URL1 (wipDailyStatistics): {url_1}") print(f"URL2 (schedulingAdjustPage): {url_2}") print(f"URL3 (workOrderPriorityQueue): {url_3}") print(f"URL4 (conciseDailyQuery): {url_4}") print(f"URL5 (linesideWarehouseInfo): {url_5}") # 并发分页拉取函数 def fetch_page_1(page): data_1 = { "endDate": daY.strftime('%Y-%m-%d 23:59:59'), "startDate": daY.strftime('%Y-%m-%d 00:00:00'), "page": page, "rows": 1000 } resp = session.post(url_1, json=data_1, cookies=cookies, headers=headers, timeout=60) if resp.status_code == 200: return resp.json().get('bo', {}).get('rows', []) else: return [] def fetch_page_4(page): headers_4 = headers.copy() headers_4["Content-Type"] = "application/json" data_4 = { "startDate": daY.strftime('%Y-%m-%d 00:00:00'), "endDate": daY.strftime('%Y-%m-%d 23:59:59'), "page": page, "row": 10000, "selectColList": [ "ALL", "smtB", "smtA", "plugInUnit", "inspectionWelding", "assembling", "coating", "millingPlate", "clean", "ict", "test", "coiledFiber", "backplane", "powerModuleMake", "powerModuleTest", "moduleHighTemperature", "modulePressurization", "moduleTest1", "moduleTest2", "highTemperatureDistribution", "highTemperatureStandard", "componentAssembly", "componentCommission", "powerCompleteMachineAssembly", "powerCompleteMachineTest", "scrapping", "scrapped", "inMaintenance", "hotTest", "vibrationTest", "visualTest", "eolTest" ], "surplusCrafts": [], "realTimeQueryFlag": "true", "rtaskNo": "" } resp = session.post(url_4, json=data_4, cookies=cookies, headers=headers_4, timeout=30) if resp.status_code == 200: return resp.json().get('bo', {}).get('rows', []) else: return [] # 在并发请求部分添加url_5的处理 def fetch_page_5(): payload = { "current": 1, "size": 3000, "searchCount": True, "t": { "billInfoType": None, "billTypeCode": "XBRK_YA", "billNo": None, "statusCode": None, "srcBillNo": None, "productionBatch": None, "warehouseIds": warehouse_ids, # 使用动态选择的仓库ID "dateStart": (daY - timedelta(days=7)).strftime('%Y-%m-%d 00:00:00'), "dateEnd": daY.strftime('%Y-%m-%d 23:59:59'), "attribute1": "目的仓库", "itemNo": None } } resp = session.post(url_5, json=payload, cookies=cookies, headers=headers, timeout=30) if resp.status_code == 200: return resp.json().get('bo', {}).get('rows', []) else: print(f"url_5请求失败") return [] # 并发拉取url_2和url_3 def fetch_url2(): print(f"\n----- 开始请求2 (schedulingAdjustPage) -----") print(f"请求时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"请求URL: {url_2}") print(f"请求头: {headers}") resp = session.get(url_2, headers=headers, cookies=cookies) print(f"响应状态码: {resp.status_code}") print(f"获取到数据行数: {len(resp.json().get('bo', {}).get('rows', []))}") return resp def fetch_url3(): data_3 = { "releaseDateStart": "", "releaseDateEnd": "", "deliveryDateStart": "", "deliveryDateEnd": "", "page": 1, "rows": 300, } print(f"\n----- 开始请求3 (workOrderPriorityQueue) -----") print(f"请求时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"请求URL: {url_3}") print(f"请求参数: {data_3}") print(f"请求头: {headers}") resp = session.post(url_3, headers=headers, json=data_3, cookies=cookies) print(f"响应状态码: {resp.status_code}") print(f"获取到数据行数: {len(resp.json().get('bo', {}).get('rows', []))}") return resp max_pages = 10 # 估算最大页数，根据实际情况可调大/小 with concurrent.futures.ThreadPoolExecutor(max_workers=16) as executor: # url_2和url_3 future2 = executor.submit(fetch_url2) future3 = executor.submit(fetch_url3) # url_1分页 futures1 = [executor.submit(fetch_page_1, p) for p in range(1, max_pages+1)] # url_4分页 futures4 = [executor.submit(fetch_page_4, p) for p in range(1, max_pages+1)] response2 = future2.result() response3 = future3.result() # 添加url_5的请求 print(f"\n----- 开始请求5 (linesideWarehouseInfo) -----") print(f"请求时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"请求URL: {url_5}") print(f"请求头: {headers}") future5 = executor.submit(fetch_page_5) all_rows_5 = future5.result() print(f"获取到数据行数: {len(all_rows_5)}") if all_rows_5: print(f"首条数据示例: {all_rows_5[0]}") # 收集url_1分页数据 all_rows_1 = [] for f in futures1: rows = f.result() if rows: all_rows_1.extend(rows) if len(rows) < 1000: break # 收集url_4分页数据 all_rows_4 = [] for f in futures4: rows = f.result() if rows: all_rows_4.extend(rows) if len(rows) < 1000: break # url_1数据 df1 = pd.DataFrame(all_rows_1) df1.rename(columns=RENAME, inplace=True) # === 新增：合并工艺属性 === df_product = DataProcessor.fetch_url6_data(factory_id) if not df_product.empty: # 确保字段名一致，假设df1和df_product都是“单板代码” df1 = pd.merge(df1, df_product[['单板代码', '工艺属性']], on='单板代码', how='left') # url_4数据 df4 = pd.DataFrame(all_rows_4) df4.rename(columns=RENAME, inplace=True) # 获取 url_3 的数据 data3_json = response3.json() rows3 = data3_json.get('bo', {}).get('rows', []) df3 = pd.DataFrame(rows3) df3.rename(columns=RENAME, inplace=True) # 只保留“计划跟踪单”和“队列状态”两列 df3 = df3[["计划跟踪单", "队列状态"]] # 将“队列状态”为“空”的值更改为“已加入队列优先级，待触发需求单” df3["队列状态"] = df3["队列状态"].replace("空", "待触发需求单") # url_2数据处理 data2_json = response2.json() rows2 = data2_json.get('bo', {}).get('rows', []) pf = pd.DataFrame(rows2) pf.rename(columns=RENAME, inplace=True) pf.rename(columns={ 'prodplanId': '批次', '线体': '维护线体', 'workOrderStatus': '指令状态', 'stockStatus': '备料状态', 'craftMain': '工序' }, inplace=True) # 在“过滤只保留特定工序”前面，保留线体作为一个新增的列 pf['线体'] = pf['维护线体'] # 将维护线体保留为一个新的列 # 按批次分组，取第一个线体 pf['线体'] = pf.groupby('批次')['维护线体'].transform('first') # 过滤只保留特定工序 pf_filtered = pf[pf['工序'].isin(['Test', 'DIP', 'SMT'])].copy() # 创建新的列，组合维护线体和指令状态 pf_filtered.loc[:, '组合'] = pf_filtered['维护线体'] + ': ' + pf_filtered['指令状态'] # 新增字段：SMT备料 pf_filtered.loc[:, 'SMT备料'] = pf_filtered.groupby(['批次', '维护线体'])['备料状态'].transform(lambda x: x.iloc[0]).fillna('') # 处理数据合并逻辑 def merge_rows(group): # 初始化 DIP 和 Test 列 dip_data = group[group['工序'] == 'DIP']['组合'].iloc[0] if 'DIP' in group['工序'].values else '' test_data = group[group['工序'] == 'Test']['组合'].iloc[0] if 'Test' in group['工序'].values else '' # 获取 SMT 行 smt_rows = group[group['工序'] == 'SMT'].copy() if smt_rows.empty: # 如果没有 SMT 行，创建一行空的 SMT 行用于合并 smt_rows = pd.DataFrame([{ '批次': group['批次'].iloc[0], '线体': group['线体'].iloc[0], # 使用批次的第一个线体 '组合': '', 'SMT备料': '', 'DIP': dip_data, 'Test': test_data }]) else: # 删除维护线体相同的重复行 smt_rows = smt_rows.drop_duplicates(subset=['线体']) smt_rows['DIP'] = dip_data smt_rows['Test'] = test_data return smt_rows # 按批次分组并应用合并逻辑 pf_final = pf_filtered.groupby('批次').apply(merge_rows).reset_index(drop=True) # 修正列名 pf_final = pf_final[['批次', '线体', '组合', 'SMT备料', 'DIP', 'Test']] # 合并数据框并重命名列 print(f"\n----- 开始合并df1和pf_final -----") print(f"合并前df1行数: {len(df1)}") print(f"合并前pf_final行数: {len(pf_final)}") # 在合并前先去除重复批次 df1 = df1.drop_duplicates(subset=['批次'], keep='first') pf_final = pf_final.drop_duplicates(subset=['批次'], keep='first') print(f"去重后df1行数: {len(df1)}") print(f"去重后pf_final行数: {len(pf_final)}") df_1 = pd.merge(df1, pf_final, on='批次', how='left') print(f"合并后df_1行数: {len(df_1)}") print(f"合并后重复批次检查: {df_1.duplicated(subset=['批次']).sum()}个重复批次") df_1.rename(columns={ '线体': '线体', '组合': 'SMT线体及状态', 'DIP': 'DIP线体及状态', 'Test': 'Test线体及状态' }, inplace=True) df_1.fillna('', inplace=True) # 读取 delivery_date 表的数据 print(f"\n----- 开始读取delivery_date表数据 -----") df_sql = pd.read_sql("SELECT * FROM delivery_date", con=engine) print(f"获取到delivery_date表数据行数: {len(df_sql)}") # 保留批次相同且 create_time 最大的行 df_sql = df_sql.sort_values('create_time').drop_duplicates('批次', keep='last') # 通过共同列批次，进行左链接 print(f"\n----- 开始合并df_1和df_sql -----") print(f"合并前df_1行数: {len(df_1)}") print(f"合并前df_sql行数: {len(df_sql)}") df = pd.merge(df_1, df_sql, on='批次', how='left') print(f"合并后df行数: {len(df)}") # 新增一个字段"是否急件" df['是否急件'] = df['单板交期'].apply(lambda x: '是' if pd.notna(x) else '') print(f"新增'是否急件'字段后行数: {len(df)}") # ============================================================== # 从 SQL 表中导出计调员数据和周期类别 query = "SELECT 工厂, 单板代码, 计调员, 周期类别 FROM part_category" part_category_df = pd.read_sql(query, con=engine) # 提取单板代码前 12 位作为新的列 part_category_df['单板代码12位'] = part_category_df['单板代码'].str[:12] # 在 df 中提取单板代码前 12 位作为新的列 df['单板代码12位'] = df['单板代码'].str[:12] # 进行左连接，同时匹配工厂和单板代码12位 df = pd.merge(df, part_category_df[['工厂', '单板代码12位', '计调员', '周期类别']], on=['工厂', '单板代码12位'], how='left') # 填充默认值 df['计调员'] = df['计调员'].fillna('') df['周期类别'] = df['周期类别'].fillna('') # 创建 part_category_dict part_category_dict = part_category_df.set_index(['工厂', '单板代码12位'])['计调员'].to_dict() # 调用 assign_responsible_person 方法分配责任人 df['计调员'] = df.apply(lambda row: DataProcessor.assign_responsible_person(row, part_category_dict) if row['计调员'] == '' else row['计调员'], axis=1) # 在“df['计调员'] = df['计调员'].fillna('')”后面新增，SQL导出lineteamleader表格 query = "SELECT 线体, 小组长, 大组长, 科室 FROM lineteamleader" lineteamleader_df = pd.read_sql(query, con=engine) # 以df为主表，连接列为线体 df = pd.merge(df, lineteamleader_df, on='线体', how='left') # 计调员与科室分开列展示 print(f"\n----- 计调员和科室信息 -----") print(f"计调员示例: {df['计调员'].iloc[0] if len(df) > 0 else '无数据'}") print(f"科室示例: {df['科室'].iloc[0] if len(df) > 0 else '无数据'}") df3 = df3.groupby('计划跟踪单').last().reset_index() df = pd.merge(df, df3, on='计划跟踪单', how='left', suffixes=('', '_queue')) # ============================================================== # 处理工艺组合路径中的逗号 df['工艺组合路径'] = df['工艺组合路径'].apply(lambda x: x.replace(',', '') if x is not None else x) # 按计划跟踪单排序 df = df.sort_values(by=['计划跟踪单'], ascending=[True]) # 过滤外协标志为"是"的行 df = df[~(df['外协标志'] == '是')] # 计算结存明细 columns_to_check = ['备料区', '贴片B面', '贴片A面', '铣板', '短插', '检焊', '单板装配', '涂覆', '涂覆装配', '清洗', '在线测试', '单板测试', '背板', '高温标模', '提单', '部件维修', '电源模块', '部件装配', '电源整机装配', '电源整机测试', 'WDM配送', 'WDM装配'] def calculate_stock_detail(row): details = [] for col in columns_to_check: if row[col] > 0: details.append(f"{col}结存{row[col]}块") return '; '.join(details) df['结存明细'] = df.apply(calculate_stock_detail, axis=1) # 判断是否贴片 def is_smt(path): if path is None or not path: return '否' first_char = path[0] if any(first_char in group for group in ('GXZηrY', '12STQ')): return '整机指令' if first_char in 'GXZηrY' else '是' return '否' df['是否贴片'] = df['工艺组合路径'].apply(is_smt) # 转换释放日期为日期时间格式 df['释放日期'] = pd.to_datetime(df['释放日期'], errors='coerce') # 计算在线周期 def calculate_online_duration(row): current_date = pd.Timestamp.now() release_date = row['释放日期'] if pd.notna(release_date): duration = current_date - release_date return f"{duration.days + duration.seconds / (24 * 60 * 60):.2f}" return None df['在线周期'] = df.apply(calculate_online_duration, axis=1) # 计算部件结存比例并保留两位小数 def calculate_stock_ratio(row): if row['是否贴片'] == '整机指令': return 0.0 else: total_stock = row.get('部件在线', 0) + row.get('备料区', 0) planned_quantity = row.get('计划数量', 1) if planned_quantity == 0: return 0.0 ratio = (total_stock / planned_quantity) return round(ratio, 3) df['部件结存比例'] = df.apply(calculate_stock_ratio, axis=1) # 添加 '部件比例' 列，将 '部件结存比例' 转换为带百分号的字符串格式 df['部件比例'] = df['部件结存比例'].apply(lambda x: f"{x * 100:.2f}%") # 计算全部结存比例并保留两位小数 df['全部结存比例'] = df.apply(lambda row: round(row['在线总数'] / row['计划数量'], 3) if row['计划数量'] != 0 else 0, axis=1) # 计算结存状态 def calculate_stock_status(row): plan_quantity = row['计划数量'] if row['是否贴片'] == '整机指令': return '整机' if row['备料区'] == plan_quantity: return '备料区' if (row['备料区'] + row['贴片B面'] + row['贴片A面']) >= plan_quantity / 7: return 'SMT' if row['涂覆'] >= plan_quantity / 7: return '涂覆' if (row['铣板'] + row['短插'] + row['检焊'] + row['单板装配'] + row['涂覆装配'] + row['清洗']) >= plan_quantity / 7: return '装焊车间' if row['在线测试'] >= plan_quantity / 7: return 'ICT' if row['单板测试'] >= plan_quantity / 7: return '功能' if row['盘纤'] >= plan_quantity / 7: return '盘纤' if row['部件维修'] >= plan_quantity / 7: return '部件维修' if row['背板'] >= plan_quantity / 7: return '背板' if row['高温标模'] >= plan_quantity / 7: return '高温' if (row['电源整机装配'] + row['电源整机测试'] + row['WDM配送'] + row['WDM装配'] + row['部件装配'] + row['高温标模']) >= plan_quantity / 7: return '整机' if row['提单'] >= plan_quantity / 7: return '提单' return '入库' df['结存状态'] = df.apply(calculate_stock_status, axis=1) # 过滤在线总数大于0的行 df = df[df['在线总数'] > 0] # 增加两个新字段 df['SMT结存'] = df['贴片B面'] + df['贴片A面'] df['DIP结存'] = df['铣板'] + df['短插'] + df['检焊'] + df['单板装配'] + df['涂覆装配'] + df['清洗'] # 添加基础部件列并计算配套数量 df['base_part'] = df['计划跟踪单'].str.split('-z').str[0] base_part_counts = df['base_part'].value_counts() def calculate_accompanying_quantity(row): base_part = row['base_part'] if '-z-' in row['计划跟踪单']: return 0 return base_part_counts.get(base_part, 0) df['配套数量'] = df.apply(calculate_accompanying_quantity, axis=1) df = df.drop(columns=['base_part']) # 1. 添加"计划跟踪单简"列 df['计划跟踪单简'] = df['计划跟踪单'].apply( lambda x: str(x).split('-', 1)[0] if pd.notna(x) and '-' in str(x) else str(x) ) # 判断是否过装焊 df['是否过装焊'] = df['工艺组合路径'].str.contains( r'[68BA\+]', regex=True).map({True: '是', False: '否'}) # 1. 从数据库中读取 gongxu90time 表的数据 query = "SELECT 批次, 转交时间 FROM gongxu90time" gongxu90time_df = pd.read_sql(query, con=engine) # 2. 筛选每个批次中转交时间最大的记录 max_transfer_time_df = gongxu90time_df.groupby('批次')['转交时间'].max().reset_index() max_transfer_time_df.rename(columns={'转交时间': '最后一道工序转交时间'}, inplace=True) # 3. 将筛选后的数据合并到 df 表中 df = pd.merge(df, max_transfer_time_df, on='批次', how='left') # 计算前工序结存时间（天） df['前工序结存时间'] = ((df['最后一道工序转交时间'] - df['释放日期']).dt.total_seconds() / (3600 24)).round(2) # 计算本工序结存时间（天） df['本工序结存时间'] = ((pd.Timestamp.now() - df['最后一道工序转交时间']).dt.total_seconds() / (3600 * 24)).round(2) if all_rows_5: try: df5 = pd.DataFrame(all_rows_5) # 新增：按批次去重 df5 = df5.drop_duplicates(subset=['productionBatch'], keep='last') print(f"成功获取url_5数据，共{len(df5)}条记录") # 确保必要的列存在 required_columns = ['productionBatch', 'statusCode', 'toWarehouseId'] if all(col in df5.columns for col in required_columns): df5.rename(columns=RENAME, inplace=True) # 只保留需要的列 df5 = df5[['批次', '单据状态', '目的仓库']] # 合并到主数据框 df = pd.merge(df, df5, on='批次', how='left') print("成功合并url_5数据") else: print("url_5数据缺少必要列，实际列名:", df5.columns) df['单据状态'] = '' df['目的仓库'] = '' except Exception as e: print(f"处理url_5数据异常：{str(e)}") traceback.print_exc() df['单据状态'] = '' df['目的仓库'] = '' else: print("未获取到url_5的数据") df['单据状态'] = '' df['目的仓库'] = '' # ===== 新增：优先处理备料区大于0的批次，直接按preferred_lines选线体 ===== # ...existing code... preferred_lines = [ 'SMTB线体', 'SMTA线体', '插件线体', '检焊线体', '装配线体', '涂覆线体', '铣板线体', '清洗线体', 'ICT线体', 'TEST线体', '盘纤线体', '背板线体', '电源模块制作线体', '电源模块测试线体', '模块高温线体', '模块耐压线体', '模块测试1线体', '模块测试2线体', '高温配送线体', '高温标模线体', '部件装配线体', '部件调试线体', '定制电源整机装配线体', '定制电源整机测试线体', '报废中线体', '已报废线体', '维修中线体', '高温测试线体', 'EOL测试线体', '视觉测试线体', '振动测试线体' ] batch_line_dict = {} # 1. 找出df4中所有“结存”结尾的列 jiecun_cols = [col for col in df4.columns if col.endswith('结存')] def get_max_jiecun_line(row): max_val = -1 max_col = None for col in jiecun_cols: val = row.get(col, 0) if pd.notna(val) and val > max_val: max_val = val max_col = col # 根据最大结存列名，找对应线体（如SMTB结存->SMTB线体） if max_col: line_col = max_col.replace('结存', '线体') return row.get(line_col, '') return '' # 2. 计算最大结存线体并覆盖df['线体'] if not df4.empty and '批次' in df4.columns: df4['最大结存线体'] = df4.apply(get_max_jiecun_line, axis=1) for _, row in df4.iterrows(): batch = row['批次'] max_line = row['最大结存线体'] if pd.notna(max_line) and str(max_line).strip(): mask = df['批次'] == batch df.loc[mask, '线体'] = max_line else: df['最大结存线体'] = '' # 3. 用备料区的线体覆盖（优先级更高） if not df4.empty and '批次' in df4.columns: for _, row in df4.iterrows(): batch = row['批次'] mask = (df['批次'] == batch) & (df['结存状态'] == '备料区') if mask.any(): first_line = "" for col in preferred_lines: val = row.get(col, "") if pd.notna(val) and str(val).strip(): first_line = val break if first_line: df.loc[mask, '线体'] = first_line else: print(f"处理数据时出错: {str(e)}") # ===== 结束新增代码 ===== # print(df.columns) df = df.drop_duplicates(subset=['批次']) return df except Exception as e: return None # 定义 assign_responsible_person 函数 @staticmethod def assign_responsible_person(row, part_category_dict=None): """ 根据 row 中的数据分配责任人（计调员）。参数: row (pd.Series): 包含单板信息的一行数据。 part_category_dict (dict, optional): 工厂和单板代码12位到计调员的映射字典。返回: str: 分配的责任人名称，如果没有匹配则返回"未分配"。 """ # 首先尝试从part_category_dict中查找 if part_category_dict is not None: try: # 提取单板代码前12位 board_code_12 = row.get('单板代码', '')[:12] if row.get('单板代码') else '' factory = row.get('工厂', '') # 如果在字典中找到匹配项，直接返回 if (factory, board_code_12) in part_category_dict and part_category_dict[(factory, board_code_12)]: return part_category_dict[(factory, board_code_12)] except Exception as e: print(f"从part_category_dict查找责任人时出错: {str(e)}") # 如果字典中没有找到或出错，使用规则分配 # 规则 1: 如果工厂是 "河源"，且工艺路径包含 "背板"，则返回 "徐捷如" if row['工厂'] == "河源" and (row['工艺路径'] is not None) and "背板" in row['工艺路径']: return "徐捷如" # 规则 2: 如果工厂是 "河源"，且单板名称匹配正则表达式，则返回 "何东" elif row['工厂'] == "河源" and re.search(r'^PPC|^PA|^PD|^PN|^ZXSDR BS8800 C100 PSUB0', row['单板名称']): return "何东" # 规则 3: 如果工厂是 "河源"，且产品大类在 ["ES", "MSR(R10)", "CLS", "IC", "MSTP"] 中，则返回 "钟慧珊" elif row['工厂'] == "河源" and row['产品大类'] in ["ES", "MSR(R10)", "CLS", "IC", "MSTP"]: return "钟慧珊" # 规则 4: 如果工厂是 "河源"，且产品大类是 "5G-WS"，则返回 "无线" elif row['工厂'] == "河源" and row['产品大类'] in ["5G-WS"]: return "无线" # 规则 5: 如果工厂是 "河源"，且产品大类在 ["CR", "CSR(R10)", "DCCS", "IPRAN-C", "ECSR"] 中，则返回 "孙树冬" elif row['工厂'] == "河源" and row['产品大类'] in ["CR", "CSR(R10)", "DCCS", "IPRAN-C", "ECSR"]: return "孙树冬" # 规则 6: 如果工厂是 "河源"，且产品大类在 ["5G_CMxU", "5G_COLT", "5G_CONT", "A10", "CHS", "CPE", "DSL", "EODN", "J10", "ONT", "POL", "PON_MxU", "PON_OLT", "PONG", "POND", "OTN", "ITPS", "MSAP", "CLOUD", "CT", "ESI", "IAD", "iCDN", "uSmartView", "VCS", "VCT"] 中，则返回 "肖应仲" elif row['工厂'] == "河源" and row['产品大类'] in ["5G_CMxU", "5G_COLT", "5G_CONT", "A10", "CHS", "CPE", "DSL", "EODN", "J10", "ONT", "POL", "PON_MxU", "PON_OLT", "PONG", "POND", "OTN", "ITPS", "MSAP", "CLOUD", "CT", "ESI", "IAD", "iCDN", "uSmartView", "VCS", "VCT"]: return "肖应仲" # 规则 7: 如果工厂是 "河源"，且产品大类在 ["C-SPN", "IPRAN-A", "SPN", "D-SPN", "IPRAN", "IPTN"] 中，则返回 "徐捷如" elif row['工厂'] == "河源" and row['产品大类'] in ["C-SPN", "IPRAN-A", "SPN", "D-SPN", "IPRAN", "IPTN"]: return "徐捷如" # 规则 8: 如果工厂是 "河源"，且产品大类在 ["CDCG", "DC_P", "DPS", "POWER", "PS_S", "SSE", "TSE", "3GPLAT", "DC_S", "SE"] 中，则返回 "祁思佳" elif row['工厂'] == "河源" and row['产品大类'] in ["CDCG", "DC_P", "DPS", "POWER", "PS_S", "SSE", "TSE", "3GPLAT", "DC_S", "SE"]: return "祁思佳" # 规则 9: 如果工厂是 "河源"，且产品大类是 "RRS"，则返回 "何东" elif row['工厂'] == "河源" and row['产品大类'] in ["RRS"]: return "何东" # 规则 10: 如果工厂是 "深圳"，且产品大类在 ["ITPS", "RSS"] 中，则返回 "江启一" elif row['工厂'] == "深圳" and row['产品大类'] in ["ITPS", "RSS"]: return "江启一" # 规则 11: 如果工厂是 "深圳"，且产品大类在 ["OTN", "WDM"] 中，则返回 "杨港会" elif row['工厂'] == "深圳" and row['产品大类'] in ["OTN", "WDM"]: return "杨港会" # 规则 12: 如果工厂是 "深圳"，且产品大类是 "CP"，则返回 "周旭升" elif row['工厂'] == "深圳" and row['产品大类'] in ["CP"]: return "周旭升" # 如果没有匹配到任何规则，则返回"未分配" else: return "未分配" class FactorySelector(tk.Toplevel): def init(self, parent): super().init(parent) self.result = [] self.title("选择工厂") self._setup_ui() def _setup_ui(self): ttk.Label(self, text="请选择需要查询的工厂：").pack(pady=10) # 工厂代码与工厂ID的映射 factories = { '53': '深圳工厂', '55': '河源工厂', '58': '南京工厂', '全部': '全部' } # 创建按钮 for code, name in factories.items(): ttk.Button( self, text=name, command=lambda x=code: self._select(x) ).pack(pady=2) # 在窗口底部添加备注 ttk.Label(self, text="工厂权限只能获取IMES已有权限！！！", foreground="red").pack(pady=10) def _select(self, factory): if factory == "全部": self.result = ['53', '55', '58'] else: self.result = [factory] self.destroy() class DataViewer: def init(self): # 确保root窗口只创建一次 self.root = tk.Tk() self.root.title("计调科数据查询系统") self.root.geometry("1400x800") # 初始化基本属性 self.current_dir = os.path.dirname(os.path.abspath(file)) self.last_update_time = None self.auto_update_id = None self.cache_expiry_hours = 10000 # 缓存过期时间为24小时 # === 先初始化缓存路径 === self.init_cache_paths() # 👈 先调用 init_cache_paths() 来初始化 cache_file # 定义列顺序和宽度 self.column_order = [ '序号', '批次', '单板名称', '计划数量', '工艺组合路径', '结存状态', '线体', '在线周期', '前工序结存时间', '本工序结存时间', '项目', '部件比例', '队列状态', '工艺属性', '单据状态', '目的仓库', 'SMT线体及状态', 'SMT备料', 'DIP线体及状态', '计划跟踪单', '单板代码', '释放日期', '计调员', '科室', '大组长', '小组长', '单板交期', '提出人', '备料区', '提单', '入库', '部件入库', '部件在线', '单板测试', '在线总数', '产品大类', '结存明细', '周期类别', '工厂' ] self.detail_widths = { '序号': 30, '批次': 60, '单板名称': 200, '计划数量': 60, '工艺组合路径': 100, '结存状态': 50, '线体': 70, '在线周期': 60, '前工序结存时间': 70, '本工序结存时间': 70, '项目': 50, '部件比例': 60, '队列状态': 110, '工艺属性': 100, '单据状态': 110, '目的仓库': 110, 'SMT线体及状态': 110, 'SMT备料': 50, 'DIP线体及状态': 110, '计划跟踪单': 130, '单板代码': 120, '释放日期': 80, '计调员': 50,'科室': 50, '大组长': 50, '小组长': 100, '单板交期': 80, '提出人': 55, '备料区': 40, '提单': 40, '入库': 40, '部件入库': 40, '部件在线': 40, '单板测试': 40, '在线总数': 40, '产品大类': 40, '结存明细': 300, '周期类别': 40, '工厂': 40 } # 初始化 server_boards self.server_boards = [ "ZXRSS SW6905P5A", "ZXRSS MBI65G50A", "ZXRSS MB69I05A", "ZXRSS MCU(MB69H03A)", "ZXRSS MB69H03A", "ZXRSS MB65I06B", "ZXRSS MC6506AB","ZXRSS MB65H03A", "ZXRSS MCU(MB65H03A)", "ZXRSS R593X_MB", "ZXRSS MB53A05A", "ZXRSS MB53H03A", "ZXRSS MB53Z3FA", "ZXRSS MB53Z3FA_S", "ZXUSP SPR30_1", "ZXUSP SPR30", "ZXRSS MCU(MB53Z03A)","ZXRSS MB53Z03A", "ZXRSS MI05102A", "ZXRSS R53G4X", "ZXRSS R593XH_MB", "ZXRSS R5XG5_MB", "ZXRSS MB53I06A", "ZXRSS MI05501A", "ZXRSS MB73I06A", "ZXRSS MB23A06A", "ZXRSS MB23I06A", "ZXRSS MPUFA", "ZXRSS MPUFA_CM", "ZXRSS R85G4X", "ZXRAST MPEMA", "ZXRAST MPUFB", "ZXRSS SPLMA" ] # 创建数据库引擎 self.engine = create_engine( f'mysql+pymysql://{mysql_user}:{mysql_password}@{mysql_host}:{mysql_port}/{mysql_database}?charset=utf8mb4', echo=True ) # 权限检查 self.check_ip_permission() # 其他初始化... self.batch_cycle = BatchCycle() self.init_cache_paths() # 👈 确保 init_cache_paths() 只调用一次 self._load_config() self.df = None self.delivery_test_df = None self.selected_factories = self.selected_factories # 从配置加载 self.setup_status_labels() self.setup_ui() self.setup_menu_buttons() self.setup_detail_view() self.setup_delivery_view() self.daily_output_view = DailyOutputView(self) self.show_detail_view() # 插入点：自动更新初始化 self.auto_update_id = None self.auto_update_var = tk.BooleanVar(value=True) self.last_update_time = None self.cache_expiry_hours = 10000 # 缓存过期时间（小时） self.load_user_config() self.schedule_auto_update() # 添加自动更新开关按钮到菜单栏 self.auto_update_check = ttk.Checkbutton( self.menu_bar, text="启用后台自动更新", variable=self.auto_update_var, command=self.toggle_auto_update ) self.auto_update_check.pack(side=tk.RIGHT, padx=5) # 窗口关闭绑定 self.root.protocol("WM_DELETE_WINDOW", self.on_closing) # ✅ 尝试加载缓存数据 if self.load_cached_data() and self.df is not None and not self.df.empty: self.update_ui_after_data_load() print(f"缓存数据加载成功，准备显示 {len(self.df)} 条记录") else: print("未找到有效缓存数据，请点击'更新数据'按钮获取最新数据") self.first_data_load_silent() self.daily_output_view = DailyOutputView(self) self.current_date = datetime.now().strftime('%Y-%m-%d') # 设置当前日期 def init_cache_paths(self): """初始化缓存路径（兼容打包环境）""" try: # 获取应用数据目录（跨平台） if sys.platform == "win32": appdata = os.getenv('APPDATA') self.cache_dir = os.path.join(appdata, 'PCBDataViewer') else: # Linux/Mac home = os.path.expanduser("~") self.cache_dir = os.path.join(home, '.pcbdataviewer') # 创建缓存目录 os.makedirs(self.cache_dir, exist_ok=True) # 设置缓存文件路径 self.cache_file = os.path.join(self.cache_dir, 'data_cache.pkl') self.config_file = os.path.join(self.cache_dir, 'settings.json') print(f"✅ 缓存目录: {self.cache_dir}") print(f"✅ 缓存文件路径: {self.cache_file}") except Exception as e: print(f"⚠️ 初始化缓存路径失败: {str(e)}") # 回退到当前目录 self.cache_dir = os.path.dirname(os.path.abspath(file)) self.cache_file = os.path.join(self.cache_dir, 'data_cache.pkl') self.config_file = os.path.join(self.cache_dir, 'settings.json') print(f"⚠️ 回退缓存路径: {self.cache_file}") def update_status_label(self): """更新状态栏信息""" if hasattr(self, 'last_update_time') and self.last_update_time: self.status_label.config( text=f"最后更新时间：{self.last_update_time.strftime('%Y-%m-%d %H:%M:%S')}" ) else: self.status_label.config(text="最后更新时间：--") def setup_status_labels(self): """创建状态标签和最后更新时间标签""" # 创建底部状态栏 self.status_bar = ttk.Frame(self.root) self.status_bar.pack(side=tk.BOTTOM, fill=tk.X, padx=5, pady=2) # 创建状态标签 self.status_label = ttk.Label(self.status_bar, text="就绪") self.status_label.pack(side=tk.LEFT) # 创建最后更新时间标签 self.last_update_label = ttk.Label(self.root, text="最后更新时间：--") self.last_update_label.pack(side=tk.BOTTOM, anchor='e', padx=10, pady=2) def load_cached_data(self): """从本地文件加载缓存数据""" try: if os.path.exists(self.cache_file) and os.path.getsize(self.cache_file) > 0: with open(self.cache_file, 'rb') as f: cache_data = pickle.load(f) # 检查缓存是否过期 last_update = cache_data.get('last_update_time') if last_update and datetime.now() - last_update > timedelta(hours=self.cache_expiry_hours): print("缓存数据已过期") return False # 验证数据完整性 raw_data = cache_data.get('raw_data') if raw_data is None or raw_data.empty: print("缓存数据为空") return False self.df = raw_data self.last_update_time = last_update self.selected_factories = cache_data.get('selected_factories', []) print(f"缓存数据加载成功，共{len(self.df)}条记录") return True else: print("缓存文件不存在或为空") return False except (pickle.UnpicklingError, EOFError) as e: print(f"读取缓存文件失败（格式错误或文件不完整）: {str(e)}") return False except Exception as e: print(f"加载缓存数据时出错: {str(e)}") return False def check_cache_status(self): """检查缓存状态""" try: if os.path.exists(self.cache_file): file_time = datetime.fromtimestamp(os.path.getmtime(self.cache_file)).strftime('%Y-%m-%d %H:%M:%S') file_size = os.path.getsize(self.cache_file) status = { 'exists': True, 'last_modified': file_time, 'size': file_size, 'size_kb': round(file_size / 1024, 2), 'is_valid': False } try: with open(self.cache_file, 'rb') as f: cache_data = pickle.load(f) if cache_data.get('raw_data') and not cache_data['raw_data'].empty: status['is_valid'] = True status['record_count'] = len(cache_data['raw_data']) except (pickle.UnpicklingError, EOFError, Exception): status['is_valid'] = False return status else: return {'exists': False} except Exception as e: print(f"⚠️ 检查缓存状态时出错: {str(e)}") return {'exists': False, 'error': str(e)} def _process_output_data(self, data, progress_frame, progress): try: if not data: messagebox.showwarning("警告", "未获取到日产出目标数据") self._cleanup_ui(progress_frame, progress) return # ✅ 更新最后更新时间 self.last_update_time = datetime.now() # 转换为DataFrame并处理数据 df = pd.DataFrame(data) if '产品大类' not in df.columns: df['产品大类'] = '' # ... 其他代码 ... # 保存数据到缓存 self.save_cached_data() self.raw_data = df self.apply_filters() # 清理UI self._cleanup_ui(progress_frame, progress) messagebox.showinfo("成功", f"成功获取 {len(df)} 条日产出目标数据") except Exception as e: self._show_update_error(str(e), progress_frame, progress) def _save_config(self): """保存配置到文件（兼容打包环境）""" try: config = { 'selected_factories': self.selected_factories, 'last_update_time': self.last_update_time } with open(self.config_file, 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) return True except Exception as e: print(f"保存配置文件失败: {str(e)}") return False def _load_config(self): """加载持久化配置（兼容打包环境）""" try: if os.path.exists(self.config_file): with open(self.config_file, 'r', encoding='utf-8') as f: config = json.load(f) self.selected_factories = config.get('selected_factories', []) self.last_update_time = config.get('last_update_time') else: self.selected_factories = [] self._save_config() return True except Exception as e: print(f"加载配置文件失败: {str(e)}") self.selected_factories = [] return False def schedule_detail_auto_update(self): """使用独立线程定时更新单板明细视图""" if hasattr(self, 'detail_update_timer'): self.detail_update_timer.cancel() # 启动新定时器 self.detail_update_timer = threading.Timer(30 * 60, self.auto_update_detail_data) self.detail_update_timer.daemon = True self.detail_update_timer.start() def save_last_update_time(self, view_name): """按视图保存最后更新时间""" self.last_update_times[view_name] = datetime.now() def get_last_update_time(self, view_name): """获取指定视图的最后更新时间""" return self.last_update_times.get(view_name, None) def auto_update_detail_data(self): """执行单板明细视图的自动更新数据（带日志）""" try: logger.info("开始单板明细视图自动更新") self.safe_status_update("正在后台自动更新单板明细数据...") self.update_data(auto=True) logger.info("单板明细视图自动更新成功") except Exception as e: logger.error(f"单板明细视图自动更新失败: {str(e)}") self.safe_status_update(f"单板明细自动更新失败: {str(e)}") finally: self.schedule_detail_auto_update() def setup_ui(self): """初始化用户界面 - 只创建一次主窗口和框架""" # 顶部菜单栏 self.menu_bar = ttk.Frame(self.root) self.menu_bar.pack(side=tk.TOP, fill=tk.X, padx=5, pady=2) # 主内容区域 self.content_frame = ttk.Frame(self.root) self.content_frame.pack(fill=tk.BOTH, expand=True) # 底部状态栏 self.status_bar = ttk.Frame(self.root) self.status_bar.pack(side=tk.BOTTOM, fill=tk.X) self.status_label = ttk.Label(self.status_bar, text="就绪") self.status_label.pack(side=tk.LEFT) # 确保所有视图容器都正确初始化 self.detail_frame = ttk.Frame(self.content_frame) self.delivery_frame = ttk.Frame(self.content_frame) self.chart_frame = ttk.Frame(self.content_frame) self.material_order_frame = ttk.Frame(self.content_frame) self.delivery_test_frame = ttk.Frame(self.content_frame) self.batch_cycle_frame = ttk.Frame(self.content_frame) # 最后更新时间标签 self.last_update_label = ttk.Label(self.root, text="最后更新时间：--") self.last_update_label.pack(side=tk.BOTTOM, anchor='e', padx=10, pady=2) def initialize_data(self): """初始化数据（确保在UI初始化后调用）""" try: # 尝试加载缓存数据 - 使用新的统一路径 if self.load_cached_data(): self.update_ui_after_data_load() print(f"缓存数据加载成功，准备显示 {len(self.df)} 条记录") else: print("未找到有效缓存数据，请点击'更新数据'按钮获取最新数据") self.first_data_load_silent() except Exception as e: print(f"⚠️ 初始化数据时出错: {str(e)}") self.status_label.config(text="数据初始化失败，请点击'更新数据'按钮") traceback.print_exc() return False def first_data_load_silent(self): """静默首次数据加载""" try: self.update_data_internal() if self.df is not None and not self.df.empty: self.last_update_time = datetime.now() self.save_cached_data() return True else: print("获取数据失败或数据为空") return False except Exception as e: print(f"首次加载失败: {str(e)}") traceback.print_exc() return False def first_data_load(self): """首次数据加载（改进版）""" try: self.first_run_btn.pack_forget() self.safe_status_update("正在获取初始数据...") self.update_data() except Exception as e: self.safe_status_update(f"首次加载失败: {str(e)}") # 重新显示按钮以便重试 if hasattr(self, 'first_run_btn') and self.first_run_btn.winfo_exists(): self.first_run_btn.pack(pady=20) def schedule_auto_update(self): """安排自动更新任务（使用 tkinter 的 after 方法）""" try: if hasattr(self, 'auto_update_id') and self.auto_update_id: self.root.after_cancel(self.auto_update_id) print("取消现有自动更新任务") if self.auto_update_var.get(): next_update = 30 * 60 * 1000 # 30分钟 next_time_str = (datetime.datetime.now() + datetime.timedelta(minutes=30)).strftime('%H:%M') self.auto_update_id = self.root.after(next_update, self.auto_update_data) self.safe_status_update(f"已开启自动更新（下次更新: {next_time_str}）") print(f"自动更新已安排，下次更新时间: {next_time_str}") except Exception as e: logging.error(f"设置自动更新失败: {str(e)}") self.safe_status_update("设置自动更新失败") traceback.print_exc() def auto_update_data(self): """执行自动更新数据（后台线程中执行）""" try: print(f"\n=== 开始自动更新数据 [{datetime.datetime.now():%Y-%m-%d %H:%M:%S}] ===") self.safe_status_update("正在后台自动更新数据...") # 在后台线程中执行数据更新 threading.Thread(target=self._background_update, daemon=True).start() except Exception as e: print(f"自动更新失败: {str(e)}") traceback.print_exc() self.safe_status_update(f"自动更新失败: {str(e)}") self.schedule_auto_update() def _background_update(self): """后台线程中执行实际的数据更新逻辑""" try: # 调用你的 update_data 方法（你已有的方法） self.update_data(auto=True) # 更新缓存时间 self.last_update_time = datetime.datetime.now() self.save_cached_data() # 保存缓存 # 更新UI self.root.after(0, self._update_ui_after_background_update, True) except Exception as e: print(f"后台更新失败: {str(e)}") self.root.after(0, lambda: messagebox.showerror("错误", f"自动更新失败: {str(e)}")) finally: # 重新安排下一次更新 self.root.after(0, self.schedule_auto_update) def auto_update_data(self): """执行自动更新（增强版）""" try: print(f"\n=== 开始自动更新数据 [{datetime.now():%Y-%m-%d %H:%M:%S}] ===") self.safe_status_update("正在自动更新数据...") # 执行数据更新 self.update_data(auto=True) # 记录成功日志 print(f"数据更新成功完成 [{datetime.now():%Y-%m-%d %H:%M:%S}]") self.safe_status_update(f"自动更新完成 {datetime.now():%H:%M}") return True except Exception as e: print(f"自动更新失败: {str(e)}") traceback.print_exc() self.safe_status_update(f"自动更新失败: {str(e)}") # 尝试重新调度，但增加延迟避免快速失败循环 self.root.after(60 * 1000, self.schedule_auto_update) return False finally: # 确保无论成功失败都重新调度 print("重新调度下一次自动更新...") self.schedule_auto_update() def safe_status_update(self, message): """安全地更新状态栏信息""" try: if hasattr(self, 'status_label') and self.status_label.winfo_exists(): self.status_label.config(text=message) except Exception as e: print(f"更新状态栏时出错: {str(e)}") def cleanup(self): """清理资源""" if hasattr(self, 'auto_update_id') and self.auto_update_id is not None: try: self.root.after_cancel(self.auto_update_id) except ValueError: # 忽略无效的计时器ID错误 pass def check_ip_permission(self): """IP或工号权限校验，满足其一即可""" try: # 获取本地IP地址 local_ip = socket.gethostbyname(socket.gethostname()) with self.engine.connect() as connection: # 1. 先查IP权限 query = "SELECT * FROM permission WHERE IP = :ip" result_ip = connection.execute(text(query), {'ip': local_ip}).fetchone() if result_ip: self.has_maintenance_permission = bool(int(result_ip.权限)) print(f"权限检查结果 - IP: {local_ip}, 权限值: {result_ip.权限}, 维护权限: {self.has_maintenance_permission}") return # 2. 如果IP无权限，查工号 from Login2 import parse_cookies cookies = parse_cookies() emp_no = cookies.get("ZTEDPGSSOUser") if cookies else None if not emp_no: messagebox.showerror("权限错误", "没有权限使用（未获取到工号），请找计调科开通权限！") self.root.destroy() sys.exit() query = "SELECT * FROM permission WHERE id = :emp_no" result_id = connection.execute(text(query), {'emp_no': emp_no}).fetchone() if result_id: self.has_maintenance_permission = bool(int(result_id.权限)) print(f"权限检查结果 - 工号: {emp_no}, 权限值: {result_id.权限}, 维护权限: {self.has_maintenance_permission}") return # 3. 两者都无权限 messagebox.showerror("权限错误", "没有权限使用，请找计调科开通权限！") self.root.destroy() sys.exit() except Exception as e: messagebox.showerror("错误", f"检查权限时出错：{str(e)}") self.root.destroy() sys.exit() def setup_menu_buttons(self): """设置顶部菜单按钮""" # 系统操作按钮组 sys_frame = ttk.Frame(self.menu_bar) sys_frame.pack(side=tk.LEFT, padx=5) ttk.Button(sys_frame, text="选择工厂", command=self.select_factories).pack(side=tk.LEFT, padx=2) ttk.Button(sys_frame, text="更新数据", command=self.update_data).pack(side=tk.LEFT, padx=2) # 分隔符 ttk.Separator(self.menu_bar, orient=tk.VERTICAL).pack(side=tk.LEFT, fill=tk.Y, padx=5) # 视图切换按钮组 view_frame = ttk.Frame(self.menu_bar) view_frame.pack(side=tk.LEFT, padx=5) ttk.Button(view_frame, text="单板明细", command=lambda: self.switch_view("单板明细")).pack(side=tk.LEFT, padx=2) ttk.Button(view_frame, text="单板交期", command=lambda: self.switch_view("单板交期")).pack(side=tk.LEFT, padx=2) ttk.Button(view_frame, text="图表查询", command=lambda: self.switch_view("图表查询")).pack(side=tk.LEFT, padx=2) ttk.Button(view_frame, text="料单提单查询", command=lambda: self.switch_view("料单提单查询")).pack(side=tk.LEFT, padx=2) ttk.Button(view_frame, text="产品信息配置", command=lambda: self.switch_view("产品信息配置")).pack(side=tk.LEFT, padx=2) ttk.Button(view_frame, text="批次周期", command=lambda: self.switch_view("批次周期")).pack(side=tk.LEFT, padx=2) ttk.Button(view_frame, text="产出计划表", command=lambda: self.switch_view("产出计划表")).pack(side=tk.LEFT, padx=2) def update_data_internal(self): """内部数据更新方法（不显示对话框）""" try: # 清理旧数据 self.clear_data() # 获取新数据 processor = DataProcessor() if self.selected_factories: all_data = [] for factory_id in self.selected_factories: df_factory = DataProcessor.process_data(factory_id) if df_factory is not None and not df_factory.empty: all_data.append(df_factory) if all_data: self.df = pd.concat(all_data, ignore_index=True) else: self.df = pd.DataFrame() else: # 如果没有选择工厂，默认获取所有工厂数据 self.df = DataProcessor.process_data('53') # 默认深圳工厂 if self.df is None: self.df = pd.DataFrame() # 同时更新产品信息配置数据 try: self.delivery_test_df = DataProcessor.fetch_url6_data() except Exception as e: print(f"更新产品信息配置数据时出错: {str(e)}") self.delivery_test_df = pd.DataFrame() if self.df is not None and not self.df.empty: self.last_update_time = datetime.now() self.save_cached_data() # 保存缓存 return True else: print("获取数据失败或数据为空") return False except Exception as e: print(f"内部数据更新失败: {str(e)}") return False def setup_application(self): """初始化应用程序""" self.setup_gui() self.setup_column_selector() self.setup_context_menu() self.setup_detail_view() self.setup_delivery_view() # self.setup_chart_view() # 尝试加载缓存数据 if self.load_cached_data(): # 更新UI显示 self.update_ui_after_data_load() else: self.status_label.config(text="未找到有效缓存数据，请点击'更新数据'按钮获取最新数据") # 设置自动更新 self.schedule_auto_update() # 默认显示单板明细视图 self.show_detail_view() def update_ui_after_data_load(self): """加载数据后更新UI（增强版）""" try: if self.df is not None: # 确保UI组件已初始化 if hasattr(self, 'search_widgets') and self.search_widgets: self.update_combobox_values() if hasattr(self, 'column_selector') and self.column_selector.winfo_exists(): self.update_column_selector_options() # 更新最后更新时间标签 self.update_last_update_label() # 更新状态栏 self.update_status_label() # 显示数据（如果当前是明细视图） if hasattr(self, 'current_view') and self.current_view == "单板明细": self.root.after(100, self.show_all_detail_data) except Exception as e: print(f"更新UI时出错: {str(e)}") traceback.print_exc() self.safe_status_update("更新UI时出错") def update_last_update_label(self): """更新最后更新时间标签""" try: if not hasattr(self, 'last_update_label') or not self.last_update_label.winfo_exists(): print("最后更新时间标签未初始化或已销毁") return if hasattr(self, 'last_update_time') and self.last_update_time: self.last_update_label.config( text=f"最后更新时间：{self.last_update_time.strftime('%Y-%m-%d %H:%M:%S')}" ) else: self.last_update_label.config(text="最后更新时间：--") except Exception as e: print(f"更新最后更新时间标签时出错: {str(e)}") traceback.print_exc() def schedule_auto_update(self): """安排自动更新任务""" if hasattr(self, 'auto_update_var') and self.auto_update_var.get(): # 取消之前的计划任务（如果有） if hasattr(self, 'auto_update_id') and self.auto_update_id: self.root.after_cancel(self.auto_update_id) # 安排新的自动更新任务（30分钟后） self.auto_update_id = self.root.after( 30 * 60 * 1000, # 30分钟 self.auto_update_data ) def auto_update_data(self): """执行自动更新数据""" try: # 在状态栏显示更新状态 if hasattr(self, 'status_label') and self.status_label.winfo_exists(): self.status_label.config(text="正在后台自动更新数据...") # 在独立线程中执行更新 threading.Thread( target=self._perform_background_update, daemon=True ).start() except Exception as e: print(f"自动更新失败: {str(e)}") if hasattr(self, 'status_label') and self.status_label.winfo_exists(): self.status_label.config(text="自动更新失败") # 重新安排下一次更新 self.schedule_auto_update() def _perform_background_update(self): """在后台线程中执行数据更新""" try: # 执行实际的数据更新逻辑 self.update_data(auto=True) # 更新完成后保存数据 self.save_cached_data() # 更新UI显示 if hasattr(self, 'root') and self.root.winfo_exists(): self.root.after(0, self.update_ui_after_data_load) self.root.after(0, lambda: self.safe_status_update( f"数据已自动更新（{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}）" )) except Exception as e: print(f"后台更新出错: {str(e)}") traceback.print_exc() if hasattr(self, 'root') and self.root.winfo_exists(): self.root.after(0, lambda: self.safe_status_update("后台更新出错")) finally: # 无论成功失败，都重新安排下一次更新 if hasattr(self, 'root') and self.root.winfo_exists(): self.root.after(0, self.schedule_auto_update) def on_closing(self): """处理窗口关闭事件""" # 取消所有计划任务 if hasattr(self, 'auto_update_id') and self.auto_update_id: self.root.after_cancel(self.auto_update_id) # 保存当前配置 if hasattr(self, 'auto_update_var'): self.save_user_config() # 关闭窗口 self.root.destroy() def setup_gui(self): """初始化GUI组件""" # 添加关闭窗口事件处理 self.root.protocol("WM_DELETE_WINDOW", self.on_closing) # 添加自动更新选项 self.auto_update_var = tk.BooleanVar(value=True) # 默认启用自动更新 auto_update_check = ttk.Checkbutton( self.menu_bar, text="启用后台自动更新", variable=self.auto_update_var, command=self.toggle_auto_update ) auto_update_check.pack(side=tk.RIGHT, padx=5) # 加载用户配置 self.load_user_config() def toggle_auto_update(self): """切换自动更新设置""" if self.auto_update_var.get(): self.schedule_auto_update() self.status_label.config(text="已启用后台自动更新") else: if hasattr(self, 'auto_update_id') and self.auto_update_id: self.root.after_cancel(self.auto_update_id) self.status_label.config(text="已禁用后台自动更新") # 保存用户配置 self.save_user_config() def save_user_config(self): """保存用户配置（自动更新状态）""" try: config_file = os.path.join(current_dir, 'user_config.json') config = { 'auto_update': self.auto_update_var.get() } with open(config_file, 'w') as f: json.dump(config, f) except Exception as e: logging.error(f"保存用户配置失败: {str(e)}") def load_user_config(self): """加载用户配置""" try: config_file = os.path.join(current_dir, 'user_config.json') if os.path.exists(config_file): with open(config_file, 'r') as f: config = json.load(f) self.auto_update_var.set(config.get('auto_update', True)) except Exception as e: logging.error(f"加载用户配置失败: {str(e)}") self.auto_update_var.set(True) def setup_column_selector(self): """设置列选择器""" # 如果已经创建了列选择器，直接返回 if hasattr(self, 'column_selector'): return self.column_selector_frame = ttk.Frame(self.root, height=30) self.column_selector_frame.pack(side=tk.TOP, fill=tk.X, padx=5, pady=5) ttk.Label(self.column_selector_frame, text="选择展示列:").pack(side=tk.LEFT) # 创建多选框或下拉菜单 self.column_selector = ttk.Combobox(self.column_selector_frame, state="readonly") self.column_selector.pack(side=tk.LEFT, padx=5) self.column_selector.bind("<<ComboboxSelected>>", self.update_columns) # 初始化列选择器的选项，默认全部选中 self.update_column_selector_options() # 默认选中所有列 self.selected_columns = set(self.detail_widths.keys()) self.update_tree_columns() def update_column_selector_options(self): """更新列选择器的选项""" try: if not hasattr(self, 'column_selector'): print("列选择器未初始化") return if not hasattr(self, 'detail_widths') or not self.detail_widths: print("列宽度未定义") return if not self.column_selector.winfo_exists(): print("列选择器已被销毁") return columns = list(self.detail_widths.keys()) self.column_selector['values'] = columns if columns: self.column_selector.current(0) # 默认选中第一个列 else: self.column_selector.set('') except Exception as e: print(f"更新列选择器选项时出错: {str(e)}") traceback.print_exc() def update_columns(self, event=None): """根据用户选择更新表格列""" selected_column = self.column_selector.get() if selected_column: if selected_column in self.selected_columns: self.selected_columns.remove(selected_column) # 取消选中 else: self.selected_columns.add(selected_column) # 添加选中 self.update_tree_columns() # 更新表格显示的列 def update_tree_columns(self): """更新表格中展示的列""" if self.detail_tree is not None: # 按照 column_order 的顺序更新显示列 display_columns = [col for col in self.column_order if col in self.selected_columns] self.detail_tree["displaycolumns"] = display_columns def update_data(self, auto=False): """在后台线程中更新数据，不阻塞UI""" # 根据模式设置不同的状态消息 if auto: self.safe_status_update("正在自动更新数据...") print(f"\n=== 自动更新开始 === [{datetime.now():%Y-%m-%d %H:%M:%S}]") else: self.safe_status_update("正在手动更新数据...") # 只在手动更新时显示进度条 if not auto: progress_frame = ttk.Frame(self.status_bar) progress_frame.pack(side=tk.RIGHT, padx=10) progress = ttk.Progressbar(progress_frame, mode='indeterminate', length=100) progress.pack(side=tk.RIGHT) progress.start() # 更新按钮状态指示 update_buttons = [] for widget in self.menu_bar.winfo_children(): if isinstance(widget, ttk.Frame): for btn in widget.winfo_children(): if isinstance(btn, ttk.Button) and btn['text'] == "更新数据": btn['text'] = "更新中..." btn['state'] = 'disabled' update_buttons.append(btn) # 定义后台更新函数 def background_update(): try: # 清理旧数据 self.clear_data() processor = DataProcessor() # 并行处理多个工厂的数据获取 if self.selected_factories: # 使用并行处理获取多个工厂数据 with concurrent.futures.ThreadPoolExecutor(max_workers=len(self.selected_factories)) as executor: # 创建任务 future_to_factory = { executor.submit(DataProcessor.process_data, factory_id): factory_id for factory_id in self.selected_factories } # 收集结果 all_dfs = [] for future in concurrent.futures.as_completed(future_to_factory): factory_id = future_to_factory[future] try: df_factory = future.result() if df_factory is not None and not df_factory.empty: all_dfs.append(df_factory) # 使用after方法安全更新状态 self.root.after(0, lambda f=factory_id: self.safe_status_update(f"已获取工厂 {f} 数据")) else: print(f"工厂 {factory_id} 返回空数据") except Exception as e: print(f"获取工厂 {factory_id} 数据时出错: {str(e)}") traceback.print_exc() # 合并所有数据 if all_dfs: self.df = pd.concat(all_dfs, ignore_index=True) else: self.df = pd.DataFrame() else: # 调用无参数的process_data() self.df = DataProcessor.process_data() # 同时更新产品信息配置数据 if hasattr(self, '_delivery_test_initialized'): try: self.delivery_test_df = DataProcessor.fetch_url6_data() except Exception as e: print(f"更新产品信息配置数据时出错: {str(e)}") traceback.print_exc() # 更新成功后的操作 if self.df is not None and not self.df.empty: self.last_update_time = datetime.now() # 保存数据到缓存 self.save_cached_data() # 使用after方法在主线程中安全更新UI self.root.after(0, self._update_ui_after_background_update, auto) else: error_msg = "未获取到有效数据，请检查网络连接和登录状态" # 使用after方法在主线程中安全显示错误 self.root.after(0, lambda: self._show_update_error(error_msg, auto)) except Exception as e: error_details = f"更新数据时发生错误：{str(e)}" print(error_details) traceback_info = traceback.format_exc() print(traceback_info) # 提取更有用的错误信息 error_lines = traceback_info.split('\n') relevant_error = next((line for line in reversed(error_lines) if line.strip() and not line.startswith(' ')), str(e)) # 使用after方法在主线程中安全显示错误 self.root.after(0, lambda: self._show_update_error(relevant_error, auto)) finally: # 使用after方法在主线程中安全清理UI self.root.after(0, lambda: self._cleanup_update_ui(progress_frame, progress, update_buttons)) # 启动后台线程 threading.Thread(target=background_update, daemon=True).start() def save_cached_data(self): """将当前数据保存到缓存文件""" try: if self.df is None or self.df.empty: print("数据为空，跳过缓存保存") return False print(f"准备保存缓存到: {self.cache_file}") with open(self.cache_file, 'wb') as f: pickle.dump({ 'raw_data': self.df, 'last_update_time': self.last_update_time, 'selected_factories': self.selected_factories }, f) print("缓存保存成功") return True except Exception as e: print(f"缓存数据时出错: {str(e)}") return False def _update_ui_after_background_update(self, auto): """后台更新完成后更新UI""" try: # 更新最后更新时间标签 if hasattr(self, 'last_update_label'): self.last_update_label.config( text=f"最后更新时间：{self.last_update_time.strftime('%Y-%m-%d %H:%M:%S')}" ) # 更新搜索框选项 self.update_combobox_values() # 刷新当前视图（如果是单板明细） if self.current_view == "单板明细": self._display_detail_data(self.df) elif self.current_view == "产品信息配置" and hasattr(self, 'delivery_test_df'): if self.delivery_test_df is not None and not self.delivery_test_df.empty: self._display_delivery_test_data(self.delivery_test_df) # 显示成功状态 self.safe_status_update("数据更新成功") except Exception as e: print(f"更新UI时出错: {str(e)}") traceback.print_exc() def _show_update_error(self, error_msg, auto): """显示更新错误""" if not auto: messagebox.showerror("错误", f"数据更新失败！\n\n详细错误：\n{error_msg}\n\n完整错误已记录到控制台") self.safe_status_update(f"数据更新失败: {error_msg}") def _cleanup_update_ui(self, progress_frame, progress, update_buttons): """清理更新UI元素""" # 停止进度条并移除 if progress.winfo_exists(): progress.stop() if progress_frame.winfo_exists(): progress_frame.destroy() # 恢复按钮状态 for btn in update_buttons: if btn.winfo_exists(): btn['text'] = "更新数据" btn['state'] = 'normal' # 视图相关方法应该集中放置 def switch_view(self, view_name): """切换视图并更新状态栏""" # 标准化视图名称 view_name = view_name.replace("detail", "单板明细").replace("delivery", "单板交期").replace("chart", "图表查询") # 隐藏所有视图 for frame_name in ['detail_frame', 'delivery_frame', 'chart_frame', 'material_order_frame', 'delivery_test_frame', 'batch_cycle_frame']: if hasattr(self, frame_name): frame = getattr(self, frame_name) if frame and hasattr(frame, 'pack_forget'): frame.pack_forget() # 隐藏产出计划表视图 if hasattr(self, 'daily_output_view') and self.daily_output_view: if hasattr(self.daily_output_view, 'frame') and hasattr(self.daily_output_view.frame, 'pack_forget'): self.daily_output_view.frame.pack_forget() # 显示选中的视图并确保初始化 if view_name == "单板明细": if not hasattr(self, 'detail_tree'): self.setup_detail_view() self.detail_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "单板明细" elif view_name == "单板交期": if not hasattr(self, 'delivery_tree'): self.setup_delivery_view() self.delivery_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "单板交期" self.update_delivery_data() elif view_name == "图表查询": if not hasattr(self, 'chart_tree'): self.setup_chart_view() self.chart_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "图表查询" elif view_name == "料单提单查询": if not hasattr(self, 'material_order_tree'): self._setup_material_order_view() self._material_order_initialized = True self.material_order_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "料单提单查询" elif view_name == "产品信息配置": if not hasattr(self, 'delivery_test_tree'): self._setup_delivery_test_view() self._delivery_test_initialized = True self.delivery_test_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "产品信息配置" if not hasattr(self, 'delivery_test_df') or self.delivery_test_df.empty: self.update_delivery_test_data() elif view_name == "批次周期": if not hasattr(self, 'batch_cycle_tree'): self._setup_batch_cycle_view() self._batch_cycle_initialized = True self.batch_cycle_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "批次周期" self.update_batch_cycle_data() elif view_name == "产出计划表": if hasattr(self, 'daily_output_view'): self.daily_output_view.frame.pack(fill=tk.BOTH, expand=True) self.current_view = "产出计划表" # 检查是否有数据，如果没有则尝试加载缓存或更新数据 if not hasattr(self.daily_output_view, 'raw_data') or self.daily_output_view.raw_data.empty: # 尝试加载缓存数据，如果失败则更新数据 if not self.daily_output_view.load_cached_data(): # 显示加载对话框 loading_dialog = SimpleLoadingDialog(self.root) self.root.update() # 在后台线程中加载数据 def load_data_thread(): try: self.daily_output_view.update_data() finally: # 确保对话框被销毁 self.root.after(0, loading_dialog.destroy) threading.Thread(target=load_data_thread, daemon=True).start() elif view_name == "单板明细": if not hasattr(self, 'detail_tree'): self.setup_detail_view() self.detail_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "单板明细" self.update_last_update_label() self.update_combobox_values() self.show_all_detail_data() # 统一更新状态栏 self.safe_status_update(f"当前视图: {view_name}") def run(self): """运行应用程序主循环""" try: # 确保所有UI初始化完成 if not hasattr(self, 'root') or not self.root.winfo_exists(): raise AttributeError("root窗口未正确初始化") # 显示主窗口 self.root.deiconify() # 启动主循环 self.root.mainloop() except Exception as e: print(f"运行主循环时出错: {str(e)}") traceback.print_exc() raise finally: # 确保清理资源 self.cleanup() def check_update_time(self): """检查数据更新时间""" if self.last_update_time is None: return True now = datetime.now() time_diff = now - self.last_update_time # 如果超过30分钟 return time_diff.total_seconds() > 1800 # 30分钟 = 1800秒 def clear_data(self): """清理数据""" if self.df is not None: del self.df self.df = None class CollapsiblePane(ttk.Frame): """可折叠面板""" def init(self, parent, title="", *args, **kwargs): ttk.Frame.init(self, parent, *args, **kwargs) self.title = title self.show = tk.BooleanVar(value=True) self._setup_ui() def _setup_ui(self): """初始化UI""" # 标题栏 self.title_frame = ttk.Frame(self) self.title_frame.pack(fill=tk.X, expand=1) # 标题样式 self.title_label = ttk.Label( self.title_frame, text=self.title, font=('Arial', 10, 'bold') ) self.title_label.pack(side=tk.LEFT, padx=5) # 折叠按钮 self.toggle_btn = ttk.Checkbutton( self.title_frame, width=2, variable=self.show, command=self._toggle, style='Toolbutton' ) self.toggle_btn.pack(side=tk.RIGHT, padx=5) # 内容区域 self.content = ttk.Frame( self, relief="groove", borderwidth=1, padding=(5,5,5,5) ) def _toggle(self): """切换显示/隐藏""" if self.show.get(): self.content.pack(fill=tk.X, expand=1, pady=(5,0)) else: self.content.pack_forget() def content_frame(self): """获取内容区域frame""" return self.content def setup_detail_view(self): """设置单板明细视图""" print(f"Debug: search_vars exists - {hasattr(self, 'search_vars')}") # 调试输出 print(f"Debug: search_widgets exists - {hasattr(self, 'search_widgets')}") # 调试输出 main_frame = ttk.Frame(self.detail_frame) main_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 创建可折叠搜索面板 self.search_pane = self.CollapsiblePane(main_frame, title="高级搜索，右侧浅蓝色点击可收缩") self.search_pane.pack(fill=tk.X, pady=(0,5)) # 自动识别搜索框 auto_frame = ttk.Frame(self.search_pane.content_frame()) auto_frame.pack(fill=tk.X, padx=5, pady=10) ttk.Label(auto_frame, text="自动识别:").pack(side=tk.LEFT, padx=(0,5)) self.auto_search_text = tk.Text(auto_frame, height=3, width=30, font=('Arial', 10)) self.auto_search_text.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0,5)) self.auto_search_text.bind('<Return>', self.auto_identify_paste_data) # 定义搜索字段（使用self.column_order之前必须确保它已定义） search_fields = [ ('在线周期', 'range', 1, 0), ('部件结存比例', 'range', 1, 1), ('全部结存比例', 'range', 1, 2), ('计调员', 'combobox', 2, 0), ('小组长', 'combobox', 2, 1), ('大组长', 'combobox', 2, 2), ('释放日期', 'date', 3, 0), ('工艺组合路径', 'entry', 3, 1), ('结存状态', 'combobox', 3, 2), ('线体', 'combobox', 4, 0), ('SMT线体及状态', 'entry', 4, 1), ('DIP线体及状态', 'entry', 4, 2), ('是否急件', 'entry', 5, 0), ('产品大类', 'combobox', 5, 1), ('结存明细', 'entry', 5, 2) ] self.search_fields = search_fields # 创建主容器用于搜索字段 search_container = ttk.Frame(self.search_pane.content_frame()) search_container.pack(fill=tk.X, padx=5, pady=5) # 创建搜索字段 - 统一使用pack布局 self.search_vars = {} self.search_widgets = {} entry_width = 15 range_width = entry_width // 2 - 1 # 使用Frame作为行容器 current_row = None row_frames = [] for i, (field, field_type, _, _) in enumerate(search_fields): # 每5个字段换一行 if i % 5 == 0: current_row = ttk.Frame(search_container) current_row.pack(fill=tk.X, pady=2) row_frames.append(current_row) # 创建字段容器 field_frame = ttk.Frame(current_row) field_frame.pack(side=tk.LEFT, padx=5, pady=2, fill=tk.X, expand=True) # 创建标签 ttk.Label(field_frame, text=f"{field}:").pack(side=tk.LEFT, padx=(0, 5)) # 根据字段类型创建控件 if field_type == 'combobox': self.search_vars[field] = tk.StringVar() widget = ttk.Combobox(field_frame, textvariable=self.search_vars[field], width=entry_width) # 设置下拉列表的值 if field == '计调员': values = self.get_unique_values('计调员') widget['values'] = values if values else ['暂无数据'] elif field == '小组长': values = self.get_unique_values('小组长') widget['values'] = values if values else ['暂无数据'] elif field == '大组长': values = self.get_unique_values('大组长') widget['values'] = values if values else ['暂无数据'] elif field == '产品大类': values = self.get_unique_values('产品大类') widget['values'] = values if values else ['暂无数据'] elif field == '结存状态': values = self.get_unique_values('结存状态') widget['values'] = values if values else ['暂无数据'] elif field == '线体': values = self.get_unique_values('线体') widget['values'] = values if values else ['暂无数据'] widget.configure(state='normal') widget.bind('<Return>', lambda e, f=field: self.handle_combobox_input(e, f)) widget.bind('<<ComboboxSelected>>', lambda e, f=field: self.handle_combobox_select(e, f)) widget.pack(side=tk.LEFT, fill=tk.X, expand=True) self.search_widgets[field] = widget elif field_type == 'date': self.search_vars[field] = tk.StringVar() widget = ttk.Entry(field_frame, textvariable=self.search_vars[field], width=entry_width) # 设置默认值为当前日期 today_str = datetime.now().strftime('%Y-%m-%d') widget.insert(0, today_str) widget.bind('<Button-1>', lambda e, w=widget: self.show_date_picker(w)) widget.pack(side=tk.LEFT, fill=tk.X, expand=True) self.search_widgets[field] = widget elif field_type == 'range': self.search_vars[field] = (tk.StringVar(), tk.StringVar()) range_frame = ttk.Frame(field_frame) range_frame.pack(side=tk.LEFT, fill=tk.X, expand=True) widget_min = ttk.Entry(range_frame, textvariable=self.search_vars[field][0], width=range_width) widget_min.pack(side=tk.LEFT, padx=(2, 5)) ttk.Label(range_frame, text="-").pack(side=tk.LEFT) widget_max = ttk.Entry(range_frame, textvariable=self.search_vars[field][1], width=range_width) widget_max.pack(side=tk.LEFT, padx=(2, 5)) self.search_widgets[field] = (widget_min, widget_max) else: # 普通输入框 self.search_vars[field] = tk.StringVar() widget = ttk.Entry(field_frame, textvariable=self.search_vars[field], width=entry_width) widget.pack(side=tk.LEFT, fill=tk.X, expand=True) self.search_widgets[field] = widget # 按钮区域 - 使用pack btn_frame = ttk.Frame(self.search_pane.content_frame()) btn_frame.pack(fill=tk.X, pady=5) ttk.Button(btn_frame, text="搜索", command=self.search_detail).pack(side=tk.LEFT, padx=5) ttk.Button(btn_frame, text="重置", command=self.reset_detail_search).pack(side=tk.LEFT, padx=5) ttk.Button(btn_frame, text="导出Excel", command=self.export_to_excel).pack(side=tk.LEFT, padx=5) # 表格框架 tree_frame = ttk.Frame(main_frame) tree_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 创建表格和滚动条 self.detail_tree = ttk.Treeview(tree_frame, columns=self.column_order, show='headings') # 设置列宽和标题 for col, width in self.detail_widths.items(): self.detail_tree.heading(col, text=col, anchor='w', command=lambda c=col: self.treeview_sort_column(c)) self.detail_tree.column(col, width=width, anchor='w', stretch=False) # 添加滚动条 vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.detail_tree.yview) hsb = ttk.Scrollbar(tree_frame, orient="horizontal", command=self.detail_tree.xview) self.detail_tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) # 使用grid布局（在tree_frame内部使用grid是可以的） self.detail_tree.grid(row=0, column=0, sticky="nsew") vsb.grid(row=0, column=1, sticky="ns") hsb.grid(row=1, column=0, sticky="ew") # 配置网格权重 tree_frame.grid_columnconfigure(0, weight=1) tree_frame.grid_rowconfigure(0, weight=1) # 默认选中所有列 self.selected_columns = set(self.column_order) self.update_tree_columns() # 添加Ctrl+C快捷键复制选中行 self.detail_tree.bind('<Control-c>', self.copy_selected_rows) self.setup_context_menu() # 设置上下文菜单 # 服务器主板筛选框架 - 使用pack server_frame = ttk.Frame(self.search_pane.content_frame()) server_frame.pack(fill=tk.X, padx=5, pady=5) self.server_filter_var = tk.BooleanVar(value=False) ttk.Checkbutton(server_frame, text="服务器主板筛选(全选/取消)", variable=self.server_filter_var).pack(side=tk.LEFT, padx=5) # 创建选择变量和复选框 self.server_vars = {} self.select_all_var = tk.BooleanVar() # 默认展开搜索面板 self.search_pane.show.set(True) self.search_pane._toggle() # 添加以下方法到DataViewer类 def toggle_all_servers(self): """全选/取消全选所有服务器主板""" state = self.select_all_var.get() for var in self.server_vars.values(): var.set(state) def search_server_boards(self): """根据选中的服务器主板查找数据""" if self.df is None: messagebox.showwarning("警告", "请先更新数据！") return # 获取选中的主板 selected_boards = [board for board, var in self.server_vars.items() if var.get()] if not selected_boards: messagebox.showinfo("提示", "请至少选择一个主板型号") return # 查找包含这些主板名称的行 mask = self.df['单板名称'].str.contains('|'.join(selected_boards), case=False, na=False) result_df = self.df[mask] if result_df.empty: messagebox.showinfo("查询结果", "未找到匹配的主板数据") return # 显示查询结果 self.display_search_results(result_df, [f"选中主板: {len(selected_boards)}个"]) def search_by_tracking_simple(self): """根据计划跟踪单简查找相关行""" if self.df is None: messagebox.showwarning("警告", "请先更新数据！") return tracking_simple = self.tracking_simple_var.get().strip() if not tracking_simple: messagebox.showinfo("提示", "请输入计划跟踪单简") return # 查找匹配的行 mask = self.df['计划跟踪单简'] == tracking_simple result_df = self.df[mask] if result_df.empty: messagebox.showinfo("查询结果", f"未找到计划跟踪单简为 {tracking_simple} 的数据") return # 显示查询结果 self.display_search_results(result_df, [f"计划跟踪单简: {tracking_simple}"]) def copy_selected_rows(self, event=None): selected_items = self.detail_tree.selection() if not selected_items: return data = [] for item in selected_items: values = self.detail_tree.item(item)['values'] data.append('\t'.join(str(v) for v in values)) copy_text = '\n'.join(data) self.root.clipboard_clear() self.root.clipboard_append(copy_text) def get_unique_values(self, field): """获取字段的唯一值列表""" try: if not hasattr(self, 'df') or self.df is None: print(f"获取{field}唯一值失败: 数据框未初始化") return [] if field not in self.df.columns: print(f"获取{field}唯一值失败: 字段不存在") return [] # 过滤掉空值并转换为字符串 values = self.df[field].dropna().astype(str).unique().tolist() # 排序并返回非空值 return sorted([v for v in values if v and v != 'nan' and v.strip()]) except Exception as e: print(f"获取{field}唯一值时出错: {str(e)}") traceback.print_exc() return [] def show_date_picker(self, widget): """显示日期选择器""" def set_date(): selected_date = cal.selection_get().strftime('%Y-%m-%d') widget.delete(0, tk.END) widget.insert(0, selected_date) top.destroy() # 获取输入框在屏幕上的位置 x = widget.winfo_rootx() y = widget.winfo_rooty() + widget.winfo_height() top = tk.Toplevel(self.root) top.title("选择日期") # 设置日历窗口位置 top.geometry(f"+{x}+{y}") # 设置日历默认选择当前日期 today = datetime.now().date() cal = Calendar(top, selectmode='day', date_pattern='yyyy-mm-dd', year=today.year, month=today.month, day=today.day) cal.pack(padx=10, pady=10) ttk.Button(top, text="确定", command=set_date).pack(pady=5) # 3. 修改search_detail方法，集成服务器主板筛选和计划跟踪单简匹配 def search_detail(self, event=None): """单板明细搜索功能（含计划跟踪单关联查询）""" try: if self.df is None or self.df.empty: messagebox.showwarning("警告", "请先更新数据！") return # 清空当前显示 for item in self.detail_tree.get_children(): self.detail_tree.delete(item) # 自动识别搜索逻辑 auto_search_text = self.auto_search_text.get("1.0", tk.END).strip() if auto_search_text: parts = re.findall(r'\S+', auto_search_text) if parts: mask = self.df.apply( lambda row: any( part in str(row[col]) for part in parts for col in ['批次', '单板名称', '单板代码', '计划跟踪单', '项目'] ), axis=1 ) result_df = self.df[mask] else: result_df = pd.DataFrame() else: result_df = self.df.copy() # 条件搜索逻辑（范围、日期、普通文本） used_conditions = [] for field, var in self.search_vars.items(): if isinstance(var, tuple): # 范围查询 min_val = var[0].get().strip() max_val = var[1].get().strip() if min_val or max_val: used_conditions.append(f"{field}: {min_val} - {max_val}") if '比例' in field: if min_val: min_val = self.parse_percentage(min_val) result_df = result_df[result_df[field].astype(float) >= min_val] if max_val: max_val = self.parse_percentage(max_val) result_df = result_df[result_df[field].astype(float) <= max_val] elif '周期' in field: if min_val: min_val = float(min_val) result_df = result_df[result_df[field].astype(float) >= min_val] if max_val: max_val = float(max_val) result_df = result_df[result_df[field].astype(float) <= max_val] else: value = var.get().strip() if value: used_conditions.append(f"{field}: {value}") if field == '释放日期': try: search_date = pd.to_datetime(value).strftime('%Y-%m-%d') result_df = result_df[result_df['释放日期'].dt.strftime('%Y-%m-%d') == search_date] except Exception: messagebox.showwarning("警告", "请输入正确的日期格式（YYYY-MM-DD）") return else: values = [v.strip() for v in value.split(',') if v.strip()] if values: mask = result_df[field].astype(str).str.contains('|'.join(values), na=False) result_df = result_df[mask] # 服务器主板筛选 if self.server_filter_var.get(): server_mask = result_df['单板名称'].str.contains('|'.join(self.server_boards), case=False, na=False) result_df = result_df[server_mask] used_conditions.append("服务器主板筛选: 是") # ------------ 新增：计划跟踪单关联查询 ------------ if not result_df.empty and '计划跟踪单简' in result_df.columns: tracking_simples = result_df['计划跟踪单简'].unique() related_mask = self.df['计划跟踪单简'].isin(tracking_simples) result_df = self.df[related_mask].drop_duplicates(subset=['批次', '计划跟踪单']) # -------------------------------------------------- if result_df.empty: messagebox.showinfo("查询结果", f"未找到匹配数据\n使用的搜索条件：\n" + '\n'.join(used_conditions)) return # 显示结果 self.display_search_results(result_df, used_conditions) except Exception as e: messagebox.showerror("错误", f"搜索数据时出错：{str(e)}\n请检查输入条件是否正确") def auto_identify_paste_data(self, event=None): # 添加 event=None 以兼容两种触发方式 try: # 检查是否有数据（包括缓存数据） has_data = (hasattr(self, 'df') and self.df is not None and not self.df.empty) # 如果没有数据但有缓存文件，尝试加载缓存 if not has_data and os.path.exists(os.path.join(self.current_dir, 'data_cache.pkl')): self.load_cached_data() has_data = (hasattr(self, 'df') and self.df is not None and not self.df.empty) if not has_data: messagebox.showwarning("警告", "请先更新数据！") return # 获取粘贴的内容 paste_data = self.auto_search_text.get("1.0", tk.END).strip() if not paste_data: return # 如果数据为空，直接返回 # 使用正则表达式分割数据（支持空格、换行、制表符等） parts = re.findall(r'\S+', paste_data) # 如果没有有效数据，直接返回 if not parts: messagebox.showinfo("提示", "未找到有效数据") return # 清空现有数据 for item in self.detail_tree.get_children(): self.detail_tree.delete(item) # 查找匹配的数据，并确保每个批次只插入一次 matched_batches = set() # 用于记录已经匹配的批次 matched_rows = [] for part in parts: for _, row in self.df.iterrows(): # 使用 self.df 作为数据源 batch = row['批次'] if batch in matched_batches: continue # 如果批次已经匹配过，跳过 # 只在特定列中匹配（批次、单板名称、单板代码、计划跟踪单、项目） search_columns = ['批次', '单板名称', '单板代码', '计划跟踪单', '项目'] for col in search_columns: if part in str(row[col]): matched_rows.append(row) matched_batches.add(batch) # 记录已经匹配的批次 break # 匹配到后跳出当前行的匹配 # 显示匹配的数据 if matched_rows: row_count = 0 for row in matched_rows: row_count += 1 values = [row_count] + list(row) self.detail_tree.insert('', tk.END, values=values, tags=('matched',)) # 高亮显示匹配的行 self.detail_tree.tag_configure('matched', background='yellow') messagebox.showinfo("提示", f"找到 {row_count} 条匹配记录") else: messagebox.showinfo("提示", "未找到匹配数据") except Exception as e: messagebox.showerror("错误", f"自动识别数据时出错：{str(e)}") def parse_percentage(self, value): """解析百分比输入""" try: if value.endswith('%'): return float(value.strip('%')) / 100 return float(value) except ValueError: raise ValueError("请输入正确的百分比格式，例如 '20%' 或 '0.2'") # 4. 修改reset_detail_search方法，重置服务器主板筛选 def reset_detail_search(self): """重置单板明细搜索条件""" try: # 1. 清空自动识别文本框 self.auto_search_text.delete("1.0", tk.END) # 2. 重置所有搜索变量 for field, var in self.search_vars.items(): if isinstance(var, tuple): # 范围查询 var[0].set('') var[1].set('') elif field == '释放日期': # 日期字段重置为当前日期 var.set(datetime.now().strftime('%Y-%m-%d')) else: # 其他字段 var.set('') # 3. 重置服务器主板筛选 self.server_filter_var.set(False) # 4. 清空当前显示 for item in self.detail_tree.get_children(): self.detail_tree.delete(item) # 5. 显示所有数据 if self.df is not None and not self.df.empty: # 按默认排序显示数据 result_df = self.df.sort_values(by=['释放日期', '计划跟踪单'], ascending=[True, True]) # 显示数据 self._display_detail_data(result_df) messagebox.showinfo("重置完成", f"已重置所有搜索条件，显示全部 {len(result_df)} 条记录") else: messagebox.showwarning("警告", "没有可显示的数据，请先更新数据") except Exception as e: messagebox.showerror("错误", f"重置搜索条件时出错：{str(e)}") def _display_detail_data(self, df): """显示单板明细数据到表格（增加去重处理）""" try: # 清空现有数据 for item in self.detail_tree.get_children(): self.detail_tree.delete(item) # 添加去重处理 - 按批次保留第一条记录 df = df.drop_duplicates(subset=['批次'], keep='first') # 显示数据 row_count = 0 for _, row in df.iterrows(): row_count += 1 values = [row_count] # 序号 # 按照 column_order 的顺序添加数据（跳过序号列） for col in self.column_order[1:]: if col in row.index: val = row[col] if pd.isna(val): val = '' elif isinstance(val, (datetime, pd.Timestamp)): if col == '释放日期': val = val.strftime('%Y-%m-%d %H:%M:%S') else: val = val.strftime('%Y-%m-%d') values.append(str(val)) else: values.append('') # 检查是否已存在相同批次的记录 existing_items = [self.detail_tree.item(item)['values'][1] for item in self.detail_tree.get_children()] if str(row['批次']) not in existing_items: self.detail_tree.insert('', tk.END, values=values) else: print(f"发现重复批次: {row['批次']}，已跳过") except Exception as e: print(f"显示数据时出错: {str(e)}") messagebox.showerror("错误", f"显示数据时出错：{str(e)}") def select_factories(self): factory_selector = FactorySelector(self.root) self.root.wait_window(factory_selector) self.selected_factories = factory_selector.result print(f"选择的工厂ID: {self.selected_factories}") # 调试输出 self._save_config() # 保存选择的工厂 if self.selected_factories: self.update_data() def setup_context_menu(self): """设置右键菜单""" self.context_menu = tk.Menu(self.root, tearoff=0) def show_context_menu(event): tree = event.widget region = tree.identify_region(event.x, event.y) if region == "cell" or region == "heading": item = tree.identify_row(event.y) column = tree.identify_column(event.x) self.context_menu.delete(0, tk.END) if region == "cell": self.context_menu.add_command(label="复制单元格", command=lambda: self.copy_cell(tree, item, column)) self.context_menu.add_command(label="复制整行", command=lambda: self.copy_row(tree, item)) self.context_menu.add_command(label="复制整列", command=lambda: self.copy_column(tree, column)) self.context_menu.add_separator() self.context_menu.add_command(label="全选", command=lambda: self.select_all(tree)) self.context_menu.add_command(label="取消全选", command=lambda: self.deselect_all(tree)) self.context_menu.add_command(label="全选并复制", command=lambda: self.select_all_and_copy(tree)) self.context_menu.post(event.x_root, event.y_root) # 为所有树形视图都绑定右键菜单 if getattr(self, 'detail_tree', None) is not None: self.detail_tree.bind('<Button-3>', show_context_menu) if getattr(self, 'delivery_tree', None) is not None: self.delivery_tree.bind('<Button-3>', show_context_menu) if getattr(self, 'delivery_test_tree', None) is not None: self.delivery_test_tree.bind('<Button-3>', show_context_menu) def copy_cell(self, tree, item, column): """复制单元格内容""" col_id = int(column[1:]) - 1 value = tree.item(item)['values'][col_id] self.root.clipboard_clear() self.root.clipboard_append(str(value)) messagebox.showinfo("提示", "已复制单元格内容到剪贴板") def copy_row(self, tree, item): """复制整行内容""" values = tree.item(item)['values'] copy_text = '\t'.join(str(v) for v in values) self.root.clipboard_clear() self.root.clipboard_append(copy_text) messagebox.showinfo("提示", "已复制整行内容到剪贴板") def copy_column(self, tree, column): """复制整列内容""" col_id = int(column[1:]) - 1 values = [] for item in tree.get_children(): value = tree.item(item)['values'][col_id] values.append(str(value)) copy_text = '\n'.join(values) self.root.clipboard_clear() self.root.clipboard_append(copy_text) messagebox.showinfo("提示", "已复制整列内容到剪贴板") def on_closing(self): """处理窗口关闭事件""" if messagebox.askokcancel("退出", "确定要退出程序吗？"): # 取消所有计划任务 if hasattr(self, 'auto_update_id') and self.auto_update_id: self.root.after_cancel(self.auto_update_id) self.root.destroy() def show_all_detail_data(self): """显示所有单板明细数据""" try: if self.df is None or self.df.empty: return # 按默认排序显示数据 result_df = self.df.sort_values(by=['释放日期', '计划跟踪单'], ascending=[True, True]) # 显示数据 self._display_detail_data(result_df) except Exception as e: messagebox.showerror("错误", f"显示数据时出错：{str(e)}") def export_to_excel(self): """导出当前查询结果到Excel""" try: if not self.detail_tree.get_children(): messagebox.showwarning("警告", "没有数据可以导出！") return # 获取保存文件路径 file_path = filedialog.asksaveasfilename( defaultextension=".xlsx", filetypes=[("Excel files", ".xlsx")], initialfile=f"单板明细_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" ) if not file_path: return # 创建数据框 data = [] columns = [self.detail_tree.heading(col)["text"] for col in self.detail_tree["columns"]] for item in self.detail_tree.get_children(): values = self.detail_tree.item(item)["values"] data.append(values) df = pd.DataFrame(data, columns=columns) # 导出到Excel df.to_excel(file_path, index=False) messagebox.showinfo("成功", "数据已成功导出到Excel文件！") except Exception as e: messagebox.showerror("错误", f"导出Excel时出错：{str(e)}") def display_search_results(self, result_df, used_conditions=None): """显示搜索结果""" # 默认按释放日期和计划跟踪单排序 result_df = result_df.sort_values(by=['释放日期', '计划跟踪单'], ascending=[True, True]) # 清空现有数据 self.detail_tree.delete(self.detail_tree.get_children()) row_count = 0 for _, row in result_df.iterrows(): row_count += 1 values = [row_count] for col in self.detail_tree['columns'][1:]: val = row.get(col, '') if pd.isna(val): val = '' elif isinstance(val, (datetime, pd.Timestamp)): # 修改这里 if col == '释放日期': val = val.strftime('%Y-%m-%d %H:%M:%S') else: val = val.strftime('%Y-%m-%d') values.append(str(val)) self.detail_tree.insert('', tk.END, values=values) # 构建消息内容 message = f"找到 {row_count} 条匹配记录" if used_conditions: message += "\n\n使用的搜索条件：\n" + '\n'.join(used_conditions) messagebox.showinfo("查询完成", message) def handle_combobox_input(self, event, field): """处理下拉框手动输入""" widget = event.widget value = widget.get().strip() if value: current_values = list(widget['values']) if value not in current_values: current_values.append(value) widget['values'] = current_values def handle_combobox_select(self, event, field): """处理下拉框选择""" widget = event.widget # 可以在这里添加多选逻辑 pass def select_all(self, tree): """全选功能""" for item in tree.get_children(): tree.selection_add(item) tree.focus_set() # 设置焦点到 Treeview 上 def deselect_all(self, tree): """取消全选功能""" tree.selection_remove(tree.selection()) def update_combobox_values(self): """更新所有下拉框的选项""" try: if not hasattr(self, 'search_widgets') or not self.search_widgets: print("搜索控件未初始化") return if self.df is not None: # 更新计调员下拉框 if '计调员' in self.search_widgets and hasattr(self.search_widgets['计调员'], 'winfo_exists') and self.search_widgets['计调员'].winfo_exists(): values = self.get_unique_values('计调员') self.search_widgets['计调员']['values'] = values if values else ['暂无数据'] # 更新小组长下拉框 if '小组长' in self.search_widgets and hasattr(self.search_widgets['小组长'], 'winfo_exists') and self.search_widgets['小组长'].winfo_exists(): values = self.get_unique_values('小组长') self.search_widgets['小组长']['values'] = values if values else ['暂无数据'] # 更新大组长下拉框 if '大组长' in self.search_widgets and hasattr(self.search_widgets['大组长'], 'winfo_exists') and self.search_widgets['大组长'].winfo_exists(): values = self.get_unique_values('大组长') self.search_widgets['大组长']['values'] = values if values else ['暂无数据'] # 更新产品大类下拉框 if '产品大类' in self.search_widgets and hasattr(self.search_widgets['产品大类'], 'winfo_exists') and self.search_widgets['产品大类'].winfo_exists(): values = self.get_unique_values('产品大类') self.search_widgets['产品大类']['values'] = values if values else ['暂无数据'] # 更新结存状态下拉框 if '结存状态' in self.search_widgets and hasattr(self.search_widgets['结存状态'], 'winfo_exists') and self.search_widgets['结存状态'].winfo_exists(): values = self.get_unique_values('结存状态') self.search_widgets['结存状态']['values'] = values if values else ['暂无数据'] # 更新线体下拉框 if '线体' in self.search_widgets and hasattr(self.search_widgets['线体'], 'winfo_exists') and self.search_widgets['线体'].winfo_exists(): values = self.get_unique_values('线体') self.search_widgets['线体']['values'] = values if values else ['暂无数据'] except Exception as e: print(f"更新下拉框选项时出错: {str(e)}") traceback.print_exc() def treeview_sort_column(self, col): """列排序功能""" if col not in self.sort_order: self.sort_order[col] = 'asc' else: self.sort_order[col] = 'desc' if self.sort_order[col] == 'asc' else 'asc' # 更新列标题显示排序方向 for column in self.detail_tree['columns']: if column != col: self.detail_tree.heading(column, text=column.replace(' ▲', '').replace(' ▼', '')) current_text = self.detail_tree.heading(col)['text'].replace(' ▲', '').replace(' ▼', '') arrow = ' ▲' if self.sort_order[col] == 'asc' else ' ▼' self.detail_tree.heading(col, text=current_text + arrow) # 获取当前显示的数据并排序 data = [] for item in self.detail_tree.get_children(): values = self.detail_tree.item(item)['values'] data.append(values) # 转换为DataFrame进行排序 df = pd.DataFrame(data, columns=self.detail_tree['columns']) # 确保单板交期字段是日期类型并格式化 if col == '单板交期': df[col] = pd.to_datetime(df[col], errors='coerce') # 始终按照释放日期和计划跟踪单排序，然后是当前选择的列 sort_columns = [] sort_ascending = [] # 如果当前排序列不是释放日期或计划跟踪单，添加默认排序 if col not in ['释放日期', '计划跟踪单']: sort_columns.extend(['释放日期', '计划跟踪单']) sort_ascending.extend([True, True]) # 添加当前排序列 sort_columns.append(col) sort_ascending.append(self.sort_order[col] == 'asc') # 执行排序 df = df.sort_values(by=sort_columns, ascending=sort_ascending, na_position='last') # 将 NaT 转换为空字符串 if col == '单板交期': df[col] = df[col].dt.strftime('%Y-%m-%d').fillna('') # 更新显示 self.detail_tree.delete(self.detail_tree.get_children()) for index, row in df.iterrows(): self.detail_tree.insert('', tk.END, values=list(row)) def setup_delivery_view(self): """设置单板交期视图""" # 确保 delivery_frame 存在 if not hasattr(self, 'delivery_frame'): self.delivery_frame = ttk.Frame(self.content_frame) # 清空现有内容（如果有的话） for widget in self.delivery_frame.winfo_children(): widget.destroy() # 添加查询输入框和按钮在上方 search_frame = ttk.Frame(self.delivery_frame) search_frame.pack(fill=tk.X, padx=5, pady=5) ttk.Label(search_frame, text="批次:").pack(side=tk.LEFT) self.batch_search_var = tk.StringVar() ttk.Entry(search_frame, textvariable=self.batch_search_var, width=15).pack(side=tk.LEFT, padx=5) ttk.Label(search_frame, text="交期:").pack(side=tk.LEFT) self.date_search_var = tk.StringVar() date_entry = ttk.Entry(search_frame, textvariable=self.date_search_var, width=15) date_entry.pack(side=tk.LEFT, padx=5) date_entry.bind('<Button-1>', lambda e: self.show_date_picker(date_entry)) ttk.Button(search_frame, text="查询", command=self.search_delivery).pack(side=tk.LEFT, padx=5) ttk.Button(search_frame, text="重置", command=self.reset_delivery_search).pack(side=tk.LEFT, padx=5) ttk.Button(search_frame, text="导出Excel", command=self.export_delivery_to_excel).pack(side=tk.LEFT, padx=5) # 添加新增和批量新增按钮 add_button = ttk.Button(search_frame, text="新增", command=self.add_delivery) bulk_add_button = ttk.Button(search_frame, text="批量新增", command=self.bulk_add_delivery) if self.has_maintenance_permission: add_button.configure(state='normal') bulk_add_button.configure(state='normal') print("已启用维护按钮") else: add_button.configure(state='disabled') bulk_add_button.configure(state='disabled') messagebox.showwarning("权限提示", "当前只有更新和查询权限，没有维护交期权限！") print("已禁用维护按钮") add_button.pack(side=tk.LEFT, padx=5) bulk_add_button.pack(side=tk.LEFT, padx=5) ttk.Button(search_frame, text="更新数据", command=self.update_delivery_data).pack(side=tk.LEFT, padx=5) # 创建表格框架 tree_frame = ttk.Frame(self.delivery_frame) tree_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 定义交期表格列 delivery_columns = ['create_time', '批次', '单板交期', '提出人', '项目', '备注', '预留字段1', '预留字段2', '修改时间', '操作'] self.delivery_tree = ttk.Treeview(tree_frame, columns=delivery_columns, show='headings') # 设置列宽和标题 column_widths = {'create_time': 150, '批次': 100, '单板交期': 120, '提出人': 100, '项目': 200, '备注': 100, '预留字段1': 100, '预留字段2': 100, '修改时间': 150, '操作': 80} for col in delivery_columns: self.delivery_tree.heading(col, text=col, anchor='w') self.delivery_tree.column(col, width=column_widths[col], anchor='w') # 添加滚动条 vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.delivery_tree.yview) hsb = ttk.Scrollbar(tree_frame, orient="horizontal", command=self.delivery_tree.xview) self.delivery_tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) # 布局 self.delivery_tree.grid(row=0, column=0, sticky='nsew') vsb.grid(row=0, column=1, sticky='ns') hsb.grid(row=1, column=0, sticky='ew') # 配置网格权重 tree_frame.grid_columnconfigure(0, weight=1) tree_frame.grid_rowconfigure(0, weight=1) # 添加修改按钮 self.delivery_tree.bind('<ButtonRelease-1>', self.on_tree_select) def update_delivery_data(self): """在后台线程中从数据库更新交期数据""" # 创建状态指示器 self.safe_status_update("正在后台更新交期数据...") # 添加进度指示器 progress_frame = ttk.Frame(self.status_bar) progress_frame.pack(side=tk.RIGHT, padx=10) progress = ttk.Progressbar(progress_frame, mode='indeterminate', length=100) progress.pack(side=tk.RIGHT) progress.start() # 更新按钮状态 update_buttons = [] for widget in self.delivery_frame.winfo_children(): if isinstance(widget, ttk.Frame): for btn in widget.winfo_children(): if isinstance(btn, ttk.Button) and btn['text'] == "更新数据": btn['text'] = "更新中..." btn['state'] = 'disabled' update_buttons.append(btn) # 定义后台更新函数 def background_update(): try: with self.engine.connect() as connection: query = "SELECT * FROM delivery_date" df_delivery = pd.read_sql(query, connection) # 使用after方法在主线程中安全更新UI self.root.after(0, lambda: self._update_delivery_ui(df_delivery)) # 更新最后更新时间 self.last_update_time = datetime.now() except Exception as e: # 使用after方法在主线程中安全显示错误 self.root.after(0, lambda: messagebox.showerror("错误", f"更新交期数据时出错：{str(e)}")) finally: # 使用after方法在主线程中安全清理UI self.root.after(0, lambda: self._cleanup_progress(progress_frame, progress, update_buttons)) # 启动后台线程 threading.Thread(target=background_update, daemon=True).start() def _update_delivery_ui(self, df_delivery): """更新交期数据UI""" try: # 清空现有数据 for item in self.delivery_tree.get_children(): self.delivery_tree.delete(item) # 插入新数据 for _, row in df_delivery.iterrows(): self.delivery_tree.insert('', tk.END, values=( row['create_time'], row['批次'], row['单板交期'], row['提出人'], row['项目'], row['备注'], row['预留字段1'], row['预留字段2'], row['修改时间'], '修改' )) # 更新状态 self.safe_status_update(f"交期数据更新成功，共 {len(df_delivery)} 条记录") except Exception as e: print(f"更新交期UI时出错: {str(e)}") traceback.print_exc() def _cleanup_progress(self, progress_frame, progress, update_buttons): """清理进度指示器和恢复按钮状态""" try: # 停止进度条并移除 if progress.winfo_exists(): progress.stop() if progress_frame.winfo_exists(): progress_frame.destroy() # 恢复按钮状态 for btn in update_buttons: if btn.winfo_exists(): btn['text'] = "更新数据" btn['state'] = 'normal' except Exception as e: print(f"清理UI元素时出错: {str(e)}") def search_delivery(self): """查询交期数据""" # 添加检查更新时间的逻辑 # if self.check_update_time(): # if messagebox.askyesno("提示", "上次更新数据已超过30分钟，是否先更新数据？"): # self.update_delivery_data() # return batch = self.batch_search_var.get().strip() date = self.date_search_var.get().strip() # 使用参数化查询避免SQL注入 query = "SELECT * FROM delivery_date WHERE 1=1" params = {} if batch: query += " AND 批次 = :batch" params['batch'] = batch if date: query += " AND 单板交期 = :date" params['date'] = date try: with self.engine.connect() as connection: # 使用参数化查询 from sqlalchemy import text df_delivery = pd.read_sql(text(query), connection, params=params) # 清空现有数据 self.delivery_tree.delete(self.delivery_tree.get_children()) # 插入查询结果 for _, row in df_delivery.iterrows(): self.delivery_tree.insert('', tk.END, values=( row['create_time'], row['批次'], row['单板交期'], row['提出人'], row['项目'], row['备注'], row['预留字段1'], row['预留字段2'], row['修改时间'], '修改' )) # 显示查询结果统计 if len(df_delivery) == 0: messagebox.showinfo("查询结果", "未找到匹配的数据") else: messagebox.showinfo("查询结果", f"找到 {len(df_delivery)} 条匹配记录") except Exception as e: messagebox.showerror("错误", f"查询数据时出错：{str(e)}") def reset_delivery_search(self): """重置交期搜索条件""" # 清空搜索条件 self.batch_search_var.set('') # 日期字段重置为当前日期 self.date_search_var.set(datetime.now().strftime('%Y-%m-%d')) # 清空搜索结果 self.delivery_tree.delete(self.delivery_tree.get_children()) def add_delivery(self): """新增单条交期数据""" def save(): batches = batch_text.get("1.0", tk.END).strip().splitlines() date = date_var.get().strip() proposer = proposer_var.get().strip() remark = remark_var.get().strip() field6 = field6_var.get().strip() field7 = field7_var.get().strip() field8 = field8_var.get().strip() create_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') if not batches or not date or not proposer: messagebox.showwarning("警告", "请填写所有必填项！") return try: with self.engine.begin() as connection: for batch in batches: query = text("INSERT INTO delivery_date (批次, 单板交期, 提出人, create_time, 项目, 备注, 预留字段1, 预留字段2) VALUES (:batch, :date, :proposer, :create_time, :remark, :field6, :field7, :field8)") params = { 'batch': batch.strip(), 'date': date, 'proposer': proposer, 'create_time': create_time, 'remark': remark or None, 'field7': field7 or None, 'field8': field8 or None } connection.execute(query, params) self.update_delivery_data() messagebox.showinfo("成功", "数据上传成功！") dialog.destroy() except Exception as e: messagebox.showerror("错误", f"保存数据时出错：{str(e)}") dialog = tk.Toplevel(self.root) dialog.title("新增交期数据") batch_text = tk.Text(dialog, width=30, height=5) # 调整为多行输入 batch_text.grid(row=0, column=1, padx=5, pady=5) date_var = tk.StringVar() proposer_var = tk.StringVar() remark_var = tk.StringVar() field6_var = tk.StringVar() field7_var = tk.StringVar() field8_var = tk.StringVar() ttk.Label(dialog, text="批次:").grid(row=0, column=0, padx=5, pady=5) ttk.Label(dialog, text="单板交期:").grid(row=1, column=0, padx=5, pady=5) date_entry = ttk.Entry(dialog, textvariable=date_var) date_entry.grid(row=1, column=1, padx=5, pady=5) date_entry.bind('<Button-1>', lambda e: self.show_date_picker(date_entry)) ttk.Label(dialog, text="提出人:").grid(row=2, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=proposer_var).grid(row=2, column=1, padx=5, pady=5) ttk.Label(dialog, text="项目:").grid(row=3, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=remark_var).grid(row=3, column=1, padx=5, pady=5) ttk.Label(dialog, text="备注:").grid(row=4, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=field6_var).grid(row=4, column=1, padx=5, pady=5) ttk.Label(dialog, text="预留字段1:").grid(row=5, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=field7_var).grid(row=5, column=1, padx=5, pady=5) ttk.Label(dialog, text="预留字段2:").grid(row=6, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=field8_var).grid(row=6, column=1, padx=5, pady=5) ttk.Button(dialog, text="保存", command=save).grid(row=7, column=0, columnspan=2, pady=10) def bulk_add_delivery(self): """批量新增交期数据""" def save(): data = batch_text.get("1.0", tk.END).strip() if not data: messagebox.showwarning("警告", "请提供批量数据！") return try: with self.engine.begin() as connection: for line in data.splitlines(): parts = line.split() if len(parts) < 3: messagebox.showwarning("警告", "数据格式不正确！请按格式：批次单板交期提出人 [项目]") return batch, date, proposer = parts[:3] remark = parts[3] if len(parts) > 3 else None field6 = parts[4] if len(parts) > 4 else None field7 = parts[5] if len(parts) > 5 else None field8 = parts[6] if len(parts) > 6 else None create_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') query = text("INSERT INTO delivery_date (批次, 单板交期, 提出人, create_time, 项目, 备注, 预留字段1, 预留字段2) VALUES (:batch, :date, :proposer, :create_time, :remark, :field6, :field7, :field8)") params = { 'batch': batch, 'date': date, 'proposer': proposer, 'create_time': create_time, 'remark': remark, 'field6': field6, 'field7': field7, 'field8': field8 } connection.execute(query, params) self.update_delivery_data() messagebox.showinfo("成功", "批量数据上传成功！") dialog.destroy() except Exception as e: messagebox.showerror("错误", f"保存数据时出错：{str(e)}") dialog = tk.Toplevel(self.root) dialog.title("批量新增交期数据") # 添加格式说明 format_label = ttk.Label(dialog, text="格式: 批次单板交期(YYYY/MM/DD) 提出人 [项目] [备注] [预留字段1] [预留字段2]") format_label.pack(pady=5) batch_text = tk.Text(dialog, width=150, height=30) batch_text.pack(padx=10, pady=10) ttk.Button(dialog, text="保存", command=save).pack(pady=10) def show_detail_view(self): """显示单板明细视图（确保数据正确显示）""" # 确保视图已初始化 if not hasattr(self, 'detail_tree'): self.setup_detail_view() # 添加对matching_frame的检查 if hasattr(self, 'matching_frame'): self.matching_frame.pack_forget() self.detail_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "单板明细" # 如果数据已加载，立即显示 if hasattr(self, 'df') and self.df is not None and not self.df.empty: self.show_all_detail_data() def export_delivery_to_excel(self): """导出交期数据到Excel""" try: if not self.delivery_tree.get_children(): messagebox.showwarning("警告", "没有数据可以导出！") return # 获取保存文件路径 file_path = filedialog.asksaveasfilename( defaultextension=".xlsx", filetypes=[("Excel files", ".xlsx")], initialfile=f"交期数据_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" ) if not file_path: return # 创建数据框 data = [] columns = [self.delivery_tree.heading(col)["text"] for col in self.delivery_tree["columns"]] for item in self.delivery_tree.get_children(): values = self.delivery_tree.item(item)["values"] data.append(values) df = pd.DataFrame(data, columns=columns) # 导出到Excel df.to_excel(file_path, index=False) messagebox.showinfo("成功", "数据已成功导出到Excel文件！") except Exception as e: messagebox.showerror("错误", f"导出Excel时出错：{str(e)}") def on_tree_select(self, event): """处理Treeview选择事件""" item = self.delivery_tree.selection()[0] column = self.delivery_tree.identify_column(event.x) if column == '#10': # 检查是否点击了操作列 if not self.has_maintenance_permission: messagebox.showwarning("权限提示", "没有修改权限！") return values = self.delivery_tree.item(item, "values") self.show_edit_dialog(item, values) # 传递 item def show_edit_dialog(self, item, values): """显示编辑对话框""" dialog = tk.Toplevel(self.root) dialog.title("编辑交期数据") # 创建变量 # 如果日期值为空，则使用当前日期 date_value = values[2] if values[2] else datetime.now().strftime('%Y-%m-%d') date_var = tk.StringVar(value=date_value) proposer_var = tk.StringVar(value=values[3]) remark_var = tk.StringVar(value=values[4]) field6_var = tk.StringVar(value=values[5]) field7_var = tk.StringVar(value=values[6]) field8_var = tk.StringVar(value=values[7]) # 布局 ttk.Label(dialog, text="单板交期:").grid(row=0, column=0, padx=5, pady=5) date_entry = ttk.Entry(dialog, textvariable=date_var) date_entry.grid(row=0, column=1, padx=5, pady=5) date_entry.bind('<Button-1>', lambda e: self.show_date_picker(date_entry)) ttk.Label(dialog, text="提出人:").grid(row=1, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=proposer_var).grid(row=1, column=1, padx=5, pady=5) ttk.Label(dialog, text="项目:").grid(row=2, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=remark_var).grid(row=2, column=1, padx=5, pady=5) ttk.Label(dialog, text="备注:").grid(row=3, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=field6_var).grid(row=3, column=1, padx=5, pady=5) ttk.Label(dialog, text="预留字段1:").grid(row=4, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=field7_var).grid(row=4, column=1, padx=5, pady=5) ttk.Label(dialog, text="预留字段2:").grid(row=5, column=0, padx=5, pady=5) ttk.Entry(dialog, textvariable=field8_var).grid(row=5, column=1, padx=5, pady=5) ttk.Button(dialog, text="保存", command=lambda: self.save_edit(item, date_var, proposer_var, remark_var, field6_var, field7_var, field8_var, dialog)).grid(row=6, column=0, columnspan=2, pady=10) def save_edit(self, item, date_var, proposer_var, remark_var, field6_var, field7_var, field8_var, dialog): """保存编辑后的数据""" # 更新数据库和Treeview modify_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') self.delivery_tree.item(item, values=( self.delivery_tree.item(item, "values")[0], # create_time self.delivery_tree.item(item, "values")[1], # 批次 date_var.get(), proposer_var.get(), remark_var.get(), field6_var.get(), field7_var.get(), field8_var.get(), modify_time, '修改' )) # 更新数据库 try: with self.engine.begin() as connection: query = text("UPDATE delivery_date SET 单板交期=:date, 提出人=:proposer, 项目=:remark, 备注=:field6, 预留字段1=:field7, 预留字段2=:field8, 修改时间=:modify_time WHERE 批次=:batch") params = { 'date': date_var.get(), 'proposer': proposer_var.get(), 'remark': remark_var.get(), 'field6': field6_var.get(), 'field7': field7_var.get(), 'field8': field8_var.get(), 'modify_time': modify_time, 'batch': self.delivery_tree.item(item, "values")[1] } connection.execute(query, params) messagebox.showinfo("成功", "数据已更新！") dialog.destroy() except Exception as e: messagebox.showerror("错误", f"更新数据时出错：{str(e)}") def select_all_and_copy(self, tree): """全选并复制功能""" self.select_all(tree) self.copy_all(tree) def copy_all(self, tree): """复制所有内容到剪贴板""" data = [] columns = [tree.heading(col)["text"] for col in tree["columns"]] data.append('\t'.join(columns)) # 添加列标题 for item in tree.get_children(): values = tree.item(item)['values'] data.append('\t'.join(str(v) for v in values)) copy_text = '\n'.join(data) self.root.clipboard_clear() self.root.clipboard_append(copy_text) messagebox.showinfo("提示", "已复制所有内容到剪贴板") def setup_chart_view(self): """设置图表视图""" # 添加按钮来绘制图表 self.button_frame = ttk.Frame(self.chart_frame) self.button_frame.pack(side=tk.TOP, anchor='nw', padx=10, pady=10) self.plot_button = ttk.Button(self.button_frame, text="贴片备料结存", command=self.check_and_plot) self.plot_button.pack(side=tk.LEFT, padx=5) self.smt_plot_button = ttk.Button(self.button_frame, text="SMT结存", command=self.check_and_plot_smt) self.smt_plot_button.pack(side=tk.LEFT, padx=5) self.dip_plot_button = ttk.Button(self.button_frame, text="DIP结存", command=self.check_and_plot_dip) self.dip_plot_button.pack(side=tk.LEFT, padx=5) self.ict_plot_button = ttk.Button(self.button_frame, text="ICT结存", command=self.check_and_plot_ict) self.ict_plot_button.pack(side=tk.LEFT, padx=5) self.test_plot_button = ttk.Button(self.button_frame, text="Test结存", command=self.check_and_plot_test) self.test_plot_button.pack(side=tk.LEFT, padx=5) self.coating_plot_button = ttk.Button(self.button_frame, text="涂覆结存", command=self.check_and_plot_coating) self.coating_plot_button.pack(side=tk.LEFT, padx=5) self.batch_cycle_plot_button = ttk.Button(self.button_frame,text="批次周期图表",command=self.plot_batch_cycle_by_dispatcher) self.batch_cycle_plot_button.pack(side=tk.LEFT, padx=5) # 初始化图表画布 self.chart_canvas = ttk.Frame(self.chart_frame) self.chart_canvas.pack(fill=tk.BOTH, expand=True) def check_and_plot_smt(self): """检查数据更新时间并绘制SMT结存图表""" if self.check_update_time(): if messagebox.askyesno("提示", "上次更新数据已超过30分钟，是否先更新数据？"): self.update_data() return self.plot_stock_trend('SMT结存', 'SMT结存总量', 'SMT结存数量', '#bed742', apply_filter=True) def plot_smt_stock_trend_smt(self): """绘制SMT结存趋势图表""" if self.df is None: messagebox.showwarning("警告", "请先更新数据！") return try: # 数据预处理 df_filtered = self.df[ (self.df['SMT线体及状态'] != '') & (self.df['SMT结存'] > 0) ].copy() df_filtered = df_filtered.drop_duplicates(subset=['批次'], keep='first') df_filtered['释放日期'] = pd.to_datetime( df_filtered['释放日期'], errors='coerce' ).dt.date df_filtered = df_filtered.dropna(subset=['释放日期']) # 分组聚合 self.df_grouped = df_filtered.groupby('释放日期').agg( SMT结存总量=('SMT结存', 'sum'), 批次数量=('批次', 'nunique') ).reset_index().sort_values('释放日期') # 创建图表 fig, ax1 = plt.subplots(figsize=(12, 7)) # 绘制柱状图 main_plot = ax1.bar( self.df_grouped['释放日期'].astype(str), self.df_grouped['SMT结存总量'], width=0.6, color='#bed742', label='SMT结存总量' ) # 创建第二个Y轴 ax2 = ax1.twinx() line_plot, = ax2.plot( self.df_grouped['释放日期'].astype(str), self.df_grouped['批次数量'], color='#D62728', marker='s', linestyle='--', label='批次数量' ) # 设置图表样式 ax1.set_title( f'SMT结存趋势\n' f'总数量：{self.df_grouped["SMT结存总量"].sum():,} | 总批次：{self.df_grouped["批次数量"].sum()}', fontsize=16, # 增大字体大小 fontproperties='SimHei', pad=20 ) ax1.set_xlabel('释放日期', fontproperties='SimHei') ax1.set_ylabel('SMT结存数量', fontproperties='SimHei', color='#4C72B0') ax2.set_ylabel('批次数量', fontproperties='SimHei', color='#D62728') # 设置坐标轴颜色 ax1.tick_params(axis='y', colors='#bed742') ax2.tick_params(axis='y', colors='#D62728') plt.xticks(rotation=45, ha='right') ax1.grid(axis='y', linestyle=':', alpha=0.7) # 合并图例 lines = [main_plot, line_plot] labels = [l.get_label() for l in lines] ax1.legend(lines, labels, loc='upper left', prop={'family': 'SimHei'}) # 添加数值标签 for i, (x, y) in enumerate(zip(self.df_grouped['释放日期'], self.df_grouped['批次数量'])): ax2.annotate(f'{y}', (i, y), textcoords="offset points", xytext=(10, -10), ha='center', fontsize=8, color='red') for bar in main_plot: height = bar.get_height() ax1.annotate(f'{int(height):,}', xy=(bar.get_x() + bar.get_width() / 2, height), xytext=(0, 3), # 3 points vertical offset textcoords="offset points", ha='center', va='bottom', fontsize=8, color='blue') # 渲染图表 self._render_chart(fig) except Exception as e: messagebox.showerror("错误", f"绘图时发生异常：{str(e)}") def _render_chart(self, fig): """渲染图表到界面（增强版）""" # 清除旧图表 for widget in self.chart_canvas.winfo_children(): widget.destroy() # 创建新画布 canvas = FigureCanvasTkAgg(fig, master=self.chart_canvas) canvas.draw() # 布局画布 canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True, padx=10, pady=10) # ✅ 安全获取坐标轴 axes = fig.get_axes() if len(axes) == 2: ax1, ax2 = axes else: ax1 = axes[0] ax2 = None # 没有第二个坐标轴时设为 None # 添加数值标签（仅在存在第二个坐标轴时才添加） if ax2 is not None: for i, (x, y) in enumerate(zip(self.df_grouped['释放日期'], self.df_grouped['批次数量'])): ax2.annotate(f'{y}', (i, y), textcoords="offset points", xytext=(10, -10), ha='center', fontsize=8, color='red') # 为柱状图添加数值标签 for bar in ax1.patches: height = bar.get_height() ax1.annotate( f'{int(height):,}', xy=(bar.get_x() + bar.get_width() / 2, height), xytext=(0, 3), textcoords="offset points", ha='center', va='bottom', fontsize=8, color='blue' ) def check_and_plot(self): """检查数据更新时间并绘制贴片备料结存图表""" if self.check_update_time(): if messagebox.askyesno("提示", "上次更新数据已超过30分钟，是否先更新数据？"): self.update_data() return self.plot_stock_trend('备料区', '备料区总量', '备料区数量', '#4C72B0', apply_filter=True) def show_chart_view(self): """显示图表视图""" # 隐藏其他视图框架，显示图表框架 self.matching_frame.pack_forget() self.detail_frame.pack_forget() self.delivery_frame.pack_forget() self.chart_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "图表查询" try: # 检查数据更新时间并绘制贴片备料结存图表 if self.check_update_time(): if messagebox.askyesno("提示", "上次更新数据已超过30分钟，是否先更新数据？"): self.update_data() return # 如果用户选择更新数据，则退出当前方法 # 调用通用的绘图方法来绘制“备料区”相关的图表 self.plot_stock_trend('备料区', '备料区总量', '备料区数量', '#4C72B0', apply_filter=True) except Exception as e: messagebox.showerror("错误", f"显示图表视图时出错：{str(e)}") def check_and_plot_dip(self): """检查数据更新时间并绘制DIP结存图表""" if self.check_update_time(): if messagebox.askyesno("提示", "上次更新数据已超过30分钟，是否先更新数据？"): self.update_data() return self.plot_stock_trend('DIP结存', 'DIP结存总量', 'DIP结存数量', '#FF5733', apply_filter=False) def check_and_plot_ict(self): """检查数据更新时间并绘制ICT结存图表""" if self.check_update_time(): if messagebox.askyesno("提示", "上次更新数据已超过30分钟，是否先更新数据？"): self.update_data() return self.plot_stock_trend('在线测试', 'ICT结存总量', 'ICT结存数量', '#33FF57', apply_filter=False) def check_and_plot_test(self): """检查数据更新时间并绘制Test结存图表""" if self.check_update_time(): if messagebox.askyesno("提示", "上次更新数据已超过30分钟，是否先更新数据？"): self.update_data() return self.plot_stock_trend('单板测试', 'Test结存总量', 'Test结存数量', '#3357FF', apply_filter=False) def check_and_plot_coating(self): """检查数据更新时间并绘制涂覆结存图表""" if self.check_update_time(): if messagebox.askyesno("提示", "上次更新数据已超过30分钟，是否先更新数据？"): self.update_data() return self.plot_stock_trend('涂覆', '涂覆结存总量', '涂覆结存数量', '#FF33A1', apply_filter=False) def plot_batch_cycle_by_dispatcher(self): """显示计调员的批次条数统计图表，使用明细数据""" try: # 检查是否已加载明细数据 if not hasattr(self, 'batch_cycle_cached_df') or self.batch_cycle_cached_df is None or self.batch_cycle_cached_df.empty: messagebox.showinfo("提示", "请先加载明细数据") return # 获取阈值 try: threshold = int(self.cycle_threshold_var.get()) except ValueError: threshold = 8 # 从缓存中取数据并过滤 df_filtered = self.batch_cycle_cached_df[self.batch_cycle_cached_df['在线周期'] > threshold] # 统计每个计调员的条数 dispatcher_count = df_filtered.groupby('计调员').size().reset_index(name='条数') dispatcher_count = dispatcher_count.sort_values(by='条数', ascending=False) # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False # 绘图 fig, ax = plt.subplots(figsize=(12, 6)) bars = ax.bar(dispatcher_count['计调员'], dispatcher_count['条数'], color='#4C72B0', label='条数') ax.set_title(f'计调员批次条数统计（明细数据） - {self.current_date}', fontsize=14) ax.set_xlabel('计调员', fontsize=12) ax.set_ylabel('条数', fontsize=12) ax.grid(axis='y', linestyle='--', alpha=0.7) # 添加数值标签 for bar in bars: height = bar.get_height() ax.annotate(f'{int(height)}', xy=(bar.get_x() + bar.get_width() / 2, height), xytext=(0, 3), textcoords="offset points", ha='center', va='bottom', fontsize=9) plt.xticks(rotation=45, ha='right') plt.tight_layout() plt.show() except Exception as e: messagebox.showerror("错误", f"绘图时出错: {str(e)}") traceback.print_exc() def plot_stock_trend(self, field, total_label, ylabel, color, apply_filter=True): """通用绘制结存趋势图表""" if self.df is None: messagebox.showwarning("警告", "请先更新数据！") return try: # 数据预处理 if apply_filter: df_filtered = self.df[ (self.df['SMT线体及状态'] != '') & (self.df[field] > 0) ].copy() else: df_filtered = self.df[self.df[field] > 0].copy() df_filtered = df_filtered.drop_duplicates(subset=['批次'], keep='first') df_filtered['释放日期'] = pd.to_datetime( df_filtered['释放日期'], errors='coerce' ).dt.date df_filtered = df_filtered.dropna(subset=['释放日期']) # 分组聚合 self.df_grouped = df_filtered.groupby('释放日期').agg( 总量=(field, 'sum'), 批次数量=('批次', 'nunique') ).reset_index().sort_values('释放日期') # 创建图表 fig, ax1 = plt.subplots(figsize=(12, 7)) # 绘制柱状图 main_plot = ax1.bar( self.df_grouped['释放日期'].astype(str), self.df_grouped['总量'], width=0.6, color=color, label=total_label ) # 创建第二个Y轴 ax2 = ax1.twinx() line_plot, = ax2.plot( self.df_grouped['释放日期'].astype(str), self.df_grouped['批次数量'], color='#D62728', marker='s', linestyle='--', label='批次数量' ) # 设置图表样式 ax1.set_title( f'{ylabel}释放日期趋势\n' f'总数量：{self.df_grouped["总量"].sum():,} | 总批次：{self.df_grouped["批次数量"].sum()}', fontsize=16, fontproperties='SimHei', pad=20 ) ax1.set_xlabel('释放日期', fontproperties='SimHei', labelpad=20) ax1.set_ylabel(ylabel, fontproperties='SimHei', color=color) ax2.set_ylabel('批次数量', fontproperties='SimHei', color='#D62728') # 设置坐标轴颜色和旋转 ax1.set_xticks(range(len(self.df_grouped['释放日期']))) ax1.set_xticklabels(self.df_grouped['释放日期'].astype(str), rotation=45, ha='right') ax1.tick_params(axis='y', colors=color) ax2.tick_params(axis='y', colors='#D62728') ax1.grid(axis='y', linestyle=':', alpha=0.7) # 合并图例 lines = [main_plot, line_plot] labels = [l.get_label() for l in lines] ax1.legend(lines, labels, loc='upper left', prop={'family': 'SimHei'}) # 添加数值标签 for i, (x, y) in enumerate(zip(self.df_grouped['释放日期'], self.df_grouped['批次数量'])): ax2.annotate(f'{y}', (i, y), textcoords="offset points", xytext=(10, -10), ha='center', fontsize=8, color='red') for bar in main_plot: height = bar.get_height() ax1.annotate(f'{int(height):,}', xy=(bar.get_x() + bar.get_width() / 2, height), xytext=(0, 3), textcoords="offset points", ha='center', va='bottom', fontsize=8, color=color) # 渲染图表 self._render_chart(fig) except Exception as e: messagebox.showerror("错误", f"绘图时发生异常：{str(e)}") def show_material_order_view(self): """显示料单提单查询视图""" self.material_order_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "料单提单查询" # 只初始化一次界面内容 if not hasattr(self, '_material_order_initialized'): self._setup_material_order_view() self._material_order_initialized = True def show_delivery_test_view(self): """显示产品信息配置视图""" self.delivery_test_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "产品信息配置" # 只初始化一次界面内容 if not hasattr(self, '_delivery_test_initialized'): self._setup_delivery_test_view() self._delivery_test_initialized = True # 只在首次或无数据时获取，否则直接显示 if hasattr(self, 'delivery_test_df') and self.delivery_test_df is not None and not self.delivery_test_df.empty: self._display_delivery_test_data(self.delivery_test_df) else: self.update_delivery_test_data() def show_batch_cycle_view(self): """显示批次周期视图""" self.batch_cycle_frame.pack(fill=tk.BOTH, expand=True) self.current_view = "批次周期" # 只初始化一次界面内容 if not hasattr(self, '_batch_cycle_initialized'): self._setup_batch_cycle_view() self._batch_cycle_initialized = True # 查询并显示数据 self.update_batch_cycle_data() def _setup_batch_cycle_view(self): """初始化批次周期界面内容（包含日期选择器、阈值输入框和图表按钮）""" # 创建搜索框架 search_frame = ttk.Frame(self.batch_cycle_frame) search_frame.pack(fill=tk.X, padx=5, pady=5) # 添加日期选择器 ttk.Label(search_frame, text="选择日期:").pack(side=tk.LEFT, padx=5) self.date_var = tk.StringVar() self.date_picker = DateEntry( search_frame, textvariable=self.date_var, date_pattern='yyyy-mm-dd', width=12 ) self.date_picker.pack(side=tk.LEFT, padx=5) # 添加在线周期阈值输入框 ttk.Label(search_frame, text="在线周期 >").pack(side=tk.LEFT, padx=5) self.cycle_threshold_var = tk.StringVar(value="8") # 默认值为8 self.cycle_threshold_entry = ttk.Entry( search_frame, textvariable=self.cycle_threshold_var, width=5 ) self.cycle_threshold_entry.pack(side=tk.LEFT, padx=5) # 刷新按钮 ttk.Button(search_frame, text="刷新数据", command=self.update_batch_cycle_data).pack(side=tk.LEFT, padx=5) ttk.Button(search_frame, text="导出Excel", command=self.export_batch_cycle_to_excel).pack(side=tk.LEFT, padx=5) # 图表按钮 ttk.Button(search_frame, text="显示图表", command=self.plot_batch_cycle_by_dispatcher).pack(side=tk.LEFT, padx=5) # 创建表格框架 tree_frame = ttk.Frame(self.batch_cycle_frame) tree_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 定义表格列 columns = [ '日期', '批次', '计划跟踪单', '单板代码', '单板名称', '释放日期', '计划数量', '工艺路径', '部件在线', '备料区', '产品大类', '在线周期', '工厂', '在线比例', '计调员', '周期类别' ] # 设置 Treeview self.batch_cycle_tree = ttk.Treeview(tree_frame, columns=columns, show='headings') for col in columns: self.batch_cycle_tree.heading(col, text=col, anchor='w') self.batch_cycle_tree.column(col, width=100, anchor='w') # 滚动条 vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.batch_cycle_tree.yview) hsb = ttk.Scrollbar(tree_frame, orient="horizontal", command=self.batch_cycle_tree.xview) self.batch_cycle_tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) self.batch_cycle_tree.grid(row=0, column=0, sticky='nsew') vsb.grid(row=0, column=1, sticky='ns') hsb.grid(row=1, column=0, sticky='ew') tree_frame.grid_columnconfigure(0, weight=1) tree_frame.grid_rowconfigure(0, weight=1) # 初始化日期选择器的可用日期 dates = self.batch_cycle.get_available_dates() if dates: self.date_picker.set_date(dates[0]) # 默认选第一个日期 self.update_batch_cycle_data() """初始化批次周期界面内容""" # 在批次周期视图初始化部分添加日期选择器 def init_batch_cycle_ui(self): # 创建批次周期视图 self.batch_cycle_view = QWidget() layout = QVBoxLayout(self.batch_cycle_view) # 创建日期选择区域 date_layout = QHBoxLayout() date_label = QLabel("选择日期:") self.date_combo = QComboBox() self.date_combo.setMinimumWidth(150) # 创建更新按钮 self.update_btn = QPushButton("刷新数据") self.update_btn.setFixedWidth(100) self.update_btn.clicked.connect(self.update_batch_cycle_data) date_layout.addWidget(date_label) date_layout.addWidget(self.date_combo) date_layout.addStretch() date_layout.addWidget(self.update_btn) # 创建表格视图 self.batch_cycle_table = QTableView() self.batch_cycle_table.setSortingEnabled(True) # 添加到主布局 layout.addLayout(date_layout) layout.addWidget(self.batch_cycle_table) # 初始化批次周期数据模型 self.batch_cycle_model = None # 初始化日期选择器 self.init_date_combo() # 默认加载最新日期的数据 self.load_batch_cycle_data() return self.batch_cycle_view def init_date_combo(self): """初始化日期选择器""" # 获取可用日期列表 self.batch_cycle = BatchCycle() dates = self.batch_cycle.get_available_dates() # 添加到下拉框 self.date_combo.clear() self.date_combo.addItems([date.strftime("%Y-%m-%d") for date in dates]) # 默认选中第一个（最新日期） if dates: self.date_combo.setCurrentIndex(0) def load_batch_cycle_data(self, selected_date=None): """加载批次周期数据""" # 如果已有数据且未指定日期，则不刷新 if self.batch_cycle_model and not selected_date: return # 获取选中的日期 if not selected_date: selected_date = self.date_combo.currentText() # 查询数据 df = self.batch_cycle.get_batch_cycle_data(selected_date) # 创建模型并设置数据 self.batch_cycle_model = PandasModel(df) self.batch_cycle_table.setModel(self.batch_cycle_model) # 调整列宽 self.batch_cycle_table.resizeColumnsToContents() def update_batch_cycle_data(self, event=None): selected_date = self.date_picker.get_date().strftime('%Y-%m-%d') try: threshold = int(self.cycle_threshold_var.get()) except ValueError: threshold = 8 df = self.batch_cycle.query_batch_cycle(selected_date, threshold) self._update_batch_cycle_ui(df) def _update_batch_cycle_ui(self, df): # 缓存当前数据供绘图使用 self.batch_cycle_cached_df = df # 如果有 Treeview，更新显示 if hasattr(self, 'batch_cycle_tree'): self.batch_cycle_tree.delete(self.batch_cycle_tree.get_children()) for index, row in df.iterrows(): self.batch_cycle_tree.insert("", "end", values=tuple(row)) def export_batch_cycle_to_excel(self): """导出批次周期数据到Excel""" if not hasattr(self, 'batch_cycle_tree') or not self.batch_cycle_tree.get_children(): messagebox.showwarning("警告", "没有数据可以导出") return try: # 获取保存文件路径 file_path = filedialog.asksaveasfilename( defaultextension=".xlsx", filetypes=[("Excel files", ".xlsx")], initialfile=f"批次周期数据_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" ) if not file_path: return # 收集数据 data = [] columns = [self.batch_cycle_tree.heading(col)["text"] for col in self.batch_cycle_tree["columns"]] data.append(columns) # 添加列标题 for item in self.batch_cycle_tree.get_children(): values = self.batch_cycle_tree.item(item)["values"] data.append(values) # 创建DataFrame并导出 df = pd.DataFrame(data[1:], columns=data[0]) df.to_excel(file_path, index=False) messagebox.showinfo("成功", "数据已成功导出到Excel文件！") except Exception as e: messagebox.showerror("错误", f"导出Excel时出错：{str(e)}") def show_batch_cycle_context_menu(self, event): """显示批次周期表格的右键菜单""" tree = event.widget item = tree.identify_row(event.y) if not item: return # 创建右键菜单 context_menu = tk.Menu(self.root, tearoff=0) context_menu.add_command(label="复制单元格", command=lambda: self.copy_cell(tree, item, tree.identify_column(event.x))) context_menu.add_command(label="复制整行", command=lambda: self.copy_row(tree, item)) context_menu.add_separator() context_menu.add_command(label="全选", command=lambda: self.select_all(tree)) context_menu.add_command(label="取消全选", command=lambda: self.deselect_all(tree)) context_menu.add_command(label="全选并复制", command=lambda: self.select_all_and_copy(tree)) # 显示菜单 context_menu.post(event.x_root, event.y_root) def _setup_delivery_test_view(self): """初始化产品信息配置界面内容""" # 创建搜索框架 search_frame = ttk.Frame(self.delivery_test_frame) search_frame.pack(fill=tk.X, padx=5, pady=5) # 添加搜索控件 ttk.Label(search_frame, text="搜索:").pack(side=tk.LEFT) self.delivery_test_search_var = tk.StringVar() ttk.Entry(search_frame, textvariable=self.delivery_test_search_var, width=30).pack(side=tk.LEFT, padx=5) ttk.Button(search_frame, text="搜索", command=self.search_delivery_test).pack(side=tk.LEFT, padx=5) ttk.Button(search_frame, text="刷新数据", command=self.update_delivery_test_data).pack(side=tk.LEFT, padx=5) ttk.Button(search_frame, text="导出Excel", command=self.export_delivery_test_to_excel).pack(side=tk.LEFT, padx=5) # 创建表格框架 tree_frame = ttk.Frame(self.delivery_test_frame) tree_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 定义表格列 columns = [ '单板代码', '单板名称', '工艺属性', '工厂', 'lastUpdatedBy', 'lastUpdatedDate' ] column_names = { '工厂': '工厂', 'lastUpdatedBy': '更新人', 'lastUpdatedDate': '更新日期' } # 列宽设置 column_widths = { '单板代码': 150, "单板名称": 150, "工艺属性": 100, 'factoryName': 80, 'lastUpdatedBy': 80, 'lastUpdatedDate': 100 } # 创建表格 self.delivery_test_tree = ttk.Treeview(tree_frame, columns=columns, show='headings') # 设置列标题和宽度 for col in columns: display_name = column_names.get(col, col) self.delivery_test_tree.heading(col, text=display_name, anchor='w') width = column_widths.get(col, 100) self.delivery_test_tree.column(col, width=width, anchor='w') # 添加滚动条 vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.delivery_test_tree.yview) hsb = ttk.Scrollbar(tree_frame, orient="horizontal", command=self.delivery_test_tree.xview) self.delivery_test_tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) # 布局 self.delivery_test_tree.grid(row=0, column=0, sticky='nsew') vsb.grid(row=0, column=1, sticky='ns') hsb.grid(row=1, column=0, sticky='ew') # 配置网格权重 tree_frame.grid_columnconfigure(0, weight=1) tree_frame.grid_rowconfigure(0, weight=1) # 添加右键菜单 self.delivery_test_tree.bind('<Button-3>', self.show_delivery_test_context_menu) def update_delivery_test_data(self): """在后台线程中从API获取产品信息配置数据""" # 创建状态指示器 self.safe_status_update("正在后台更新产品信息配置数据...") # 添加进度指示器 progress_frame = ttk.Frame(self.status_bar) progress_frame.pack(side=tk.RIGHT, padx=10) progress = ttk.Progressbar(progress_frame, mode='indeterminate', length=100) progress.pack(side=tk.RIGHT) progress.start() # 更新按钮状态 update_buttons = [] for widget in self.delivery_test_frame.winfo_children(): if isinstance(widget, ttk.Frame): for btn in widget.winfo_children(): if isinstance(btn, ttk.Button) and btn['text'] == "刷新数据": btn['text'] = "更新中..." btn['state'] = 'disabled' update_buttons.append(btn) # 定义后台更新函数 def background_update(): try: # 获取数据 df = DataProcessor.fetch_url6_data() if df is not None and 'factoryName' in df.columns: df.rename(columns={'factoryName': '工厂'}, inplace=True) # 使用after方法在主线程中安全更新UI self.root.after(0, lambda: self._update_delivery_test_ui(df)) except Exception as e: # 使用after方法在主线程中安全显示错误 error_msg = str(e) self.root.after(0, lambda: messagebox.showerror("错误", f"获取产品信息配置数据时出错：{error_msg}")) print(f"获取产品信息配置数据时出错：{error_msg}") traceback.print_exc() finally: # 使用after方法在主线程中安全清理UI self.root.after(0, lambda: self._cleanup_progress(progress_frame, progress, update_buttons)) # 启动后台线程 threading.Thread(target=background_update, daemon=True).start() def _update_delivery_test_ui(self, df): """更新产品信息配置UI""" try: if df is None or df.empty: messagebox.showinfo("提示", "未获取到产品信息配置数据") return # 清空现有数据 for item in self.delivery_test_tree.get_children(): self.delivery_test_tree.delete(item) # 保存数据到实例变量，以便搜索使用 self.delivery_test_df = df # 显示数据 self._display_delivery_test_data(df) # 更新最后更新时间 self.last_update_time = datetime.now() if hasattr(self, 'last_update_label') and self.last_update_label.winfo_exists(): self.last_update_label.config( text=f"最后更新时间：{self.last_update_time.strftime('%Y-%m-%d %H:%M:%S')}" ) messagebox.showinfo("成功", f"成功获取 {len(df)} 条产品信息配置数据") except Exception as e: print(f"更新产品信息配置UI时出错: {str(e)}") traceback.print_exc() def _display_delivery_test_data(self, df, max_rows=100000): """显示产品信息配置数据到表格""" # 限制显示行数，避免过多数据导致界面卡顿 if len(df) > max_rows: messagebox.showinfo("提示", f"数据量较大，仅显示前 {max_rows} 条记录") df = df.head(max_rows) # 清空现有数据 for item in self.delivery_test_tree.get_children(): self.delivery_test_tree.delete(item) # 插入数据 for _, row in df.iterrows(): values = [] for col in self.delivery_test_tree['columns']: val = row.get(col, '') if pd.isna(val): val = '' elif isinstance(val, (datetime, pd.Timestamp)): val = val.strftime('%Y-%m-%d %H:%M:%S') values.append(str(val)) self.delivery_test_tree.insert('', tk.END, values=values) def search_delivery_test(self): """搜索产品信息配置数据""" if not hasattr(self, 'delivery_test_df') or self.delivery_test_df.empty: messagebox.showinfo("提示", "请先获取数据") return search_text = self.delivery_test_search_var.get().strip().lower() if not search_text: # 如果搜索框为空，显示所有数据 self._display_delivery_test_data(self.delivery_test_df) return # 在多个列中搜索 search_columns = ['单板代码', '单板名称', '工艺属性', '工厂', 'lastUpdatedBy', 'lastUpdatedDate'] # 创建搜索条件 mask = pd.Series(False, index=self.delivery_test_df.index) for col in search_columns: if col in self.delivery_test_df.columns: # 将列转换为字符串并转为小写进行搜索 mask |= self.delivery_test_df[col].astype(str).str.lower().str.contains(search_text, na=False) # 应用过滤 filtered_df = self.delivery_test_df[mask] if filtered_df.empty: messagebox.showinfo("提示", "未找到匹配的数据") return # 显示过滤后的数据 self._display_delivery_test_data(filtered_df) messagebox.showinfo("搜索结果", f"找到 {len(filtered_df)} 条匹配记录") def export_delivery_test_to_excel(self): """导出产品信息配置数据到Excel""" if not hasattr(self, 'delivery_test_df') or self.delivery_test_df.empty: messagebox.showinfo("提示", "没有数据可以导出") return try: # 获取保存文件路径 file_path = filedialog.asksaveasfilename( defaultextension=".xlsx", filetypes=[("Excel files", ".xlsx")], initialfile=f"产品信息配置数据_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" ) if not file_path: return # 导出到Excel self.delivery_test_df.to_excel(file_path, index=False) messagebox.showinfo("成功", "数据已成功导出到Excel文件！") except Exception as e: messagebox.showerror("错误", f"导出Excel时出错：{str(e)}") def show_delivery_test_context_menu(self, event): """显示产品信息配置表格的右键菜单""" tree = event.widget item = tree.identify_row(event.y) if not item: return # 创建右键菜单 context_menu = tk.Menu(self.root, tearoff=0) context_menu.add_command(label="复制单元格", command=lambda: self.copy_cell(tree, item, tree.identify_column(event.x))) context_menu.add_command(label="复制整行", command=lambda: self.copy_row(tree, item)) context_menu.add_separator() context_menu.add_command(label="全选", command=lambda: self.select_all(tree)) context_menu.add_command(label="取消全选", command=lambda: self.deselect_all(tree)) context_menu.add_command(label="全选并复制", command=lambda: self.select_all_and_copy(tree)) # 显示菜单 context_menu.post(event.x_root, event.y_root) def _setup_material_order_view(self): """初始化料单提单查询界面内容（支持多行查询）""" search_frame = ttk.Frame(self.material_order_frame) search_frame.pack(fill=tk.X, padx=5, pady=5) # 多行输入框 ttk.Label(search_frame, text="输入单板代码（每行一个）:").pack(side=tk.LEFT) self.material_order_text = tk.Text(search_frame, height=10, width=30) self.material_order_text.pack(side=tk.LEFT, padx=5) self.material_order_text.bind('<KeyRelease>', self.on_material_input) # 绑定输入事件 # 结果表格 tree_frame = ttk.Frame(self.material_order_frame) tree_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 动态列配置 self.material_order_tree = ttk.Treeview(tree_frame) self.material_order_tree.pack(fill=tk.BOTH, expand=True) # 滚动条 vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.material_order_tree.yview) hsb = ttk.Scrollbar(tree_frame, orient="horizontal", command=self.material_order_tree.xview) self.material_order_tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) vsb.pack(side=tk.RIGHT, fill=tk.Y) hsb.pack(side=tk.BOTTOM, fill=tk.X) def on_material_input(self, event=None): """处理多行输入事件（防抖）""" try: # 防抖：输入后300ms再查，期间有新输入则重置计时 if hasattr(self, 'material_input_after_id') and self.material_input_after_id: try: self.root.after_cancel(self.material_input_after_id) except ValueError: # 忽略无效的计时器ID错误 pass if hasattr(self, 'root') and self.root.winfo_exists(): self.material_input_after_id = self.root.after(300, self._do_material_search) else: print("无法设置延迟搜索：根窗口不存在或已销毁") except Exception as e: print(f"处理物料输入时出错: {str(e)}") traceback.print_exc() def _do_material_search(self): """执行物料搜索（防抖后调用）""" try: # 检查组件是否存在 if not hasattr(self, 'material_order_text') or not self.material_order_text.winfo_exists(): print("物料输入框不存在或已销毁") return codes = self.get_input_codes() # 增加：只有codes变化时才查 if hasattr(self, '_last_codes') and self._last_codes == codes: return self._last_codes = codes if codes: # 检查是否有有效的搜索代码 if not any(code.strip() for code in codes): print("没有有效的搜索代码") return # 执行搜索 self.search_material_order(codes) except Exception as e: print(f"执行物料搜索时出错: {str(e)}") traceback.print_exc() # 安全地显示错误消息 if hasattr(self, 'root') and self.root.winfo_exists(): try: messagebox.showerror("错误", f"搜索失败: {str(e)}") except Exception as msg_error: print(f"显示错误消息失败: {str(msg_error)}") def get_input_codes(self): """获取并验证输入代码""" try: if not hasattr(self, 'material_order_text') or not self.material_order_text.winfo_exists(): print("物料输入框未初始化或已销毁") return None input_text = self.material_order_text.get("1.0", tk.END).strip() codes = [code.strip() for code in input_text.splitlines() if code.strip()] return codes if codes else None except Exception as e: print(f"获取输入代码时出错: {str(e)}") traceback.print_exc() return None # 在material_order查询方法中 from sqlalchemy import text def search_material_order(self, codes): try: # 生成近7天日期（含今天） end_date = datetime.now().date() start_date = end_date - timedelta(days=6) # 修正为6天前，共7天 # 打印调试信息 print(f"查询日期范围: {start_date} 至 {end_date}") print(f"查询单板代码: {codes}") # 构建参数化查询 with self.engine.connect() as conn: sql = text(""" SELECT 日期, 单板代码, 单板名称, SUM(提单) AS 提单总数 FROM query WHERE 单板代码 IN :codes AND 日期 BETWEEN :start_date AND :end_date GROUP BY 日期, 单板代码, 单板名称 ORDER BY 日期 """) # 使用正确的参数格式 result = conn.execute(sql, { "codes": tuple(codes), # 转换为元组 "start_date": start_date, "end_date": end_date }) df = pd.DataFrame(result.fetchall(), columns=result.keys()) print("查询结果样例:\n", df.head()) # 调试输出 # 处理空数据情况 if df.empty: print("数据库查询返回空结果") self.material_order_tree.delete(self.material_order_tree.get_children()) messagebox.showinfo("提示", "未找到相关数据") return # 创建透视表（保证日期列顺序） date_columns = sorted(df['日期'].unique()) df_grouped = df.groupby(['单板名称', '单板代码', '日期'], as_index=False)['提单总数'].sum() pivot = df_grouped.pivot_table( index=['单板名称', '单板代码'], columns='日期', values='提单总数', aggfunc='sum', fill_value=0 ).reindex(columns=date_columns, fill_value=0) # 更新Treeview self.update_material_tree(pivot, date_columns) except Exception as e: messagebox.showerror("错误", f"查询失败: {str(e)}") logging.error(traceback.format_exc()) def update_material_tree(self, pivot_df, date_columns): """更新表格显示""" # 清空旧数据 self.material_order_tree.delete(self.material_order_tree.get_children()) # 配置列 columns = ['单板名称', '单板代码'] + date_columns self.material_order_tree.config(columns=columns) # 设置列属性，宽度加大 for col in columns: if col in ['单板名称', '单板代码']: width = 180 else: width = 100 # 日期列宽度 self.material_order_tree.heading(col, text=col, anchor='w') self.material_order_tree.column(col, width=width, anchor='w', minwidth=80, stretch=True) # 插入数据 for idx, row in pivot_df.reset_index().iterrows(): values = [row['单板名称'], row['单板代码']] + [row[col] for col in date_columns] self.material_order_tree.insert('', 'end', values=values) def on_closing(): """处理窗口关闭事件""" if messagebox.askokcancel("退出", "确定要退出程序吗？"): sys.exit() # 确保程序退出 def get_base_path(): """获取基础路径""" try: if getattr(sys, 'frozen', False): # PyInstaller创建的临时文件夹 base_path = sys.executable else: # 正常运行时的脚本位置 base_path = os.path.dirname(os.path.abspath(file)) return base_path except Exception as e: logging.error(f"获取基础路径失败: {str(e)}") return os.path.dirname(os.path.abspath(file)) class DailyOutputView: """产出计划表视图""" def init(self, parent): self.parent = parent self.frame = ttk.Frame(parent.content_frame) self.cache_file = os.path.join(os.path.dirname(os.path.abspath(file)), 'daily_output_cache.pkl') self.last_update_time = None self.cache_expiry_hours = 1000 # 缓存过期时间 self._setup_ui() # ✅ 先初始化UI组件 self.load_cached_data() # ✅ 再加载缓存数据 def save_cached_data(self): """将当前数据保存到缓存文件""" try: with open(self.cache_file, 'wb') as f: pickle.dump({ 'raw_data': self.raw_data, 'last_update_time': self.last_update_time, 'selected_factories': self.parent.selected_factories }, f) print(f"缓存数据已保存到: {self.cache_file}") return True except Exception as e: print(f"保存缓存失败: {str(e)}") return False def show_context_menu(self, event): """显示右键菜单""" tree = event.widget item = tree.identify_row(event.y) column = tree.identify_column(event.x) # 创建右键菜单 context_menu = tk.Menu(self.frame, tearoff=0) # 添加菜单项 if item: # 如果点击在某一行上 context_menu.add_command(label="复制单元格", command=lambda: self.copy_cell(tree, item, column)) context_menu.add_command(label="复制整行", command=lambda: self.copy_row(tree, item)) context_menu.add_separator() # 添加列操作 if column != "#0": # 不是序号列 context_menu.add_command(label="复制整列", command=lambda: self.copy_column(tree, column)) context_menu.add_separator() # 添加全局操作 context_menu.add_command(label="全选", command=lambda: self.select_all(tree)) context_menu.add_command(label="取消全选", command=lambda: self.deselect_all(tree)) context_menu.add_command(label="全选并复制", command=lambda: self.select_all_and_copy(tree)) context_menu.add_separator() context_menu.add_command(label="导出到Excel", command=self.export_to_excel) # 显示菜单 context_menu.post(event.x_root, event.y_root) def _setup_ui(self): """初始化UI""" # 创建主框架 self.frame = ttk.Frame(self.parent.content_frame) # 筛选控件框架 filter_frame = ttk.Frame(self.frame) filter_frame.pack(fill=tk.X, padx=5, pady=5) # 工厂筛选 ttk.Label(filter_frame, text="工厂:").pack(side=tk.LEFT) self.factory_var = tk.StringVar() self.factory_cb = ttk.Combobox(filter_frame, textvariable=self.factory_var, state="readonly") self.factory_cb.pack(side=tk.LEFT, padx=5) # 线体筛选 ttk.Label(filter_frame, text="线体:").pack(side=tk.LEFT) self.line_var = tk.StringVar() self.line_cb = ttk.Combobox(filter_frame, textvariable=self.line_var, state="readonly") self.line_cb.pack(side=tk.LEFT, padx=5) # 产品大类筛选 ttk.Label(filter_frame, text="产品大类:").pack(side=tk.LEFT) self.product_var = tk.StringVar() self.product_cb = ttk.Combobox(filter_frame, textvariable=self.product_var, state="readonly") self.product_cb.pack(side=tk.LEFT, padx=5) # 责任人筛选 ttk.Label(filter_frame, text="责任人:").pack(side=tk.LEFT) self.responsible_var = tk.StringVar() self.responsible_cb = ttk.Combobox(filter_frame, textvariable=self.responsible_var, state="readonly") self.responsible_cb.pack(side=tk.LEFT, padx=5) # 创建 Treeview 控件 base_columns = ["批次", "线体", "工厂", "产品大类", "单板代码", "单板名称", "计划数量", "剩余数量", "计划结束时间", "责任人"] # 生成未来7天日期列（只显示月-日） date_columns = [(datetime.now() + timedelta(days=i)).strftime('%m-%d') for i in range(7)] columns = base_columns + date_columns # 初始化 Treeview self.tree = ttk.Treeview(self.frame, columns=columns, show='headings') # 设置列宽 column_widths = { "批次": 60, "线体": 40, "工厂": 40, "产品大类": 60, "单板代码": 150, "单板名称": 200, "计划数量": 60, "剩余数量": 60, "计划结束时间": 100, "责任人": 60 } # 设置列标题和宽度 for col in columns: self.tree.heading(col, text=col) width = column_widths.get(col, 100) self.tree.column(col, width=width, anchor='center') # 设置居中显示 # 添加滚动条 vsb = ttk.Scrollbar(self.frame, orient="vertical", command=self.tree.yview) hsb = ttk.Scrollbar(self.frame, orient="horizontal", command=self.tree.xview) self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) # 创建主布局框架 main_frame = ttk.Frame(self.frame) main_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # 顶部按钮工具栏 toolbar = ttk.Frame(main_frame) toolbar.pack(fill=tk.X, pady=(0, 5)) ttk.Button(toolbar, text="刷新数据", command=self.update_data).pack(side=tk.LEFT, padx=2) ttk.Button(toolbar, text="导出Excel", command=self.export_to_excel).pack(side=tk.LEFT, padx=2) ttk.Button(toolbar, text="应用筛选", command=self.apply_filters).pack(side=tk.LEFT, padx=2) # 创建专门用于 tree 的 Frame tree_frame = ttk.Frame(main_frame) tree_frame.pack(fill=tk.BOTH, expand=True) # 使用 pack 布局放置 tree 和滚动条 self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) vsb.pack(side=tk.RIGHT, fill=tk.Y) hsb.pack(side=tk.BOTTOM, fill=tk.X) # 设置表格显示行数 self.tree.configure(height=25) # 绑定右键菜单 self.tree.bind('<Button-3>', self.show_context_menu) def load_cached_data(self): """从本地文件加载缓存的数据""" try: if os.path.exists(self.cache_file): with open(self.cache_file, 'rb') as f: cache_data = pickle.load(f) # 检查缓存是否过期 last_update = cache_data.get('last_update_time') if last_update and datetime.now() - last_update > timedelta(hours=self.cache_expiry_hours): print("产出计划表缓存数据已过期") return False # 验证数据完整性 raw_data = cache_data.get('raw_data') if raw_data is None or raw_data.empty: print("产出计划表缓存数据为空") return False self.raw_data = raw_data self.last_update_time = last_update # 更新UI self._update_ui_from_cache() print(f"从缓存加载产出计划表数据成功，数据行数: {len(self.raw_data)}, 上次更新时间: {self.last_update_time}") return True return False except Exception as e: print(f"加载产出计划表缓存数据时出错: {str(e)}") traceback.print_exc() return False def _update_ui_from_cache(self): """从缓存数据更新UI""" try: if not hasattr(self, 'raw_data') or self.raw_data.empty: return # 更新筛选下拉框 if '工厂' in self.raw_data.columns: factories = sorted(self.raw_data['工厂'].unique()) self.factory_cb['values'] = ['全部'] + list(factories) self.factory_cb.current(0) if '线体' in self.raw_data.columns: lines = sorted(self.raw_data['线体'].dropna().unique()) self.line_cb['values'] = ['全部'] + list(lines) self.line_cb.current(0) if '产品大类' in self.raw_data.columns: product_types = sorted(self.raw_data['产品大类'].unique()) self.product_cb['values'] = ['全部'] + list(product_types) self.product_cb.current(0) if '责任人' in self.raw_data.columns: responsible_persons = sorted(self.raw_data['责任人'].dropna().unique()) self.responsible_cb['values'] = ['全部'] + list(responsible_persons) self.responsible_cb.current(0) # 应用当前筛选条件 self.apply_filters() # 更新状态信息 if hasattr(self.parent, 'safe_status_update'): self.parent.safe_status_update(f"已加载产出计划表缓存数据（{self.last_update_time.strftime('%Y-%m-%d %H:%M:%S')}）") except Exception as e: print(f"从缓存更新UI时出错: {str(e)}") traceback.print_exc() def update_data(self): """在后台线程中从日产出目标获取数据并更新视图""" # ✅ 强制忽略缓存，从数据库重新获取数据 if hasattr(self, 'raw_data'): del self.raw_data # 删除旧缓存数据 # 创建状态指示器 if hasattr(self.parent, 'safe_status_update'): self.parent.safe_status_update("正在后台更新产出计划表数据...") # 添加进度指示器 progress_frame = ttk.Frame(self.frame) progress_frame.pack(side=tk.TOP, padx=10, pady=5) progress = ttk.Progressbar(progress_frame, mode='indeterminate', length=100) progress.pack(side=tk.TOP) progress.start() # 更新按钮状态 update_buttons = [] for widget in self.frame.winfo_children(): if isinstance(widget, ttk.Frame): for btn in widget.winfo_children(): if isinstance(btn, ttk.Button) and btn['text'] == "更新数据": btn['text'] = "更新中..." btn['state'] = 'disabled' update_buttons.append(btn) # 定义后台更新函数 def background_update(): try: from 日产出目标 import main as get_daily_output data = get_daily_output() # 使用after方法在主线程中安全更新UI self.parent.root.after(0, lambda: self._process_output_data(data, progress_frame, progress, update_buttons)) except Exception as e: # 使用after方法在主线程中安全显示错误 error_msg = str(e) self.parent.root.after(0, lambda: self._show_update_error(error_msg, progress_frame, progress, update_buttons)) # 启动后台线程 threading.Thread(target=background_update, daemon=True).start() def show_daily_output_view(self): """显示产出计划表视图""" if hasattr(self, 'daily_output_view'): self.daily_output_view.frame.pack(fill=tk.BOTH, expand=True) self.current_view = "产出计划表" # 检查是否有数据，如果没有则尝试加载缓存或更新数据 if not hasattr(self.daily_output_view, 'raw_data') or self.daily_output_view.raw_data.empty: # 尝试加载缓存数据，如果失败则更新数据 if not self.daily_output_view.load_cached_data(): # 显示加载对话框 loading_dialog = SimpleLoadingDialog(self.root) self.root.update() # 在后台线程中加载数据 def load_data_thread(): try: self.daily_output_view.update_data() finally: # 确保对话框被销毁 self.root.after(0, loading_dialog.destroy) threading.Thread(target=load_data_thread, daemon=True).start() def _process_output_data(self, data, progress_frame, progress, update_buttons): """处理产出计划表数据并更新UI""" try: if not data: messagebox.showwarning("警告", "未获取到日产出目标数据") self._cleanup_ui(progress_frame, progress, update_buttons) return # 转换为DataFrame并确保产品大类列存在 df = pd.DataFrame(data) if '产品大类' not in df.columns: df['产品大类'] = '' # 确保日期格式正确 if '计划结束时间' in df.columns: df['计划结束时间'] = pd.to_datetime(df['计划结束时间']) # 更新筛选下拉框 if '工厂' in df.columns: factories = sorted(df['工厂'].unique()) self.factory_cb['values'] = ['全部'] + list(factories) self.factory_cb.current(0) # 添加线体选项 if '线体' in df.columns: lines = sorted(df['线体'].dropna().unique()) self.line_cb['values'] = ['全部'] + list(lines) self.line_cb.current(0) if '产品大类' in df.columns: product_types = sorted(df['产品大类'].unique()) self.product_cb['values'] = ['全部'] + list(product_types) self.product_cb.current(0) # 获取part_category_dict用于责任人分配 try: with self.parent.engine.connect() as connection: query = "SELECT 工厂, 单板代码, 计调员 FROM part_category" part_category_df = pd.read_sql(query, con=connection) # 提取单板代码前12位 part_category_df['单板代码12位'] = part_category_df['单板代码'].str[:12] # 创建字典 part_category_dict = part_category_df.set_index(['工厂', '单板代码12位'])['计调员'].to_dict() # 为每行分配责任人 def assign_responsible_with_db(row): # 先从数据库获取 try: board_code_12 = row.get('单板代码', '')[:12] if row.get('单板代码') else '' factory = row.get('工厂', '') if (factory, board_code_12) in part_category_dict: return part_category_dict[(factory, board_code_12)] except: pass # 如果数据库中没有，使用规则分配 return self.assign_responsible_person(row) df['责任人'] = df.apply(assign_responsible_with_db, axis=1) except Exception as e: print(f"获取责任人数据失败: {str(e)}") # 如果数据库查询失败，使用规则分配 df['责任人'] = df.apply(self.assign_responsible_person, axis=1) # 更新责任人下拉框 responsible_persons = sorted(df['责任人'].dropna().unique()) self.responsible_cb['values'] = ['全部'] + list(responsible_persons) self.responsible_cb.current(0) # 保存原始数据 self.raw_data = df # 应用当前筛选条件 self.apply_filters() # 保存数据到缓存 self.save_cached_data() # 清理UI self._cleanup_ui(progress_frame, progress, update_buttons) messagebox.showinfo("成功", f"成功获取 {len(df)} 条日产出目标数据") except Exception as e: self._show_update_error(str(e), progress_frame, progress, update_buttons) def _show_update_error(self, error_msg, progress_frame, progress, update_buttons): """显示更新错误""" messagebox.showerror("错误", f"更新产出计划表数据失败: {error_msg}") self._cleanup_ui(progress_frame, progress, update_buttons) def _cleanup_ui(self, progress_frame, progress, update_buttons): """清理UI元素""" try: # 停止进度条并移除 if progress.winfo_exists(): progress.stop() if progress_frame.winfo_exists(): progress_frame.destroy() # 恢复按钮状态 for btn in update_buttons: if btn.winfo_exists(): btn['text'] = "刷新数据" btn['state'] = 'normal' except Exception as e: print(f"清理UI元素时出错: {str(e)}") def assign_responsible_person(self, row): """分配责任人""" # 规则 1: 如果工厂是 "河源"，且工艺路径包含 "背板"，则返回 "徐捷如" if row.get('工厂', '') == "河源" and (row.get('工艺路径', '') is not None) and "背板" in row.get('工艺路径', ''): return "徐捷如" # 规则 2: 如果工厂是 "河源"，且单板名称匹配正则表达式，则返回 "何东" elif row.get('工厂', '') == "河源" and re.search(r'^PPC|^PA|^PD|^PN|^ZXSDR BS8800 C100 PSUB0', row.get('单板名称', '')): return "何东" # 规则 3: 如果工厂是 "河源"，且产品大类在 ["ES", "MSR(R10)", "CLS", "IC", "MSTP"] 中，则返回 "钟慧珊" elif row.get('工厂', '') == "河源" and row.get('产品大类', '') in ["ES", "MSR(R10)", "CLS", "IC", "MSTP"]: return "钟慧珊" # 规则 4: 如果工厂是 "河源"，且产品大类是 "5G-WS"，则返回 "无线" elif row.get('工厂', '') == "河源" and row.get('产品大类', '') in ["5G-WS"]: return "无线" # 规则 5: 如果工厂是 "河源"，且产品大类在 ["CR", "CSR(R10)", "DCCS", "IPRAN-C", "ECSR"] 中，则返回 "孙树冬" elif row.get('工厂', '') == "河源" and row.get('产品大类', '') in ["CR", "CSR(R10)", "DCCS", "IPRAN-C", "ECSR"]: return "孙树冬" # 规则 6: 如果工厂是 "河源"，且产品大类在 ["5G_CMxU", "5G_COLT", "5G_CONT", "A10", "CHS", "CPE", "DSL", "EODN", "J10", "ONT", "POL", "PON_MxU", "PON_OLT", "PONG", "POND", "OTN", "ITPS", "MSAP", "CLOUD", "CT", "ESI", "IAD", "iCDN", "uSmartView", "VCS", "VCT"] 中，则返回 "肖应仲" elif row.get('工厂', '') == "河源" and row.get('产品大类', '') in ["5G_CMxU", "5G_COLT", "5G_CONT", "A10", "CHS", "CPE", "DSL", "EODN", "J10", "ONT", "POL", "PON_MxU", "PON_OLT", "PONG", "POND", "OTN", "ITPS", "MSAP", "CLOUD", "CT", "ESI", "IAD", "iCDN", "uSmartView", "VCS", "VCT"]: return "肖应仲" # 规则 7: 如果工厂是 "河源"，且产品大类在 ["C-SPN", "IPRAN-A", "SPN", "D-SPN", "IPRAN", "IPTN"] 中，则返回 "徐捷如" elif row.get('工厂', '') == "河源" and row.get('产品大类', '') in ["C-SPN", "IPRAN-A", "SPN", "D-SPN", "IPRAN", "IPTN"]: return "徐捷如" # 规则 8: 如果工厂是 "河源"，且产品大类在 ["CDCG", "DC_P", "DPS", "POWER", "PS_S", "SSE", "TSE", "3GPLAT", "DC_S", "SE"] 中，则返回 "祁思佳" elif row.get('工厂', '') == "河源" and row.get('产品大类', '') in ["CDCG", "DC_P", "DPS", "POWER", "PS_S", "SSE", "TSE", "3GPLAT", "DC_S", "SE"]: return "祁思佳" # 规则 9: 如果工厂是 "河源"，且产品大类是 "RRS"，则返回 "何东" elif row.get('工厂', '') == "河源" and row.get('产品大类', '') in ["RRS"]: return "何东" # 规则 10: 如果工厂是 "深圳"，且产品大类在 ["ITPS", "RSS"] 中，则返回 "江启一" elif row.get('工厂', '') == "深圳" and row.get('产品大类', '') in ["ITPS", "RSS"]: return "江启一" # 规则 11: 如果工厂是 "深圳"，且产品大类在 ["OTN", "WDM"] 中，则返回 "杨港会" elif row.get('工厂', '') == "深圳" and row.get('产品大类', '') in ["OTN", "WDM"]: return "杨港会" # 规则 12: 如果工厂是 "深圳"，且产品大类是 "CP"，则返回 "周旭升" elif row.get('工厂', '') == "深圳" and row.get('产品大类', '') in ["CP"]: return "周旭升" # 如果没有匹配到任何规则，则返回 "未分配" else: return "未分配" def show_context_menu(self, event): """显示右键菜单""" tree = event.widget region = tree.identify_region(event.x, event.y) if region == "cell" or region == "heading": item = tree.identify_row(event.y) column = tree.identify_column(event.x) # 创建右键菜单 context_menu = tk.Menu(self.frame, tearoff=0) if region == "cell": context_menu.add_command(label="复制单元格", command=lambda: self.copy_cell(tree, item, column)) context_menu.add_command(label="复制整行", command=lambda: self.copy_row(tree, item)) context_menu.add_command(label="复制整列", command=lambda: self.copy_column(tree, column)) context_menu.add_separator() context_menu.add_command(label="全选", command=lambda: self.select_all(tree)) context_menu.add_command(label="取消全选", command=lambda: self.deselect_all(tree)) context_menu.add_command(label="全选并复制", command=lambda: self.select_all_and_copy(tree)) context_menu.post(event.x_root, event.y_root) def copy_cell(self, tree, item, column): """复制单元格内容""" col_id = int(column[1:]) - 1 value = tree.item(item)['values'][col_id] self.parent.root.clipboard_clear() self.parent.root.clipboard_append(str(value)) messagebox.showinfo("提示", "已复制单元格内容到剪贴板") def copy_row(self, tree, item): """复制整行内容""" values = tree.item(item)['values'] copy_text = '\t'.join(str(v) for v in values) self.parent.root.clipboard_clear() self.parent.root.clipboard_append(copy_text) messagebox.showinfo("提示", "已复制整行内容到剪贴板") def copy_column(self, tree, column): """复制整列内容""" col_id = int(column[1:]) - 1 values = [] for item in tree.get_children(): value = tree.item(item)['values'][col_id] values.append(str(value)) copy_text = '\n'.join(values) self.parent.root.clipboard_clear() self.parent.root.clipboard_append(copy_text) messagebox.showinfo("提示", "已复制整列内容到剪贴板") def select_all(self, tree): """全选功能""" for item in tree.get_children(): tree.selection_add(item) tree.focus_set() # 设置焦点到 Treeview 上 def deselect_all(self, tree): """取消全选功能""" tree.selection_remove(tree.selection()) def select_all_and_copy(self, tree): """全选并复制功能""" self.select_all(tree) self.copy_all(tree) def copy_all(self, tree): """复制所有内容到剪贴板""" data = [] columns = [tree.heading(col)["text"] for col in tree["columns"]] data.append('\t'.join(columns)) # 添加列标题 for item in tree.get_children(): values = tree.item(item)['values'] data.append('\t'.join(str(v) for v in values)) copy_text = '\n'.join(data) self.parent.root.clipboard_clear() self.parent.root.clipboard_append(copy_text) messagebox.showinfo("提示", "已复制所有内容到剪贴板") def edit_responsible(self, item): """编辑责任人""" # 获取当前责任人 current_responsible = self.tree.set(item, "责任人") # 创建编辑对话框 dialog = tk.Toplevel(self.parent.root) dialog.title("编辑责任人") dialog.geometry("300x150") dialog.transient(self.parent.root) dialog.grab_set() # 居中显示 dialog.update_idletasks() width = dialog.winfo_width() height = dialog.winfo_height() x = (dialog.winfo_screenwidth() // 2) - (width // 2) y = (dialog.winfo_screenheight() // 2) - (height // 2) dialog.geometry(f'+{x}+{y}') # 创建下拉框 ttk.Label(dialog, text="选择责任人:").pack(pady=(20, 5)) responsible_var = tk.StringVar(value=current_responsible) combo = ttk.Combobox(dialog, textvariable=responsible_var, values=self.responsible_options, width=20) combo.pack(pady=5) # 保存按钮 def save_responsible(): new_responsible = responsible_var.get() self.tree.set(item, "责任人", new_responsible) dialog.destroy() ttk.Button(dialog, text="保存", command=save_responsible).pack(pady=10) def calculate_daily_output(self, row): """计算每日入库数量""" # 获取剩余数量 remaining = row.get('剩余数量', row.get('计划数量', 0)) if remaining <= 0: return [0] 7 # 获取计划开始和结束时间 start_time_str = row.get('计划开始时间', '') end_date_str = row.get('计划结束时间', '') if not end_date_str or pd.isna(end_date_str): return [0] * 7 try: # 如果没有计划开始时间，使用当前时间 if not start_time_str or pd.isna(start_time_str): start_time = pd.Timestamp.now() else: start_time = pd.to_datetime(start_time_str) end_time = pd.to_datetime(end_date_str) except: return [0] * 7 # 如果开始时间晚于结束时间，返回全0 if start_time > end_time: return [0] * 7 # 计算总计划时间（秒） total_seconds = (end_time - start_time).total_seconds() if total_seconds <= 0: return [0] * 7 # 获取未来7天的日期对象 today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) date_objects = [today + timedelta(days=i) for i in range(7)] # 每天结束时间（23:59:59） date_end_times = [(date + timedelta(days=1) - timedelta(seconds=1)) for date in date_objects] # 计算每日数量 daily_output = [] # 先计算每天的原始产出数量 for i, date in enumerate(date_objects): # 当天开始时间 day_start = date if i == 0 else date.replace(hour=0, minute=0, second=0) # 如果是第一天且计划开始时间在当天，使用计划开始时间 if i == 0 and start_time.date() == date.date() and start_time > day_start: day_start = start_time # 当天结束时间 day_end = date_end_times[i] # 如果计划结束时间在当天，使用计划结束时间 if end_time.date() == date.date() and end_time < day_end: day_end = end_time # 如果当天不在计划范围内 if day_end < start_time or day_start > end_time: daily_output.append(0) continue # 计算当天有效工作时间（秒） day_seconds = (day_end - day_start).total_seconds() # 计算当天时间占比 day_ratio = day_seconds / total_seconds # 计算当天产出数量 day_qty = round(remaining * day_ratio) daily_output.append(day_qty) # 确保总和不超过剩余数量 total_output = sum(daily_output) if total_output > remaining: # 如果超过，按比例缩减每天的产出 scale_factor = remaining / total_output if total_output > 0 else 0 daily_output = [round(qty * scale_factor) for qty in daily_output] # 处理舍入误差，确保总和等于剩余数量 total_adjusted = sum(daily_output) if total_adjusted < remaining: # 将差值添加到最后一个非零日期 for i in range(len(daily_output) - 1, -1, -1): if daily_output[i] > 0: daily_output[i] += remaining - total_adjusted break return daily_output def apply_filters(self): """应用筛选条件""" if not hasattr(self, 'raw_data'): return # 清空现有数据 for item in self.tree.get_children(): self.tree.delete(item) # 应用筛选 df = self.raw_data.copy() # 工厂筛选 factory = self.factory_var.get() if factory and factory != '全部': df = df[df['工厂'] == factory] # 线体筛选 line = self.line_var.get() if line and line != '全部': df = df[df['线体'] == line] # 产品大类筛选 (修正重复代码) product = self.product_var.get() if product and product != '全部': df = df[df['产品大类'] == product] # 获取part_category_dict try: with self.parent.engine.connect() as connection: query = "SELECT 工厂, 单板代码, 计调员 FROM part_category" part_category_df = pd.read_sql(query, con=connection) # 提取单板代码前12位 part_category_df['单板代码12位'] = part_category_df['单板代码'].str[:12] # 创建字典 part_category_dict = part_category_df.set_index(['工厂', '单板代码12位'])['计调员'].to_dict() except Exception as e: print(f"获取part_category_dict失败: {str(e)}") part_category_dict = {} # 责任人筛选 - 确保对所有工厂都有效 (修正缩进) responsible = self.responsible_var.get() if responsible and responsible != '全部': # 直接按责任人名称筛选，不考虑工厂 df['责任人'] = df['责任人'].astype(str).str.strip() df = df[df['责任人'].str.strip() == responsible.strip()] print(f"责任人筛选结果: {len(df)}条记录") # 添加未来7天日期列和日期对象 today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) date_columns = [(today + timedelta(days=i)).strftime('%m-%d') for i in range(7)] date_objects = [today + timedelta(days=i) for i in range(7)] # 添加每天结束时间（23:59:59） date_end_times = [(date + timedelta(days=1) - timedelta(seconds=1)) for date in date_objects] # 直接显示所有数据，不按线体分组 for i, (_, row) in enumerate(df.iterrows()): # 从part_category_dict获取责任人 responsible = "" try: # 提取单板代码前12位 board_code_12 = row.get('单板代码', '')[:12] if row.get('单板代码') else '' factory = row.get('工厂', '') # 尝试从字典获取 if (factory, board_code_12) in part_category_dict: responsible = part_category_dict[(factory, board_code_12)] except Exception as e: print(f"从part_category_dict获取责任人失败: {str(e)}") # 如果没有找到，则使用assign_responsible_person if not responsible: responsible = self.assign_responsible_person(row) # 计算每日入库数量 daily_output = self.calculate_daily_output(row) values = [ row.get('批次', ''), row.get('线体', ''), row.get('工厂', ''), row.get('产品大类', ''), row.get('单板代码', ''), row.get('单板名称', ''), row.get('计划数量', 0), row.get('剩余数量', 0), row.get('计划结束时间', ''), responsible ] + daily_output # 格式化日期 if isinstance(values[8], pd.Timestamp): values[8] = values[8].strftime('%Y-%m-%d') item = self.tree.insert("", tk.END, text=str(i+1), values=values) # 设置责任人值 self.tree.set(item, "责任人", responsible) # 绑定双击事件来编辑责任人 self.tree.tag_bind(item, '<Double-1>', lambda e, item=item: self.edit_responsible(item)) def export_to_excel(self): """导出数据到Excel""" if not self.tree.get_children(): messagebox.showwarning("警告", "没有数据可以导出") return try: # 获取保存路径 file_path = filedialog.asksaveasfilename( defaultextension=".xlsx", filetypes=[("Excel文件", "*.xlsx")], initialfile=f"产出计划表_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" ) if not file_path: return # 收集所有数据 data = [] columns = ["序号"] + list(self.tree["columns"]) data.append(columns) for idx, item in enumerate(self.tree.get_children()): values = [str(idx+1)] + list(self.tree.item(item, 'values')) data.append(values) # 创建DataFrame并导出 df = pd.DataFrame(data[1:], columns=data[0]) # 格式化数值列 numeric_cols = ["计划数量", "剩余数量"] for col in numeric_cols: if col in df.columns: df[col] = df[col].apply(lambda x: int(x) if pd.notna(x) else 0) df.to_excel(file_path, index=False) messagebox.showinfo("成功", f"数据已成功导出到:\n{file_path}") except Exception as e: messagebox.showerror("错误", f"导出失败: {str(e)}") def main(): try: print("程序启动中...") check_and_update_version() print("版本检查完成") app = DataViewer() # 删除 root 参数 print("DataViewer实例创建完成") # 设置关闭窗口事件处理 app.root.protocol("WM_DELETE_WINDOW", on_closing) print("准备进入主循环") app.run() print("主循环结束") except Exception as e: print(f"程序运行出错: {e}") traceback.print_exc() sys.exit() if name == "main": main()

<think>我们正在处理一个复杂的桌面应用程序，它涉及多个视图、数据获取、缓存管理、权限控制等功能......

images（c2691+c3640+c7200）.zip

CISCO-c7200.zip

c7200 iso文件

GNS3路由c7200镜像文件

思科c7200.bin镜像文件

GNS3 IOS镜像 C2691

c7200-adventerprisek9-mz.152-4.S7.image

c7200-adventerprisek9_sna-mz.150-1.M.bin

c7200-advipservicesk9-mz.124-9.T1.bin

思科 cisco ios 镜像 c2691-advsecurityk9-mz.124-11.T2.bin

unzip-c2691-advsecurityk9-mz.124-11.T2.rar

C3640,C3745,C7200 IOS IMAGE

GNS3-思科c3640/c3745/c7200路由器IOS

Cisco c3640 c3745 c7200路由器IOS

unzip-c3640-js-mz.124-10.rar

c7200-adventerprisek9-mz.124-11T.zip

c2600,c2691,c3620,c3660ios

c2691-advsecurityk9-mz.124-11.T2

GNS3模拟器C7200镜像

c7200 + GNS3

GNS3 全套IOS c1720-c3640

Router c7200 IOS

C7200IMAGE

cisco 路由器 3640.3745.7200 ios镜像

Cisco 路由器3640镜像文件

C3745和C7200GNS3兼容IOS文件

c7200 bin文件

gns3 帧中继实验配置工程文件 包含2691 ios 镜像文件

cisco 2691 IOS

unzip-c3640-ik9o3s-mz.124-10.rar

c7200-adventerprisek9-mz.124-24.T5.bin

MySQL详细安装、配置过程，多图，详解

最新资源

gns3 帧中继实验配置工程文件包含2691 ios 镜像文件