crossborder/crossborder/zhejiang/crawl_gov_zhejiang_full.py

import argparse
import os
import random
import re
import time
import sys
from datetime import datetime, timedelta
from pathlib import Path
from urllib.parse import urljoin

from faker import Faker
from selenium import webdriver
from selenium.common.exceptions import StaleElementReferenceException
from selenium.webdriver import FirefoxOptions
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.base_country_code import extract_year_month
from crossborder.utils.dingtalk import send_dingtalk_message
from crossborder.zhejiang import download_dir
from crossborder.zhejiang import gov_commodity_zhejiang_city
from crossborder.zhejiang import gov_commodity_zhejiang_country
from crossborder.zhejiang import gov_commodity_zhejiang_import_export
from crossborder.utils import base_country_code, base_mysql
from crossborder.utils.log import  get_logger

log = get_logger(__name__)

def configure_stealth_options():
    """增强型反检测配置[1,4](@ref)"""
    opts = FirefoxOptions()
    print("当前下载路径:", Path(download_dir).resolve())
    # 文件下载配置
    opts.set_preference("browser.download.dir", download_dir)
    opts.set_preference("browser.download.folderList", 2)
    opts.set_preference("browser.download.manager.showWhenStarting", False)
    opts.set_preference("browser.helperApps.neverAsk.saveToDisk",
                        "application/octet-stream, application/vnd.ms-excel")  # 覆盖常见文件类型
    opts.set_preference("browser.download.manager.useWindow", False)  # 禁用下载管理器窗口
    opts.set_preference("browser.download.manager.showAlertOnComplete", False)  # 关闭完成提示

    # 反检测参数
    opts.set_preference("dom.webdriver.enabled", False)
    opts.set_preference("useAutomationExtension", False)
    opts.add_argument("--disable-blink-features=AutomationControlled")

    # 动态指纹
    fake = Faker()
    opts.set_preference("general.useragent.override", fake.firefox())
    opts.set_preference("intl.accept_languages", "zh-CN,zh;q=0.9")

    # 视口配置
    opts.add_argument("--width=1440")
    opts.add_argument("--height=900")
    opts.add_argument("--headless")
    return opts

def crawl_by_year_tabs(driver, base_url, year_month):
    """按年份Tab导航采集数据"""
    years = ['2023年', '2024年', '2025年']
    WebDriverWait(driver, 30).until(
        EC.presence_of_element_located((By.CLASS_NAME, "portlet"))
    )

    year_tabs = driver.find_elements(By.XPATH, '//ul[@class="nav_sj"]//li//a')
    for tab in year_tabs:
        year_text = tab.text.strip()
        if int(year_text[:4]) <= 2022:
            log.info(f"{year_text} 后的数据无需下载")
            continue

        year_url = tab.get_attribute("href")
        if not year_url.startswith(('http://', 'https://')):
            year_url = base_url.split('//')[0] + '//' + base_url.split('/')[2] + year_url

        # 新标签页打开年份页面
        driver.execute_script("window.open(arguments[0]);", year_url)
        driver.switch_to.window(driver.window_handles[-1])
        log.info(f"\n正在处理 {year_text} 年份页面")

        process_month_tabs(driver, year_text, base_url, year_month)

        # 返回主窗口
        driver.close()
        driver.switch_to.window(driver.window_handles[0])


def get_current_and_previous_month(text):
    """
    将类似 "2025年-五月" 或 "2025年十二月" 的字符串解析为：
    - 当前年份
    - 当前中文月份
    - 上一个月的 (年份, 中文月份)

    :param text: 输入文本，如 "2025年-五月"
    :return: tuple(current_year, current_month, previous_year, previous_month)
    """

    # 中文月份映射表
    month_map = {
        '一': 1, '二': 2, '三': 3, '四': 4, '五': 5, '六': 6,
        '七': 7, '八': 8, '九': 9, '十': 10, '十一': 11, '十二': 12
    }

    reverse_month_map = {v: k for k, v in month_map.items()}

    # 提取年份和中文月份
    match = re.search(r'(\d{4})年[-\s]?([一二三四五六七八九十][一|二]?月)', text)
    if not match:
        raise ValueError(f"无法从 '{text}' 提取有效的年份和月份")

    year_str = match.group(1)
    chinese_month_str = match.group(2).replace('月', '')

    # 处理中文月份转数字
    if chinese_month_str in month_map:
        current_month_num = month_map[chinese_month_str]
    elif chinese_month_str == '十月':
        current_month_num = 10
    else:
        raise ValueError(f"不支持的中文月份格式: {chinese_month_str}")

    current_year_num = int(year_str)

    # 计算上个月
    if current_month_num > 1:
        previous_year_num = current_year_num
        previous_month_num = current_month_num - 1
    else:
        previous_year_num = current_year_num - 1
        previous_month_num = 12

    # 转换为中文月份
    previous_month_chinese = reverse_month_map[previous_month_num]

    return (
        f"{current_year_num}年",
        chinese_month_str + '月',
        f"{previous_year_num}年",
        previous_month_chinese + '月'
    )

def process_month_tabs(driver, year, base_url, year_month):
    """处理月份Tab导航（动态获取真实存在的月份）"""
    # 显式等待容器加载
    WebDriverWait(driver, 30).until(
        EC.presence_of_element_located((By.CLASS_NAME, "portlet"))
    )

    target_months = ['一月', '二月', '三月', '四月', '五月', '六月',
                     '七月', '八月', '九月', '十月', '十一月', '十二月']
    processed_months = set()  # 已处理月份记录
    retry_count = 0

    y1, m1, y2, m2 = get_current_and_previous_month(year_month)

    while retry_count < 3:
        try:
            # 全量获取所有月份Tab
            month_items = driver.find_elements(By.XPATH, '//ul[@class="nav_tab"]//li')
            if not month_items:
                log.info(f"{year}年没有月份Tab，停止处理")
                break

            all_found = True
            month_text = ''
            found = False
            for i,item in enumerate(month_items):
                a_tag = item.find_element(By.XPATH, './/a')
                month_text = a_tag.text.strip()

                if month_text in processed_months:
                    continue

                if not month_text in target_months:
                    continue  # 跳过已处理月份

                log.info(f"点击月份Tab：{year}-{month_text}")
                if year_month is not None:
                    if y1 != year or y2 != year:
                        retry_count += 1
                        break
                    if not (y1 == year and m1 == month_text) and not (y2 == year and m2 == month_text):
                        log.info(f"{year}年 {month_text} 月份跳过, auto tar: {year_month}")
                        continue
                a_tag.click()

                # 处理详情页逻辑
                WebDriverWait(driver, 30).until(
                    EC.presence_of_element_located((By.CLASS_NAME, "portlet"))
                )
                detail_link_arr = get_behind_detail_link(driver, base_url)
                if not detail_link_arr:
                    log.info(f"{year}-{month_text} 未找到详情链接")
                for detail_link in detail_link_arr:
                    log.info(f"{year}-{month_text} 详情链接：{detail_link}")
                    driver.get(detail_link)
                    download_file_from_detail_page(driver)
                    driver.back()
                    WebDriverWait(driver, 30).until(
                        EC.presence_of_element_located((By.CLASS_NAME, "portlet"))
                    )

                processed_months.add(month_text)
                found = True

            if not found:
                log.info(f"{year}年未找到 {month_text} Tab")
                all_found = False

            if all_found:
                log.info(f"{year}年所有目标月份处理完成")
                break
            else:
                # 部分月份未找到，重新获取元素
                log.info(f"第 {retry_count} 次重试获取月份Tab...")

        except StaleElementReferenceException:
            log.info("页面刷新，重新获取月份Tab列表...")
            time.sleep(2)

    log.info(f"{year}年最终处理的月份：{processed_months}")

def get_behind_detail_link(driver, base_url):
   """获取点击月份Tab后 conList_ul 下所有 li 的 a 标签完整链接"""
   href_arr = []
   try:
       elements = WebDriverWait(driver, 30).until(
           EC.element_to_be_clickable((By.XPATH, '//ul[@class="conList_ul"]/li/a'))
       )
       elements = elements.find_elements(By.XPATH, '//ul[@class="conList_ul"]/li/a')
       for element in elements:
           href = element.get_attribute("href")
           full_url = urljoin(base_url, href)  # 自动处理相对路径
           href_arr.append(full_url)
       return href_arr
   except Exception as e:
       log.info(f"获取详情链接失败: {str(e)}")
       return []

def download_file_from_detail_page(driver):
    WebDriverWait(driver, 30).until(
        EC.presence_of_element_located((By.CLASS_NAME, "portlet"))
    )

    try:
        elements = driver.find_elements(By.XPATH, '//div[@class="easysite-news-content"]//div[@id="easysiteText"]//p//a')
        if not elements:
            log.info("详情页未找到目标文件链接")
            return

        for download_btn in elements:
            file_name = download_btn.text.strip()
            if not file_name:
                continue
            file_url = download_btn.get_attribute("href")

            if not file_url.lower().endswith(('.xls', '.xlsx')):
                log.info(f"跳过非 Excel 文件: {file_url}")
                continue

            log.info(f"正在下载: {file_name} → {file_url}")

            # 记录下载前的文件列表
            existing_files = set(f.name for f in Path(download_dir).glob('*'))
            # 随机点击延迟
            time.sleep(random.uniform(1, 3))
            download_btn.click()

            downloaded_file = wait_for_download_complete(existing_files=existing_files)

            year, start_month, month = extract_year_and_month(file_name)
            final_path = Path(download_dir) / year / month / f"{file_name}"
            if os.path.exists(final_path):
                log.info(f"文件已存在：{file_name} 正在覆盖...")
                os.unlink(final_path)

            final_dir = Path(download_dir) / year / month
            final_dir.mkdir(parents=True, exist_ok=True)
            log.info(f"√ 正在移动文件 {downloaded_file} 至 {final_path}")
            downloaded_file.rename(final_path)
            log.info(f"√ 下载成功：{final_path}")

    except Exception as e:
        log.info(f"详情页处理异常: {str(e)}")

def extract_year_and_month(file_name):
    # 支持两种格式：
    #  - 2025年1-2月xxx
    #  - 2025年3月xxx
    match = re.search(r"(\d{4})年(\d{1,2})(?:-(\d{1,2}))?月", file_name)

    if match:
        year = match.group(1)
        start_month = match.group(2)
        end_month = match.group(3) if match.group(3) else start_month

        return year, start_month.zfill(2), end_month.zfill(2)
    else:
        raise ValueError(f"无法从文件名中提取年份和月份：{file_name}")


def convert_to_chinese_uppercase(num):
    if not 1 <= num <= 12:
        return None  # 超出范围的数字返回 None 或根据需要处理
    if num < 10:
        return '零一二三四五六七八九'[num]
    elif num == 10:
        return '十'
    elif num == 11:
        return '十一'
    elif num == 12:
        return '十二'
    return None

def detect_latest_month(driver, url):
    driver.get(url)
    current_date = datetime.now()
    for offset in range(0, 3):
        check_date = current_date - timedelta(days=offset * 30)
        check_year = check_date.year
        month = check_date.month
        check_month = convert_to_chinese_uppercase(month)

        target_title = f"{check_month}月"
        try:
            WebDriverWait(driver, 10).until(
                EC.presence_of_element_located((By.XPATH, f'//ul[@class="nav_tab"]//li/a[normalize-space()="{target_title}"]'))
            )
            log.info(f"已找到最新月份数据 {check_year}-{check_month}")
            # 看是否已存表，已存则跳过；
            count = base_mysql.get_code_exist(f'{check_year}-{month:02d}', '330000')
            if count > 0:
                log.info(f"count: {count} -> 已存在 {check_year}-{check_month} 数据，跳过")
                continue
            return f"{check_year}年-{check_month}月"
        except:
            log.error(f"未找到 {target_title}")
            continue
    log.error("三个月内未找到有效数据")
    return None

def extract_year_month_chinese(text):
    """
    支持格式：
    - 2025年四月
    - 2025年-四月
    - 2024年十二月
    - 2023年-二月
    """
    # 中文月份映射表
    month_map = {
        '一': '01',
        '二': '02',
        '三': '03',
        '四': '04',
        '五': '05',
        '六': '06',
        '七': '07',
        '八': '08',
        '九': '09',
        '十': '10',
        '十一': '11',
        '十二': '12'
    }

    # 正则匹配年份和中文月份
    match = re.search(r"(\d{4})年-?([一|二|三|四|五|六|七|八|九|十]{1,2}[一|二]?月)", text)

    if not match:
        raise ValueError(f"无法从文本中提取年份和月份: {text}")

    year = match.group(1)  # 提取年份

    # 提取中文月份并处理成数字
    chinese_month = match.group(2).replace('月', '')
    if chinese_month in month_map:
        month = month_map[chinese_month]
    else:
        # 特殊处理 "十月"
        if chinese_month == '十月':
            month = '10'
        else:
            raise ValueError(f"不支持的中文月份格式: {chinese_month}")

    return year, month

def crawl_with_selenium(url, mark):
    driver = webdriver.Firefox(options=configure_stealth_options())

    year_month = None
    if 'auto' == mark:
        res = detect_latest_month(driver, url)
        if res is None:
            log.info("浙江省海关没有最新数据更新")
            return None
        year_month = res
        print(f"检测到最新有效数据：{year_month}")

    base_url = 'http://hangzhou.customs.gov.cn'
    try:
        # 注入反检测脚本
        driver.execute_script("""
            Object.defineProperty(navigator, 'webdriver', { 
                get: () => undefined 
            });
            window.alert = () => {};
        """)

        # 页面加载策略
        driver.get(url)

        # 按年份导航
        crawl_by_year_tabs(driver, base_url, year_month)

    finally:
        driver.quit()
        log.info(f"浙江省海关全量数据下载任务完成")
        # 等待5s后执行
        time.sleep(5)
        hierarchical_traversal(download_dir, year_month)
        log.info("浙江省海关类章、国家、城市所有文件处理完成！")
        time.sleep(5)
        base_mysql.update_shandong_yoy('浙江省')
        log.info("浙江省海关城市同比sql处理完成")
    return 'finish', year_month

def wait_for_download_complete(timeout=30, existing_files=None):
    """
    监控下载目录，等待文件下载完成并返回新下载的文件。

    :param timeout: 超时时间（秒）
    :param existing_files: 下载前已存在的文件列表
    :return: 新下载的文件路径
    """
    start_time = time.time()
    temp_exts = ('.part', '.crdownload')

    if existing_files is None:
        existing_files = set(f.name for f in Path(download_dir).glob('*'))

    while (time.time() - start_time) < timeout:
        # 获取有效文件列表
        valid_files = []
        for f in Path(download_dir).glob('*'):
            if (f.name not in existing_files and
                not f.name.endswith(temp_exts) and
                f.stat().st_size > 0):
                valid_files.append(f)

        # 等待最新文件稳定
        if valid_files:
            return max(valid_files, key=lambda x: x.stat().st_mtime)
        time.sleep(2)
    raise TimeoutError("文件下载超时")

def hierarchical_traversal(root_path, year_month):
    """分层遍历：省份->年份->月目录"""
    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):
        # 构造完整的路径：download/shandong/2025/03
        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']}")
                path = md['path']
                if year_month is not None:
                    year, month = extract_year_month_chinese(year_month)
                    parts = path.parts
                    if year_dir.name != year or parts[-1] != month:
                        log.info(f"浙江省海关已处理 {year_month} 数据，返回")
                        return
                gov_commodity_zhejiang_import_export.process_folder(path)
                gov_commodity_zhejiang_country.process_folder(path)
                gov_commodity_zhejiang_city.process_folder(path)

def main():
    try:
        parser = argparse.ArgumentParser(description='海关数据智能抓取系统')
        parser.add_argument('--year', type=int, default=None, help='终止年份（如2023），未指定时抓取最新两个月')
        args = parser.parse_args()

        start_time = time.time()
        if args.year == 2023:
            log.info("正在全量采集浙江省海关数据")
            crawl_with_selenium('http://hangzhou.customs.gov.cn/hangzhou_customs/575609/zlbd/575612/575612/6430241/6430315/index.html', 'all')
            duration = time.time() - start_time
            minutes, seconds = divmod(duration, 60)
            send_dingtalk_message(f'【浙江省海关】全量数据采集完成，耗时 {int(minutes)}分{seconds:.1f}秒')
        else:
            log.info("正在增量采集浙江省海关数据")
            res = crawl_with_selenium('http://hangzhou.customs.gov.cn/hangzhou_customs/575609/zlbd/575612/575612/6430241/6430315/index.html','auto')
            if res is not None:
                r1, r2 = res
                if r1 == 'finish':
                    duration = time.time() - start_time
                    minutes, seconds = divmod(duration, 60)
                    send_dingtalk_message(f'【浙江省海关】 {r2} 增量数据采集完成，耗时 {int(minutes)}分{seconds:.1f}秒')

    except Exception as e:
        send_dingtalk_message(f"【浙江省海关】发生错误：{e}")

if __name__ == '__main__':
    main()