crossborder/zhejiang/gov_commodity_zhejiang_import_export.py

from pathlib import Path

import re
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

CUSTOM_COMMODITY_REPLACEMENTS = {
    '稻谷及大米': '稻谷、大米及大米粉',
    '有机发光二极管平板显示模组': '有机发光二极管（OLED）平板显示模组',
}

PRESERVE_PARENTHESES_KEYWORDS = {
    '汽车（包括底盘）',
}

def clean_commodity_name(name, preserve_keywords=None):
    """
    自定义清洗商品名称逻辑，支持条件保留中文括号内容

    :param name: 商品名称字符串
    :param preserve_keywords: 需要保留括号的关键词集合
    :return: 清洗后的商品名称
    """
    name = str(name).strip().replace('(', '（').replace(')', '）')

    # 去除非必要符号
    name = re.sub(r'[#*?]', '', name)
    name = re.sub(r'_x000D_', '', name)

    # 判断是否需要保留括号内容
    if preserve_keywords:
        for keyword in preserve_keywords:
            if keyword == name:
                # 匹配到关键词时，不移除括号内容
                return name

    # 默认移除中文括号及内容
    name = re.sub(r'（[^）]*）', '', name)
    return name.strip()

def get_df_import_export(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()

    flag = True
    sheet_name = base_country_code.find_sheet_by_keyword(file_path, "主出商品")
    if not sheet_name:
        log.info(f"{file_path} 单文件未找到包含 主出商品 sheet")
        # 23年1-11月数据要在多文件里找
        for file_path in file_paths:
            if '主要出口商品' in file_path.name:
                file_path = file_path
                flag = False
                break
    if not sheet_name and flag:
        log.info(f"{path} 中未找到 主出商品 sheet或文件")
        return None

    if flag:
        df = pd.read_excel(xls, sheet_name=sheet_name, header=None).iloc[2:]
    else:
        df = pd.read_excel(file_path, header=None).iloc[1:]
    temp_df = df[[0, 1]].rename(columns={0: 'commodity', 1: 'export'})
    temp_df['commodity'] = (
        temp_df['commodity']
        .astype(str)
        .apply(lambda x: clean_commodity_name(x, preserve_keywords=PRESERVE_PARENTHESES_KEYWORDS))
        .replace(CUSTOM_COMMODITY_REPLACEMENTS, regex=False)
    )
    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])

    flag_2 = True
    sheet_name = base_country_code.find_sheet_by_keyword(file_path, "主进商品")
    if not sheet_name:
        log.info(f"{file_path} 单文件未找到包含 主进商品 sheet")
        # 23年1-11月数据要在多文件里找
        for file_path in file_paths:
            if '主要进口商品' in file_path.name:
                file_path = file_path
                flag_2 = False
                break
    if not sheet_name and flag_2:
        log.info(f"{path} 中未找到 主进商品 sheet或文件")
        return None

    if flag_2:
        df = pd.read_excel(xls, sheet_name=sheet_name, header=None).iloc[2:]
    else:
        df = pd.read_excel(file_path, header=None).iloc[1:]
    temp_df = df[[0, 1]].rename(columns={0: 'commodity', 1: 'import'})
    temp_df['commodity'] = (
        temp_df['commodity']
        .astype(str)
        .apply(lambda x: clean_commodity_name(x, preserve_keywords=PRESERVE_PARENTHESES_KEYWORDS))
        .replace(CUSTOM_COMMODITY_REPLACEMENTS, regex=False)
    )
    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])

    return import_df, export_df


def process_folder(path):
    res = get_df_import_export(path, None)
    if not res:
        log.info(f"{path} 目录里文件未找到包含 主出、主进商品 sheet")
        return
    import_df, export_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()

    if not month == 1:
        previous_month_dir = base_country_code.get_previous_month_dir(path)
        res = get_df_import_export(previous_month_dir, year_month)
        if not res:
            log.info(f"{path} 上月目录里文件未找到包含 主出、主进商品 sheet")
            return
        prev_import_df, prev_export_df = res

        # 上月数据分组
        prev_import = prev_import_df.groupby('commodity')['import'].sum().reset_index()
        prev_export = prev_export_df.groupby('commodity')['export'].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)
        log.info(f"合并文件: {path}*********{previous_month_dir}")

    # 合并进出口数据
    merged_df = pd.merge(curr_import, curr_export, on='commodity', how='outer')
    save_to_database(merged_df, year, month)

def save_to_database(merged_df, year, month):
    year_month = f'{year}-{month:02d}'
    processed_commodities = set()
    sql_arr = []
    try:
        for _, row in merged_df.iterrows():
            commodity_name = str(row['commodity']).strip()
            if commodity_name == '消费品' or commodity_name == '劳动密集型产品':
                log.info(f'{commodity_name} 商品不存在，跳过')
                continue
            commodity_code, commodity_name_fix = base_mysql.get_commodity_id(commodity_name)
            if not commodity_code:
                log.info(f"未找到商品名称 '{commodity_name}' 对应的 ID")
                continue
            if not commodity_name_fix or commodity_name_fix in processed_commodities:
                log.info(f"已处理过 '{commodity_name_fix}'，传入name：{commodity_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(
                    (0 if pd.isna(monthly_import) else monthly_import) +
                    (0 if pd.isna(monthly_export) else monthly_export),
                    4
                )
            else:
                monthly_import = round(row['import'] / 10000, 4)
                monthly_export = round(row['export'] / 10000, 4)
                monthly_total = round(
                    (0 if pd.isna(monthly_import) else monthly_import) +
                    (0 if pd.isna(monthly_export) else monthly_export),
                    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) VALUES "
                   f"('{year}', '{year_month}', '330000', '浙江省', '{commodity_code}', '{commodity_name_fix}', {format_sql_value(monthly_total)}, {format_sql_value(monthly_export)}, {format_sql_value(monthly_import)}, now());")
            sql_arr.append(sql)

            processed_commodities.add(commodity_name_fix)
            # log.info(f'{commodity_name} -> {commodity_name_fix}')

    except Exception as e:
        log.info(f"{year_month} prov_commodity_trade 生成 SQL 文件时发生异常: {str(e)}")

    log.info(f"√ {year_month} prov_commodity_trade 成功生成 SQL 文件 size {len(sql_arr)} ")
    # 解析完后生成sql文件批量入库
    base_mysql.bulk_insert(sql_arr)
    log.info(f"√ {year_month} prov_commodity_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} | 省份：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)

    # root = Path(base_country_code.download_dir)/'2023'/'01'
    # process_folder(root)
    log.info("浙江杭州海关类章所有文件处理完成！")