price_crawler.py

import os
import time
from datetime import date, timedelta
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

# --- 配置区 ---
CHROME_DRIVER_PATH = r"F:\1.PyCharm project\251013_Python crawling government data_ZZJuan\chromedriver.exe"
DOWNLOAD_DIRECTORY = os.path.join(os.getcwd(), "downloads")
TARGET_URL = "https://fgw.ln.gov.cn/fgw/xxgk/jgjc/mrjg/index.shtml"

def daterange(start_date, end_date):
    for n in range(int((end_date - start_date).days) + 1):
        yield start_date + timedelta(n)

# 1. 增强浏览器配置
def setup_driver():
    """初始化并返回一个配置好的、增强反检测的Chrome WebDriver"""
    chrome_options = Options()
    if not os.path.exists(DOWNLOAD_DIRECTORY):
        os.makedirs(DOWNLOAD_DIRECTORY)
    prefs = {"download.default_directory": DOWNLOAD_DIRECTORY}
    chrome_options.add_experimental_option("prefs", prefs)

    # 添加用户配置，避免被识别为自动化程序
    chrome_options.add_argument("--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36")
    chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
    chrome_options.add_experimental_option("useAutomationExtension", False)

    service = Service(CHROME_DRIVER_PATH)
    driver = webdriver.Chrome(service=service, options=chrome_options)

    # 执行JS代码移除webdriver标记
    driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
        "source": """
            Object.defineProperty(navigator, 'webdriver', {
                get: () => undefined
            })
        """
    })
    return driver

# 3. 修改日期输入方式
def perform_query(driver, date_str):
    """使用JavaScript执行日期输入，兼容非标准控件"""
    try:
        wait = WebDriverWait(driver, 10)

        # 1. 定位日期输入框并通过JS设置值
        date_input = wait.until(EC.visibility_of_element_located((By.NAME, "tnxtime")))
        driver.execute_script(f"arguments[0].value = '{date_str}';", date_input)
        driver.execute_script("arguments[0].dispatchEvent(new Event('input'));", date_input)
        print(f"  - 已通过JS输入日期: {date_str}")

        # 2. 定位查询按钮并点击
        query_button = wait.until(EC.element_to_be_clickable((By.NAME, "btncxjq")))
        query_button.click()
        print("  - 已点击查询按钮")

        # 3. 等待结果加载
        wait.until(EC.visibility_of_element_located((By.ID, "Button3")))
        time.sleep(1)
        print("  - 页面已刷新")
        return True
    except Exception as e:
        print(f"  - 错误：执行查询时出错: {e}")
        return False

def rename_latest_file(date_str, file_type, files_before):
    time.sleep(3)
    files_after = os.listdir(DOWNLOAD_DIRECTORY)
    new_files = list(set(files_after) - set(files_before))
    new_files = [f for f in new_files if not f.endswith('.crdownload')]
    if new_files:
        latest_file = new_files[0]
        new_filename = f"{date_str}-{file_type}.xls"
        old_path = os.path.join(DOWNLOAD_DIRECTORY, latest_file)
        new_path = os.path.join(DOWNLOAD_DIRECTORY, new_filename)
        if os.path.exists(new_path):
            os.remove(new_path)
        os.rename(old_path, new_path)
        print(f"  - 文件已重命名为: {new_filename}")
    else:
        print(f"  - 警告：未找到新的下载文件 ({file_type})")

def download_files_for_date(driver, date_str):
    try:
        wait = WebDriverWait(driver, 10)

        # --- 下载农副产品表格 ---
        print("  - 正在处理'农副产品'表格...")
        files_before_agri = os.listdir(DOWNLOAD_DIRECTORY)
        # 1. 点击详情 (Button1)
        detail_button_agri = wait.until(EC.element_to_be_clickable((By.ID, "Button1")))
        detail_button_agri.click()
        print("    - 已点击'农副产品'详情按钮")

        # 2. 在新页面导出 (Button4)
        wait.until(EC.staleness_of(detail_button_agri)) # 等待页面跳转
        export_button_agri = wait.until(EC.element_to_be_clickable((By.ID, "Button4")))
        export_button_agri.click()
        print("    - 已点击'农副产品'导出按钮")
        rename_latest_file(date_str, "农副产品", files_before_agri)

        # 3. 返回 (btncxjq)
        back_button_agri = wait.until(EC.element_to_be_clickable((By.ID, "btncxjq")))
        back_button_agri.click()
        print("    - 已点击'返回'按钮")
        # 等待主页面的一个明确元素重新出现，以确认已成功返回
        wait.until(EC.visibility_of_element_located((By.ID, "Button1")))

        # --- 下载蔬菜表格 ---
        print("  - 正在处理'蔬菜'表格...")
        files_before_veg = os.listdir(DOWNLOAD_DIRECTORY)
        # 1. 点击详情 (Button2)
        detail_button_veg = wait.until(EC.element_to_be_clickable((By.ID, "Button2")))
        detail_button_veg.click()
        print("    - 已点击'蔬菜'详情按钮")

        # 2. 在新页面导出 (Button4)
        wait.until(EC.staleness_of(detail_button_veg)) # 等待页面跳转
        export_button_veg = wait.until(EC.element_to_be_clickable((By.ID, "Button4")))
        export_button_veg.click()
        print("    - 已点击'蔬菜'导出按钮")
        rename_latest_file(date_str, "蔬菜", files_before_veg)

        # 3. 返回 (btncxjq)
        back_button_veg = wait.until(EC.element_to_be_clickable((By.ID, "btncxjq")))
        back_button_veg.click()
        print("    - 已点击'返回'按钮")
        # 再次等待主页面元素出现
        wait.until(EC.visibility_of_element_located((By.ID, "Button1")))

        return True
    except Exception as e:
        print(f"  - 错误：下载文件时出错: {e}")
        # 尝试返回主 iframe，以防出错后卡在子页面
        try:
            driver.switch_to.default_content()
            iframe_elements = driver.find_elements(By.TAG_NAME, "iframe")
            if iframe_elements:
                driver.switch_to.frame(iframe_elements[0])
        except Exception as ie:
            print(f"  - 尝试返回主框架时出错: {ie}")
        return False

# 2. 改进iframe切换逻辑
def main():
    start_date_str = input("请输入开始日期 (格式 YYYY-MM-DD): ")
    end_date_str = input("请输入结束日期 (格式 YYYY-MM-DD): ")
    try:
        start_date = date.fromisoformat(start_date_str)
        end_date = date.fromisoformat(end_date_str)
    except ValueError:
        print("日期格式错误，请使用 YYYY-MM-DD 格式。")
        return

    driver = setup_driver()
    try:
        driver.get(TARGET_URL)
        wait = WebDriverWait(driver, 20)

        try:
            iframe_elements = wait.until(EC.presence_of_all_elements_located((By.TAG_NAME, "iframe")))
            if not iframe_elements:
                raise Exception("未找到任何iframe元素")

            found_iframe = False
            for iframe in iframe_elements:
                try:
                    driver.switch_to.frame(iframe)
                    wait.until(EC.presence_of_element_located((By.NAME, "btncxjq")))
                    found_iframe = True
                    print("成功切换到内容框架 (iframe)。")
                    break
                except:
                    driver.switch_to.default_content()
                    continue

            if not found_iframe:
                raise Exception("在所有iframe中均未找到查询按钮'btncxjq'")

        except Exception as e:
            print(f"致命错误：无法找到或切换到 iframe: {e}")
            return

        for single_date in daterange(start_date, end_date):
            current_date_str = single_date.strftime("%Y-%m-%d")
            print(f"--- 正在处理日期: {current_date_str} ---")

            if perform_query(driver, current_date_str):
                if not download_files_for_date(driver, current_date_str):
                    print(f"  - 下载失败，跳过日期: {current_date_str}")
            else:
                print(f"  - 查询失败，跳过此日期。")

    finally:
        print("\n所有任务已完成。")
        driver.quit()

if __name__ == "__main__":
    main()