from decimal import Decimal from pathlib import Path import pandas as pd from openpyxl import load_workbook from db_helper import DBHelper from utils.constants import DOWNLOAD_DIR, GUANGDONG_CITY from utils.log import log from utils.parse_utils import traverse_and_process, extract_year_month_from_path, get_previous_month_dir # 配置日志 PROV_CODE = "440000" PROV_NAME = "广东省" download_dir = DOWNLOAD_DIR / "guangdong" db = DBHelper() def match_customs_file(filename, customs_name, year, month): """匹配海关文件""" filename_lower = filename.lower() if customs_name == "广州海关": return "7地市进出口综合统计" in filename_lower elif customs_name == "深圳海关": return "深圳海关综合统计资料" in filename_lower elif customs_name == "汕头海关": return "5市报表" in filename_lower elif customs_name == "黄埔海关": return "东莞市进出口企业性质总值表" in filename_lower elif customs_name == "江门海关": if "江门市" in filename_lower or "阳江市" in filename_lower: return "外贸进出口有关情况统计表" in filename_lower elif customs_name == "湛江海关": if "湛江市" in filename_lower or "茂名市" in filename_lower: return "外贸进出口数据" in filename_lower return False def process_guangzhou_customs(file_path, year, month): """处理广州海关数据""" try: # 读取Excel文件 wb = load_workbook(file_path, data_only=True) sheet = wb.worksheets[0] # 查找包含月份的表头行 month_str = f"{year}年{month}月" header_row = None for i in range(1, 4): # 检查前3行 row_values = [str(cell.value).strip() if cell.value else "" for cell in sheet[i]] if any(month_str in val for val in row_values): header_row = i break if header_row is None: log.error(f"未找到 {month_str} 的表头") return pd.DataFrame() # 确定数据列位置 data_cols = [] for cell in sheet[header_row]: if cell.value and month_str in str(cell.value): data_cols.append(cell.column - 1) # 转换为0-based索引 if len(data_cols) < 6: log.error(f"未找到足够的 {month_str} 数据列") return pd.DataFrame() # 提取7地市数据 results = [] target_cities = ["广州市", "深圳市", "东莞市", "汕头市", "江门市", "湛江市", "茂名市"] for row in sheet.iter_rows(min_row=header_row + 1): city_cell = row[0].value if city_cell and "广东省" in str(city_cell): city_name = str(city_cell).replace("广东省", "").strip() if city_name in target_cities: try: # 获取各列值 total = row[data_cols[0]].value export = row[data_cols[1]].value import_val = row[data_cols[2]].value yoy_total = row[data_cols[3]].value yoy_export = row[data_cols[4]].value yoy_import = row[data_cols[5]].value # 转换数据类型 def convert_value(val): if isinstance(val, (int, float)): return Decimal(str(val)) elif isinstance(val, str) and val.replace(".", "").isdigit(): return Decimal(val) return Decimal(0) # 添加到结果 results.append({ "city_name": city_name, "monthly_total": convert_value(total), "monthly_import": convert_value(import_val), "monthly_export": convert_value(export), "yoy_import_export": convert_value(yoy_total), "yoy_import": convert_value(yoy_import), "yoy_export": convert_value(yoy_export) }) except Exception as e: log.error(f"处理城市 {city_name} 出错: {e}") return pd.DataFrame(results) except Exception as e: log.error(f"处理广州海关文件出错: {str(e)}") return pd.DataFrame() def process_shenzhen_customs(file_path, year, month): """处理深圳海关数据""" try: wb = load_workbook(file_path, data_only=True) results = [] # 处理深圳和惠州两个sheet for city, sheet_name in [("深圳市", "深圳市进出口（贸易方式）"), ("惠州市", "惠州市进出口（贸易方式）")]: try: if sheet_name in wb.sheetnames: sheet = wb[sheet_name] else: log.warning(f"未找到sheet: {sheet_name}") continue # 查找总值行 total_row_idx = None for i, row in enumerate(sheet.iter_rows(values_only=True), 1): if row and "总值" in str(row[0]): total_row_idx = i break if total_row_idx is None: log.error(f"未找到总值行: {sheet_name}") continue # 查找包含月份的表头 month_str = f"{year}年{month}月" header_row = None data_col = None for i, row in enumerate(sheet.iter_rows(max_row=3, values_only=True), 1): if any(month_str in str(cell) for cell in row if cell): header_row = i for col_idx, cell_val in enumerate(row): if cell_val and month_str in str(cell_val): data_col = col_idx break break if data_col is None: log.error(f"未找到 {month_str} 列") continue # 获取数据值 (亿元转换为万元) total_value = sheet.cell(row=total_row_idx, column=data_col + 1).value yoy_value = sheet.cell(row=total_row_idx, column=data_col + 2).value if total_value is None or yoy_value is None: log.error(f"{city} 数据为空") continue # 转换数据类型 def convert_value(val): if isinstance(val, (int, float)): return Decimal(str(val)) elif isinstance(val, str) and val.replace(".", "").isdigit(): return Decimal(val) return Decimal(0) # 添加到结果 results.append({ "city_name": city, "monthly_total": convert_value(total_value) * Decimal('10000'), "monthly_import": None, # 没有单独的进口/出口数据 "monthly_export": None, "yoy_import_export": convert_value(yoy_value), "yoy_import": Decimal(0), "yoy_export": Decimal(0) }) except Exception as e: log.error(f"处理 {city} 数据出错: {str(e)}") return pd.DataFrame(results) except Exception as e: log.error(f"处理深圳海关文件出错: {str(e)}") return pd.DataFrame() def process_shantou_customs(file_path, year, month): """处理汕头海关数据 (逻辑同广州海关)""" log.info(f"处理汕头海关文件: {file_path.name}") return process_guangzhou_customs(file_path, year, month) def process_huangpu_customs(file_path, year, month): """处理黄埔海关数据""" try: wb = load_workbook(file_path, data_only=True) sheet = wb.active # 查找合计行 total_row_idx = None for i, row in enumerate(sheet.iter_rows(values_only=True), 1): if row and "合计" in str(row[0]): total_row_idx = i break if total_row_idx is None: log.error("未找到合计行") return pd.DataFrame() # 查找包含月份的表头 month_str = f"{year}年{month}月" header_row = None data_cols = [] for i in range(1, 4): # 检查前3行 row_values = [str(cell.value) if cell.value else "" for cell in sheet[i]] if any(month_str in val and "人民币" in val for val in row_values): header_row = i for col_idx, val in enumerate(row_values): if val and month_str in val and "人民币" in val: data_cols.append(col_idx) break if len(data_cols) < 6: log.error(f"未找到足够的 {month_str} 人民币数据列") return pd.DataFrame() # 获取合计行数据 row_values = [cell.value for cell in sheet[total_row_idx]] # 转换数据类型 def convert_value(val): if isinstance(val, (int, float)): return Decimal(str(val)) elif isinstance(val, str) and val.replace(".", "").isdigit(): return Decimal(val) return Decimal(0) # 提取数据 results = [{ "city_name": "东莞市", "monthly_total": convert_value(row_values[data_cols[0]]), # 进出口 "monthly_export": convert_value(row_values[data_cols[1]]), # 出口 "monthly_import": convert_value(row_values[data_cols[2]]), # 进口 "yoy_import_export": convert_value(row_values[data_cols[3]]), # 进出口同比 "yoy_export": convert_value(row_values[data_cols[4]]), # 出口同比 "yoy_import": convert_value(row_values[data_cols[5]]) # 进口同比 }] return pd.DataFrame(results) except Exception as e: log.error(f"处理黄埔海关文件出错: {str(e)}") return pd.DataFrame() def process_jiangmen_customs(file_path, year, month): """处理江门海关数据""" try: wb = load_workbook(file_path, data_only=True) sheet = wb.active # 从文件名确定城市 city_name = "江门市" if "江门市" in file_path.name else "阳江市" target_row_name = "江门市进出口商品" if city_name == "江门市" else "阳江市进出口商品总值" # 查找目标行 target_row_idx = None for i, row in enumerate(sheet.iter_rows(values_only=True), 1): if row and target_row_name in str(row[0]): target_row_idx = i break if target_row_idx is None: log.error(f"未找到 {target_row_name} 行") return pd.DataFrame() # 查找包含月份的表头 month_str = f"{year}年{month}月" header_row = None data_cols = [] for i in range(1, 4): # 检查前3行 row_values = [str(cell.value) if cell.value else "" for cell in sheet[i]] if any(month_str in val for val in row_values): header_row = i for col_idx, val in enumerate(row_values): if val and month_str in val: data_cols.append(col_idx) break if len(data_cols) < 6: log.error(f"未找到足够的 {month_str} 数据列") return pd.DataFrame() # 获取目标行数据 row_values = [cell.value for cell in sheet[target_row_idx]] # 转换数据类型 def convert_value(val): if isinstance(val, (int, float)): return Decimal(str(val)) elif isinstance(val, str) and val.replace(".", "").isdigit(): return Decimal(val) return Decimal(0) # 提取数据 (亿元转换为万元) return pd.DataFrame([{ "city_name": city_name, "monthly_total": convert_value(row_values[data_cols[0]]) * Decimal('10000'), # 进出口 "monthly_export": convert_value(row_values[data_cols[1]]) * Decimal('10000'), # 出口 "monthly_import": convert_value(row_values[data_cols[2]]) * Decimal('10000'), # 进口 "yoy_import_export": convert_value(row_values[data_cols[3]]), # 进出口同比 "yoy_export": convert_value(row_values[data_cols[4]]), # 出口同比 "yoy_import": convert_value(row_values[data_cols[5]]) # 进口同比 }]) except Exception as e: log.error(f"处理江门海关文件出错: {str(e)}") return pd.DataFrame() def process_zhanjiang_customs(file_path, year, month): """处理湛江海关数据""" try: wb = load_workbook(file_path, data_only=True) sheet = wb.worksheets[0] # 从文件名确定城市 city_name = "湛江市" if "湛江市" in file_path.name else "茂名市" # 查找月度数据表格 table_start_row = None month_str = f"{year}年{month}月" for i, row in enumerate(sheet.iter_rows(values_only=True), 1): if row and any(month_str in str(cell) for cell in row if cell): table_start_row = i break if table_start_row is None: log.error(f"未找到 {month_str} 月度数据表") return pd.DataFrame() # 查找目标行（城市名所在行） target_row_idx = None for i in range(table_start_row, table_start_row + 20): # 在后续行中查找 row_val = sheet.cell(row=i, column=1).value if row_val and city_name in str(row_val): target_row_idx = i break if target_row_idx is None: log.error(f"未找到 {city_name} 数据行") return pd.DataFrame() # 提取数据 results = [] for col in [2, 3, 4, 5, 6, 7]: # 依次为进出口、出口、进口、进出口同比、出口同比、进口同比 cell_value = sheet.cell(row=target_row_idx, column=col).value results.append(cell_value) # 转换数据类型 def convert_value(val): if isinstance(val, (int, float)): return Decimal(str(val)) elif isinstance(val, str) and val.replace(".", "").isdigit(): return Decimal(val) return Decimal(0) return pd.DataFrame([{ "city_name": city_name, "monthly_total": convert_value(results[0]), "monthly_export": convert_value(results[1]), "monthly_import": convert_value(results[2]), "yoy_import_export": convert_value(results[3]), "yoy_export": convert_value(results[4]), "yoy_import": convert_value(results[5]) }]) except Exception as e: log.error(f"处理湛江海关文件出错: {str(e)}") return pd.DataFrame() def get_customs_processor(customs_name): """获取不同海关的处理函数""" processors = { "广州海关": process_guangzhou_customs, "深圳海关": process_shenzhen_customs, "汕头海关": process_shantou_customs, "黄埔海关": process_huangpu_customs, "江门海关": process_jiangmen_customs, "湛江海关": process_zhanjiang_customs } return processors.get(customs_name) def parse_excel(current_dir): """主解析入口（优化为单参数模式） Args: current_dir (str): 当前月份数据目录（格式：/年份/省份/月份） """ current_path = Path(current_dir) year, month = extract_year_month_from_path(current_path) log.info(f"开始处理 {year}年{month}月 数据: {current_dir}") try: # 动态获取前月目录 prev_dir = get_previous_month_dir(current_path) if month != 1 else None if prev_dir: log.info(f"上个月数据目录: {prev_dir}") # 获取当前目录下所有Excel文件 excel_files = list(current_path.glob("*.xls*")) if not excel_files: log.warning(f"当前目录下未找到Excel文件: {current_dir}") return # 按海关处理每个文件 all_results = pd.DataFrame() customs_map = { "广州海关": [], "深圳海关": [], "汕头海关": [], "黄埔海关": [], "江门海关": [], "湛江海关": [] } # 组织文件到不同的海关 for file_path in excel_files: for customs_name in customs_map.keys(): if match_customs_file(file_path.name, customs_name, year, month): customs_map[customs_name].append(file_path) log.info(f"匹配到 {customs_name} 文件: {file_path.name}") break # 处理每个海关的文件 for customs_name, file_list in customs_map.items(): processor = get_customs_processor(customs_name) if not processor: continue for file_path in file_list: # 特殊处理深圳海关和江门海关的2月数据(缺少1月数据) if customs_name in ["深圳海关", "江门海关"] and month == 2: # 获取2月份完整数据 df_full = processor(file_path, year, month) if df_full.empty: continue # 创建1月份数据 (取2月份数据的一半) df_half = df_full.copy() for col in ['monthly_total', 'monthly_import', 'monthly_export']: df_half[col] = df_half[col] / 2 # 设置1月份数据 df_half['month'] = 1 # 设置2月份数据 (取2月份数据的一半) df_full['month'] = 2 for col in ['monthly_total', 'monthly_import', 'monthly_export']: df_full[col] = df_full[col] / 2 # 合并数据 df_customs = pd.concat([df_half, df_full]) else: # 正常处理数据 df_customs = processor(file_path, year, month) if not df_customs.empty: df_customs['month'] = month if not df_customs.empty: all_results = pd.concat([all_results, df_customs]) # 如果没有获取到数据 if all_results.empty: log.warning(f"未处理到有效数据: {current_dir}") return # 添加公共字段 all_results['prov_code'] = PROV_CODE all_results['prov_name'] = PROV_NAME all_results['year'] = year all_results['month'] = all_results.get('month', month) all_results['crossborder_year_month'] = all_results['year'].astype(str) + '-' + all_results['month'].astype( str).str.zfill(2) # 添加城市编码 def get_city_code(row): return GUANGDONG_CITY.get(row['city_name'], '0000') all_results['city_code'] = all_results.apply(get_city_code, axis=1) # 排序并删除重复项 all_results = all_results.sort_values(by=['city_code', 'crossborder_year_month']) all_results = all_results.drop_duplicates(subset=['crossborder_year_month', 'city_code'], keep='last') # 打印处理结果 log.info(f"处理完成，共获得 {len(all_results)} 条数据") # 选择入库字段 final_df = all_results[[ 'crossborder_year_month', 'prov_code', 'prov_name', 'city_code', 'city_name', 'monthly_total', 'monthly_import', 'monthly_export', 'yoy_import_export', 'yoy_import', 'yoy_export' ]].copy() # 打印前几条数据 log.info(f"处理后数据示例:\n{final_df.head()}") # 这里调用DBHelper入库（实际使用时请取消注释） """ from db_helper import DBHelper db = DBHelper() db.bulk_insert( final_df, 't_yujin_crossborder_prov_region_trade', conflict_columns=['crossborder_year_month', 'city_code'], update_columns=['monthly_total', 'monthly_import', 'monthly_export', 'yoy_import_export', 'yoy_import', 'yoy_export'] ) """ log.info(f"{current_dir}数据已全部成功处理") except Exception as e: log.error(f"处理失败：{current_dir}，错误：{str(e)}") raise # 遍历目录的函数（原样保留） # 测试入口 if __name__ == "__main__": traverse_and_process(download_dir, parse_excel, province_name="guangdong")