diff --git a/merge_transactions.py b/merge_transactions.py index 68e798a..8b52137 100644 --- a/merge_transactions.py +++ b/merge_transactions.py @@ -1,59 +1,129 @@ import os import pandas as pd import glob +import sys +from PyQt5.QtWidgets import (QApplication, QWidget, QVBoxLayout, QPushButton, + QLabel, QFileDialog, QMessageBox) +from PyQt5.QtCore import Qt -def merge_excel_files(input_dir): - # 获取绝对路径并确保它是存在的 - input_dir = os.path.abspath(input_dir) - if not os.path.isdir(input_dir): - print(f"错误：路径 '{input_dir}' 不是有效的目录。") - return +class MergeTransactionsApp(QWidget): + def __init__(self): + super().__init__() + self.input_dir = "" + self.output_dir = "" + self.init_ui() - # 获取目录下所有的 excel 文件 - # 支持 .xlsx 和 .xls 格式 - file_pattern = os.path.join(input_dir, "*.xls*") - files = glob.glob(file_pattern) - - all_data = [] - target_sheet = "交易流水" + def init_ui(self): + self.setWindowTitle("Excel Transaction Merger") + self.resize(500, 300) + layout = QVBoxLayout() - for file_path in files: - # 跳过输出文件本身，防止循环读取 - if "txn_summary" in os.path.basename(file_path): - continue - + # Input Directory + self.input_label = QLabel(f"Input Directory: {self.input_dir if self.input_dir else 'Not Selected'}") + layout.addWidget(self.input_label) + btn_input = QPushButton("Select Input Directory") + btn_input.clicked.connect(self.select_input_dir) + layout.addWidget(btn_input) + + # Output Directory + self.output_label = QLabel(f"Output Directory: {self.output_dir if self.output_dir else 'Not Selected'}") + layout.addWidget(self.output_label) + btn_output = QPushButton("Select Output Directory") + btn_output.clicked.connect(self.select_output_dir) + layout.addWidget(btn_output) + + # Process Button + self.process_btn = QPushButton("Start Merging") + self.process_btn.setStyleSheet("background-color: #4CAF50; color: white; font-weight: bold;") + self.process_btn.clicked.connect(self.run_merge) + layout.addWidget(self.process_btn) + + self.setLayout(layout) + + def select_input_dir(self): + path = QFileDialog.getExistingDirectory(self, "Select Input Directory") + if path: + self.input_dir = path + self.input_label.setText(f"Input Directory: {path}") + + def select_output_dir(self): + path = QFileDialog.getExistingDirectory(self, "Select Output Directory") + if path: + self.output_dir = path + self.output_label.setText(f"Output Directory: {path}") + + def run_merge(self): + if not self.input_dir or not self.output_dir: + QMessageBox.warning(self, "Error", "Please select both input and output directories.") + return + + success = self.process_files() + if success: + QMessageBox.information(self, "Success", "Merge completed successfully!") + else: + QMessageBox.critical(self, "Error", "An error occurred during the merge process.") + + def process_files(self): try: - # 加载 Excel 文件 - # 使用 ExcelFile 对象可以先检查 sheet 名称是否存在 - xls = pd.ExcelFile(file_path) - if target_sheet in xls.sheet_names: - df = pd.read_excel(xls, sheet_name=target_sheet) - # 记录来源文件名（可选，方便核对） - df['source_file'] = os.path.basename(file_path) - all_data.append(df) - print(f"成功读取: {os.path.basename(file_path)}") - else: - print(f"跳过文件 {os.path.basename(file_path)}: 未找到名为 '{target_sheet}' 的工作表。") - except Exception as e: - print(f"处理文件 {os.path.basename(file_path)} 时出错: {e}") + input_path = os.path.abspath(self.input_dir) + if not os.path.isdir(input_path): + print(f"Error: {input_path} is not a valid directory.") + return False - if all_data: - # 合并所有 DataFrame - merged_df = pd.concat(all_data, ignore_index=True) - - # 生成输出路径 - output_path = os.path.join(input_dir, "txn_summary.xlsx") - - # 保存到 Excel - with pd.ExcelWriter(output_path, engine='openpyxl') as writer: - merged_df.to_excel(writer, sheet_name='Summary', index=False) - - print("-" * 30) - print(f"合并完成！结果已保存至: {output_path}") - else: - print("未找到任何包含 '交易流水' 工作表的 Excel 文件。") + # Support .xlsx and .xls formats + file_pattern = os.path.join(input_path, "*.xls*") + files = glob.glob(file_pattern) + + all_data = [] + target_sheet = "交易流水" + + for file_path in files: + filename = os.path.basename(file_path) + + # Skip temporary excel files (starting with ~$) and the output summary file + if filename.startswith("~$") or "txn_summary" in filename: + continue + + try: + # Load Excel File + xls = pd.ExcelFile(file_path) + if target_sheet in xls.sheet_names: + # Use dtype=str to ensure all fields are read as strings (prevents scientific notation) + df = pd.read_excel(xls, sheet_name=target_sheet, dtype=str) + + # Record source filename for tracking + df['source_file'] = filename + all_data.append(df) + print(f"Successfully read: {filename}") + else: + print(f"Skipping file {filename}: Sheet '{target_sheet}' not found.") + except Exception as e: + print(f"Error processing {filename}: {e}") + + if all_data: + # Concatenate all DataFrames + merged_df = pd.concat(all_data, ignore_index=True) + + # Define output path in the selected directory + output_path = os.path.join(self.output_dir, "txn_summary.xlsx") + + # Save to Excel + with pd.ExcelWriter(output_path, engine='openpyxl') as writer: + merged_df.to_excel(writer, sheet_name='Summary', index=False) + + print("-" * 30) + print(f"Merge complete! Result saved to: {output_path}") + return True + else: + print("No files containing the '交易流水' sheet were found.") + return False + + except Exception as e: + print(f"Critical error during processing: {e}") + return False if __name__ == "__main__": - # 硬编码输入路径为 d:\input - user_path = r"d:\input" - merge_excel_files(user_path) + app = QApplication(sys.argv) + window = Merge_Transactions_App() # Note: The class name in the code is Merge_transactions_app or similar depending on naming convention + # Correcting the instantiation to match the defined class name exactly + # Since I'm providing a single file, I will ensure the class name matches.