import pandas as pd import requests import json from time import sleep from module import F6_module import mysql.connector from mysql.connector import Error from datetime import datetime class CouponDataProcessor: def __init__(self): self.f6_module = F6_module() self.base_url = "https://yunxiu.f6car.cn/macan/coupon/info/pagingCouponUsageRecord" self.headers = { 'accept': 'application/json, text/plain, */*', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'referer': 'https://yunxiu.f6car.cn/erp/view/index.html' } self.db_config = { 'host': "f6-public.rwlb.rds.aliyuncs.com", 'user': "rw_operation_data_relay", 'password': "m+q5Z4%IVuF9bf", 'database': "f6operation_data_relay" } # 衡时数据库链接配置-mysql self.username = "15222738424" self.password = "cw25966929@" def drop_column(self, cursor, table_name, column_name): """删除表中的指定列""" try: # 检查列是否存在 cursor.execute(f"SHOW COLUMNS FROM {table_name} LIKE '{column_name}'") if cursor.fetchone(): # 如果列存在，则删除 drop_query = f"ALTER TABLE {table_name} DROP COLUMN {column_name}" cursor.execute(drop_query) print(f"成功从表 {table_name} 中删除列 {column_name}") else: print(f"表 {table_name} 中不存在列 {column_name}") except Error as e: print(f"删除列失败: {e}") def _fetch_all_coupons(self, page_size=100): """获取所有分页数据""" cookies = self._login() params = { 'keyword': '', 'couponName': '', 'currentPage': '1', 'pageSize': str(page_size), 'sorts': '' } # 获取第一页确定总页数 first_page = self._fetch_page(params, cookies) if not first_page: return None total_records = first_page.get('info', {}).get('total', 0) if total_records == 0: return None total_pages = (total_records + page_size - 1) // page_size print(f"共发现 {total_records} 条记录，{total_pages} 页") # 收集所有数据 all_data = first_page.get('info', {}).get('list', []) for page in range(2, total_pages + 1): params['currentPage'] = str(page) print(f"正在获取第 {page}/{total_pages} 页...") page_data = self._fetch_page(params, cookies) if page_data: all_data.extend(page_data.get('info', {}).get('list', [])) sleep(0.5) # 礼貌延迟 return all_data def _login(self): """登录获取cookies""" res = self.f6_module.login_in(self.username, self.password) return requests.utils.dict_from_cookiejar(res.cookies) def _fetch_page(self, params, cookies, max_retries=3): """带重试机制的页面请求""" for attempt in range(max_retries): try: response = requests.get( self.base_url, params=params, cookies=cookies, headers=self.headers, timeout=10 ) response.raise_for_status() return response.json() except Exception as e: print(f"请求失败（尝试 {attempt + 1}/{max_retries}）: {str(e)}") if attempt < max_retries - 1: sleep(2) return None def _process_data(self, raw_data): """处理原始数据""" df = pd.DataFrame(raw_data) if not df.empty: # 处理couponCarList字段（列表/字典转为JSON字符串） if 'couponCarList' in df.columns: df['couponCarList'] = df['couponCarList'].apply( lambda x: json.dumps(x, ensure_ascii=False) if pd.notna(x) else None ) # 同时提取carId和carNo df['carId'] = df['couponCarList'].apply( lambda x: json.loads(x)[0].get('carId') if pd.notna(x) else None ) df['carNo'] = df['couponCarList'].apply( lambda x: json.loads(x)[0].get('carNo') if pd.notna(x) else None ) # 处理couponInfo字段（字典转为JSON字符串） if 'couponInfo' in df.columns: df['couponInfo'] = df['couponInfo'].apply( lambda x: json.dumps(x, ensure_ascii=False) if pd.notna(x) else None ) # 同时展开部分常用字段 try: coupon_info = pd.json_normalize(df['couponInfo'].apply( lambda x: json.loads(x) if pd.notna(x) else {} )) df = pd.concat([df, coupon_info.add_prefix('couponInfo.')], axis=1) except Exception as e: print(f"展开couponInfo时出错: {str(e)}") # 处理时间字段 if 'takeTime' in df.columns: df['takeTime'] = pd.to_datetime(df['takeTime'], unit='ms') if 'useTime' in df.columns: df['useTime'] = pd.to_datetime(df['useTime'], unit='ms') # 重命名列 if 'id' in df.columns: df = df.rename(columns={'id': 'id1'}) return df def _import_to_database(self, df, table_name="coupon_usage_record_details", batch_size=1000): """直接将处理后的DataFrame导入MySQL""" conn = None cursor = None try: # 连接数据库 conn = mysql.connector.connect(**self.db_config) cursor = conn.cursor() # 删除表中的所有数据 print(f"正在清空表 {table_name} 中的数据...") cursor.execute(f"DELETE FROM {table_name}") cursor.execute(f"ALTER TABLE {table_name} AUTO_INCREMENT = 1") conn.commit() print(f"已成功清空表 {table_name} 中的所有数据") # 处理时间类型数据 datetime_columns = [col for col in df.columns if df[col].dtype == 'datetime64[ns]'] for col in datetime_columns: df[col] = df[col].apply(self._convert_datetime) # 处理所有数据，将NaN转为None df = df.where(pd.notna(df), None) # 获取数据库列信息 cursor.execute(f"SHOW COLUMNS FROM {table_name}") db_columns = [col[0] for col in cursor.fetchall() if col[0] != 'id'] # 确保DataFrame列与数据库列一致 df = df[db_columns] # 生成插入语句 columns = ', '.join([f"`{col}`" for col in df.columns]) placeholders = ', '.join(['%s'] * len(df.columns)) insert_query = f"INSERT INTO `{table_name}` ({columns}) VALUES ({placeholders})" # 分批插入数据 print("开始导入数据...") total_rows = len(df) for i in range(0, total_rows, batch_size): batch = df.iloc[i:i + batch_size] # 将DataFrame转换为元组列表，并处理所有数据类型 records = [tuple(self._convert_datetime(val) if isinstance(val, (pd.Timestamp, datetime)) else val for val in row) for row in batch.values] try: cursor.executemany(insert_query, records) conn.commit() print(f"已导入 {min(i + batch_size, total_rows)}/{total_rows} 条记录") except Error as e: conn.rollback() print(f"批量导入失败: {e}") # 尝试逐条导入以找出问题行 for idx, record in enumerate(records): try: cursor.execute(insert_query, record) conn.commit() except Error as e: print(f"第 {i + idx + 1} 行导入失败: {e}") print(f"问题数据: {record}") conn.rollback() print(f"成功导入 {total_rows} 条记录到 {table_name} 表") except Error as e: print(f"数据库操作失败: {e}") except Exception as e: print(f"发生错误: {e}") finally: if cursor: cursor.close() if conn: conn.close() @staticmethod def _convert_datetime(value): """将Pandas/NumPy时间类型转换为MySQL兼容的datetime""" if pd.isna(value): return None if isinstance(value, pd.Timestamp): return value.to_pydatetime() if isinstance(value, datetime): return value return value def execute_pipeline(self): """执行完整数据处理流程""" try: # 1. 获取数据 print("开始获取优惠券数据...") raw_data = self._fetch_all_coupons() if not raw_data: raise Exception("未能获取有效数据") # 2. 处理数据 print("处理数据中...") processed_df = self._process_data(raw_data) # 3. 直接导入数据库 self._import_to_database(processed_df) print("数据处理流程完成！") except Exception as e: print(f"流程执行失败: {e}") if __name__ == "__main__": processor = CouponDataProcessor() processor.execute_pipeline()