130 lines
5.0 KiB
Python
130 lines
5.0 KiB
Python
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
|
|
|
|
class MergeTransactionsApp(QWidget):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.input_dir = ""
|
|
self.output_dir = ""
|
|
self.init_ui()
|
|
|
|
def init_ui(self):
|
|
self.setWindowTitle("Excel Transaction Merger")
|
|
self.resize(500, 300)
|
|
layout = QVBoxLayout()
|
|
|
|
# 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:
|
|
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
|
|
|
|
# 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__":
|
|
app = QApplication(sys.argv)
|
|
window = MergeTransactionsApp()
|
|
window.show()
|
|
sys.exit(app.exec_())
|