crossborder/jiangsu/gov_commodity_jiangsu_country.py

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

# 排除地区名单
EXCLUDE_REGIONS = ["亚洲", "非洲", "欧洲", "拉丁美洲", "北美洲", "大洋洲", "南极洲",
                   "东南亚国家联盟", "欧洲联盟", "亚太经济合作组织",
                   "区域全面经济伙伴关系协定（RCEP）成员国", "共建“一带一路”国家和地区"]

def get_df(path, year_month):
    global df,  df_type
    file_paths = list(Path(path).glob('*'))
    if not file_paths:
        log.info("未找到任何文件")
        return
    if len(file_paths) == 1:
        file_path = file_paths[0]
        log.info(f"处理单文件: {file_path.name}")
        xls = pd.ExcelFile(file_path)

        df = pd.read_excel(xls, sheet_name=1, header=None).iloc[5:]
        df_type = 0

    else:
        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:]
                df_type = 1
                break
    return df, df_type

def process_folder(path):
    year, month = base_country_code.extract_year_month_from_path(path)
    year_month = f'{year}-{month:02d}'

    sql_arr = []
    try:
        df, df_type = get_df(path, year_month)
        if df_type == 0:
            country_name_index = 0
            col_total_index, col_monthly_export_index, col_monthly_import_index = 1, 3, 5
        else:
            country_name_index = 1
            col_total_index, col_monthly_export_index, col_monthly_import_index = 2, 4, 6

        for index, row in df.iterrows():
            if index < 4:
                continue

            # 提取国家名称并去除括号内容
            country_name = str(row.values[country_name_index]).strip()
            if country_name.endswith("）") or country_name.endswith(")"):
                country_name = country_name.rsplit("（")[0] or country_name.rsplit("(")[0]

            # 过滤掉排除地区
            if country_name in EXCLUDE_REGIONS:
                continue

            # 获取国家编码
            country_code = base_country_code.COUNTRY_CODE_MAPPING.get(country_name)
            if not country_code:
                log.info(f"{year_month} 未找到国家 '{country_name}' 对应的编码")
                continue

            # 提取数据并格式化
            monthly_export, monthly_import, monthly_total = value_row(row, col_total_index, col_monthly_export_index, col_monthly_import_index)
            if df_type == 0:
                monthly_export, monthly_import, monthly_total = round(float(monthly_export) * 10000, 4), round(float(monthly_import) * 10000, 4), round(float(monthly_total) * 10000, 4)
            yoy_export, yoy_import, yoy_import_export = 0, 0, 0

            # 构建 SQL
            sql = (
                f"INSERT INTO t_yujin_crossborder_prov_country_trade "
                f"(crossborder_year, crossborder_year_month, prov_code, prov_name, country_code, country_name, "
                f"monthly_total, monthly_export, monthly_import, yoy_import_export, yoy_import, yoy_export, create_time) "
                f"VALUES ('{year}', '{year_month}', '320000', '江苏省', '{country_code}', '{country_name}', "
                f"{format_sql_value(monthly_total)}, {format_sql_value(monthly_export)}, {format_sql_value(monthly_import)}, '{yoy_import_export}', '{yoy_import}', "
                f"'{yoy_export}', NOW())"
                f"ON DUPLICATE KEY UPDATE create_time = now() ;"
            )
            sql_arr.append(sql)
    except Exception as e:
        log.info(f"{year_month} 处理时发生异常: {str(e)}")

    log.info(f"√ {year_month} 成功生成 SQL 条数: {len(sql_arr)}")
    # 批量插入数据库
    base_mysql.bulk_insert(sql_arr)
    log.info(f"√ {year_month} prov_country_trade SQL 存表完成！")

def value_row(row, col_total_index, col_monthly_export_index, col_monthly_import_index):
    def value_special_handler(value):
        if pd.isna(value) or value == "--":
            return "0"
        else:
            return value.strip()

    monthly_total = value_special_handler(str(row.values[col_total_index]))
    monthly_export = value_special_handler(str(row.values[col_monthly_export_index]))
    monthly_import = value_special_handler(str(row.values[col_monthly_import_index]))

    return monthly_export, monthly_import, monthly_total


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} | 省份：jiangsu")

        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(base_country_code.download_dir)
    log.info("江苏南京海关国别所有文件处理完成！")