From eda6d4ad6648ff010e589b314e2b3c5fdc3ded67 Mon Sep 17 00:00:00 2001 From: wangqifan Date: Sun, 10 Aug 2025 11:31:55 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E6=88=90or=E5=9C=B0=E5=9D=80=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E8=BD=AC=E6=8D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 4 ++ browser_manager.py | 91 +++++++++++++++++++++++++++++++++++ config.py | 7 +++ data/o-address-sample.csv | 3 ++ main.py | 68 ++++++++++++++++++++++++++ modules/or_address_handler.py | 65 +++++++++++++++++++++++++ requirements.txt | 3 ++ 7 files changed, 241 insertions(+) create mode 100644 .gitignore create mode 100644 browser_manager.py create mode 100644 config.py create mode 100644 data/o-address-sample.csv create mode 100644 main.py create mode 100644 modules/or_address_handler.py create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a2b7c52 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +__pycache__/*.pyc +chrome_user_data/* +modules/__pycache__/*.pyc +data/o-address.csv diff --git a/browser_manager.py b/browser_manager.py new file mode 100644 index 0000000..c45cc09 --- /dev/null +++ b/browser_manager.py @@ -0,0 +1,91 @@ +# browser_manager.py + +import subprocess +import time +from selenium import webdriver +from selenium.webdriver.chrome.options import Options +from selenium.webdriver.chrome.service import Service +from webdriver_manager.chrome import ChromeDriverManager +from config import CHROME_DEBUGGER_PORT, USER_DATA_DIR + +def get_chrome_driver(use_debugger=False, url=None): + """ + 获取 Chrome WebDriver 实例。 + + :param use_debugger: 是否使用 debugger 模式连接到现有浏览器。 + :param url: 初始加载的 URL。 + :return: WebDriver 实例。 + """ + chrome_options = Options() + + if use_debugger: + # 连接到已经启动的 Chrome 浏览器 + debugger_address = f"127.0.0.1:{CHROME_DEBUGGER_PORT}" + chrome_options.add_experimental_option("debuggerAddress", debugger_address) + print(f"连接到现有的 Chrome 实例: {debugger_address}") + else: + # 启动一个新的 Chrome 浏览器 + # 为了后续可以复用，也指定 user-data-dir + chrome_options.add_argument(f"--user-data-dir={USER_DATA_DIR}") + chrome_options.add_argument("--start-maximized") + chrome_options.add_argument("--disable-infobars") + chrome_options.add_argument("--disable-extensions") + chrome_options.add_argument("--disable-popup-blocking") + print("启动一个新的 Chrome 实例。") + + try: + service = Service(ChromeDriverManager().install()) + driver = webdriver.Chrome(service=service, options=chrome_options) + if url and not use_debugger: + driver.get(url) + return driver + except Exception as e: + print(f"启动浏览器时出错: {e}") + if not use_debugger: + print("尝试以调试模式启动一个新的 Chrome 实例...") + # 如果正常启动失败，可以尝试手动启动一个调试模式的浏览器 + start_chrome_in_debug_mode() + time.sleep(5) # 等待浏览器启动 + return get_chrome_driver(use_debugger=True) + return None + +def start_chrome_in_debug_mode(): + """ + 在后台启动一个带有远程调试端口的 Chrome 浏览器。 + """ + try: + command = ( + f'start chrome.exe --remote-debugging-port={CHROME_DEBUGGER_PORT} ' + f'--user-data-dir="{USER_DATA_DIR}"' + ) + subprocess.Popen(command, shell=True) + print(f"已启动 Chrome 调试模式，端口: {CHROME_DEBUGGER_PORT}") + except FileNotFoundError: + print("错误: 'chrome.exe' 未找到。请确保 Chrome 已安装并已添加到系统路径中。") + except Exception as e: + print(f"启动调试模式的 Chrome 时出错: {e}") + +if __name__ == '__main__': + # 这是一个简单的使用示例 + + # 示例1: 启动一个新的浏览器并访问百度 + # print("--- 示例1: 启动新浏览器 ---") + # driver1 = get_chrome_driver(url="https://www.baidu.com") + # if driver1: + # time.sleep(5) + # print("页面标题:", driver1.title) + # driver1.quit() + + # 示例2: 首先手动启动一个调试模式的浏览器，然后连接它 + print("\n--- 示例2: 连接到调试模式的浏览器 ---") + print(f"请先手动执行以下命令启动Chrome: \nchrome.exe --remote-debugging-port={CHROME_DEBUGGER_PORT} --user-data-dir=\"{USER_DATA_DIR}\"") + input("按 Enter 键继续...") + + driver2 = get_chrome_driver(use_debugger=True) + if driver2: + print("成功连接到浏览器!") + print("当前页面标题:", driver2.title) + # 注意：通过 debugger 连接时，不要用 quit() 关闭浏览器，否则会关闭整个浏览器 + # driver2.close() # close() 只会关闭当前标签页 + else: + print("连接浏览器失败。") diff --git a/config.py b/config.py new file mode 100644 index 0000000..090705d --- /dev/null +++ b/config.py @@ -0,0 +1,7 @@ +# config.py + +# 用于 Chrome debugger 模式的端口 +CHROME_DEBUGGER_PORT = 9222 + +# Chrome 用户数据目录 +USER_DATA_DIR = "chrome_user_data" diff --git a/data/o-address-sample.csv b/data/o-address-sample.csv new file mode 100644 index 0000000..a278ce9 --- /dev/null +++ b/data/o-address-sample.csv @@ -0,0 +1,3 @@ +address,main_address +0x01354beccc5053b36ff0167df51090d4d735a0b2, +0x01354beccc5053b36ff0167df51090d4d735a0b2, diff --git a/main.py b/main.py new file mode 100644 index 0000000..fc1e384 --- /dev/null +++ b/main.py @@ -0,0 +1,68 @@ +# main.py + +import argparse +from browser_manager import get_chrome_driver +from modules.or_address_handler import process_addresses + +# 模块映射 +MODULES = { + "or-address": process_addresses, + # 在这里可以添加更多的模块 + # "another-module": another_handler_function, +} + +def main(): + """ + 主函数，用于解析参数并执行相应的自动化任务。 + """ + # 1. 设置命令行参数解析 + parser = argparse.ArgumentParser(description="Selenium 自动化脚本启动器") + parser.add_argument( + "--module", + type=str, + required=True, + choices=MODULES.keys(), + help="要执行的业务模块名称。" + ) + parser.add_argument( + "--debug", + action="store_true", + help="使用此参数连接到已存在的 Chrome 调试实例。" + ) + args = parser.parse_args() + + # 2. 根据参数选择浏览器启动方式 + use_debugger = args.debug + driver = get_chrome_driver(use_debugger=use_debugger) + + if not driver: + print("无法获取 WebDriver，程序退出。") + return + + try: + # 3. 根据参数调用相应的业务模块 + module_name = args.module + handler_function = MODULES.get(module_name) + + if handler_function: + print(f"--- 开始执行模块: {module_name} ---") + handler_function(driver) + print(f"--- 模块: {module_name} 执行完毕 ---") + else: + # 这段代码理论上不会执行，因为 argparse 的 choices 已经做了限制 + print(f"错误: 未知的模块 '{module_name}'。") + + except Exception as e: + print(f"执行自动化任务时发生严重错误: {e}") + finally: + # 4. 任务结束后的清理工作 + print("所有任务执行完毕。") + # 如果不是连接的调试模式，则关闭浏览器 + if not use_debugger: + print("关闭浏览器...") + driver.quit() + else: + print("任务完成，保持调试浏览器开启。") + +if __name__ == "__main__": + main() diff --git a/modules/or_address_handler.py b/modules/or_address_handler.py new file mode 100644 index 0000000..eff3b5a --- /dev/null +++ b/modules/or_address_handler.py @@ -0,0 +1,65 @@ +# modules/or_address_handler.py + +import pandas as pd +import os +import time +from selenium.webdriver.common.by import By +from selenium.webdriver.support.ui import WebDriverWait +from selenium.webdriver.support import expected_conditions as EC + +def process_addresses(driver): + """ + 处理 o-address.csv 文件，访问 Etherscan 并更新地址信息。 + :param driver: Selenium WebDriver 实例。 + """ + # 定义文件路径 + csv_file_path = os.path.join('data', 'o-address.csv') + + # 1. 读取 CSV 文件 + if not os.path.exists(csv_file_path): + print(f"错误: 文件未找到 at {csv_file_path}") + return + + df = pd.read_csv(csv_file_path) + print(f"成功读取 {len(df)} 条地址。") + + # 2. 循环处理每个地址 + for index, row in df.iterrows(): + address = row['address'] + + # 检查 main_address 是否已经有值，如果有则跳过 + # 使用 pd.isna() 检查 NaN，并检查是否为空字符串 + if 'main_address' in df.columns and pd.notna(row['main_address']) and row['main_address']: + print(f"地址 {address} 已有 main_address，跳过。") + continue + + # 构建 URL 并访问 + url = f"https://etherscan.io/address/{address}" + print(f"正在访问: {url}") + driver.get(url) + + try: + # 等待页面加载并找到 #mainaddress 元素 + wait = WebDriverWait(driver, 20) # 最多等待20秒 + main_address_element = wait.until( + EC.presence_of_element_located((By.ID, "mainaddress")) + ) + + # 获取并保存值 + main_address_value = main_address_element.text + df.loc[index, 'main_address'] = main_address_value + print(f" -> 成功获取 main_address: {main_address_value}") + + except Exception as e: + print(f" -> 无法为地址 {address} 找到 main_address。") + df.loc[index, 'main_address'] = "Not Found" # 标记为未找到 + + # 每次处理后都保存一次，防止中途中断 + df.to_csv(csv_file_path, index=False) + + # 短暂延时，避免请求过于频繁 + time.sleep(2) + + # 3. 所有处理完成后，再次保存最终结果 + df.to_csv(csv_file_path, index=False) + print(f"\n处理完成！结果已保存到 {csv_file_path}") diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..5164988 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +selenium==4.34.2 +webdriver-manager==4.0.2 +pandas==2.3.1