crossborder/crossborder/guangdong/selenium_guangdong_download.py

import argparse
import random
import re
import time
from datetime import datetime, timedelta

import numpy as np
import pandas as pd
from selenium import webdriver
from selenium.common import TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait

from crossborder.utils import base_mysql
from crossborder.utils.db_helper import DBHelper
from crossborder.utils.constants import DOWNLOAD_DIR, COUNTRY_CODE_MAPPING
from crossborder.utils.dingtalk import send_dingtalk_message
from crossborder.utils.download_utils import configure_stealth_options, generate_month_sequence
from crossborder.utils.log import  get_logger

log = get_logger(__name__)
from crossborder.utils.parse_utils import clean_county_name, convert_wan_to_yuan, clean_commodity_name


BASE_URL = "http://gdfs.customs.gov.cn/guangdong_sub/zwgk62/sjgb59/6b4cdb3f-1.html"
download_dir = DOWNLOAD_DIR / "guangdong"



PROV_CODE = "440000"
PROV_NAME = "广东省"

db = DBHelper()

#广东省海关数据    页面无下载按钮，这里数据为在线读取页面表格数据

def detect_latest_month(driver):
    """三级回溯智能检测最新有效月份（修正年/月匹配逻辑）"""
    driver.get(BASE_URL)
    current_date = datetime.now()

    for offset in range(0, 3):
        check_date = current_date - timedelta(days=offset * 30)
        check_year = check_date.year
        check_month = check_date.month

        # 构建正则表达式：兼容“1至X月”和“X月”两种格式，并允许前后有空格
        pattern = re.compile(
            rf'（5）{check_year}\s*年\s*(?:1至)?{check_month}\s*月广东省外贸进出口主要国别（地区）总值表（人民币值）',
            re.IGNORECASE
        )

        try:
            # 使用 Python 端的正则匹配所有含“广东省”的链接 title
            elements = WebDriverWait(driver, 10).until(
                EC.presence_of_all_elements_located((By.XPATH, '//a[contains(@title,"广东省")]'))
            )

            for element in elements:
                title = element.get_attribute("title")
                if pattern.search(title):
                    log.info(f"已找到最新月份数据 {check_year}-{check_month}")
                    return check_year, check_month

            log.error(f"未找到匹配项（正则：{pattern.pattern}）")
        except TimeoutException:
            log.info(f"页面加载超时或无匹配项（{check_year}-{check_month}）")
            continue

    raise Exception("三个月内未找到有效数据")


def process_month_data(driver, year, month):
    """处理月度数据：支持三种表格类型"""
    patterns = [
        (re.compile(rf'（5）{year}\s*年\s*(1-)?{month}\s*月广东省外贸进出口主要国别（地区）总值表（人民币值）'), 'country'),
        (re.compile(rf'（6）{year}\s*年\s*(1-)?{month}\s*月广东省出口重点商品总值表（人民币值）'), 'export_commodity'),
        (re.compile(rf'（7）{year}\s*年\s*(1-)?{month}\s*月广东省进口重点商品总值表（人民币值）'), 'import_commodity')
    ]

    found_count = 0
    commodity_data = {'export': [], 'import': []}  # 存储商品数据等待合并
    links = driver.find_elements(By.XPATH, '//a[contains(@title,"广东省")]')

    for link in links:
        title = link.get_attribute("title")
        for pattern, table_type in patterns:
            if pattern.search(title):
                log.info(f"处理表格: {title}")
                url = link.get_attribute("href")

                # 新标签页处理
                driver.execute_script("window.open(arguments[0]);", url)
                driver.switch_to.window(driver.window_handles[-1])
                try:
                    WebDriverWait(driver, 15).until(
                        EC.presence_of_element_located((By.XPATH, "//table[@border='1']"))
                    )

                    # 根据表格类型处理数据
                    if table_type == 'country':
                        data = parse_page_country_data(driver, year, month)
                        df_country = pd.DataFrame(data)
                        db.bulk_insert(
                            df_country,
                            't_yujin_crossborder_prov_country_trade',
                            conflict_columns=['crossborder_year_month', 'prov_code', 'country_code'],
                            update_columns=['monthly_total', 'monthly_import', 'monthly_export',
                                            'yoy_import_export', 'yoy_import', 'yoy_export']
                        )

                        found_count += 1
                    else:
                        data_type = 'export' if table_type == 'export_commodity' else 'import'
                        commodity_data[data_type] = parse_page_commodity_data(driver, data_type, year, month)
                        found_count += 1
                except Exception as e:
                    log.info(f"表格处理失败: {e}")

    # 将数据返回，而不是在内部合并
    return found_count, commodity_data



def parse_page_country_data(driver, year, month):
    """解析目标页面的表格数据"""
    data = []

    try:
        # 等待表格加载
        WebDriverWait(driver, 15).until(
            EC.presence_of_element_located((By.XPATH, "//table[@border='1']"))
        )
        table = driver.find_element(By.XPATH, "//table[@border='1' and @bordercolor='#000000']")
        rows = table.find_elements(By.TAG_NAME, 'tr')

        # 检测表格的列数
        header_row = rows[2]  # 假设表头在第3行
        header_row.find_elements(By.TAG_NAME, 'td')


        # 数据行从第4行开始（跳过表头）
        for row in rows[4:]:
            cols = [col.text.strip() for col in row.find_elements(By.TAG_NAME, 'td')]
            country_name = cols[0]

            if (country_name == '广东外贸总值' or
                    country_name == '东盟' or
                    country_name == '欧盟' or
                    country_name == '总计' or
                    country_name == '广东外贸总额' or
                    country_name == '广东外贸总计' or
                    country_name == '总值'
            ):
                continue

            if month == 2:
                # 处理合并后的1月和2月数据
                monthly_total = convert_wan_to_yuan(cols[1])
                monthly_export = convert_wan_to_yuan(cols[3])
                monthly_import = convert_wan_to_yuan(cols[5])

                # 将2月的数据除以2，并生成1月和2月的数据
                for m in [1, 2]:
                    adjusted_monthly_total = monthly_total / 2
                    adjusted_monthly_export = monthly_export / 2
                    adjusted_monthly_import = monthly_import / 2
                    adjusted_yoy_total = 0
                    adjusted_yoy_export = 0
                    adjusted_yoy_import = 0

                    country_name_clean = clean_county_name(country_name)
                    country_code = COUNTRY_CODE_MAPPING.get(country_name_clean)

                    data.append({
                        'crossborder_year': year,
                        'crossborder_year_month': f"{year}-{m:02d}",
                        'prov_code': PROV_CODE,
                        'prov_name': PROV_NAME,
                        'country_code': country_code,
                        'country_name': country_name_clean,
                        'monthly_total': adjusted_monthly_total,
                        'monthly_export': adjusted_monthly_export,
                        'monthly_import': adjusted_monthly_import,
                        'yoy_import_export': adjusted_yoy_total,
                        'yoy_export': adjusted_yoy_export,
                        'yoy_import': adjusted_yoy_import
                    })
            else:
                # 原逻辑处理13列的情况
                monthly_total = convert_wan_to_yuan(cols[3])
                monthly_export = convert_wan_to_yuan(cols[7])
                monthly_import = convert_wan_to_yuan(cols[11])
                yoy_total = parse_number(cols[4])
                yoy_export = parse_number(cols[8])
                yoy_import = parse_number(cols[12])

                country_name_clean = clean_county_name(country_name)
                country_code = COUNTRY_CODE_MAPPING.get(country_name_clean)

                data.append({
                    'crossborder_year': year,
                    'crossborder_year_month': f"{year}-{month:02d}",
                    'prov_code': PROV_CODE,
                    'prov_name': PROV_NAME,
                    'country_code': country_code,
                    'country_name': country_name_clean,
                    '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.info(f"解析表格失败: {e}")
        raise
    finally:
        driver.close()
        driver.switch_to.window(driver.window_handles[0])

    return data



def parse_page_commodity_data(driver, data_type, year, month):
    """解析商品表通用函数"""
    data = []
    try:
        table = driver.find_element(By.XPATH, "//table[@border='1' and @bordercolor='#000000']")
        rows = table.find_elements(By.TAG_NAME, 'tr')

        # 检测表格的列数
        header_row = rows[2]  # 假设表头在第3行
        cols = header_row.find_elements(By.TAG_NAME, 'td')
        num_cols = len(cols)

        # 数据行从第4行开始（跳过表头）
        for row in rows[4:]:
            cols = [col.text.strip() for col in row.find_elements(By.TAG_NAME, 'td')]
            if len(cols) < 3:
                continue

            # 清洗商品名称（处理&nbsp;和空格）
            name = clean_commodity_name(cols[0])
            if name == '肉类':
                name = '肉类（包括杂碎）'

            if month == 2:
                # 处理合并后的1月和2月数据
                value = convert_wan_to_yuan(cols[1])


                # 将2月的数据除以2，并生成1月和2月的数据
                for m in [1, 2]:
                    adjusted_value = value / 2
                    adjusted_yoy = 0  # 同比置为0

                    data.append({
                        'commodity_name': name,
                        'commodity_code': db.get_commodity_id(name),
                        'monthly_export' if data_type == 'export' else 'monthly_import': adjusted_value,
                        'crossborder_year_month': f"{year}-{m:02d}"
                    })
            else:
                # 原逻辑处理5列的情况
                value = convert_wan_to_yuan(cols[3] if data_type == 'export' else cols[3])


                data.append({
                    'commodity_name': name,
                    'commodity_code': db.get_commodity_id(name),
                    'monthly_export' if data_type == 'export' else 'monthly_import': value,
                    'crossborder_year_month': f"{year}-{month:02d}"
                })

    except Exception as e:
        log.info(f"解析商品表失败: {e}")
        raise
    finally:
        driver.close()
        driver.switch_to.window(driver.window_handles[0])

    return data


def merge_commodity_data(import_data, export_data, year, month):
    """
    根据commodity_code合并进出口数据（支持不同商品的存在情况）
    :param year:
    :param month:
    :param import_data: 进口数据列表（含commodity_code）
    :param export_data: 出口数据列表（含commodity_code）
    :return: 合并后的DataFrame
    """
    # 转换数据为DataFrame
    df_import = pd.DataFrame(import_data)
    df_export = pd.DataFrame(export_data)

    # 合并逻辑（全外连接保留所有商品）
    merged_df = pd.merge(
        df_import,
        df_export,
        on=['commodity_code',  'commodity_name',  'crossborder_year_month'],
        how='outer'
    )

    # 计算总量（可选，根据表结构需求）

    #进出口总值计算
    merged_df['monthly_total'] = merged_df['monthly_import'].fillna(0) + merged_df['monthly_export'].fillna(0)
    merged_df['monthly_total'] = merged_df['monthly_total'].replace(0, np.nan)

    merged_df['crossborder_year'] = year
    #不为空是填充传入年月,1.2月数据在上级已经构建好
    merged_df['crossborder_year_month'] = merged_df['crossborder_year_month'].fillna(f"{year}-{month:02d}")
    merged_df['prov_code'] = PROV_CODE
    merged_df['prov_name'] = PROV_NAME

    return merged_df

def parse_number(text):
    """转换文本为浮点数（处理空值、负号）"""
    text = text.strip().replace(',', '')
    if not text or text == '-':
        return None
    try:
        return float(text)
    except ValueError:
        return None


# 优化后的代码逻辑：

def reverse_crawler(driver, target_months):
    """逆向分页抓取核心逻辑"""
    processed_months = set()
    page = 1
    for year, month in target_months:
        log.info(f"\n开始处理 {year}年{month}月数据".center(50, "="))
        WebDriverWait(driver, 15).until(EC.presence_of_element_located((By.CLASS_NAME, "conList_ul")))

        current_page = 1
        found_tables = 0
        export_data = []
        import_data = []

        while True:
            random_sleep(base=2, variance=3)
            try:
                log.info(f"当前页面：{driver.current_url}, 第{page}页")
                found, commodity_data = process_month_data(driver, year, month)
                found_tables += found

                # 累积商品数据
                if commodity_data['export']:
                    export_data.extend(commodity_data['export'])
                if commodity_data['import']:
                    import_data.extend(commodity_data['import'])

                # 完成三个表格采集
                if found_tables >= 3:
                    # 确保同时有进口和出口数据
                    if export_data and import_data:
                        final_df = merge_commodity_data(export_data, import_data, year, month)
                        final_df = final_df.replace({np.nan: None})
                        db.bulk_insert(df=final_df, table_name='t_yujin_crossborder_prov_commodity_trade',
                                      conflict_columns=['commodity_code', 'crossborder_year_month'],
                                      update_columns=['monthly_import', 'monthly_export', 'monthly_total'])
                    log.info(f"已完成{year}年{month}月全部表格采集")
                    processed_months.add((year, month))
                    break

                log.info(f"第{page}页已采集表格数：{found_tables}/3,前往下一页采集")

                # 分页点击逻辑
                WebDriverWait(driver, 15).until(
                    EC.element_to_be_clickable((By.XPATH, '//a[@class="pagingNormal next"]'))
                ).click()

                current_page += 1
                page += 1

            except TimeoutException:
                log.info(f"未找到更多分页，已采集表格数：{found_tables}/3")
                break
            except Exception as e:
                log.info(f"分页异常：{str(e)}")
                handle_retry(driver)
                break

    return processed_months



def random_sleep(base=2, variance=5):
    """智能随机等待"""
    sleep_time = base + random.random() * variance
    time.sleep(sleep_time)

def handle_retry(driver):
    """异常恢复处理"""
    try:
        driver.refresh()
        WebDriverWait(driver, 15).until(
            EC.presence_of_element_located((By.CLASS_NAME, "conList_ul"))
        )
        log.info("浏览器异常已恢复")
    except:
        log.info("需要人工干预的严重错误")
        raise


def main():
    """主入口（优化广东海关数据采集逻辑）"""
    parser = argparse.ArgumentParser(description='海关数据智能抓取系统')
    parser.add_argument('--year', type=int, default=None,
                        help='终止年份（如2023），未指定时抓取最新两个月')
    args = parser.parse_args()
    start_time = time.time()
    target_months = []  # 初始化目标月份列表
    data_collected = False  # 数据采集状态标记
    log.info("【广东海关】数据抓取开始".center(66, "*"))

    driver = None  # 初始化避免未定义

    try:
        # 1. 初始化浏览器
        driver = webdriver.Firefox(options=configure_stealth_options(download_dir))
        log.info("浏览器初始化完成")

        # 2. 检测最新有效月份
        valid_year, valid_month = detect_latest_month(driver)
        log.info(f"【广东海关】最新数据：{valid_year}年{valid_month:02d}月")

        # 3. 数据存在性检查（仅在未指定年份时执行）
        if not args.year:
            db = DBHelper()
            count = db.get_code_exist(f'{valid_year}-{valid_month:02d}', PROV_CODE)
            if count > 0:
                log.error(f"数据库已存在【广东省】 {valid_year}-{valid_month:02d} 商品贸易数据，本次抓取终止")
                return

        # 4. 生成目标月份序列
        if args.year:
            target_months = generate_month_sequence(
                start_year=valid_year,
                start_month=valid_month,
                end_year=args.year,
                skip_january=True
            )
        else:
            # 未指定年份时只抓最近两个月份
            target_months = generate_month_sequence(
                start_year=valid_year,
                start_month=valid_month
            )

        log.info(f"【广东海关】目标采集月份序列：{len(target_months)}个月份")

        # 5. 执行数据采集
        reverse_crawler(driver, target_months)
        data_collected = True
        log.info(f"【广东海关】成功采集 {len(target_months)} 个月份数据")

        # 6. 数据清洗入库（如有需要可添加）
        # log.info("\n【广东海关】数据清洗入库中...")
        # traverse_and_process(...)

    except Exception as e:
        # 捕获并记录所有异常
        log.exception(f"【广东海关】采集过程中发生错误: {str(e)}")
        send_dingtalk_message(f"【广东海关数据采集异常】{str(e)}")

    finally:
        # 确保浏览器退出
        if driver:
            driver.quit()
            log.info("浏览器已退出")

        # 7. 只有在成功采集数据时才发送通知
        if data_collected:
            duration = time.time() - start_time
            minutes, seconds = divmod(duration, 60)
            message = (f"【广东海关】{len(target_months)}个月份数据采集完成"
                       f"，总耗时：{int(minutes)}分{seconds:.1f}秒")
            send_dingtalk_message(message)

        log.info("【广东海关】处理流程结束".center(66, "*"))


if __name__ == "__main__":
    main()