import logging # 导入日志模块，用于记录日志信息 import os # 导入操作系统模块，用于处理文件和目录操作 from SQLiteInterface import SQLiteInterface # 从SQLiteInterface模块导入SQLiteInterface类 import pandas as pd # 导入Pandas模块，用于数据处理 import chardet # 导入chardet模块，用于检测文件编码 import shutil # 导入shutil模块，用于文件操作 import glob # 导入glob模块，用于文件模式匹配 # 通过装饰器统一捕捉并记录函数中可能抛出的任何异常 def exception_handler(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logging.error(f"An error occurred in {func.__name__}: {e}") return None return wrapper # 定义FetchDataEngineAlipay类，用于处理数据抓取、解析和存储 class FetchDataEngineAlipay: # 初始化方法，设置数据源类型、数据类型、API配置和文件路径等 def __init__(self, db_interface, source_type='API', data_type='Fund', api_config=None, file_path=None): self.db_interface = db_interface # 数据库接口实例 self.source_type = source_type # 数据源类型（API或文件） self.data_type = data_type # 数据类型（基金或股票） self.api_config = api_config or {} # API配置 self.file_path = file_path or os.path.join(os.getcwd(), 'DataSource') # 文件路径 # 主执行方法，负责数据的抓取、解析和存储 @exception_handler def execute(self): raw_data = self._fetch_data() # 抓取数据 if not raw_data: logging.error("Failed to fetch data.") return False parsed_data = self._parse_data(raw_data) # 解析数据 if not parsed_data: logging.error("Failed to parse data.") return False success = self._save_data(parsed_data) # 存储数据 if not success: logging.error("Failed to save data.") return False logging.info("Data fetch, parse, and save successful.") return True # 根据数据源类型抓取数据 @exception_handler def _fetch_data(self): if self.source_type == 'API': return self._fetch_api_data() elif self.source_type == 'File': return self._fetch_file_data() # 根据数据源类型解析数据 @exception_handler def _parse_data(self, raw_data): if self.source_type == 'API': return self._parse_api_data(raw_data) elif self.source_type == 'File': return self._parse_file_data(raw_data) # 根据数据源类型存储数据 @exception_handler def _save_data(self, parsed_data): if self.source_type == 'API': return self._save_api_data(parsed_data) elif self.source_type == 'File': return self._save_file_data(parsed_data) # 抓取API数据 @exception_handler def _fetch_api_data(self): if self.data_type == 'Fund': return self._fetch_fund_data() elif self.data_type == 'Stock': return self._fetch_stock_daily_data() # 抓取文件数据 @exception_handler def _fetch_file_data(self): file_patterns = { 'alipay': ('alipay_record*.csv', self._clean_csv_file, {"header_keyword": "交易时间"}), } cleaned_files_with_types = [] for source_type, (pattern, processing_func, kwargs) in file_patterns.items(): file_path_pattern = os.path.join(self.file_path, pattern) for file_path in glob.glob(file_path_pattern): cleaned_file = processing_func(file_path, **kwargs) if cleaned_file: cleaned_files_with_types.append((cleaned_file, source_type)) return cleaned_files_with_types # 解析API数据 @exception_handler def _parse_api_data(self, raw_data): if self.data_type == 'Fund': return {'nav_data': self._fetch_nav_from_js(raw_data), 'fund_info': self._fetch_fund_info_from_js(raw_data)} elif self.data_type == 'Stock': return self._parse_stock_data(raw_data) # 解析文件数据 @exception_handler def _parse_file_data(self, raw_data): parsed_data_list = [] for cleaned_file, source_type in raw_data: parsed_data = self._read_file(cleaned_file, source_type) if parsed_data is not None: parsed_data_list.append((parsed_data, source_type)) return parsed_data_list # 存储API数据 @exception_handler def _save_api_data(self, parsed_data): if self.data_type == 'Fund': return self._persist_fund_data(parsed_data) elif self.data_type == 'Stock': return self._persist_stock_data(parsed_data) # 存储文件数据 @exception_handler def _save_file_data(self, parsed_data): for data, source_type in parsed_data: if source_type == 'alipay': self._persist_alipay_transactions(data) logging.info("Successfully saving File data.") return True # 读取文件数据 @exception_handler def _read_file(self, file_path, source_type): file_extension = os.path.splitext(file_path)[1].lower() if file_extension == '.csv': return self._read_csv(file_path, source_type) elif file_extension in ['.xls', '.xlsx']: return self.read_excel(file_path, source_type) else: raise ValueError(f"Unsupported file type: {file_extension}") # 移动处理后的文件到已处理文件夹 @exception_handler def _move_processed_file(self, processed_file_path): processed_folder = os.path.join(self.file_path, 'Processed') if not os.path.exists(processed_folder): os.makedirs(processed_folder) shutil.move(processed_file_path, os.path.join(processed_folder, os.path.basename(processed_file_path))) # 清理CSV文件 @exception_handler def _clean_csv_file(self, file_path, header_keyword="交易时间", footer_keyword=None): with open(file_path, 'r', encoding='GB2312', errors='replace') as f: lines = f.readlines() header_found = False cleaned_lines = [] for line in lines: line = line.strip() if header_keyword in line and not header_found: header_found = True header_fields = line.split(',') cleaned_header_fields = [field.strip() for field in header_fields] cleaned_lines.append(','.join(cleaned_header_fields)) elif header_found: data_fields = line.split(',') cleaned_data_fields = [field.strip() for field in data_fields] cleaned_lines.append(','.join(cleaned_data_fields)) if not header_found: raise ValueError(f"Header keyword '{header_keyword}' not found in file: {file_path}") if cleaned_lines: cleaned_file_path = file_path.replace('.csv', '_cleaned.csv') with open(cleaned_file_path, 'w', encoding='UTF-8') as f: f.write("\n".join(cleaned_lines)) self._move_processed_file(file_path) return cleaned_file_path else: raise ValueError("No columns to parse from file") # 读取CSV文件 @exception_handler def _read_csv(self, file_path, source_type): with open(file_path, 'rb') as f: result = chardet.detect(f.read()) encoding = result['encoding'] df = pd.read_csv(file_path, encoding=encoding) amount_column = '金额' if source_type != 'cmbpay' else '交易金额' df[amount_column] = df[amount_column].apply(str) df[amount_column] = df[amount_column].str.replace(r'[￥,¥]', '').str.strip() df[amount_column] = pd.to_numeric(df[amount_column], errors='coerce') df['交易时间'] = pd.to_datetime(df['交易时间'], errors='coerce').dt.strftime('%Y-%m-%d %H:%M:%S') if source_type == 'cmbpay': amount_column2 = '联机余额' df[amount_column2] = df[amount_column2].apply(str).str.replace(r'[￥,¥]', '').str.strip() df[amount_column2] = pd.to_numeric(df[amount_column2], errors='coerce') df['记账日期'] = pd.to_datetime(df['记账日期'], errors='coerce').dt.strftime('%Y-%m-%d %H:%M:%S') self._move_processed_file(file_path) return df # 持久化存储支付宝交易数据 @exception_handler def _persist_alipay_transactions(self, alipay_df): for index, row in alipay_df.iterrows(): existing_record = self.db_interface.read( 'SELECT * FROM AlipayCashFlowStatement WHERE date = ? AND amount = ? AND description = ?', (row['交易时间'], row['金额'], row['商品说明']) ) if not existing_record: self.db_interface.write( 'INSERT INTO AlipayCashFlowStatement (date, category, counterparty, description, transaction_type, amount, payment_method, status, note, cashflow_account_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', (row['交易时间'], row['交易分类'], row['交易对方'], row['商品说明'], row['收/支'], row['金额'], row['收/付款方式'], row['交易状态'], row['备注'], 'alipay') ) # 主程序 db_interface = SQLiteInterface() fetch_engine = FetchDataEngineAlipay( db_interface=db_interface, source_type='File', data_type='Alipay', file_path='./DataSource' ) fetched_data = fetch_engine.execute() print("Fetched data:", fetched_data)