crossborder/zhejiang/gov_commodity_zhejiang_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_country(path, year_month):
    file_paths = list(Path(path).glob('*'))
    if not file_paths:
        log.info("未找到任何文件")
        return None

    file_path = file_paths[0]
    log.info(f"处理文件: {file_path.name}")

    xls = pd.ExcelFile(file_path)
    import_df = pd.DataFrame()
    export_df = pd.DataFrame()
    total_df = pd.DataFrame()

    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:
        df = pd.read_excel(xls, sheet_name=1, header=None)
    else:
        df = pd.read_excel(file_path, header=None)
    temp_df = df[[0, 1]].rename(columns={0: 'commodity', 1: '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[[0, 2]].rename(columns={0: 'commodity', 2: '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[[0, 3]].rename(columns={0: 'commodity', 3: '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):
    res = get_df_country(path, None)
    if not res:
        log.info(f"{path} 目录里文件未找到包含 国别 sheet")
        return
    import_df, export_df, total_df = res

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

    # 当月数据分组清洗
    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_country(previous_month_dir, year_month)
        if not res:
            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')

    sql_arr = []
    # try:
    for _, row in merged_df.iterrows():
        country_name = str(row['commodity']).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

        # 提取数据并格式化
        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_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}', '330000', '浙江省', '{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()) 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 存表完成！\n")

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(base_country_code.download_dir)

    root = Path(base_country_code.download_dir) / '2024' / '07'
    process_folder(root)
    log.info("浙江杭州海关国别所有文件处理完成！")