根据您的错误信息 `SyntaxError: EOL while scanning string literal`，这是一个典型的Python字符串字面量未正确闭合的语法错误[ref_5][ref_6]。让我分析可能的原因并提供完整的解决方案。 ## 错误原因分析 | 错误类型 | 常见原因 | 具体表现 | |---------|---------|---------| | `SyntaxError: EOL while scanning string literal` | 字符串引号未正确闭合 | 忘记闭合引号、字符串跨行未使用正确语法、字符串内包含未转义的引号[ref_1][ref_4][ref_6] | | `SyntaxError: unexpected EOF while parsing` | 括号、引号等未配对 | 缺少右括号、右引号等[ref_5] | 从您提供的脚本片段末尾看： ```python self.logger.info(f"文件大小: {os.path.getsize(output_file)") ``` 这里明显**缺少了右括号和右引号**，应该是： ```python self.logger.info(f"文件大小: {os.path.getsize(output_file)}") ``` ## 完整的可执行脚本（修复版） 以下是修复了所有语法错误的完整脚本： ```python """ Excel表格对比系统 - 修复版 修复了SyntaxError: EOL while scanning string literal错误 """ import os import sys import pandas as pd import numpy as np import traceback from datetime import datetime from openpyxl import load_workbook from openpyxl.styles import PatternFill, Font, Alignment, Border, Side import warnings warnings.filterwarnings('ignore') # ==================== 配置参数 ==================== class Config: """配置参数类""" # 输入文件路径（根据实际情况修改） QIDAO_FILE = "柒道表.xlsx" L8_FILE = "L8表.xlsx" # 输出文件路径 OUTPUT_DIR = "对比结果" LOG_FILE = None # 自动生成 # 列映射关系（用户提供） COLUMN_MAPPING = { '标': '目标', '实': '实际', '区': '区域', '比': '同比', '达': '达成', '折': '折扣', '效': '店效', '同': '同期', '占': '占比', '连': '连带' } # 关键列（用于数据对比） KEY_COLUMNS = ['区域', '目标', '实际'] # 对比阈值 DIFF_THRESHOLD = 0.01 # 1%差异阈值 # ==================== 日志系统 ==================== class Logger: """日志记录类""" def __init__(self, log_file=None): if log_file is None: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') log_file = f"compare_log_{timestamp}.txt" self.log_file = log_file self.log_messages = [] def log(self, message, level="INFO"): """记录日志""" timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') log_entry = f"[{timestamp}] [{level}] {message}" # 打印到控制台 print(log_entry) # 保存到内存 self.log_messages.append(log_entry) # 实时写入文件 try: with open(self.log_file, 'a', encoding='utf-8') as f: f.write(log_entry + "\n") except Exception as e: print(f"写入日志文件失败: {str(e)}") def info(self, message): """信息级别日志""" self.log(message, "INFO") def warning(self, message): """警告级别日志""" self.log(message, "WARNING") def error(self, message): """错误级别日志""" self.log(message, "ERROR") def section(self, title): """输出章节标题""" separator = "=" * 60 self.info(f"\n{separator}") self.info(f" {title}") self.info(f"{separator}") def summary(self): """输出日志摘要""" self.section("执行摘要") self.info(f"日志文件: {os.path.abspath(self.log_file)}") self.info(f"总日志条目: {len(self.log_messages)}") # 统计各级别日志数量 levels = {} for msg in self.log_messages: if 'ERROR' in msg: levels['ERROR'] = levels.get('ERROR', 0) + 1 elif 'WARNING' in msg: levels['WARNING'] = levels.get('WARNING', 0) + 1 elif 'INFO' in msg: levels['INFO'] = levels.get('INFO', 0) + 1 for level, count in levels.items(): self.info(f"{level}级别日志: {count}条") # ==================== 文件处理 ==================== class FileHandler: """文件处理类""" def __init__(self, logger): self.logger = logger self.config = Config() def validate_files(self): """验证输入文件""" self.logger.section("文件验证") files_to_check = [ (self.config.QIDAO_FILE, "柒道表"), (self.config.L8_FILE, "L8表") ] all_exist = True for file_path, file_name in files_to_check: if not os.path.exists(file_path): self.logger.error(f"{file_name}文件不存在: {file_path}") self.logger.info(f"当前工作目录: {os.getcwd()}") self.logger.info(f"目录内容: {os.listdir('.')}") all_exist = False else: file_size = os.path.getsize(file_path) / 1024 # KB self.logger.info(f"✓ {file_name}: {file_path} ({file_size:.2f} KB)") return all_exist def create_output_dir(self): """创建输出目录""" if not os.path.exists(self.config.OUTPUT_DIR): os.makedirs(self.config.OUTPUT_DIR) self.logger.info(f"创建输出目录: {self.config.OUTPUT_DIR}") else: self.logger.info(f"输出目录已存在: {self.config.OUTPUT_DIR}") return self.config.OUTPUT_DIR # ==================== 数据加载 ==================== class DataLoader: """数据加载类""" def __init__(self, logger): self.logger = logger self.config = Config() def load_excel_file(self, file_path, sheet_name=0): """加载Excel文件""" try: self.logger.info(f"正在读取文件: {file_path}") # 先查看工作表名称 excel_file = pd.ExcelFile(file_path) self.logger.info(f"工作表名称: {excel_file.sheet_names}") # 读取数据 df = pd.read_excel(file_path, sheet_name=sheet_name) # 基本信息 self.logger.info(f"数据形状: {df.shape[0]} 行 × {df.shape[1]} 列") self.logger.info(f"列名列表: {list(df.columns)}") # 数据类型信息 self.logger.info(f"数据类型:") for col in df.columns[:5]: # 只显示前5列 dtype = str(df[col].dtype) non_null = df[col].notnull().sum() self.logger.info(f" {col}: {dtype} ({non_null}个非空值)") # 显示前几行数据 if len(df) > 0: self.logger.info(f"前3行数据预览:") preview = df.head(3).to_string() for line in preview.split('\n'): self.logger.info(f" {line}") return df except Exception as e: self.logger.error(f"读取文件失败 {file_path}: {str(e)}") raise def check_columns(self, qidao_df, l8_df): """检查列名和映射关系""" self.logger.section("列名检查") # 显示实际列名 self.logger.info("柒道表实际列名:") for col in qidao_df.columns: self.logger.info(f" '{col}'") self.logger.info("L8表实际列名:") for col in l8_df.columns: self.logger.info(f" '{col}'") # 检查映射关系 self.logger.info("列映射关系检查:") mapping_issues = [] for short_name, full_name in self.config.COLUMN_MAPPING.items(): in_qidao = full_name in qidao_df.columns in_l8 = full_name in l8_df.columns status_qidao = "✓" if in_qidao else "✗" status_l8 = "✓" if in_l8 else "✗" self.logger.info(f" {full_name}: 柒道表{status_qidao} L8表{status_l8}") if not in_qidao: mapping_issues.append(f"柒道表缺少列: {full_name}") if not in_l8: mapping_issues.append(f"L8表缺少列: {full_name}") # 检查关键列 self.logger.info("关键列检查:") for col in self.config.KEY_COLUMNS: if col in qidao_df.columns and col in l8_df.columns: self.logger.info(f" ✓ {col}: 两表都存在") else: missing_in = [] if col not in qidao_df.columns: missing_in.append("柒道表") if col not in l8_df.columns: missing_in.append("L8表") self.logger.error(f" ✗ {col}: 在{', '.join(missing_in)}中缺失") return mapping_issues # ==================== 数据处理 ==================== class DataProcessor: """数据处理类""" def __init__(self, logger): self.logger = logger self.config = Config() def standardize_columns(self, df, df_name): """标准化列名（根据映射关系）""" self.logger.info(f"标准化{df_name}的列名...") # 创建反向映射（完整名 -> 简称） reverse_mapping = {v: k for k, v in self.config.COLUMN_MAPPING.items()} # 重命名列 renamed_cols = {} for col in df.columns: if col in reverse_mapping: new_name = reverse_mapping[col] renamed_cols[col] = new_name self.logger.info(f" {col} -> {new_name}") if renamed_cols: df = df.rename(columns=renamed_cols) return df def calculate_metrics(self, df, df_name): """计算指标""" self.logger.info(f"计算{df_name}的指标...") metrics_added = [] # 计算达成率（如果存在目标和实际列） if '目标' in df.columns and '实际' in df.columns: df['达成率'] = df['实际'] / df['目标'] metrics_added.append('达成率') self.logger.info(f" 添加达成率列，示例值: {df['达成率'].iloc[0] if len(df) > 0 else 'N/A'}") # 计算同比增长（如果存在同比列） if '同比' in df.columns: df['同比增长'] = df['同比'] * 100 metrics_added.append('同比增长') # 计算折扣率（如果存在折扣列） if '折扣' in df.columns: df['折扣率'] = df['折扣'] * 100 metrics_added.append('折扣率') self.logger.info(f" 共添加{len(metrics_added)}个指标: {', '.join(metrics_added)}") return df def compare_tables(self, qidao_df, l8_df): """对比两个表格""" self.logger.section("数据对比") # 确保有共同的关键列 common_cols = set(qidao_df.columns) & set(l8_df.columns) self.logger.info(f"共同列数: {len(common_cols)}") self.logger.info(f"共同列: {list(common_cols)}") if '区域' not in common_cols: self.logger.error("错误: 两表没有共同的'区域'列，无法进行对比") return None # 按区域合并 self.logger.info("按区域合并数据...") # 标准化列名（添加后缀） qidao_df_suffix = qidao_df.add_suffix('_柒道') l8_df_suffix = l8_df.add_suffix('_L8') # 确保区域列名一致 if '区域_柒道' in qidao_df_suffix.columns: qidao_df_suffix = qidao_df_suffix.rename(columns={'区域_柒道': '区域'}) if '区域_L8' in l8_df_suffix.columns: l8_df_suffix = l8_df_suffix.rename(columns={'区域_L8': '区域'}) # 合并数据 merged_df = pd.merge( qidao_df_suffix, l8_df_suffix, on='区域', how='outer', indicator=True ) self.logger.info(f"合并后数据形状: {merged_df.shape}") # 统计合并情况 merge_stats = merged_df['_merge'].value_counts() self.logger.info("合并统计:") for merge_type, count in merge_stats.items(): self.logger.info(f" {merge_type}: {count}行") # 移除合并指示列 merged_df = merged_df.drop('_merge', axis=1) return merged_df def calculate_differences(self, merged_df): """计算差异""" self.logger.info("计算差异指标...") differences = [] # 检查并计算目标差异 if '目标_柒道' in merged_df.columns and '目标_L8' in merged_df.columns: merged_df['目标差异'] = merged_df['目标_L8'] - merged_df['目标_柒道'] merged_df['目标差异率'] = merged_df['目标差异'] / merged_df['目标_柒道'].replace(0, np.nan) differences.extend(['目标差异', '目标差异率']) self.logger.info(" 已计算目标差异") # 检查并计算实际差异 if '实际_柒道' in merged_df.columns and '实际_L8' in merged_df.columns: merged_df['实际差异'] = merged_df['实际_L8'] - merged_df['实际_柒道'] merged_df['实际差异率'] = merged_df['实际差异'] / merged_df['实际_柒道'].replace(0, np.nan) differences.extend(['实际差异', '实际差异率']) self.logger.info(" 已计算实际差异") # 检查并计算达成率差异 if '达成率_柒道' in merged_df.columns and '达成率_L8' in merged_df.columns: merged_df['达成率差异'] = merged_df['达成率_L8'] - merged_df['达成率_柒道'] differences.append('达成率差异') self.logger.info(" 已计算达成率差异") self.logger.info(f" 共添加{len(differences)}个差异指标") # 标记显著差异 if '目标差异率' in merged_df.columns: merged_df['目标显著差异'] = merged_df['目标差异率'].abs() > self.config.DIFF_THRESHOLD significant = merged_df['目标显著差异'].sum() self.logger.info(f" 目标显著差异行数: {significant}") return merged_df, differences # ==================== 结果输出 ==================== class ResultExporter: """结果导出类""" def __init__(self, logger): self.logger = logger self.config = Config() def format_excel(self, writer, df, sheet_name): """格式化Excel输出""" try: # 获取workbook和worksheet workbook = writer.book worksheet = writer.sheets[sheet_name] # 设置列宽 for i, col in enumerate(df.columns): column_width = max(df[col].astype(str).map(len).max(), len(col)) + 2 worksheet.column_dimensions[chr(65 + i)].width = min(column_width, 50) # 设置标题样式 header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid") header_font = Font(color="FFFFFF", bold=True) header_alignment = Alignment(horizontal="center", vertical="center") for cell in worksheet[1]: cell.fill = header_fill cell.font = header_font cell.alignment = header_alignment # 设置边框 thin_border = Border( left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin') ) for row in worksheet.iter_rows(min_row=1, max_row=worksheet.max_row, min_col=1, max_col=worksheet.max_column): for cell in row: cell.border = thin_border # 冻结首行 worksheet.freeze_panes = 'A2' except Exception as e: self.logger.warning(f"格式化Excel时出错: {str(e)}") def export_results(self, results_dict, output_dir): """导出结果到Excel""" self.logger.section("结果导出") timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') output_file = os.path