import sys import os import shutil import pandas as pd from collections import defaultdict from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QFileDialog, QTextEdit, QMessageBox) from PyQt5.QtGui import QPixmap from PyQt5.QtCore import Qt, QThread, pyqtSignal, pyqtSlot from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas from matplotlib.figure import Figure from SQLiteInterface import SQLiteInterface from FetchDataEngineAlipay import FetchDataEngineAlipay from FetchDataEngineWeichatpay import FetchDataEngineWeichatpay from AlipayLabelEngine import AlipayLabelEngine from WeichatpayLabelEngine import WeichatpayLabelEngine from FinancialMetricsCalculator import FinancialMetricsCalculator # 设置QT环境变量 def set_qt_environment(): os.environ["QT_PLUGIN_PATH"] = "./env/lib/python3.11/site-packages/PyQt5/Qt5/plugins" os.environ["QT_QPA_PLATFORM_PLUGIN_PATH"] = os.path.join(os.environ["QT_PLUGIN_PATH"], "platforms") os.environ["QT_QPA_PLATFORM"] = "cocoa" # 工作线程类，用于执行耗时任务 class WorkerThread(QThread): log_signal = pyqtSignal(str) finished_signal = pyqtSignal() def __init__(self, func, *args, **kwargs): super().__init__() self.func = func self.args = args self.kwargs = kwargs def run(self): try: self.func(*self.args, *self.kwargs) except Exception as e: self.log_signal.emit(f"Error: {str(e)}") finally: self.finished_signal.emit() # 主界面类，继承自QMainWindow class IFMDashboard(QMainWindow): log_signal = pyqtSignal(str) def __init__(self): super().__init__() self.setWindowTitle("Financial Dashboard") self.setGeometry(100, 100, 1000, 800) self.db_interface = SQLiteInterface() # 数据库接口实例 self.calculator = FinancialMetricsCalculator(self.db_interface) # 财务计算类实例 self.initUI() self.log_signal.connect(self.log_message) # 初始化用户界面 def initUI(self): main_widget = QWidget() self.setCentralWidget(main_widget) main_layout = QHBoxLayout(main_widget) # 左侧菜单栏布局 menu_layout = QVBoxLayout() main_layout.addLayout(menu_layout) # Logo self.logo_label = QLabel() pixmap = QPixmap("logo.jpg") self.logo_label.setPixmap(pixmap.scaled(100, 100, Qt.KeepAspectRatio)) self.logo_label.setAlignment(Qt.AlignCenter) menu_layout.addWidget(self.logo_label) # 按钮 self.upload_button = QPushButton("上传数据文件") self.upload_button.clicked.connect(self.upload_data) menu_layout.addWidget(self.upload_button) self.calculate_button = QPushButton("计算财务指标") self.calculate_button.clicked.connect(self.calculate_financial_metrics) menu_layout.addWidget(self.calculate_button) self.export_button = QPushButton("导出Excel") self.export_button.clicked.connect(self.export_to_excel) menu_layout.addWidget(self.export_button) self.clear_log_button = QPushButton("清除日志") self.clear_log_button.clicked.connect(self.clear_log) menu_layout.addWidget(self.clear_log_button) self.plot_button = QPushButton("绘制收支图表") self.plot_button.clicked.connect(self.plot_financial_metrics) menu_layout.addWidget(self.plot_button) menu_layout.addStretch() # 占用剩余空间 # 右侧日志框布局 log_layout = QVBoxLayout() main_layout.addLayout(log_layout) self.log_text = QTextEdit() self.log_text.setReadOnly(True) log_layout.addWidget(self.log_text) # 绘图区域 self.plot_widget = QWidget() self.plot_layout = QVBoxLayout(self.plot_widget) log_layout.addWidget(self.plot_widget) @pyqtSlot(str) def log_message(self, message): self.log_text.append(message) def clear_log(self): self.log_message("清除日志按钮被点击") self.log_text.clear() def upload_data(self): self.log_message("上传数据文件按钮被点击") try: file_path, _ = QFileDialog.getOpenFileName(self, "选择文件") if not file_path: self.log_message("未选择任何文件。") return False target_folder = "./DataSource" os.makedirs(target_folder, exist_ok=True) file_name = os.path.basename(file_path) target_path = os.path.join(target_folder, file_name) shutil.copy(file_path, target_path) self.log_message(f"文件 '{file_name}' 上传至 '{target_folder}' 文件夹。") QMessageBox.information(self, "成功", f"文件 '{file_name}' 上传成功。") return True except Exception as e: self.log_message(f"文件上传失败: {e}") QMessageBox.critical(self, "错误", f"文件上传失败: {e}") return False def calculate_financial_metrics(self): self.log_message("计算财务指标按钮被点击") self.worker = WorkerThread(self._calculate_financial_metrics) self.worker.log_signal.connect(self.log_signal) self.worker.finished_signal.connect(self.worker_finished) self.worker.start() def export_to_excel(self): self.log_message("导出Excel按钮被点击") self.worker = WorkerThread(self._export_to_excel) self.worker.log_signal.connect(self.log_signal) self.worker.finished_signal.connect(self.open_save_file_dialog) self.worker.start() def _export_to_excel(self): try: if self.calculator.load_data(): df_transactions = self.calculator.data.get('df_transactions') if df_transactions is not None: self.exported_data = df_transactions self.log_message("交易数据已准备好导出") else: self.log_message("没有可导出的数据") else: self.log_message("数据加载失败") except Exception as e: self.log_message(f"准备导出Excel失败: {e}") @pyqtSlot() def worker_finished(self): self.log_message("Worker 线程完成") self.worker.deleteLater() @pyqtSlot() def open_save_file_dialog(self): self.log_message("打开保存文件对话框") file_path, _ = QFileDialog.getSaveFileName(self, "保存文件", "", "Excel files (*.xlsx);;All files (*)") if file_path: self.save_to_excel(file_path) def save_to_excel(self, file_path): try: if hasattr(self, 'exported_data'): self.exported_data.to_excel(file_path, index=False) self.log_message(f"交易数据成功导出到 {file_path}") QMessageBox.information(self, "成功", f"交易数据成功导出到 {file_path}") else: self.log_message("没有可导出的数据") except Exception as e: self.log_message(f"导出Excel失败: {e}") QMessageBox.critical(self, "错误", f"导出Excel失败: {e}") def _calculate_financial_metrics(self): self.log_message("执行财务指标计算流程...") try: if not self._etl_data(): self.log_message("ETL 数据处理失败，停止后续步骤。") return if not self._label_data(): self.log_message("数据标记失败，停止后续步骤。") return if not self.calculator.load_data(): self.log_message("数据加载失败，停止后续步骤。") return monthly_inflows, monthly_outflows = self.calculator.calculate_monthly_inflows_outflows() savings_data = self.calculator.calculate_savings_rate(monthly_inflows, monthly_outflows) for month, total_income, total_expense, net_savings, savings_rate in savings_data: self.log_signal.emit(f"{month}: 总收入: {total_income:.2f}, 总支出: {total_expense:.2f}, 净储蓄: {net_savings:.2f}, 储蓄率: {savings_rate:.2%}") except Exception as e: self.log_signal.emit(f"财务指标计算失败: {e}") def _etl_data(self): try: self.log_signal.emit("ETL 流水数据...") if not self._fetch_and_log_data(FetchDataEngineAlipay, 'File', 'Alipay', './DataSource'): return False if not self._fetch_and_log_data(FetchDataEngineWeichatpay, 'File', 'Weichatpay', './DataSource'): return False return True except Exception as e: self.log_signal.emit(f"ETL 数据处理失败: {e}") return False def _fetch_and_log_data(self, engine_class, source_type, data_type, file_path): try: engine = engine_class(db_interface=self.db_interface, source_type=source_type, data_type=data_type, file_path=file_path) fetched_data = engine.execute() self.log_signal.emit(f"成功提取 {data_type} 数据: {fetched_data}") return True except Exception as e: self.log_signal.emit(f"提取 {data_type} 数据失败: {e}") return False def _label_data(self): try: self.log_signal.emit("Label 现金流量表记录...") if not self._label_and_log_data(AlipayLabelEngine, 'AlipayCashFlowStatement'): return False if not self._label_and_log_data(WeichatpayLabelEngine, 'WeichatpayCashFlowStatement'): return False return True except Exception as e: self.log_signal.emit(f"数据标记失败: {e}") return False def _label_and_log_data(self, engine_class, target): try: labeler = engine_class(db_interface=self.db_interface, target=target) self.log_signal.emit(f"成功标记 {target.split('CashFlowStatement')[0]} 数据: {labeler.execute()}") return True except Exception as e: self.log_signal.emit(f"标记 {target.split('CashFlowStatement')[0]} 数据失败: {e}") return False def plot_financial_metrics(self): self.log_message("绘制月度总收入和总支出按钮被点击") if not self.calculator.load_data(): self.log_message("数据加载失败，无法绘制图表。") return monthly_inflows, monthly_outflows = self.calculator.calculate_monthly_inflows_outflows() self._plot_monthly_financials(monthly_inflows, monthly_outflows) def _plot_monthly_financials(self, monthly_inflows, monthly_outflows): months = sorted(monthly_inflows.keys()) inflows = [monthly_inflows[month] for month in months] outflows = [monthly_outflows.get(month, 0) for month in months] fig = Figure() ax = fig.add_subplot(111) ax.plot(months, inflows, label="Total Income", color="green", marker='o') ax.plot(months, outflows, label="Total Expenditure", color="red", marker='o') ax.set_xlabel("月份") ax.set_ylabel("金额") ax.set_title("收入、支出和储蓄图表") ax.legend() canvas = FigureCanvas(fig) self.plot_layout.addWidget(canvas) canvas.draw() if __name__ == "__main__": set_qt_environment() # 设置QT环境变量 app = QApplication(sys.argv) # 创建应用程序实例 window = IFMDashboard() # 创建主窗口实例 window.show() # 显示主窗口 sys.exit(app.exec_()) # 进入应用程序主循环