#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 完整的座位分配系统 包含文件校验、智能分配算法和日志输出功能 支持1-10人连坐需求，能够处理不连续座位 """ import sys import os from pathlib import Path import datetime import pandas as pd import numpy as np def check_dependencies(): """检查并安装必要的依赖包""" required_packages = { 'pandas': 'pandas>=1.3.0', 'numpy': 'numpy>=1.20.0', 'openpyxl': 'openpyxl>=3.0.0' } missing_packages = [] print("检查依赖包...") for package_name, package_spec in required_packages.items(): try: __import__(package_name) print(f"✅ {package_name} 已安装") except ImportError: missing_packages.append(package_spec) print(f"❌ {package_name} 未安装") if missing_packages: print(f"\n发现缺失依赖: {', '.join(missing_packages)}") print("正在尝试自动安装...") try: import subprocess for package in missing_packages: print(f"安装 {package}...") result = subprocess.run([sys.executable, '-m', 'pip', 'install', package], capture_output=True, text=True) if result.returncode == 0: print(f"✅ {package} 安装成功") else: print(f"❌ {package} 安装失败: {result.stderr}") return False except Exception as e: print(f"❌ 自动安装失败: {e}") print("\n请手动安装以下依赖包:") for package in missing_packages: print(f" pip install {package}") input("安装完成后按Enter键继续...") return False return True def auto_detect_files(): """自动检测和识别Excel文件""" print("\n=== 自动文件识别系统 ===") # 扫描当前目录下的所有xlsx文件 current_dir = Path('.') xlsx_files = list(current_dir.glob('*.xlsx')) # 过滤掉输出文件和示例文件 exclude_patterns = ['座位信息_最终分配', '最终座位分配日志', '_示例', '_temp', '_backup'] xlsx_files = [f for f in xlsx_files if not any(pattern in f.name for pattern in exclude_patterns)] print(f"发现 {len(xlsx_files)} 个Excel文件:") for i, file_path in enumerate(xlsx_files, 1): file_size = file_path.stat().st_size / 1024 # KB print(f" {i}. {file_path.name} ({file_size:.1f} KB)") if len(xlsx_files) == 0: print("\n❌ 未找到Excel文件") print("请确保当前目录下有Excel数据文件") return None, None if len(xlsx_files) > 2: print(f"\n⚠️ 发现 {len(xlsx_files)} 个Excel文件，超过2个") print("为避免混淆，请确保目录下只有2个数据文件") print("（程序会自动忽略输出文件和示例文件）") print("\n请手动移除多余文件，或选择要使用的文件:") for i, file_path in enumerate(xlsx_files, 1): print(f" {i}. {file_path.name}") return None, None # 分析文件结构 print(f"\n正在分析文件结构...") file_info = [] for file_path in xlsx_files: try: # 读取文件获取基本信息 df = pd.read_excel(file_path, nrows=5) # 只读前5行进行分析 info = { 'path': file_path, 'name': file_path.name, 'columns': len(df.columns), 'rows': len(df) + 1, # 加上标题行 'column_names': list(df.columns), 'sample_data': df.head(2) # 获取前2行样本数据 } file_info.append(info) print(f"\n📄 {file_path.name}:") print(f" 列数: {info['columns']}") print(f" 行数: {info['rows']}+ (仅检查前5行)") print(f" 列名: {', '.join(str(col) for col in info['column_names'][:6])}{'...' if len(info['column_names']) > 6 else ''}") except Exception as e: print(f"\n❌ 读取文件 {file_path.name} 失败: {e}") return None, None # 根据列数自动识别文件类型 personnel_file = None seat_file = None print(f"\n=== 文件类型识别 ===") for info in file_info: if info['columns'] == 5 or info['columns'] == 6: # 人员信息通常是5-6列 # 进一步检查列名是否符合人员信息格式 col_names = [str(col).lower() for col in info['column_names']] personnel_keywords = ['姓名', 'name', '证件', 'id', '手机', 'phone', '备注', 'remark'] if any(keyword in ' '.join(col_names) for keyword in personnel_keywords): personnel_file = info print(f"✅ {info['name']} -> 识别为人员信息文件 ({info['columns']}列)") else: print(f"⚠️ {info['name']} 有{info['columns']}列，但列名不符合人员信息格式") elif info['columns'] >= 10: # 座位信息通常是10-12列 # 进一步检查列名是否符合座位信息格式 col_names = [str(col).lower() for col in info['column_names']] seat_keywords = ['区域', 'area', '楼层', 'floor', '排号', 'row', '座位', 'seat'] if any(keyword in ' '.join(col_names) for keyword in seat_keywords): seat_file = info print(f"✅ {info['name']} -> 识别为座位信息文件 ({info['columns']}列)") else: print(f"⚠️ {info['name']} 有{info['columns']}列，但列名不符合座位信息格式") else: print(f"❓ {info['name']} 有{info['columns']}列，无法自动识别类型") # 验证识别结果 if personnel_file is None: print(f"\n❌ 未找到人员信息文件") print("人员信息文件应包含: 姓名、证件类型、证件号、手机号、备注等列") return None, None if seat_file is None: print(f"\n❌ 未找到座位信息文件") print("座位信息文件应包含: 区域、楼层、排号、座位号等列") return None, None print(f"\n🎉 文件识别成功!") print(f"人员信息: {personnel_file['name']}") print(f"座位信息: {seat_file['name']}") # 显示文件内容预览 print(f"\n=== 文件内容预览 ===") print(f"\n📋 人员信息文件预览 ({personnel_file['name']}):") print(personnel_file['sample_data'].to_string(index=False, max_cols=10)) print(f"\n🪑 座位信息文件预览 ({seat_file['name']}):") print(seat_file['sample_data'].to_string(index=False, max_cols=10)) return personnel_file['path'], seat_file['path'] def check_data_files(): """检查并自动识别数据文件""" print("\n=== 数据文件检查 ===") # 首先尝试自动识别 personnel_file, seat_file = auto_detect_files() if personnel_file is None or seat_file is None: print(f"\n❌ 自动识别失败") print("\n请检查以下要求:") print("1. 确保目录下有且仅有2个Excel文件") print("2. 人员信息文件应有5-6列（姓名、证件类型、证件号、手机号、备注等）") print("3. 座位信息文件应有10+列（区域、楼层、排号、座位号等）") print("4. 列名应包含相关关键词") print("\n提示: 您可以参考示例文件来准备数据") return False, None, None return True, str(personnel_file), str(seat_file) # 检查依赖 if not check_dependencies(): print("❌ 依赖检查失败") input("按Enter键退出...") sys.exit(1) class Logger: """日志记录器""" def __init__(self, log_file='seat_allocation_log.txt'): # 创建log文件夹 self.log_dir = Path('log') self.log_dir.mkdir(exist_ok=True) # 设置日志文件路径 self.log_file = self.log_dir / log_file self.logs = [] def log(self, message, print_to_console=True): """记录日志""" timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_entry = f"[{timestamp}] {message}" self.logs.append(log_entry) if print_to_console: print(message) def save_logs(self): """保存日志到文件""" try: with open(self.log_file, 'w', encoding='utf-8') as f: f.write('\n'.join(self.logs)) print(f"日志已保存到: {self.log_file}") return True except Exception as e: print(f"保存日志失败: {e}") return False def get_log_path(self, filename): """获取log文件夹中的文件路径""" return self.log_dir / filename class SeatAllocationSystem: """座位分配系统""" def __init__(self): self.logger = Logger() self.personnel_df = None self.seat_df = None self.seating_groups = [] self.row_analysis = {} self.personnel_file = None self.seat_file = None def choose_file_mode(self): """选择文件识别模式""" self.logger.log("\n=== 选择文件识别模式 ===") self.logger.log("1. 默认文件名 - 读取 '人员信息.xlsx' 和 '座位信息.xlsx'") self.logger.log("2. 智能识别 - 自动识别目录中的Excel文件") while True: try: choice = input("\n请选择文件识别模式 (1/2，直接回车选择默认): ").strip() if choice == '' or choice == '1': self.logger.log("✅ 选择默认文件名模式") return 'default' elif choice == '2': self.logger.log("✅ 选择智能识别模式") return 'smart' else: print("请输入 1 或 2") except KeyboardInterrupt: self.logger.log("\n用户取消操作") return 'default' def smart_identify_files(self): """智能识别Excel文件""" self.logger.log("\n=== 智能识别Excel文件 ===") # 获取当前目录下的所有Excel文件 excel_files = [] for ext in ['*.xlsx', '*.xls']: excel_files.extend(Path('.').glob(ext)) if len(excel_files) < 2: self.logger.log(f"❌ 当前目录只找到 {len(excel_files)} 个Excel文件，需要至少2个") return None, None self.logger.log(f"📁 找到 {len(excel_files)} 个Excel文件:") # 分析每个文件的内容 file_analysis = [] for file_path in excel_files: try: # 先读取完整文件获取真实行数 df_full = pd.read_excel(file_path) total_rows = len(df_full) # 再读取前5行进行列分析 df_sample = pd.read_excel(file_path, nrows=5) analysis = { 'path': file_path, 'name': file_path.name, 'rows': total_rows, # 使用真实行数 'cols': len(df_sample.columns), 'columns': list(df_sample.columns), 'score': 0, 'type': 'unknown' } # 判断文件类型 - 改进的识别算法 columns_str = ' '.join(str(col).lower() for col in df_sample.columns) columns_list = [str(col).lower() for col in df_sample.columns] # 座位信息文件的强特征（优先判断） seat_strong_keywords = ['座位id', '区域', '楼层', '排号', '座位号'] seat_strong_score = sum(1 for keyword in seat_strong_keywords if keyword in columns_str) # 人员信息文件的强特征 personnel_strong_keywords = ['备注'] # 备注列是人员信息文件的强特征 personnel_strong_score = sum(1 for keyword in personnel_strong_keywords if keyword in columns_str) # 如果有座位强特征，优先识别为座位文件 if seat_strong_score >= 3: # 至少包含3个座位强特征 analysis['type'] = 'seat' analysis['score'] = seat_strong_score * 10 # 给予高权重 # 如果有人员强特征且列数较少，识别为人员文件 elif personnel_strong_score > 0 and len(df_sample.columns) <= 8: analysis['type'] = 'personnel' analysis['score'] = personnel_strong_score * 10 else: # 使用原有的弱特征判断 personnel_keywords = ['姓名', '证件', '手机', 'name', 'id', 'phone'] personnel_score = sum(1 for keyword in personnel_keywords if keyword in columns_str) seat_keywords = ['区域', '楼层', '排号', '座位', 'area', 'floor', 'row', 'seat'] seat_score = sum(1 for keyword in seat_keywords if keyword in columns_str) # 结合列数进行判断 if len(df_sample.columns) <= 8 and personnel_score > 0: analysis['type'] = 'personnel' analysis['score'] = personnel_score elif len(df_sample.columns) > 8 and seat_score > 0: analysis['type'] = 'seat' analysis['score'] = seat_score elif personnel_score > seat_score: analysis['type'] = 'personnel' analysis['score'] = personnel_score elif seat_score > personnel_score: analysis['type'] = 'seat' analysis['score'] = seat_score file_analysis.append(analysis) self.logger.log(f" 📄 {file_path.name}: {total_rows}行 × {len(df_sample.columns)}列") self.logger.log(f" 列名: {', '.join(str(col) for col in df_sample.columns[:5])}{'...' if len(df_sample.columns) > 5 else ''}") self.logger.log(f" 推测类型: {analysis['type']} (得分: {analysis['score']})") except Exception as e: self.logger.log(f" ❌ {file_path.name}: 读取失败 - {e}") # 自动匹配最佳文件 personnel_files = [f for f in file_analysis if f['type'] == 'personnel'] seat_files = [f for f in file_analysis if f['type'] == 'seat'] personnel_file = None seat_file = None if personnel_files: personnel_file = max(personnel_files, key=lambda x: x['score'])['path'] self.logger.log(f"\n🎯 自动识别人员信息文件: {personnel_file.name}") if seat_files: seat_file = max(seat_files, key=lambda x: x['score'])['path'] self.logger.log(f"🎯 自动识别座位信息文件: {seat_file.name}") # 如果自动识别不完整，提供手动选择 if not personnel_file or not seat_file: self.logger.log("\n⚠️ 自动识别不完整，请手动选择:") personnel_file, seat_file = self.manual_select_files(file_analysis) return personnel_file, seat_file def manual_select_files(self, file_analysis): """手动选择文件""" self.logger.log("\n=== 手动选择文件 ===") print("\n可用的Excel文件:") for i, analysis in enumerate(file_analysis): print(f"{i+1}. {analysis['name']} ({analysis['rows']}行 × {analysis['cols']}列)") print(f" 列名: {', '.join(str(col) for col in analysis['columns'][:3])}...") personnel_file = None seat_file = None # 选择人员信息文件 while not personnel_file: try: choice = input(f"\n请选择人员信息文件 (1-{len(file_analysis)}): ").strip() idx = int(choice) - 1 if 0 <= idx < len(file_analysis): personnel_file = file_analysis[idx]['path'] self.logger.log(f"✅ 选择人员信息文件: {personnel_file.name}") else: print("请输入有效的数字") except (ValueError, KeyboardInterrupt): print("请输入有效的数字") # 选择座位信息文件 while not seat_file: try: choice = input(f"\n请选择座位信息文件 (1-{len(file_analysis)}): ").strip() idx = int(choice) - 1 if 0 <= idx < len(file_analysis): if file_analysis[idx]['path'] == personnel_file: print("不能选择相同的文件，请重新选择") continue seat_file = file_analysis[idx]['path'] self.logger.log(f"✅ 选择座位信息文件: {seat_file.name}") else: print("请输入有效的数字") except (ValueError, KeyboardInterrupt): print("请输入有效的数字") return personnel_file, seat_file def set_data_files(self, personnel_file, seat_file): """设置数据文件路径""" self.personnel_file = personnel_file self.seat_file = seat_file def identify_and_set_files(self): """识别并设置数据文件""" file_mode = self.choose_file_mode() if file_mode == 'default': # 默认文件名模式 personnel_file = Path('人员信息.xlsx') seat_file = Path('座位信息.xlsx') if not personnel_file.exists(): self.logger.log(f"❌ 错误: {personnel_file} 文件不存在") return False if not seat_file.exists(): self.logger.log(f"❌ 错误: {seat_file} 文件不存在") return False else: # 智能识别模式 personnel_file, seat_file = self.smart_identify_files() if not personnel_file or not seat_file: self.logger.log("❌ 文件识别失败") return False self.set_data_files(personnel_file, seat_file) return True def load_data(self): """加载人员信息和座位信息数据""" self.logger.log("=== 开始加载数据 ===") # 使用自动识别的文件路径 if not self.personnel_file or not self.seat_file: self.logger.log("❌ 错误: 数据文件路径未设置") return False if not Path(self.personnel_file).exists(): self.logger.log(f"❌ 错误: {self.personnel_file} 文件不存在") return False if not Path(self.seat_file).exists(): self.logger.log(f"❌ 错误: {self.seat_file} 文件不存在") return False try: self.logger.log(f"正在读取人员信息文件: {self.personnel_file}") # 读取人员信息，指定数据类型 self.personnel_df = pd.read_excel(self.personnel_file, dtype={ '姓名': 'str', '证件类型': 'str', '证件号': 'str', # 证件号作为字符串读取，避免X被转换 '手机号': 'str' }) self.logger.log(f"正在读取座位信息文件: {self.seat_file}") # 读取座位信息，指定数据类型 seat_df_raw = pd.read_excel(self.seat_file, dtype={ '区域': 'str', '楼层': 'str', '排号': 'str', '座位号': 'str', '姓名': 'str', '证件类型': 'str', '证件号': 'str', '手机号': 'str', '签发地/国籍': 'str' }) # 过滤和清洗座位数据 self.seat_df = self.filter_seat_data(seat_df_raw) # 清理文字信息中的空格 self.clean_text_data() self.logger.log(f"✅ 文件加载成功") self.logger.log(f" 人员信息: {self.personnel_df.shape[0]} 行 × {self.personnel_df.shape[1]} 列") self.logger.log(f" 座位信息: {self.seat_df.shape[0]} 行 × {self.seat_df.shape[1]} 列") return True except Exception as e: self.logger.log(f"❌ 文件加载失败: {e}") return False def clean_text_data(self): """清理文字数据中的空格""" self.logger.log("清理文字数据中的空格...") # 清理人员信息中的空格 text_columns_personnel = ['姓名', '证件类型', '证件号'] for col in text_columns_personnel: if col in self.personnel_df.columns: self.personnel_df[col] = self.personnel_df[col].astype(str).str.strip() # 清理座位信息中的空格 text_columns_seat = ['区域', '楼层', '排号', '座位号', '姓名', '证件类型', '证件号', '签发地/国籍'] for col in text_columns_seat: if col in self.seat_df.columns: # 只对非空值进行清理 mask = self.seat_df[col].notna() self.seat_df.loc[mask, col] = self.seat_df.loc[mask, col].astype(str).str.strip() self.logger.log("✅ 文字数据清理完成") def filter_seat_data(self, seat_df_raw): """过滤和清洗座位数据，移除空行和无效数据""" self.logger.log("开始过滤座位数据...") original_count = len(seat_df_raw) self.logger.log(f"原始数据行数: {original_count}") # 关键列，这些列都不能为空 key_columns = ['区域', '楼层', '排号', '座位号'] # 创建过滤条件：关键列都不能为空且不能为'nan'字符串 filter_condition = True for col in key_columns: if col in seat_df_raw.columns: # 过滤空值、NaN、'nan'字符串 col_condition = ( seat_df_raw[col].notna() & (seat_df_raw[col].astype(str).str.strip() != '') & (seat_df_raw[col].astype(str).str.lower() != 'nan') ) filter_condition = filter_condition & col_condition # 应用过滤条件 seat_df_filtered = seat_df_raw[filter_condition].copy() # 重置索引 seat_df_filtered.reset_index(drop=True, inplace=True) filtered_count = len(seat_df_filtered) removed_count = original_count - filtered_count self.logger.log(f"过滤后数据行数: {filtered_count}") self.logger.log(f"移除无效行数: {removed_count}") if removed_count > 0: self.logger.log(f"✅ 已过滤掉 {removed_count} 行无效数据（空行、示例数据等）") # 显示过滤后的数据概览 if filtered_count > 0: areas = seat_df_filtered['区域'].unique() floors = seat_df_filtered['楼层'].unique() self.logger.log(f"有效座位区域: {', '.join(areas)}") self.logger.log(f"有效座位楼层: {', '.join(floors)}") return seat_df_filtered def validate_personnel_structure(self): """校验人员信息文件结构""" self.logger.log("\n=== 人员信息结构校验 ===") required_columns = ['姓名', '证件类型', '证件号', '手机号', '备注'] validation_results = [] # 检查必需列 missing_columns = [] for col in required_columns: if col not in self.personnel_df.columns: missing_columns.append(col) if missing_columns: self.logger.log(f"❌ 缺少必需列: {missing_columns}") validation_results.append(False) else: self.logger.log("✅ 所有必需列都存在") validation_results.append(True) # 检查数据完整性 self.logger.log("\n数据完整性检查:") for col in ['姓名', '证件类型', '证件号', '手机号']: if col in self.personnel_df.columns: null_count = self.personnel_df[col].isnull().sum() if null_count > 0: self.logger.log(f"⚠️ {col} 列有 {null_count} 个空值") validation_results.append(False) else: self.logger.log(f"✅ {col} 列数据完整") validation_results.append(True) # 检查重复人员（姓名+身份证号组合） self.logger.log("\n重复人员检查:") if '姓名' in self.personnel_df.columns and '证件号' in self.personnel_df.columns: # 创建姓名+证件号的组合标识 self.personnel_df['人员标识'] = self.personnel_df['姓名'].astype(str) + '_' + self.personnel_df['证件号'].astype(str) duplicate_persons = self.personnel_df[self.personnel_df['人员标识'].duplicated()] if not duplicate_persons.empty: self.logger.log(f"⚠️ 发现重复人员:") for _, person in duplicate_persons.iterrows(): self.logger.log(f" {person['姓名']} (证件号: {person['证件号']})") validation_results.append(False) else: self.logger.log("✅ 无重复人员（姓名+证件号组合检查）") validation_results.append(True) # 单独检查重名情况（仅提示，不影响校验结果） duplicate_names = self.personnel_df[self.personnel_df['姓名'].duplicated(keep=False)] if not duplicate_names.empty: self.logger.log(f"📝 发现重名人员（但证件号不同，允许通过）:") name_groups = duplicate_names.groupby('姓名') for name, group in name_groups: self.logger.log(f" 姓名: {name}") for _, person in group.iterrows(): self.logger.log(f" 证件号: {person['证件号']}") else: self.logger.log("❌ 缺少姓名或证件号列，无法进行重复检查") validation_results.append(False) return all(validation_results) def validate_seating_groups(self): """校验连坐组的完整性和正确性""" self.logger.log("\n=== 连坐组校验 ===") validation_results = [] issues = [] i = 0 group_num = 1 while i < len(self.personnel_df): person = self.personnel_df.iloc[i] remark = person['备注'] if pd.isna(remark): # 无备注，单独坐 self.logger.log(f"✅ 第 {group_num} 组: {person['姓名']} (单独)") i += 1 else: # 有备注，检查连坐组 try: group_size = int(remark) if group_size < 1 or group_size > 10: issue = f"❌ 第 {group_num} 组: {person['姓名']} 的备注 {group_size} 超出范围 (1-10)" self.logger.log(issue) issues.append(issue) validation_results.append(False) i += 1 continue # 检查后续人员是否足够 if i + group_size > len(self.personnel_df): issue = f"❌ 第 {group_num} 组: {person['姓名']} 需要 {group_size} 人连坐，但后续人员不足" self.logger.log(issue) issues.append(issue) validation_results.append(False) i += 1 continue # 检查后续人员的备注是否为空 group_members = [] valid_group = True for j in range(group_size): member = self.personnel_df.iloc[i + j] group_members.append(member['姓名']) if j == 0: # 第一个人应该有备注 if pd.isna(member['备注']) or int(member['备注']) != group_size: issue = f"❌ 第 {group_num} 组: 组长 {member['姓名']} 的备注应该是 {group_size}" self.logger.log(issue) issues.append(issue) valid_group = False else: # 后续人员应该没有备注 if not pd.isna(member['备注']): issue = f"❌ 第 {group_num} 组: {member['姓名']} 应该没有备注（当前备注: {member['备注']}）" self.logger.log(issue) issues.append(issue) valid_group = False if valid_group: self.logger.log(f"✅ 第 {group_num} 组: {', '.join(group_members)} ({group_size}人连坐)") validation_results.append(True) else: validation_results.append(False) i += group_size except ValueError: issue = f"❌ 第 {group_num} 组: {person['姓名']} 的备注 '{remark}' 不是有效数字" self.logger.log(issue) issues.append(issue) validation_results.append(False) i += 1 group_num += 1 return all(validation_results), issues def validate_seat_structure(self): """校验座位信息结构和连续性""" self.logger.log("\n=== 座位信息结构校验 ===") required_columns = ['区域', '楼层', '排号', '座位号'] validation_results = [] # 检查必需列 missing_columns = [] for col in required_columns: if col not in self.seat_df.columns: missing_columns.append(col) if missing_columns: self.logger.log(f"❌ 缺少必需列: {missing_columns}") return False, {} self.logger.log("✅ 所有必需列都存在") # 检查数据完整性 for col in required_columns: null_count = self.seat_df[col].isnull().sum() if null_count > 0: self.logger.log(f"❌ {col} 列有 {null_count} 个空值") validation_results.append(False) else: self.logger.log(f"✅ {col} 列数据完整") validation_results.append(True) # 分析座位结构 self.logger.log("\n座位结构分析:") seat_groups = {} for _, row in self.seat_df.iterrows(): key = (row['区域'], row['楼层'], row['排号']) if key not in seat_groups: seat_groups[key] = [] seat_groups[key].append(row['座位号']) # 检查每排的座位结构和可用连续段 self.logger.log("\n座位结构分析:") for (area, floor, row_num), seats in seat_groups.items(): # 提取座位号数字 seat_numbers = [] for seat in seats: try: seat_numbers.append(int(str(seat).replace('号', ''))) except: self.logger.log(f"❌ {area}-{floor}-{row_num}: 座位号格式异常 '{seat}'") validation_results.append(False) continue seat_numbers.sort() # 分析连续段 consecutive_segments = [] if seat_numbers: current_segment = [seat_numbers[0]] for i in range(1, len(seat_numbers)): if seat_numbers[i] - seat_numbers[i-1] == 1: # 连续 current_segment.append(seat_numbers[i]) else: # 不连续，保存当前段，开始新段 consecutive_segments.append(current_segment) current_segment = [seat_numbers[i]] # 添加最后一段 consecutive_segments.append(current_segment) # 显示分析结果 if len(consecutive_segments) == 1 and len(consecutive_segments[0]) == len(seat_numbers): self.logger.log(f"✅ {area}-{floor}-{row_num}: {len(seats)} 个座位完全连续 ({min(seat_numbers)}-{max(seat_numbers)})") else: segments_info = [] max_segment_size = 0 for segment in consecutive_segments: if len(segment) == 1: segments_info.append(f"{segment[0]}") else: segments_info.append(f"{segment[0]}-{segment[-1]}") max_segment_size = max(max_segment_size, len(segment)) self.logger.log(f"📊 {area}-{floor}-{row_num}: {len(seats)} 个座位，{len(consecutive_segments)} 个连续段: {', '.join(segments_info)}") self.logger.log(f" 最大连续段: {max_segment_size} 个座位") validation_results.append(True) # 座位不连续不算错误，只是需要特殊处理 return all(validation_results), seat_groups def validate_capacity_feasibility(self, seat_groups): """校验座位容量和分配可行性""" self.logger.log("\n=== 容量和可行性校验 ===") # 统计人员总数 total_people = len(self.personnel_df) total_seats = sum(len(seats) for seats in seat_groups.values()) self.logger.log(f"总人数: {total_people}") self.logger.log(f"总座位数: {total_seats}") if total_people > total_seats: self.logger.log(f"❌ 座位不足: 需要 {total_people} 个座位，只有 {total_seats} 个") return False else: self.logger.log(f"✅ 座位充足: 剩余 {total_seats - total_people} 个座位") # 分析连坐组需求 self.logger.log("\n连坐组需求分析:") group_sizes = [] i = 0 while i < len(self.personnel_df): person = self.personnel_df.iloc[i] remark = person['备注'] if pd.isna(remark): group_sizes.append(1) i += 1 else: try: group_size = int(remark) group_sizes.append(group_size) i += group_size except: group_sizes.append(1) i += 1 max_group_size = max(group_sizes) self.logger.log(f"最大连坐组: {max_group_size} 人") # 检查是否有足够的连续座位来容纳最大连坐组 self.logger.log(f"\n连续座位可行性分析:") max_consecutive_available = 0 feasible_rows = [] for (area, floor, row_num), seats in seat_groups.items(): # 分析每排的连续座位段 seat_numbers = [] for seat in seats: try: seat_numbers.append(int(str(seat).replace('号', ''))) except: continue seat_numbers.sort() # 找出最大连续段 max_consecutive_in_row = 0 if seat_numbers: current_consecutive = 1 for i in range(1, len(seat_numbers)): if seat_numbers[i] - seat_numbers[i-1] == 1: current_consecutive += 1 else: max_consecutive_in_row = max(max_consecutive_in_row, current_consecutive) current_consecutive = 1 max_consecutive_in_row = max(max_consecutive_in_row, current_consecutive) if max_consecutive_in_row >= max_group_size: feasible_rows.append((area, floor, row_num, max_consecutive_in_row)) max_consecutive_available = max(max_consecutive_available, max_consecutive_in_row) self.logger.log(f" {area}-{floor}-{row_num}: 最大连续 {max_consecutive_in_row} 个座位") self.logger.log(f"\n全场最大连续座位: {max_consecutive_available} 个") if max_group_size > max_consecutive_available: self.logger.log(f"❌ 无法容纳最大连坐组: 需要 {max_group_size} 个连续座位，最大连续段只有 {max_consecutive_available} 个") return False else: self.logger.log(f"✅ 可以容纳最大连坐组") self.logger.log(f"可容纳最大连坐组的排数: {len(feasible_rows)} 个") # 统计各种大小的连坐组 size_counts = {} for size in group_sizes: size_counts[size] = size_counts.get(size, 0) + 1 self.logger.log("\n连坐组分布:") for size in sorted(size_counts.keys()): count = size_counts[size] if size == 1: self.logger.log(f" 单人组: {count} 个") else: self.logger.log(f" {size}人连坐组: {count} 个") return True def analyze_seating_requirements(self, use_phone_grouping=False): """分析人员连坐需求""" self.logger.log("\n=== 人员连坐需求分析 ===") if use_phone_grouping: self.logger.log("🔍 使用手机号分组模式") return self.analyze_seating_by_phone() else: self.logger.log("🔍 使用备注分组模式") return self.analyze_seating_by_remark() def analyze_seating_by_phone(self): """基于手机号分析连坐需求""" self.logger.log("根据相同手机号识别连坐组...") # 按手机号分组 phone_groups = self.personnel_df.groupby('手机号') self.seating_groups = [] for phone, group in phone_groups: group_members = [row for _, row in group.iterrows()] group_size = len(group_members) if group_size == 1: # 单人组 self.seating_groups.append({ 'type': 'single', 'size': 1, 'members': group_members, 'leader': group_members[0]['姓名'], 'grouping_method': 'phone' }) self.logger.log(f"📱 单人: {group_members[0]['姓名']} (手机号: {phone})") else: # 连坐组 leader_name = group_members[0]['姓名'] member_names = [member['姓名'] for member in group_members] self.seating_groups.append({ 'type': 'group', 'size': group_size, 'members': group_members, 'leader': leader_name, 'grouping_method': 'phone' }) self.logger.log(f"📱 连坐组: {', '.join(member_names)} (手机号: {phone}, {group_size}人)") # 统计 size_stats = {} for group in self.seating_groups: size = group['size'] size_stats[size] = size_stats.get(size, 0) + 1 self.logger.log(f"\n基于手机号识别出 {len(self.seating_groups)} 个座位组:") for size in sorted(size_stats.keys()): count = size_stats[size] if size == 1: self.logger.log(f" 单人组: {count} 个") else: self.logger.log(f" {size}人连坐组: {count} 个") return True def analyze_seating_by_remark(self): """基于备注分析连坐需求""" self.logger.log("根据备注数字识别连坐组...") self.seating_groups = [] i = 0 while i < len(self.personnel_df): person = self.personnel_df.iloc[i] remark = person['备注'] if pd.isna(remark): # 无备注，单独坐 self.seating_groups.append({ 'type': 'single', 'size': 1, 'members': [person], 'leader': person['姓名'], 'grouping_method': 'remark' }) i += 1 else: # 有备注，按备注数量连坐 group_size = int(remark) group_members = [] # 收集连坐组成员 for j in range(group_size): if i + j < len(self.personnel_df): group_members.append(self.personnel_df.iloc[i + j]) self.seating_groups.append({ 'type': 'group' if group_size > 1 else 'single', 'size': len(group_members), 'members': group_members, 'leader': person['姓名'], 'grouping_method': 'remark' }) i += group_size # 统计 size_stats = {} for group in self.seating_groups: size = group['size'] size_stats[size] = size_stats.get(size, 0) + 1 self.logger.log(f"\n基于备注识别出 {len(self.seating_groups)} 个座位组:") for size in sorted(size_stats.keys()): count = size_stats[size] if size == 1: self.logger.log(f" 单人组: {count} 个") else: self.logger.log(f" {size}人连坐组: {count} 个") return True def analyze_seat_structure_advanced(self): """高级座位结构分析，识别连续段""" self.logger.log("\n=== 高级座位结构分析 ===") seat_groups = {} for _, row in self.seat_df.iterrows(): key = (row['区域'], row['楼层'], row['排号']) if key not in seat_groups: seat_groups[key] = [] seat_groups[key].append({ 'index': row.name, '座位号': row['座位号'], 'row_data': row }) # 分析每排的连续段 self.row_analysis = {} for key, seats in seat_groups.items(): area, floor, row_num = key # 按座位号排序 def get_seat_number(seat_info): try: return int(seat_info['座位号'].replace('号', '')) except: return 0 seats.sort(key=get_seat_number) seat_numbers = [get_seat_number(seat) for seat in seats] # 找出所有连续段 consecutive_segments = [] if seat_numbers: current_segment_start = 0 for i in range(1, len(seat_numbers)): if seat_numbers[i] - seat_numbers[i-1] != 1: # 发现间隙，结束当前段 consecutive_segments.append({ 'start_idx': current_segment_start, 'end_idx': i - 1, 'size': i - current_segment_start, 'seat_numbers': seat_numbers[current_segment_start:i], 'seats': seats[current_segment_start:i] }) current_segment_start = i # 添加最后一段 consecutive_segments.append({ 'start_idx': current_segment_start, 'end_idx': len(seat_numbers) - 1, 'size': len(seat_numbers) - current_segment_start, 'seat_numbers': seat_numbers[current_segment_start:], 'seats': seats[current_segment_start:] }) self.row_analysis[key] = { 'total_seats': len(seats), 'all_seats': seats, 'consecutive_segments': consecutive_segments, 'max_consecutive': max(seg['size'] for seg in consecutive_segments) if consecutive_segments else 0 } # 显示分析结果 if len(consecutive_segments) == 1: self.logger.log(f"✅ {area}-{floor}-{row_num}: {len(seats)} 个座位完全连续") else: segments_info = [] for seg in consecutive_segments: if seg['size'] == 1: segments_info.append(f"{seg['seat_numbers'][0]}") else: segments_info.append(f"{seg['seat_numbers'][0]}-{seg['seat_numbers'][-1]}") self.logger.log(f"📊 {area}-{floor}-{row_num}: {len(seats)} 个座位，{len(consecutive_segments)} 段: {', '.join(segments_info)}") self.logger.log(f" 最大连续段: {self.row_analysis[key]['max_consecutive']} 个座位") return True def smart_seat_assignment(self): """智能座位分配算法""" self.logger.log("\n=== 开始智能座位分配 ===") # 按组大小排序，大组优先分配 sorted_groups = sorted(self.seating_groups, key=lambda x: x['size'], reverse=True) # 创建座位分配结果 seat_df_copy = self.seat_df.copy() # 记录已使用的座位 used_seats = set() assignment_log = [] unassigned_groups = [] self.logger.log(f"需要分配 {len(sorted_groups)} 个组") for group_idx, seating_group in enumerate(sorted_groups): group_size = seating_group['size'] group_type = seating_group['type'] leader = seating_group['leader'] self.logger.log(f"\n处理第 {group_idx + 1} 组: {leader} ({group_type}, {group_size} 人)") if group_size == 1: # 单人组，找任意可用座位 assigned = False for (area, floor, row_num), analysis in self.row_analysis.items(): for seat_info in analysis['all_seats']: if seat_info['index'] not in used_seats: # 分配座位 person_info = seating_group['members'][0] seat_index = seat_info['index'] # 覆盖座位表中的人员信息列 seat_df_copy.loc[seat_index, '姓名'] = str(person_info['姓名']).strip() seat_df_copy.loc[seat_index, '证件类型'] = str(person_info['证件类型']).strip() seat_df_copy.loc[seat_index, '证件号'] = str(person_info['证件号']).strip() seat_df_copy.loc[seat_index, '手机国家号'] = person_info.get('Unnamed: 3', 86) seat_df_copy.loc[seat_index, '手机号'] = str(person_info['手机号']).strip() # 只有分配备注放在第13列 seat_df_copy.loc[seat_index, '分配备注'] = str(person_info.get('备注', '')).strip() assignment_log.append({ '组号': group_idx + 1, '组类型': group_type, '组大小': group_size, '组长': leader, '姓名': person_info['姓名'], '区域': area, '楼层': floor, '排号': row_num, '座位号': seat_info['座位号'] }) used_seats.add(seat_index) self.logger.log(f" 分配到 {area}-{floor}-{row_num}: {person_info['姓名']} -> {seat_info['座位号']}") assigned = True break if assigned: break else: # 多人组，需要连续座位 # 更新可用座位分析（排除已使用的座位） current_row_analysis = {} for key, analysis in self.row_analysis.items(): available_seats = [seat for seat in analysis['all_seats'] if seat['index'] not in used_seats] if available_seats: # 重新分析连续段 available_seats.sort(key=lambda x: int(x['座位号'].replace('号', ''))) seat_numbers = [int(seat['座位号'].replace('号', '')) for seat in available_seats] # 找出连续段 consecutive_segments = [] if seat_numbers: current_segment_start = 0 for i in range(1, len(seat_numbers)): if seat_numbers[i] - seat_numbers[i-1] != 1: consecutive_segments.append({ 'start_idx': current_segment_start, 'end_idx': i - 1, 'size': i - current_segment_start, 'seats': available_seats[current_segment_start:i] }) current_segment_start = i consecutive_segments.append({ 'start_idx': current_segment_start, 'end_idx': len(seat_numbers) - 1, 'size': len(seat_numbers) - current_segment_start, 'seats': available_seats[current_segment_start:] }) current_row_analysis[key] = { 'consecutive_segments': consecutive_segments } # 寻找合适的连续座位 assigned = False for (area, floor, row_num), analysis in current_row_analysis.items(): for segment in analysis['consecutive_segments']: if segment['size'] >= group_size: # 找到合适的连续座位 seats_to_use = segment['seats'][:group_size] seat_numbers = [int(seat['座位号'].replace('号', '')) for seat in seats_to_use] self.logger.log(f" 分配到 {area}-{floor}-{row_num} (连续座位: {min(seat_numbers)}-{max(seat_numbers)})") # 分配座位 for i, seat_info in enumerate(seats_to_use): person_info = seating_group['members'][i] seat_index = seat_info['index'] seat_df_copy.loc[seat_index, '姓名'] = str(person_info['姓名']).strip() seat_df_copy.loc[seat_index, '证件类型'] = str(person_info['证件类型']).strip() seat_df_copy.loc[seat_index, '证件号'] = str(person_info['证件号']).strip() seat_df_copy.loc[seat_index, '手机国家号'] = person_info.get('Unnamed: 3', 86) seat_df_copy.loc[seat_index, '手机号'] = str(person_info['手机号']).strip() # 只有分配备注放在第13列 seat_df_copy.loc[seat_index, '分配备注'] = str(person_info.get('备注', '')).strip() assignment_log.append({ '组号': group_idx + 1, '组类型': group_type, '组大小': group_size, '组长': leader, '姓名': person_info['姓名'], '区域': area, '楼层': floor, '排号': row_num, '座位号': seat_info['座位号'] }) used_seats.add(seat_index) self.logger.log(f" {person_info['姓名']} -> {seat_info['座位号']}") assigned = True break if assigned: break if not assigned: self.logger.log(f" ❌ 无法为第 {group_idx + 1} 组分配座位") unassigned_groups.append(seating_group) # 显示未分配的组 if unassigned_groups: self.logger.log(f"\n⚠️ 有 {len(unassigned_groups)} 个组未能分配座位") for group in unassigned_groups: if group['type'] == 'single': self.logger.log(f" 未分配: {group['leader']}") else: member_names = [member['姓名'] for member in group['members']] self.logger.log(f" 未分配: {', '.join(member_names)} (连坐 {group['size']} 人)") return seat_df_copy, assignment_log def save_results(self, seat_df_result, assignment_log): """保存分配结果""" try: # 保存更新后的座位信息 output_file = self.logger.get_log_path('座位信息_最终分配.xlsx') seat_df_result.to_excel(output_file, index=False) self.logger.log(f"\n座位分配结果已保存到: {output_file}") self.logger.log(f"📋 人员信息已覆盖到座位表对应列:") self.logger.log(f" 第6列: 姓名") self.logger.log(f" 第7列: 证件类型") self.logger.log(f" 第8列: 证件号") self.logger.log(f" 第9列: 手机国家号") self.logger.log(f" 第10列: 手机号") self.logger.log(f"📝 分配备注信息:") self.logger.log(f" 第13列: 分配备注（新增列，不覆盖原数据）") self.logger.log(f"💡 座位基本信息（第1-5列）保持不变") # 保存分配日志 if assignment_log: log_df = pd.DataFrame(assignment_log) log_file = self.logger.get_log_path('最终座位分配日志.xlsx') log_df.to_excel(log_file, index=False) self.logger.log(f"分配日志已保存到: {log_file}") # 显示分配统计 self.logger.log(f"\n=== 分配统计 ===") self.logger.log(f"总共分配了 {len(assignment_log)} 个座位") # 按组大小统计 size_stats = log_df.groupby('组大小').agg({ '姓名': 'count', '组号': 'nunique' }).rename(columns={'姓名': '总人数', '组号': '组数'}) self.logger.log("\n按组大小统计:") for size, row in size_stats.iterrows(): if size == 1: self.logger.log(f" 单人组: {row['组数']} 个组, {row['总人数']} 人") else: self.logger.log(f" {size}人连坐组: {row['组数']} 个组, {row['总人数']} 人") # 验证连续性 self.logger.log("\n=== 连续性验证 ===") consecutive_check = [] for group_num in sorted(log_df['组号'].unique()): group_data = log_df[log_df['组号'] == group_num] group_size = group_data.iloc[0]['组大小'] group_leader = group_data.iloc[0]['组长'] if group_size > 1: # 多人组 # 检查是否在同一排 areas = group_data['区域'].unique() floors = group_data['楼层'].unique() rows = group_data['排号'].unique() if len(areas) == 1 and len(floors) == 1 and len(rows) == 1: # 在同一排，检查座位号是否连续 seats = group_data['座位号'].tolist() seat_numbers = [] for seat in seats: try: seat_numbers.append(int(seat.replace('号', ''))) except: seat_numbers.append(0) seat_numbers.sort() is_consecutive = all(seat_numbers[i] + 1 == seat_numbers[i+1] for i in range(len(seat_numbers)-1)) if is_consecutive: consecutive_check.append(True) self.logger.log(f"✅ 组 {group_num} ({group_leader}): {group_size}人连坐，座位连续 {seat_numbers}") else: consecutive_check.append(False) self.logger.log(f"❌ 组 {group_num} ({group_leader}): {group_size}人连坐，座位不连续 {seat_numbers}") else: consecutive_check.append(False) self.logger.log(f"❌ 组 {group_num} ({group_leader}): 不在同一排") success_rate = sum(consecutive_check) / len(consecutive_check) * 100 if consecutive_check else 100 self.logger.log(f"\n连续性检查结果: {sum(consecutive_check)}/{len(consecutive_check)} 个多人组座位连续 ({success_rate:.1f}%)") return True except Exception as e: self.logger.log(f"保存结果时出错: {e}") return False def run_validation(self): """运行完整的文件校验""" self.logger.log("=" * 60) self.logger.log("座位分配系统 - 文件校验") self.logger.log("=" * 60) # 识别并设置数据文件 if not self.identify_and_set_files(): return False # 加载数据 if not self.load_data(): return False # 人员信息结构校验 personnel_structure_valid = self.validate_personnel_structure() # 连坐组校验 seating_groups_valid, group_issues = self.validate_seating_groups() # 座位信息校验 seat_structure_valid, seat_groups = self.validate_seat_structure() # 容量可行性校验 capacity_valid = self.validate_capacity_feasibility(seat_groups) # 总结 self.logger.log("\n" + "=" * 60) self.logger.log("校验结果总结") self.logger.log("=" * 60) all_valid = all([ personnel_structure_valid, seating_groups_valid, seat_structure_valid, capacity_valid ]) self.logger.log(f"人员信息结构: {'✅ 通过' if personnel_structure_valid else '❌ 失败'}") self.logger.log(f"连坐组完整性: {'✅ 通过' if seating_groups_valid else '❌ 失败'}") self.logger.log(f"座位信息结构: {'✅ 通过' if seat_structure_valid else '❌ 失败'}") self.logger.log(f"容量可行性: {'✅ 通过' if capacity_valid else '❌ 失败'}") self.logger.log(f"\n总体校验结果: {'✅ 全部通过' if all_valid else '❌ 存在问题'}") if group_issues: self.logger.log(f"\n发现的问题:") for issue in group_issues: self.logger.log(f" {issue}") if all_valid: self.logger.log("\n🎉 文件校验通过，可以进行座位分配！") else: self.logger.log("\n⚠️ 请修复上述问题后再进行座位分配。") return all_valid def run_allocation(self): """运行完整的座位分配""" self.logger.log("\n" + "=" * 60) self.logger.log("开始座位分配") self.logger.log("=" * 60) # 选择分组方式 grouping_method = self.choose_grouping_method() # 分析人员连坐需求 if not self.analyze_seating_requirements(use_phone_grouping=(grouping_method == 'phone')): return False # 高级座位结构分析 if not self.analyze_seat_structure_advanced(): return False # 执行智能座位分配 seat_df_result, assignment_log = self.smart_seat_assignment() # 保存结果 if self.save_results(seat_df_result, assignment_log): self.logger.log("\n🎉 座位分配完成！") return True else: self.logger.log("\n❌ 座位分配失败！") return False def choose_grouping_method(self): """选择分组方式""" self.logger.log("\n=== 选择连坐分组方式 ===") self.logger.log("1. 备注分组 - 根据备注数字识别连坐组（默认）") self.logger.log("2. 手机号分组 - 根据相同手机号识别连坐组") while True: try: choice = input("\n请选择分组方式 (1/2，直接回车选择默认): ").strip() if choice == '' or choice == '1': self.logger.log("✅ 选择备注分组模式") return 'remark' elif choice == '2': self.logger.log("✅ 选择手机号分组模式") return 'phone' else: print("请输入 1 或 2") except KeyboardInterrupt: self.logger.log("\n用户取消操作") return 'remark' def main(): """主函数""" print("=" * 60) print("座位分配系统 v2.0 (智能识别版)") print("=" * 60) try: # 自动检查和识别数据文件 success, personnel_file, seat_file = check_data_files() if not success: input("\n按Enter键退出...") return print("\n开始运行座位分配系统...") system = SeatAllocationSystem() # 设置识别出的文件路径 system.set_data_files(personnel_file, seat_file) # 运行校验 if system.run_validation(): # 校验通过，询问是否继续分配 response = input("\n文件校验通过，是否开始座位分配? (Y/n): ") if response.lower() in ['', 'y', 'yes']: # 运行分配 if system.run_allocation(): print("\n🎉 座位分配完成！") print("请查看log文件夹中的输出文件:") print("- log/座位信息_最终分配.xlsx (分配结果)") print("- log/最终座位分配日志.xlsx (详细日志)") print("- log/seat_allocation_log.txt (运行日志)") else: print("\n❌ 座位分配失败！") else: print("用户取消座位分配") else: print("\n❌ 文件校验失败，请修复问题后重试") # 保存日志 system.logger.save_logs() except FileNotFoundError as e: print(f"\n❌ 文件未找到: {e}") print("请确保所有必要文件都在程序目录下") except PermissionError as e: print(f"\n❌ 文件权限错误: {e}") print("请确保程序有读写文件的权限") except Exception as e: print(f"\n❌ 程序运行出错: {e}") print("请检查数据文件格式是否正确") finally: # 等待用户确认后退出 input("\n程序结束，按Enter键退出...") if __name__ == "__main__": main()