import re from pathlib import Path import pandas as pd from utils import base_country_code, base_mysql from utils.base_country_code import format_sql_value from utils.log import log YEAR_PATTERN = re.compile(r"^\d{4}$") MONTH_PATTERN = re.compile(r"^(0[1-9]|1[0-2])$") all_records = [] def process_folder(path, all_records): file_paths = list(Path(path).glob('*')) if not file_paths: log.info("未找到任何文件") return year, month = base_country_code.extract_year_month_from_path(path) year_month = f'{year}-{month:02d}' if len(file_paths) == 1: file_path = file_paths[0] log.info(f"处理单文件: {file_path.name}") # 读取所有sheet xls = pd.ExcelFile(file_path) import_df = pd.DataFrame() export_df = pd.DataFrame() total_df = pd.DataFrame() sheet_name = base_country_code.find_sheet_by_keyword(file_path, "类章") if not sheet_name: log.info(f"{file_path} 未找到包含 类章 sheet") return skip_index = 4 if year_month == '2024-11' else 5 df = pd.read_excel(xls, sheet_name=sheet_name, header=None).iloc[skip_index:] temp_df = df[[0, 5]].rename(columns={0: 'commodity', 5: 'import'}) temp_df['import'] = pd.to_numeric(temp_df['import'].replace('--', 0), errors='coerce') temp_df['import'] = temp_df['import'] * 10000 import_df = pd.concat([import_df, temp_df]) temp_df = df[[0, 3]].rename(columns={0: 'commodity', 3: 'export'}) temp_df['export'] = pd.to_numeric(temp_df['export'].replace('--', 0), errors='coerce') temp_df['export'] = temp_df['export'] * 10000 export_df = pd.concat([export_df, temp_df]) temp_df = df[[0, 1]].rename(columns={0: 'commodity', 1: 'total'}) temp_df['total'] = pd.to_numeric(temp_df['total'].replace('--', 0), errors='coerce') temp_df['total'] = temp_df['total'] * 10000 total_df = pd.concat([total_df, temp_df]) save_to_database(import_df, export_df, total_df, year, month, all_records) else: # 2024-10 -2023-01 import_df = pd.DataFrame() export_df = pd.DataFrame() total_df = pd.DataFrame() for file in file_paths: if "商品类章" in file.name: log.info(f"处理多文件: {file.name}") file_path = Path(path) / file df = pd.read_excel(file_path, header=None).iloc[6:] temp_df = df[[1, 5]].rename(columns={1: 'commodity', 5: 'import'}) temp_df['import'] = pd.to_numeric(temp_df['import'].replace('--', 0), errors='coerce') temp_df['import'] = temp_df['import'] * 10 import_df = pd.concat([import_df, temp_df]) temp_df = df[[1, 3]].rename(columns={1: 'commodity', 3: 'export'}) temp_df['export'] = pd.to_numeric(temp_df['export'].replace('--', 0), errors='coerce') temp_df['export'] = temp_df['export'] * 10 export_df = pd.concat([export_df, temp_df]) temp_df = df[[1, 2]].rename(columns={1: 'commodity', 2: 'total'}) temp_df['total'] = pd.to_numeric(temp_df['total'].replace('--', 0), errors='coerce') temp_df['total'] = temp_df['total'] * 10 total_df = pd.concat([total_df, temp_df]) break save_to_database(import_df, export_df, total_df, year, month, all_records) def save_to_database(import_df, export_df, total_df, year, month, all_records): # 直接合并，不使用 groupby，保持原始顺序 merged_df = pd.concat( [import_df.set_index('commodity'), export_df.set_index('commodity'), total_df.set_index('commodity')], axis=1, join='outer').reset_index() merged_df = merged_df merged_df['original_order'] = merged_df.index # 保留原始顺序 merged_df = merged_df.sort_values('original_order').reset_index(drop=True) sql_arr = [] processed_commodities = set() all_records_index = 0 year_month = f'{year}-{month:02d}' for _, row in merged_df.iterrows(): commodity_name = str(row['commodity']) # commodity_name = str(row['commodity']).strip() # 找类名确定索引 result = extract_category_or_chapter(commodity_name, all_records_index) if result is None: log.info(f"未找到商品名称 '{commodity_name}' 对应的ID") continue if result >= len(all_records): log.info(f"all_records 已超限 '{commodity_name}' 跳过") continue all_records_index = result commodity_code, category_name = int(all_records[all_records_index][0]), str(all_records[all_records_index][1]) if commodity_code in processed_commodities: continue monthly_import = round(row['import'], 4) monthly_export = round(row['export'], 4) monthly_total = round(row['total'], 4) sql = (f"INSERT INTO t_yujin_crossborder_prov_commodity_trade " f"(crossborder_year, crossborder_year_month, prov_code, prov_name, commodity_code, commodity_name, monthly_total, monthly_export, monthly_import, create_time, commodity_source) VALUES " f"('{year}', '{year_month}', '320000', '江苏省', '{commodity_code}', '{category_name}', {monthly_total}, {monthly_export}, {monthly_import}, now(), 1);") sql_arr.append(sql) processed_commodities.add(commodity_code) log.info(f"√ {year_month} 成功生成SQL文件 size {len(sql_arr)} ") base_mysql.bulk_insert(sql_arr) log.info(f"√ {year_month} prov_commodity_trade SQL 存表完成!") def extract_category_or_chapter(text, all_records_index): text = text.strip() # 匹配“第一类”或“第1类” first_class_match = re.match(r'^第(一|\d+)类', text, re.IGNORECASE | re.UNICODE) if first_class_match and (first_class_match.group(1) == '1' or first_class_match.group(1) == '一'): return 0 else: return all_records_index + 1 def hierarchical_traversal(root_path, all_records): """分层遍历：省份->年份->月目录""" root = Path(root_path) # 获取所有年份目录 year_dirs = [ item for item in root.iterdir() if item.is_dir() and YEAR_PATTERN.match(item.name) ] # 按年倒序 for year_dir in sorted(year_dirs, key=lambda x: x.name, reverse=True): # 构造完整的路径：download/shandong/2025/03 log.info(f"\n年份：{year_dir.name} | 省份：jiangsu") # 提取月份目录 month_dirs = [] for item in year_dir.iterdir(): if item.is_dir() and 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'], all_records) if __name__ == '__main__': all_records = base_mysql.get_hs_all() hierarchical_traversal(base_country_code.download_dir, all_records) # root = Path(base_country_code.download_dir)/'2024'/'11' # process_folder(root, all_records) print("江苏南京海关类章所有文件处理完成！")