import time from pathlib import Path import pandas as pd from zhejiang import download_dir from utils import base_country_code, base_mysql from utils.base_country_code import format_sql_value from utils.log import log city_code_map = { "杭州地区": "330100", "宁波地区": "330200", "温州地区": "330300", "绍兴地区": "330400", "湖州地区": "330500", "嘉兴地区": "330600", "金华地区": "330700", "衢州地区": "330800", "舟山地区": "330900", "台州地区": "331000", "丽水地区": "331100" } def get_df(path, year_month): file_paths = list(Path(path).glob('*')) if not file_paths: log.info("未找到任何文件") return None flag = True file_path = file_paths[0] if len(file_paths) > 1: for file_path in file_paths: if '十一地市' in file_path.name: file_path = file_path flag = False break if flag: xls = pd.ExcelFile(file_path) df = pd.read_excel(xls, sheet_name=0, header=None) else: df = pd.read_excel(file_path, header=None) import_df = pd.DataFrame() export_df = pd.DataFrame() total_df = pd.DataFrame() temp_df = df[[1, 2]].rename(columns={1: 'commodity', 2: 'total'}) temp_df['total'] = pd.to_numeric(temp_df['total'].replace('--', 0), errors='coerce').astype(float) if year_month and year_month == '2024-07': temp_df['total'] = temp_df['total'] / 10000 total_df = pd.concat([total_df, temp_df]) temp_df = df[[1, 3]].rename(columns={1: 'commodity', 3: 'import'}) temp_df['import'] = pd.to_numeric(temp_df['import'].replace('--', 0), errors='coerce').astype(float) if year_month and year_month == '2024-07': temp_df['import'] = temp_df['import'] / 10000 import_df = pd.concat([import_df, temp_df]) temp_df = df[[1, 4]].rename(columns={1: 'commodity', 4: 'export'}) temp_df['export'] = pd.to_numeric(temp_df['export'].replace('--', 0), errors='coerce').astype(float) if year_month and year_month == '2024-07': temp_df['export'] = temp_df['export'] / 10000 export_df = pd.concat([export_df, temp_df]) return import_df, export_df, total_df def process_folder(path): year, month = base_country_code.extract_year_month_from_path(path) year_month = f'{year}-{month:02d}' sql_arr = [] res = get_df(path, None) if res is None: log.info(f"{year_month} prov_region_trade 未找到包含 地市 sheet") return import_df, export_df, total_df = res # 当月数据分组清洗 curr_import = import_df.groupby('commodity')['import'].sum().reset_index() curr_export = export_df.groupby('commodity')['export'].sum().reset_index() total_df = total_df.groupby('commodity')['total'].sum().reset_index() if not month == 1: previous_month_dir = base_country_code.get_previous_month_dir(path) res = get_df(previous_month_dir, year_month) if res is None: log.info(f"{path} 上月目录里文件未找到包含 地市 sheet") return prev_import_df, prev_export_df, prev_total_df = res # 上月数据分组 prev_import = prev_import_df.groupby('commodity')['import'].sum().reset_index() prev_export = prev_export_df.groupby('commodity')['export'].sum().reset_index() prev_total_df = prev_total_df.groupby('commodity')['total'].sum().reset_index() # 差值计算 curr_import = pd.merge(curr_import, prev_import, on='commodity', how='left') curr_import['import'] = round(curr_import['import_x'] - curr_import['import_y'], 4) curr_export = pd.merge(curr_export, prev_export, on='commodity', how='left') curr_export['export'] = round(curr_export['export_x'] - curr_export['export_y'], 4) total_df = pd.merge(total_df, prev_total_df, on='commodity', how='left') total_df['total'] = round(total_df['total_x'] - total_df['total_y'], 4) log.info(f"合并文件: {path}*********{previous_month_dir}") # 合并进出口数据 merged_df = pd.merge(curr_import, curr_export, on='commodity', how='outer') merged_df = pd.merge(merged_df, total_df, on='commodity', how='outer') for _, row in merged_df.iterrows(): city_name = str(row['commodity']).strip() city_code = city_code_map.get(city_name) if not city_code: log.info(f"未找到省 '{city_name}' 对应市编码") continue # 提取数据并格式化 if year == 2025 or (year == 2024 and month in [7, 8, 9, 10, 11, 12]): monthly_import = round(row['import'], 4) monthly_export = round(row['export'], 4) monthly_total = round(row['total'], 4) else: monthly_import = round(row['import'] / 10000, 4) monthly_export = round(row['export'] / 10000, 4) monthly_total = round(row['total'] / 10000, 4) yoy_import_export, yoy_import, yoy_export = 0, 0, 0 # 组装 SQL 语句 sql = (f"INSERT INTO t_yujin_crossborder_prov_region_trade " f"(crossborder_year, crossborder_year_month, prov_code, prov_name, city_code, city_name, monthly_total, monthly_export, monthly_import,yoy_import_export, yoy_import, yoy_export, create_time) VALUES " f"('{year}', '{year_month}', '330000', '浙江省', '{city_code}', '{city_name}', {format_sql_value(monthly_total)}, {format_sql_value(monthly_export)}, {format_sql_value(monthly_import)}, '{yoy_import_export}', '{yoy_import}', '{yoy_export}', now()) ON DUPLICATE KEY UPDATE create_time = now();\n") sql_arr.append(sql) log.info(f"√ {year_month} prov_region_trade 成功生成 SQL 文件 size {len(sql_arr)} ") base_mysql.bulk_insert(sql_arr) log.info(f"√ {year_month} prov_region_trade SQL 存表完成!") def hierarchical_traversal(root_path): root = Path(root_path) year_dirs = [ item for item in root.iterdir() if item.is_dir() and base_country_code.YEAR_PATTERN.match(item.name) ] for year_dir in sorted(year_dirs, key=lambda x: x.name, reverse=True): log.info(f"\n年份：{year_dir.name} | 省份：zhejiang") month_dirs = [] for item in year_dir.iterdir(): if item.is_dir() and base_country_code.MONTH_PATTERN.match(item.name): month_dirs.append({"path": item, "month": int(item.name)}) if month_dirs: for md in sorted(month_dirs, key=lambda x: x["month"], reverse=True): log.info(f" 月份：{md['month']:02d} | 路径：{md['path']}") process_folder(md['path']) if __name__ == '__main__': hierarchical_traversal(download_dir) log.info(f"浙江杭州海关城市所有文件处理完成！") time.sleep(5) base_mysql.update_january_yoy('浙江省') base_mysql.update_shandong_yoy('浙江省') log.info("同比sql处理完成") # root = Path(download_dir)/'2024'/'07' # process_folder(root)