crossborder/guangdong/guangdong_sub_customs_parse_excel.py

from decimal import Decimal, InvalidOperation
from pathlib import Path

import pandas as pd

from db_helper import DBHelper
from utils.constants import DOWNLOAD_DIR, GUANGDONG_CITY
from utils.log import log
from utils.parse_utils import traverse_and_process, extract_year_month_from_path, get_previous_month_dir

# 配置日志
PROV_CODE = "440000"
PROV_NAME = "广东省"
download_dir = DOWNLOAD_DIR / "guangdong"

db = DBHelper()

_zhanjiang_first_month = None



# 广州海关:万元
# 深圳海关:亿元
# 汕头海关:万元
# 黄埔海关:万元
# 江门海关:亿元
# 湛江海关:万元

def match_customs_file(filename, customs_name, year, month):
    """匹配海关文件"""
    filename_lower = filename.lower()

    if customs_name == "广州海关":
        return "7地市进出口综合统计" in filename_lower
    elif customs_name == "深圳海关":
        return "深圳海关综合统计资料" in filename_lower
    elif customs_name == "汕头海关":
        return "5市报表" in filename_lower
    elif customs_name == "黄埔海关":
        return "东莞市进出口企业性质总值表" in filename_lower
    elif customs_name == "江门海关":
        if "江门市" in filename_lower or "阳江市" in filename_lower:
            return "外贸进出口有关情况统计表" in filename_lower
    elif customs_name == "湛江海关":
        if "湛江市" in filename_lower or "茂名市" in filename_lower:
            return "外贸进出口数据" in filename_lower
    return False


def find_header_and_columns(df, year, month):
    """
    查找匹配月份的表头行并定位对应的列索引。
    支持三种基础格式：
        - "2024年12月"
        - "2024年12月-2024年12月"
        - "2024-12-01 00:00:00"
    如果都未找到，则尝试匹配特殊格式："2023年01月-2023年02月"
    """
    candidate_month_strs = [
        f"{year}年{month:02d}月",
        f"{year}年{month:02d}月-{year}年{month:02d}月",
        f"{year}-{month:02d}-01 00:00:00"
    ]

    header_row = None
    for i in range(min(3, len(df))):
        row_cells = [str(cell).strip() for cell in df.iloc[i]]
        for cell_val in row_cells:
            if any(s == cell_val for s in candidate_month_strs):
                header_row = i
                break
        if header_row is not None:
            break

    # 如果没找到常规格式，尝试特殊格式：2023年01月-2023年02月
    special_format = "2023年01月-2023年02月"
    if header_row is None:
        log.warning(f"未找到常规格式，尝试匹配特殊格式: {special_format}")
        for i in range(min(3, len(df))):
            row_cells = [str(cell).strip() for cell in df.iloc[i]]
            for cell_val in row_cells:
                if cell_val == special_format:
                    header_row = i
                    log.info(f"成功匹配特殊格式: {special_format} 行号={i}")
                    break
            if header_row is not None:
                break

    if header_row is None:
        log.error("未找到任何支持的表头格式")
        return None, []

    # 确定数据列位置（包含所有候选格式）
    data_cols = []
    for col in range(len(df.columns)):
        cell_val = str(df.iloc[header_row, col]).strip()
        if cell_val in candidate_month_strs:
            data_cols.append(col)

    if not data_cols:
        for col in range(len(df.columns)):
            cell_val = str(df.iloc[header_row, col]).strip()
            if cell_val in [special_format]:
                data_cols.append(col)

    if not data_cols:
        log.error("未找到对应的数据列")
        return header_row, []

    return header_row, data_cols



def process_guangzhou_customs(file_path, year, month,customs_type='guangzhou'):
    """处理广州海关数据"""
    try:
        # 读取Excel文件
        df = pd.read_excel(file_path, sheet_name=0, header=None)
        log.info(f"处理广州海关文件: {file_path.name}")

        header_row,data_cols = find_header_and_columns(df, year, month)

        # 提取7地市数据
        results = []

        target_cities = ["广州市", "韶关市", "佛山市", "肇庆市", "河源市",
                         "清远市", "汕头市", "梅州市", "汕尾市", "潮州市", "揭阳市", "云浮市"]

        for idx in range(header_row + 1, len(df)):
            row = df.iloc[idx]
            city_cell = str(row[0])
            if "广东省" in city_cell:
                city_name = city_cell.replace("广东省", "").strip()
                if city_name in target_cities:
                    try:
                        if len(data_cols)>3:
                            monthly_total = Decimal(str(row[data_cols[0]]))  # 进出口
                            monthly_export = Decimal(str(row[data_cols[4]]))  # 出口
                            monthly_import = Decimal(str(row[data_cols[8]]))  # 进口
                            yoy_import_export = Decimal(str(row[data_cols[1]]))  # 进出口同比
                            yoy_export = Decimal(str(row[data_cols[5]]))  # 出口同比
                            yoy_import = Decimal(str(row[data_cols[9]]))  # 进口同比
                        else:
                            monthly_total = Decimal(str(row[data_cols[0]]))
                            monthly_export = Decimal(str(row[data_cols[1]]))
                            monthly_import = Decimal(str(row[data_cols[2]]))
                            yoy_import_export = Decimal(str(row[data_cols[0]+1]))  # 进出口同比
                            yoy_export = Decimal(str(row[data_cols[1]+1]))  # 出口同比
                            yoy_import = Decimal(str(row[data_cols[2]+1]))  # 进口同比


                        results.append({
                            "city_name": city_name,
                            "monthly_total": monthly_total,
                            "monthly_import": monthly_import,
                            "monthly_export": monthly_export,
                            "yoy_import_export": yoy_import_export,
                            "yoy_import": yoy_import,
                            "yoy_export": yoy_export
                        })
                    except Exception as e:
                        log.error(f"处理行 {idx} 出错: {e}")

        return pd.DataFrame(results)

    except Exception as e:
        log.error(f"处理广州海关文件出错: {str(e)}")
        return pd.DataFrame()


def process_shenzhen_customs(file_path, year, month):
    """处理深圳海关数据（完整6指标版）"""
    try:
        log.info(f"处理深圳海关文件: {file_path.name}")
        results = []

        for city, sheet_name in [("深圳市", "深圳市进出口（贸易方式）"),
                                 ("惠州市", "惠州市进出口（贸易方式）")]:
            try:
                df = pd.read_excel(file_path, sheet_name=sheet_name, header=None)
            except:
                log.warning(f"未找到sheet: {sheet_name}")
                continue

            # 查找总值行
            total_row_idx = None
            for idx in range(len(df)):
                if "总值" in str(df.iloc[idx, 0]):
                    total_row_idx = idx
                    break

            if total_row_idx is None:
                log.error(f"未找到总值行: {sheet_name}")
                continue

            try:
                # 列索引映射（基于您提供的完整数据结构）
                # 进出口 | 出口 | 进口 的数值和同比
                monthly_total = convert_unit(str(df.iloc[total_row_idx, 1]))
                yoy_total = Decimal(str(df.iloc[total_row_idx, 2]))

                monthly_export = convert_unit(str(df.iloc[total_row_idx, 3]))
                yoy_export = Decimal(str(df.iloc[total_row_idx, 4]))

                monthly_import = convert_unit(str(df.iloc[total_row_idx, 5]))
                yoy_import = Decimal(str(df.iloc[total_row_idx, 6]))

                results.append({
                    "city_name": city,
                    "monthly_total": monthly_total,
                    "monthly_export": monthly_export,
                    "monthly_import": monthly_import,
                    "yoy_import_export": yoy_total,  # 进出口同比
                    "yoy_export": yoy_export,
                    "yoy_import": yoy_import
                })
            except Exception as e:
                log.error(f"处理 {city} 数据出错: {e}")
                # 尝试部分提取（回退方案）
                try:
                    monthly_total = Decimal(str(df.iloc[total_row_idx, 1])) * Decimal('10000')
                    results.append({
                        "city_name": city,
                        "monthly_total": monthly_total,
                        "monthly_export": None,
                        "monthly_import": None,
                        "yoy_import_export": Decimal('0'),
                        "yoy_export": Decimal('0'),
                        "yoy_import": Decimal('0')
                    })
                except:
                    log.error(f"连基础进出口总值都无法提取: {sheet_name}")

        return pd.DataFrame(results)

    except Exception as e:
        log.error(f"处理深圳海关文件出错: {str(e)}")
        return pd.DataFrame()


def process_shantou_customs(file_path, year, month):
    """处理汕头海关数据 (逻辑同广州海关)"""
    log.info(f"处理汕头海关文件: {file_path.name}")
    return process_guangzhou_customs(file_path, year, month,customs_type='shantou')


def process_huangpu_customs(file_path, year, month):
    """处理黄埔海关数据"""
    try:
        df = pd.read_excel(file_path, sheet_name=0, header=None)
        log.info(f"处理黄埔海关文件: {file_path.name}")

        # 查找合计行
        total_row_idx = None
        for idx in range(len(df)):
            if "合计" in str(df.iloc[idx, 0]):
                total_row_idx = idx
                break

        if total_row_idx is None:
            log.error("未找到合计行")
            return pd.DataFrame()

        # 查找包含月份的表头,匹配23年1月-23年多种格式

        if year == 2024 and month == 12:
            month_str = '45627'
        elif year == 2023 and month == 12:
            month_str = '45261'
        elif year == 2023 and month == 3:
            month_str = f"{year}年{month:02d}月-{year}年{month:02d}月"
        else:
            month_str = f'{year}-{month:02d}-01 00:00:00'

        header_row = None
        for i in range(min(3, len(df))):
            row_cells = [str(cell).strip() for cell in df.iloc[i]]
            if any(month_str in cell  in cell for cell in row_cells):
                header_row = i
                break

        if header_row is None:
            log.error(f"未找到 {month_str} 人民币表头")
            return pd.DataFrame()

        # 确定数据列位置
        data_cols = []
        for col in range(len(df.columns)):
            cell_val = str(df.iloc[header_row, col])
            if month_str in cell_val :
                data_cols.append(col)

        if len(data_cols) < 3:
            log.error(f"未找到足够的 {month_str} 人民币数据列")
            return pd.DataFrame()

        try:
            result = []
            # 提取数据
            row = df.iloc[total_row_idx]
            monthly_total = Decimal(str(row[data_cols[0]]))  # 进出口
            monthly_export = Decimal(str(row[data_cols[1]]))  # 出口
            monthly_import = Decimal(str(row[data_cols[2]]))  # 进口
            yoy_import_export = str(row[data_cols[0]+1])  # 进出口同比
            yoy_export = str(row[data_cols[1]+1])  # 出口同比
            yoy_import = str(row[data_cols[2]+1])  # 进口同比

            result.append({
                "crossborder_year_month": f'{year}-{month:02d}',
                "city_name": "东莞市",
                "monthly_total": monthly_total,
                "monthly_import": monthly_import,
                "monthly_export": monthly_export,
                "yoy_import_export": yoy_import_export,
                "yoy_import": yoy_import,
                "yoy_export": yoy_export
            })
            #东莞市一月数据比较特殊
            if month == 2:
                monthly_total_sum = Decimal(str(row[data_cols[0]+4]))  # 进出口
                monthly_export_sum = Decimal(str(row[data_cols[1]+4]))  # 出口
                monthly_import_sum = Decimal(str(row[data_cols[2]+4]))  # 进口
                january_monthly_total = monthly_total_sum - monthly_total
                january_monthly_export = monthly_export_sum - monthly_export
                january_monthly_import = monthly_import_sum - monthly_import
                result.append({
                    "crossborder_year_month": f'{year}-01',
                    "city_name": "东莞市",
                    "monthly_total": january_monthly_total,
                    "monthly_import": january_monthly_export,
                    "monthly_export": january_monthly_import,
                })

            return pd.DataFrame(result)
        except Exception as e:
            log.error(f"提取数据出错: {e}")
            return pd.DataFrame()

    except Exception as e:
        log.error(f"处理黄埔海关文件出错: {str(e)}")
        return pd.DataFrame()


def process_jiangmen_customs(file_path, year, month):
    """处理江门海关数据"""
    try:
        df = pd.read_excel(file_path, sheet_name=0, header=None)
        log.info(f"处理江门海关文件: {file_path.name}")

        # 从文件名确定城市
        city_name = "江门市" if "江门市" in file_path.name else "阳江市"
        target_row_name = "江门市进出口商品总值" if city_name == "江门市" else "阳江市进出口商品总值"

        # 查找目标行
        target_row_idx = None
        for idx in range(len(df)):
            if target_row_name in str(df.iloc[idx, 0]):
                target_row_idx = idx
                break

        if target_row_idx is None:
            log.error(f"未找到 {target_row_name} 行")
            return pd.DataFrame()

        # 查找包含月份的表头
        if month == 2:
           month_str = f"1-{month}月"
        else:
           month_str = f"{month}月"
        header_row = None
        for i in range(min(6, len(df))):
            if any(month_str == str(cell).strip() for cell in df.iloc[i]):
                header_row = i
                break

        if header_row is None:
            log.error(f"未找到 {month_str} 表头")
            return pd.DataFrame()

        # 确定数据列位置
        data_cols = []
        for col in range(len(df.columns)):
            cell_val = str(df.iloc[header_row, col])
            if cell_val.strip() == month_str:
                data_cols.append(col)

        if len(data_cols) < 3:
            log.error(f"未找到足够的 {month_str} 数据列")
            return pd.DataFrame()

        # 提取数据 (亿元转换为万元)
        row = df.iloc[target_row_idx]
        monthly_total = convert_unit(str(row[data_cols[0]]))
        monthly_export = convert_unit(str(row[data_cols[1]]))
        monthly_import = convert_unit(str(row[data_cols[2]]))
        yoy_import_export = str(row[data_cols[0]+1])
        yoy_export = str(row[data_cols[1]+1])
        yoy_import = str(row[data_cols[2]+1])

        return pd.DataFrame([{
            "city_name": city_name,
            "monthly_total": monthly_total,
            "monthly_import": monthly_import,
            "monthly_export": monthly_export,
            "yoy_import_export": yoy_import_export,
            "yoy_import": yoy_import,
            "yoy_export": yoy_export
        }])

    except Exception as e:
        log.error(f"处理江门海关文件出错: {str(e)}")
        return pd.DataFrame()


def process_zhanjiang_customs(file_path, year, month):
    """处理湛江海关数据 满足「是第一次调用」或者「month == 12」任意一个条件"""
    global _zhanjiang_first_month

    # 判断是否应执行核心逻辑
    if _zhanjiang_first_month is None:
        # 第一次调用，记录初始月份
        _zhanjiang_first_month = month
        should_execute = True
    else:
        # 后续调用仅在以下情况下执行：
        # - 与初次调用的 month 相同（允许多城市同时处理）
        # - 或者 month == 12
        should_execute = (month == _zhanjiang_first_month) or (month == 12)

    if not should_execute:
        log.warning(f"跳过湛江海关{year}年{month}文件: {file_path.name}")
        return pd.DataFrame()
    try:
        df = pd.read_excel(file_path, sheet_name=0, header=None)
        log.info(f"处理湛江海关文件: {file_path.name}")
        # 从文件名确定城市
        city_name = "湛江市" if "湛江市" in file_path.name else "茂名市"
        # 查找月度数据表格
        month_str = f"{year}年前{month}个月{city_name}进出口数据（月度）"

        # target_header_row = None
        #
        # # 查找表头行
        # for i in range(min(3, len(df))):  # 在前5行找表头
        #     if any(month_str in str(cell) for cell in df.iloc[i]):
        #         target_header_row = i
        #         break
        #
        # if target_header_row is None:
        #     log.error(f"未找到 {month_str} 表头")
        #     return pd.DataFrame()

        target_header_row =1

        # 确定数据列位置
        data_cols = {}
        for col in range(len(df.columns)):
            cell_val = str(df.iloc[target_header_row+1, col])
            data_cols["year_month"] = 0
            if "进出口" in cell_val :
                data_cols["total"] = col
            elif "出口" in cell_val :
                data_cols["export"] = col
            elif "进口" in cell_val :
                data_cols["import"] = col

        if len(data_cols) < 1:
            log.error(f"未找到足够的 {month_str} 数据列")
            return pd.DataFrame()

        start_row = target_header_row + 4
        end_row = start_row + month

        # 提取多行数据
        rows = df.iloc[start_row:end_row]
        results = []

        for _, row in rows.iterrows():
            try:
                year_month = str(row[data_cols["year_month"]])
                formatted_year_month = f"{year_month[:4]}-{year_month[4:]}"
                monthly_total = Decimal(str(row[data_cols["total"]]))  # 进出口
                monthly_export = Decimal(str(row[data_cols["export"]]))  # 出口
                monthly_import = Decimal(str(row[data_cols["import"]]))  # 进口
                yoy_import_export = Decimal(str(row[data_cols["total"] + 1]))  # 进出口同比
                yoy_export = Decimal(str(row[data_cols["export"] + 1]))  # 出口同比
                yoy_import = Decimal(str(row[data_cols["import"] + 1]))  # 进口同比

                results.append({
                    "crossborder_year_month":formatted_year_month,
                    "city_name": city_name,
                    "monthly_total": monthly_total,
                    "monthly_import": monthly_import,
                    "monthly_export": monthly_export,
                    "yoy_import_export": yoy_import_export,
                    "yoy_import": yoy_import,
                    "yoy_export": yoy_export
                })
            except Exception as e:
                log.error(f"解析某一行数据出错: {e}")
                continue  # 单行错误不影响整体处理

        return pd.DataFrame(results)

    except Exception as e:
        log.error(f"处理湛江海关文件出错: {str(e)}")
        return pd.DataFrame()


def get_customs_processor(customs_name):
    """获取不同海关的处理函数"""
    processors = {
        "广州海关": process_guangzhou_customs,
        "深圳海关": process_shenzhen_customs,
        "汕头海关": process_shantou_customs,
        "黄埔海关": process_huangpu_customs,
        "江门海关": process_jiangmen_customs,
        "湛江海关": process_zhanjiang_customs
    }
    return processors.get(customs_name)


def parse_excel(current_dir):
    """主解析入口（优化为单参数模式）
    Args:
        current_dir (str): 当前月份数据目录（格式：/年份/省份/月份）
    """
    current_path = Path(current_dir)
    year, month = extract_year_month_from_path(current_path)

    log.info(f"开始处理 {year}年{month}月 数据: {current_dir}")

    try:
        # 动态获取前月目录
        prev_dir = get_previous_month_dir(current_path) if month != 1 else None
        if prev_dir:
            log.info(f"上个月数据目录: {prev_dir}")

        # 获取当前目录下所有Excel文件
        excel_files = list(current_path.glob("*.xls*"))
        if not excel_files:
            log.warning(f"当前目录下未找到Excel文件: {current_dir}")
            return

        # 按海关处理每个文件
        all_results = pd.DataFrame()
        customs_map = {
            "广州海关": [],
            "深圳海关": [],
            "汕头海关": [],
            "黄埔海关": [],
            "江门海关": [],
            "湛江海关": []
        }

        # 组织文件到不同的海关
        for file_path in excel_files:
            for customs_name in customs_map.keys():
                if match_customs_file(file_path.name, customs_name, year, month):
                    customs_map[customs_name].append(file_path)
                    log.info(f"匹配到 {customs_name} 文件: {file_path.name}")
                    break

        # 处理每个海关的文件
        for customs_name, file_list in customs_map.items():
            processor = get_customs_processor(customs_name)
            if not processor:
                continue

            for file_path in file_list:
                # 特殊处理深圳海关和江门海关的2月数据(缺少1月数据)
                if customs_name in ["深圳海关", "江门海关"] and month == 2:
                    # 获取2月份完整数据
                    df_full = processor(file_path, year, month)
                    if df_full.empty:
                        continue

                    # 创建1月份数据 (取2月份数据的一半)
                    df_half = df_full.copy()
                    for col in ['monthly_total', 'monthly_import', 'monthly_export']:
                        # 注意：只有数值列才进行减半操作，避免对字符串操作
                        if col in df_half.columns:
                            df_half[col] = df_half[col] / 2

                    # 设置1月份
                    df_half['crossborder_year_month'] = f'{year}-01'

                    # 设置2月份数据 (取2月份数据的一半)
                    df_full['crossborder_year_month'] = f'{year}-02'
                    for col in ['monthly_total', 'monthly_import', 'monthly_export']:
                        if col in df_full.columns:
                            df_full[col] = df_full[col] / 2

                    # 合并数据
                    df_customs = pd.concat([df_half, df_full])
                else:
                    # 正常处理数据
                    df_customs = processor(file_path, year, month)
                    if not df_customs.empty:
                        df_customs['month'] = month

                if not df_customs.empty:
                    all_results = pd.concat([all_results, df_customs])

        # 如果没有获取到数据
        if all_results.empty:
            log.warning(f"未处理到有效数据: {current_dir}")
            return

        # 添加公共字段
        all_results['prov_code'] = PROV_CODE
        all_results['prov_name'] = PROV_NAME
        all_results['crossborder_year'] = year
        all_results['city_code'] = all_results['city_name'].astype(str).map(GUANGDONG_CITY).fillna('0000')
        all_results['month'] = all_results.get('month', month)
        if 'crossborder_year_month' in all_results.columns:
            all_results['crossborder_year_month'] = (
                all_results['crossborder_year_month']
                .replace('', pd.NA)
                .fillna(f'{year}-{month:02d}')
            )
        else:
            all_results['crossborder_year_month'] = f'{year}-{month:02d}'

        # 排序并删除重复项
        # all_results = all_results.sort_values(by=['city_code', 'crossborder_year_month'])
        # all_results = all_results.drop_duplicates(subset=['crossborder_year_month', 'city_code'], keep='last')

        # 打印处理结果
        log.info(f"处理完成，共获得广东省 {len(all_results)} 条地级市数据")

        # 选择入库字段
        final_df = all_results[[
            'crossborder_year_month', 'prov_code', 'prov_name',
            'crossborder_year','city_code', 'city_name',
            'monthly_total','monthly_import', 'monthly_export',
            'yoy_import_export','yoy_import', 'yoy_export'
        ]].copy()

        final_df = final_df.where(pd.notna(final_df), None)

        # 打印前几条数据
        # log.debug(f"处理后数据示例:\n{final_df.head()}")

        # 这里调用DBHelper入库（实际使用时请取消注释）
        from db_helper import DBHelper
        db = DBHelper()
        db.bulk_insert(
            final_df,
            't_yujin_crossborder_prov_region_trade',
            conflict_columns=['crossborder_year_month', 'city_code'],
            update_columns=['monthly_total', 'monthly_import', 'monthly_export',
                            'yoy_import_export', 'yoy_import', 'yoy_export']
        )

        log.info(f"{current_dir}数据已全部成功处理")

    except Exception as e:
        log.error(f"处理失败：{current_dir}，错误：{str(e)}")
        raise

def convert_unit(value):
    """亿元转万元，处理空值"""
    try:
        # 如果 value 不是特殊的无效值，进行转换并保留4位小数
        return round(Decimal(value) * 10000, 4) if value not in ['-', ''] else None
    except (InvalidOperation, ValueError):
        # 捕获异常，返回 None
        return None



# 测试入口
if __name__ == "__main__":

    traverse_and_process(download_dir, parse_excel, province_name="guangdong")
    db_helper = DBHelper()
    db_helper.update_prov_yoy("广东省")