TenderCrawler/SourceCode/TenderCrawler/app/jobs/data_collector.py

import importlib

from selenium import webdriver

import adapters
import utils
from adapters.data_collection_adapter_interface import IDataCollectionAdapter
from stores.data_store_interface import IDataStore
from stores.default_data_store import DefaultDataStore


class DataCollector:

    _adapter = None
    _driver = None
    _store = None

    def __init__(
        self, adapter_type: str, url: str, un: str, up: str, store: IDataStore = None
    ):
        self._adapter = self._gen_adapter(adapter_type, url)
        self._driver = self.adapter.driver
        # if type == "chinabidding":
        #     return
        self.adapter.login(un, up)
        if store:
            self._store = store
        else:
            self._store = DefaultDataStore()

    @property
    def driver(self) -> webdriver:
        return self._driver

    @property
    def store(self) -> IDataStore:
        return self._store

    @property
    def adapter(self) -> IDataCollectionAdapter:
        return self._adapter

    def set_store(self, store: IDataStore) -> None:
        self._store = store

    def collect(self, keyword: str):
        adapters.collect(self.adapter, keyword, self.store)

    def close(self):
        utils.get_logger().info(f"关闭浏览器驱动，URL: {self.adapter.url}")
        adapters.teardown(self.adapter)

    @staticmethod
    def _gen_adapter(adapter_type: str, url: str):
        adapter_model_name = utils.get_config_value(
            f"adapter.{adapter_type}.model_name"
        )
        adapter_class_name = utils.get_config_value(
            f"adapter.{adapter_type}.class_name"
        )
        if adapter_class_name:
            try:
                utils.get_logger().info(
                    f"生成适配器 TYPE：{adapter_type}，适配器: {adapter_class_name}，URL：{url}"
                )
                # 使用 importlib 动态导入模块
                adapter_module = importlib.import_module(
                    f"adapters.{adapter_model_name}"
                )
                adapter_class = getattr(adapter_module, adapter_class_name)
                adapter = adapter_class(url)
            except ImportError as e:
                raise ImportError(f"无法导入适配器模块 {adapter_model_name}") from e
            except AttributeError as e:
                raise AttributeError(
                    f"适配器模块 {adapter_model_name} 中找不到类 {adapter_class_name}"
                ) from e
        else:
            raise Exception("不支持的适配器类型")
        return adapter