crossborder/jiangsu/gov_commodity_jiangsu_import_export.py

import re
from pathlib import Path

import pandas as pd

from utils import base_country_code, base_mysql
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()
        skip_index = 4 if year_month == '2024-11' else 5
        sheet_index = 6 if year_month == '2024-11' else 4
        df = pd.read_excel(xls, sheet_name=sheet_index, 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)"
               f"ON DUPLICATE KEY UPDATE create_time = now() ;")
        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("江苏南京海关类章所有文件处理完成！")