pip install selenium oss2 kafka-python-ng

[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

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)

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()

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"開(kāi)始處理項(xiàng)【{item_id}】對(duì)應(yīng)參數(shù)【{param_value}】")
    
    # 構(gòu)建請(qǐng)求鏈接
    url = f"{config.get('engine', 'site_base_path')}/view?param={param_value}&id={item_id}"
    driver.get(url)
    
    try:
        # 簡(jiǎn)單等待頁(yè)面加載
        time.sleep(3)  # 根據(jù)需要調(diào)整或替換為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（省略具體實(shí)現(xiàn)，根據(jù)實(shí)際情況添加）
        upload_to_oss(fname)
        
        # 發(fā)送完成通知
        notify_completion(item_id, param_value, fname)
        
        logger.info(f"完成處理項(xiàng)【{item_id}】對(duì)應(yīng)參數(shù)【{param_value}】")
    except Exception as e:
        logger.error(f"處理項(xiàng)【{item_id}】對(duì)應(yīng)參數(shù)【{param_value}】時(shí)發(fā)生異常: {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):
    """發(fā)送完成通知"""
    producer.send(config.get('kafka', 'notify_topic'), {
        'itemId': item_id,
        'paramValue': param_value,
        'imagePath': image_path
    })

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"消費(fèi)消息發(fā)生異常: {ex}")