利用selenium库配合火狐驱动实现自动化,额外加pyautogui库实现并行操作;pandas库保存网页数据,实现方式如下:

from bs4 import BeautifulSoup
from selenium import webdriver
#from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import  By
from selenium.webdriver.support.ui import  WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time
import re
from selenium.webdriver.firefox.options import Options
from selenium.webdriver.firefox.firefox_binary import FirefoxBinary
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.firefox.service import Service
import pandas as pd
from selenium.webdriver import FirefoxProfile
from openpyxl.styles import Border,Side,Font,Alignment,PatternFill
import os
import pyautogui
import pyperclip
import pygetwindow as gw
import threading
import csv

#复制详情网址处理函数
def get_xq_url():
    # 复制详情网址
    # 检查是否有火狐浏览器打开
    firefox_wins = gw.getWindowsWithTitle("MOzilla Firefox")
    if not firefox_wins:
        print("未检测到打开的火狐浏览器!")
    else:
        ff_win = firefox_wins[0]
        ff_win.activate()  # 切换火狐浏览器
        time.sleep(1)

        pyautogui.click(x=324, y=47)
        time.sleep(2)
        pyautogui.hotkey("ctrl", "a")
        time.sleep(0.1)
        pyautogui.hotkey("ctrl", "c")
        xq_url = pyperclip.paste()
        print(f"详情网址为:{xq_url}")
        time.sleep(2)
        #获取完之后点击关闭按钮
        pyautogui.click(x=1882,y=69)
        time.sleep(0.5)
        # 切回原始界面
        pyautogui.keyDown("alt")
        pyautogui.press("tab")
        pyautogui.keyUp("alt")
    return xq_url

#解决多线程无法有返回值的问题
class ThreadWithResult(threading.Thread):
    def __init__(self,target,args=()):
        super().__init__(target=target,args=args)
        self.result = None

    #执行函数,捕获返回值url存入result
    def run(self):
        self.result = self._target(*self._args)

    def get_result(self):
        return self.result

#写入文件模板
def write_file(v_df,out_put_flie,md,sheent_name):
    with pd.ExcelWriter(out_put_flie, engine='openpyxl', mode=md) as writer:
        v_df.to_excel(writer, sheet_name=sheent_name, index=False)  # 输出总体文件
        ws = writer.sheets[sheent_name] #

        # 加边框
        thin = Side(style='thin', color='000000')
        border = Border(top=thin, right=thin, bottom=thin, left=thin)
        # 给所有表格加边框
        for row in ws.iter_rows(min_row=1, max_row=ws.max_row, min_col=1, max_col=ws.max_column):
            for cell in row:
                cell.border = border
        # 自适应列宽处理
        for col in ws.columns:
            col_center = col[0].column_letter
            max_len = max(len(str(cell.value)) for cell in col)
            ws.column_dimensions[col_center].width = max_len + 6

#切换最底层的iframe
def get_all_deep_iframe(driver,flag):
    while True:
        try:
            wait = WebDriverWait(driver, 15)
            inner_frame = wait.until(
                EC.presence_of_element_located((By.XPATH, f'//iframe[starts-with(@name,"{flag}")]')))
            driver.switch_to.frame(inner_frame)
        except:
            break

def get_all_deep_south_iframe(driver,flag):
    switch_count = 0 #统计成功切入的iframe数
    while True:
        try:
            wait = WebDriverWait(driver, 15)
            inner_frame = wait.until(
                EC.presence_of_element_located((By.XPATH, f'//iframe[starts-with(@name,"{flag}")]')))
            driver.switch_to.frame(inner_frame)
            switch_count +=1
        except:
            break
    return switch_count

#处理详情页左边树形菜单数据
def left_tree_get_data(driver,txt):
    page_html_loop = driver.page_source
    soup_old_loop = BeautifulSoup(page_html_loop, "html.parser")

    data_column = []
    data_column1 = []
    data_column2 = []
    # 提取表头
    header_column = [div.get_text(strip=True) for div in
                     soup_old_loop.select(
                         '#Header_myiframe0_0,#TH_Right_myiframe0 .list_gridCell_standard')]  # 获取表头

    # 提取表数据
    tr_list1 = soup_old_loop.select('tr[id^="TR_Left_myiframe0_"]')  # 获取表数据
    tr_list2 = soup_old_loop.select('tr[id^="TR_Right_myiframe0_"]')  # 获取表数据
    for tr in tr_list1:
        lock_cells = [div.get_text(strip=True) for div in tr.select('div.list_gridCell_lock')]
        data_column1.append(lock_cells)
    for tr in tr_list2:
        std_cells = [div.get_text(strip=True) for div in tr.select('div.list_gridCell_standard')]
        data_column2.append(std_cells)
    for lock_rr, std_arr in zip(data_column1, data_column2):
        new_row = lock_rr + std_arr
        data_column.append(new_row)
    df3 = pd.DataFrame(data_column, columns=header_column)
    new_df3_pivot = df3.T
    new_df3_pivot.reset_index(inplace=True)
    new_df3_pivot.rename(columns={"index": "旧字段", 0: "旧字段值", 1: "旧字段值2"}, inplace=True)
    new_df3_pivot.insert(0, "新增字段1", "字段1值")
    new_df3_pivot.insert(1, "新增字段2", f"字段2值-{txt}")
    return new_df3_pivot

#提取详情页面数据函数
def get_xq_right_data(tmp_url2,driver,wait,idx):
    if idx == 0:
        md = "w"
    else:
        md = "a"
    # 回切当前iframe
    driver.switch_to.default_content()
    driver.switch_to.frame("ObjectList")
    inner_frame2 = wait.until(EC.presence_of_element_located((By.XPATH, f'//iframe[starts-with(@name,"center")]')))
    driver.switch_to.frame(inner_frame2)
    time.sleep(1)
    # 重新获取idx
    current_spans = wait.until(EC.presence_of_all_elements_located((By.XPATH, '//span[contains(@class,"tree-title")]')))
    span_elem = current_spans[idx]
    txt = span_elem.text.strip()
    if idx == 0:          
        south_num_flag = 0 #soutn页面标识
        span_elem.click()
        # 切换最底层iframe
        get_all_deep_iframe(driver,"center")
        time.sleep(6)
        page_html = driver.page_source
        soup_old = BeautifulSoup(page_html, "html.parser")
        tr_list = soup_old.select("tr")
        res_list_xq_all =[]
        data_dict ={}
        for tr in tr_list:
            left_td_all = tr.select('td[class^="info_td_left"]')
            right_td_all = tr.select('td[class^="info_td_right"]')
            for left_td, right_td in zip(left_td_all, right_td_all):
                left_style = left_td.get("style", "")
                right_style = right_td.get("style", "")
                # 不要隐藏的字段
                if "display: none" in left_style or "display: none" in right_style:
                    continue
                name = left_td.find("span").get_text(strip=True)
                # 下拉框
                sel = right_td.find("select")
                # 输入框
                inp = right_td.find("input")
                if sel:
                    code = sel.get("initvalue", "")
                    val = code
                    opt_list = sel.find_all("option")
                    for opt in opt_list:
                        opt_val = opt.get("value", "")
                        if opt_val == code:
                            val = opt.get_text(strip=True)
                            break
                elif inp:
                    val = inp.get("value", "")
                else:
                    val = ""
                data_dict[name] = val
        column_1 = "新增字段值3"
        column_2 = "新增字段值3-详情-基本信息"
        for name,val in data_dict.items():
            res_list_xq_all.append([column_1,column_2,name,val])
            #res_list_xq_all.append(res_list_xq)
        #print(res_list_xq_all)
        new_df2 = pd.DataFrame(res_list_xq_all,columns=["新增字段1","新增字段2","旧字段","旧字段值"])
        new_df2["旧字段值2"]=""
        new_df2.to_csv(tmp_url2, mode=md, header=True,index=False,encoding='gbk')
        time.sleep(3)
    else:
        # 重新获取idx
        current_spans = wait.until(
            EC.presence_of_all_elements_located((By.XPATH, '//span[contains(@class,"tree-title")]')))
        span_elem = current_spans[idx]
        span_elem.click()
        # 切换最底层center iframe
        get_all_deep_iframe(driver,"center")
        time.sleep(6)
        new_df3_pivot=left_tree_get_data(driver, txt)
        new_df3_pivot.to_csv(tmp_url2, mode=md, header=False,index=False,encoding='gbk')
        time.sleep(1)
        #切换上一层iframe
        driver.switch_to.parent_frame()
        south_num_flag=get_all_deep_south_iframe(driver,"south")
        if south_num_flag:
            new_df3_pivot_1 = left_tree_get_data(driver, txt)
            new_df3_pivot_1.to_csv(tmp_url2, mode=md, header=False,index=False,encoding='gbk')

    return south_num_flag

def acess_old_website(tmp_url1,tmp_url2,filer_code_yp,idx):
    login_url = "登录网址" #登录网址
    USERNAME = "USERNAME"
    PASSWORD = "PASSWORD"

    fire_options = Options()
    fire_options.set_preference("dom.webdriver.enabled",False)
    fire_options.set_preference("useAutomationExtension",False)
    #忽略自签证书错误
    fire_options.accept_insecure_certs =True
    fire_options.set_preference("layout.css.devPixelsPerPx","1.0")
    fire_options.set_preference("dom.disable_open_during_load",True)
    fire_path=FirefoxBinary(r'C:\firefox.exe') #指定火狐程序
    gecko_path =r'E:\geckodriver.exe' #指定火狐驱动
    config_path =r'C:\rcml7n57.default' #指定配置路径,通过火狐浏览器网址栏输入about:profiles寻找
    profiles =FirefoxProfile(config_path)
    driver =webdriver.Firefox(options=fire_options,firefox_binary=fire_path,executable_path=gecko_path,firefox_profile=profiles)
    driver.execute_script("""
    Object.defineProperty(navigator,'webdriver',{get: () => undefined })
    document.querySelectorAll('*').forEach(item=>{item.style.pointerEvents='auto'})
    """)
    wait = WebDriverWait(driver, 15)
    try:
        driver.get(login_url)
        time.sleep(1)
        driver.execute_script("window.confirm =function(){return true;}; window.alert =function(){return true;}")
        driver.execute_script("window.alert =function(){return true;}")
        user_input = wait.until(EC.presence_of_element_located(("id","UserID"))) #driver.find_element("id", "UserID")
        user_input.clear()
        user_input.send_keys(USERNAME)
        time.sleep(1)
        pwd_input = wait.until(EC.presence_of_element_located(("xpath",'//input[@class="Dpwd" and @type="text"]')))  #driver.find_element("id", "Dpwd")
        pwd_input.click()
        time.sleep(0.5)
        pwd_hide =driver.find_element("id","Password")
        driver.execute_script(f"arguments[0].value='{PASSWORD}'",pwd_hide)
        time.sleep(0.3)
        # 登录
        login_btn = driver.find_element("xpath", "//a[@class='login_btn']")
        login_btn.click()

        #默认确定按钮
        alert = WebDriverWait(driver, 10).until(EC.alert_is_present())
        alert.accept()
        time.sleep(3)

        driver.execute_script("""
                                var e = document.evaluate("/html/body/div[2]/div[1]/div[1]/div[2]/div[4]/div/ul/li[1]/a",
                                document,null,XPathResult.FIRST_ORDERED_NODE_TYPE,null).singleNodeValue;
                                  if(e){ e.click();e.focus();console.log("执行成功");} else{
                                    console.log("执行失败,未找到元素");}
                                """)  # 一级菜单
        time.sleep(1)
        driver.execute_script("""
                                       var e = document.evaluate("/html/body/table/tbody/tr[1]/td/table/tbody/tr/td[2]/ul/li[1]/a/span[2]",
                                       document,null,XPathResult.FIRST_ORDERED_NODE_TYPE,null).singleNodeValue;
                                         if(e){ e.click();e.focus();console.log("执行成功");} else{
                                           console.log("执行失败,未找到元素");}
                                       """)  # 二级菜单
        time.sleep(1)

        driver.execute_script("""
                                               var e = document.evaluate("/html/body/table/tbody/tr[1]/td/table/tbody/tr/td[2]/ul/li[1]/ul/li[1]/a/span[2]",
                                               document,null,XPathResult.FIRST_ORDERED_NODE_TYPE,null).singleNodeValue;
                                                 if(e){ e.click();e.focus();console.log("执行成功");} else{
                                                   console.log("执行失败,未找到元素");}
                                               """)  # 三级菜单

        time.sleep(6)

        #先切入到 MainCenter
        driver.switch_to.frame("MainCenter")
        time.sleep(1)
        #内层
        inner_frame =driver.find_element("xpath",'//iframe[contains(@id,"tab_T01_iframe")]')
        driver.switch_to.frame(inner_frame)
        time.sleep(2)

        driver.execute_script("""
                                                       var e = document.evaluate("//html/body/table/tbody/tr[3]/td/form/hidden/div[1]/div[1]/fieldset/legend",
                                                       document,null,XPathResult.FIRST_ORDERED_NODE_TYPE,null).singleNodeValue;
                                                         if(e){ e.click();e.focus();console.log("执行成功");} else{
                                                           console.log("执行失败,未找到元素");}
                                                       """)  # 查询条件
        #点击查询条件
        # expand_search = wait.until(EC.element_to_be_clickable((By.XPATH,'//div[contains(.,"查询条件")]')))
        # expand_search.click()
        time.sleep(2)


        #定位字段输入框,并赋值
        input_box =driver.find_element(By.NAME,"输入框网页端名称") 
        input_box.clear()
        input_box.send_keys(filer_code_yp)
        input_box.send_keys(Keys.ENTER)

        time.sleep(8)
        #获取第一个页面字段值
        data_xxgl_js = driver.execute_script('''
        let result=[];
        for(let idx=0;idx<DZ.length;idx++){
             let rowArr = DZ[idx][2];
             if(Array.isArray(rowArr)){
             for(let j=0;j<rowArr.length;j++){
                if (rowArr[j][0] === arguments[0]){
                result.push(rowArr[j]);
                }
              }
            }
        }
          return result;
        ''',filer_code_yp)
        xxgl_cols =["匹配字段名称"]
        df = pd.DataFrame(data_xxgl_js,columns=xxgl_cols)
        #码值转换
        zt_dict ={
            '01':'临时',
            '02':'生效',
            '03':'落实担保',
            '04':'解除担保',
            '05':'终结',
            '06':'处置',
            '07':'删除'
        }
        df["状态"]=df["状态"].map(zt_dict).fillna("")

        sfzlw_dict = {
            '01':'是',
            '02':'否'
        }
        df["码值字段相关"] = df["码值字段相关"].map(sfzlw_dict).fillna("")
        new_df_pivot =new_df.T
        new_df_pivot.reset_index(inplace=True)
        new_df_pivot.rename(columns={"index":"旧字段",0:"旧字段值"},inplace=True)
        new_df_pivot.insert(0,"新增字段1","字段1值")
        new_df_pivot.insert(1,"新增字段2","字段2值")
        new_df_pivot["旧字段值2"] = ""
        new_df_pivot.to_excel(tmp_url1,index=False) #


        print("请复制详情菜单完整地址,并关闭按钮,别忘了关闭按钮,请按回车键下一步")
        t = ThreadWithResult(target=get_xq_url,args=())
        #点击详情
        menu3 = wait.until(EC.element_to_be_clickable((By.XPATH, "//*[contains(text(),'详情')]")))
        time.sleep(1.5)
        ActionChains(driver).move_to_element(menu3).pause(0.3).perform()
        print("执行并行线程前")
        time.sleep(1)
        t.start()
        ActionChains(driver).click(menu3).perform()
        print("执行并行线程后")
        #等待子线程执行完毕,返回网址
        t.join()
        xq_url = t.get_result()

        #如果鼠标获取不到则手工输入
        if not xq_url or "https:" not in xq_url:
            # 手动切换新标签
            xq_url = input("需要手工操作!请输入详情网址:")


        #进入详情页面
        driver.get(xq_url)
        time.sleep(2)
        # 回切当前iframe
        driver.switch_to.default_content()
        driver.switch_to.frame("ObjectList")
        inner_frame2 = wait.until(EC.presence_of_element_located((By.XPATH, f'//iframe[starts-with(@name,"center")]')))
        driver.switch_to.frame(inner_frame2)

        left_tree = wait.until(EC.presence_of_all_elements_located((By.XPATH, '//span[contains(@class,"tree-title")]')))

        tree_node_dict ={num: ele.text.strip() for num,ele in enumerate(left_tree)}

        #执行提取数据函数
        layer_south = get_xq_right_data(tmp_url2,driver, wait,idx)
        time.sleep(5)
        if layer_south:
            print("存在副页面")

    finally:
        driver.quit()
    return tree_node_dict


def tran_data_seed_one(is_merge_flag,tmp_url1,tmp_url2,filer_code_yp):
    if is_merge_flag == '0':
        idx = 0
        #dict_new = None
        while True:
            try:
                dict_new = acess_old_website(tmp_url1, tmp_url2, filer_code_yp, idx)
                dict_len = len(dict_new)
                print(f"{idx + 1}:{dict_new[idx]} 项执行成功;需要执行总数为:{dict_len},还剩余{dict_len - idx - 1}项")
                if idx + 1 == dict_len:
                    print("执行完毕,退出循环")
                    break
                idx += 1  # 进行下一个
                # break
            except Exception as e:
                print(f"{idx + 1}:{dict_new[idx]}项报错,错误信息:{str(e)},3秒后请重试")
                time.sleep(3)
    elif is_merge_flag == '1':
        df = pd.read_excel(tmp_url1)
        #df2 = pd.read_table(tmp_url2,header=None,sep=",",engine="python")
        with open(tmp_url2,"r",encoding="gbk") as f:
            data = list(csv.reader(f))
        max_len = max(map(len,data))
        #print(data)
        header_row=data[0]
        header =header_row+[f"字段值{i+2}" for i in range(1,(max_len-len(header_row))+1 )]
        data_row = [row + [""] * (max_len - len(row)) for idx, row in enumerate(data[1:])]
        #print(header)
        df2 =pd.DataFrame(data_row,columns=header)
        # df2.to_excel(r"E:\临时文件\新建文件夹\linshi2.xlsx")
        # df.to_excel(r"E:\临时文件\新建文件夹\linshi1.xlsx")

        total_df = pd.concat([df, df2], ignore_index=True)
        # 写入文件
        write_file(total_df, out_put_flie, "w", "Sheet1")




seed_list = [
['分类','分类值']
]
tmp_url1=r'\result.xlsx' #第一个临时文件,存外层数据
tmp_url2=r'\result2.csv' #第二个临时文件,存详情文件

for yp_zl,yp_bh in seed_list:
    out_put_flie = r"\模块-" + yp_zl + ".xlsx"
    try:
        tran_data_seed_one('0',#处理方式 #0 只执行result文件;1:只执行合并文件
                           tmp_url1, #第一个临时文件,存外层数据
                           tmp_url2, #第二个临时文件,存详情文件
                           yp_bh #押品编号
                           )
        print(f'生成result {yp_zl} 临时文件成功!')
        tran_data_seed_one('1',  # 处理方式 #0 只执行result文件;1:只执行合并文件
                           tmp_url1,  # 第一个临时文件,存外层数据
                           tmp_url2,  # 第二个临时文件,存详情文件
                           yp_bh  # 押品编号
                           )
        print(f'生成 {yp_zl} 文件成功!')
    except Exception as e:
        print(f"执行失败,失败信息{e}")

Logo

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地,聚焦技能开发、插件实践与部署教程,为开发者提供可直接落地的方案、工具与交流平台,助力高效构建与落地 AI 应用

更多推荐