使用Python和Selenium构建一个自动化图像引擎_Python

本篇指南将教你如何使用python和selenium库来构建一个自动化图像引擎，该引擎能够根据指定参数自动截取网页快照，并将生成的图片存储到云端。此工具还可以通过消息队列接收任务指令，非常适合需要批量处理网页截图的应用场景。

1. 准备环境

确保你已经安装了python和必要的库：

pip install selenium oss2 kafka-python-ng

2. 创建配置文件

创建一个简单的config.ini文件来存储你的oss和kafka设置：

[oss]
access_key_id = your_oss_access_key_id
access_key_secret = your_oss_access_key_secret
bucket_name = your_bucket_name
endpoint = http://oss-cn-hangzhou.aliyuncs.com

[kafka]
bootstrap_servers = localhost:9092
topic = your_topic_name
notify_topic = your_notify_topic
consumer_group = your_consumer_group

[engine]
driver_path = path/to/chromedriver
image_path = path/to/screenshots
param_path = path/to/params
site_base_path = https://example.com

3. 设置日志记录

为程序添加基本的日志记录功能，以便于调试：

import logging
from logging.handlers import timedrotatingfilehandler
import os

logger = logging.getlogger('image_engine')
logger.setlevel(logging.debug)

log_file = 'logs/image_engine.log'
os.makedirs('logs', exist_ok=true)
handler = timedrotatingfilehandler(log_file, when='midnight', backupcount=7, encoding='utf-8')
formatter = logging.formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setformatter(formatter)
logger.addhandler(handler)

4. 初始化selenium webdriver

初始化chrome webdriver，并设置窗口最大化：

from selenium import webdriver
from selenium.webdriver.chrome.service import service

# 读取配置文件
import configparser
config = configparser.configparser()
config.read('config.ini')

service = service(config.get('engine', 'driver_path'))
driver = webdriver.chrome(service=service)
driver.maximize_window()

5. 图像处理逻辑

编写一个函数来处理每个kafka消息，打开指定网页，等待页面加载完成，然后保存截图：

from kafka import kafkaconsumer, kafkaproducer
import json
import time
from datetime import datetime
import oss2

def process_task(msg):
    task_params = json.loads(msg.value)
    item_id = task_params['itemid']
    param_value = task_params['paramvalue']
    
    logger.info(f"开始处理项【{item_id}】对应参数【{param_value}】")
    
    # 构建请求链接
    url = f"{config.get('engine', 'site_base_path')}/view?param={param_value}&id={item_id}"
    driver.get(url)
    
    try:
        # 简单等待页面加载
        time.sleep(3)  # 根据需要调整或替换为webdriverwait
        
        # 生成截图文件名
        today = datetime.now().strftime('%y-%m-%d')
        screenshot_dir = os.path.join(config.get('engine', 'image_path'), 'images', today)
        os.makedirs(screenshot_dir, exist_ok=true)
        fname = os.path.join(screenshot_dir, f"{item_id}_{param_value}.png")
        
        driver.save_screenshot(fname)
        logger.info(f"保存截图到 {fname}")
        
        # 上传至oss（省略具体实现，根据实际情况添加）
        upload_to_oss(fname)
        
        # 发送完成通知
        notify_completion(item_id, param_value, fname)
        
        logger.info(f"完成处理项【{item_id}】对应参数【{param_value}】")
    except exception as e:
        logger.error(f"处理项【{item_id}】对应参数【{param_value}】时发生异常: {e}")

def upload_to_oss(file_path):
    """上传文件到阿里云oss"""
    auth = oss2.auth(config.get('oss', 'access_key_id'), config.get('oss', 'access_key_secret'))
    bucket = oss2.bucket(auth, config.get('oss', 'endpoint'), config.get('oss', 'bucket_name'))
    remote_path = os.path.relpath(file_path, config.get('engine', 'image_path'))
    bucket.put_object_from_file(remote_path, file_path)

def notify_completion(item_id, param_value, image_path):
    """发送完成通知"""
    producer.send(config.get('kafka', 'notify_topic'), {
        'itemid': item_id,
        'paramvalue': param_value,
        'imagepath': image_path
    })

6. 启动kafka消费者

启动kafka消费者，监听消息并调用处理函数：

if __name__ == "__main__":
    consumer = kafkaconsumer(
        config.get('kafka', 'topic'),
        bootstrap_servers=config.get('kafka', 'bootstrap_servers').split(','),
        group_id=config.get('kafka', 'consumer_group'),
        auto_offset_reset='latest',
        enable_auto_commit=true,
        value_deserializer=lambda m: m.decode('utf-8')
    )

    for msg in consumer:
        try:
            process_task(msg)
        except exception as ex:
            logger.error(f"消费消息发生异常: {ex}")