vegatable-price-trends
/
config.py

from dataclasses import dataclass
import os
import utils
import pandas as pd
import numpy as np
import datetime
'''
本模块当作全局配置文件使用, 所有的配置都将硬编码到配置类的属性

1. 配置类之间可以互相组合, 例如 Config 类的 CrawlerConfig 属性是一个 _CrawlerConfig() 对象, 而 _CrawlerConfig 类的 VegetableCrawlerConfig 属性是 _VegetableCrawlerConfig() 对象
2. 请遵守如下命名格式:
        - 配置类的命名:
            class _MyConfig:
        - 在配置类中将另一个类作为属性:
            class _MyConfig2:
                pass
            class _MyConfig:
                def __init__(self, Config):
                    self.MyConfig2: _MyConfig2 = _MyConfig2(Config)  # 当内含其他配置类时,应当实现在__init__, 并且接收Config并将它传递给其他的配置类
        - 在配置类中将非配置类作为属性:
            class _MyConfig3:
                CONST_VALUE_1 = 123
                MY_CONST_VAL = "456"
                VALUE = [7,8,9]

                def __init__(self, Config):
                    self.PROJECT_PATH = Config.PROJECT_PATH  # 当属性需要调用其他配置类的属性时,需要实现在__init__并通过接收的Config来访问

3. 实现了上述两条 那么调用配置可以按照如下方式:

URL = Config.CrawlerConfig.VegetableCrawlerConfig.URL

'''

@ utils.singleton
class _VegetableTableCleanerConfig():
    '''
    用于结构化蔬菜数据的清洗器的配置
    '''
    def __init__(self, config):
        pass


@ utils.singleton
class _CleanerConfig():
    '''
    用于所有数据清洗器的配置
    '''
    def __init__(self, config):
        self.VegetableTableCleanerConfig: _VegetableTableCleanerConfig = _VegetableTableCleanerConfig(config)

        self.TABLE_COLUMNS = config.DataPackageConfig.TABLE_COLUMNS


@ utils.singleton
class _VegetableCrawlerConfig():
    '''
    用于蔬菜价格爬虫的配置
    '''
    URL = "https://nwggfw.nyncw.sh.gov.cn:8084/ggfw/api/otherInterface/marketPrice/getData"
    TRY_AGAIN_SECONDS = 100  # 当请求失败时,再次请求的时间间隔

    def __init__(self, config):
        pass

@ utils.singleton
class _DailyWeatherCrawlerConfig():
    '''
    用于天气价格爬虫的配置
    '''
    URL = "https://tianqi.2345.com/Pc/GetHistory"

    headers = {
    "User-Agent" : """Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (HTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36"""
    }

    current_data = datetime.datetime.now()
    current_year = current_data.year
    current_month = current_data.month
    current_day = current_data.day

    @property
    def cityids(self):
        return ["58362","54938"]

    @property
    def citys(self):
        return ["上海","山东临沂"]

    def __init__(self, config):
        pass


class _ExtremeWeatherCrawlerConfig():
    '''
    用于天气价格爬虫的配置
    '''
    URL = "xxx"
    def __init__(self, config):
        pass


@ utils.singleton
class _EconomyCrawlerConfig():
    '''
    用于经济价格爬虫的配置
    '''
    URL = "xxxxx"
    def __init__(self, config):
        pass

@ utils.singleton
class _OilCrawlerConfig():
    '''
    用于汽油价格爬虫的配置
    '''
    URL = "xxxxx"
    def __init__(self, config):
        pass

@ utils.singleton
class _CrawlerConfig():
    '''
    用于所有爬虫的一些配置
    '''
    def __init__(self, config):
        self.VegetableCrawlerConfig: _VegetableCrawlerConfig = _VegetableCrawlerConfig(config)
        self.DailyWeatherCrawlerConfig: _DailyWeatherCrawlerConfig = _DailyWeatherCrawlerConfig(config)
        self.ExtremeWeatherCrawlerConfig: _ExtremeWeatherCrawlerConfig = _ExtremeWeatherCrawlerConfig(config)
        self.EconomyCrawlerConfig: _EconomyCrawlerConfig = _EconomyCrawlerConfig(config)
        self.OilCrawlerConfig: _OilCrawlerConfig = _OilCrawlerConfig(config)

        self.RESPONSE_COLUMNS = config.DataPackageConfig.RESPONSE_COLUMNS
        self.NOW_TIME_FUNC = utils.now_time_for_crawler


@ utils.singleton
class _DataManagerConfig():
    '''
    用于所有数据管理器的配置
    '''
    STORE_IN_FILE_SYSTEM = True
    STORE_IN_DATABASE = False

    def __init__(self, config):
        self.FileSystemManagerConfig: _FileSystemManagerConfig = _FileSystemManagerConfig(config)


@ utils.singleton
class _VegetablePriceResponseFileSystemManagerConfig():
    '''
    蔬菜价格
    '''
    def __init__(self, config):
        self.DATA_ROOT_DIR_PATH = os.path.join(config.PROJECT_PATH, "data")
        self.RESPONSE_MAIN_KEY = config.DataPackageConfig.RESPONSE_MAIN_KEY
        self.RESPONSE_FILE_PATH = os.path.join(self.DATA_ROOT_DIR_PATH, "vegetable_price_response.csv")

@ utils.singleton
class _DailyWeatherResponseFileSystemManagerConfig():
    '''
    日常天气
    '''
    def __init__(self, config):
        self.DATA_ROOT_DIR_PATH = os.path.join(config.PROJECT_PATH, "data")
        self.RESPONSE_FILE_PATH = os.path.join(self.DATA_ROOT_DIR_PATH, "daily_weather_response.csv")


@ utils.singleton
class _ExtremeWeatherResponseFileSystemManagerConfig():
    '''
    极端天气
    '''
    def __init__(self, config):
        self.DATA_ROOT_DIR_PATH = os.path.join(config.PROJECT_PATH, "data")
        self.RESPONSE_FILE_PATH = os.path.join(self.DATA_ROOT_DIR_PATH, "extreme_weather_response.csv")

@ utils.singleton
class _ResponseFileSystemManagerConfig():
    '''
    用于响应文件系统数据管理器的配置
    '''
    def __init__(self, config):
        self.VegetablePriceResponseFileSystemManagerConfig: _VegetablePriceResponseFileSystemManagerConfig = _VegetablePriceResponseFileSystemManagerConfig(config)
        self.DailyWeatherResponseFileSystemManagerConfig: _DailyWeatherResponseFileSystemManagerConfig = _DailyWeatherResponseFileSystemManagerConfig(config)
        self.ExtremeWeatherResponseFileSystemManagerConfig: _ExtremeWeatherResponseFileSystemManagerConfig = _ExtremeWeatherResponseFileSystemManagerConfig(config)
        self.RESPONSE_COLUMNS = config.DataPackageConfig.RESPONSE_COLUMNS
        self.RESPONSE_MAIN_KEY = config.DataPackageConfig.RESPONSE_MAIN_KEY


@ utils.singleton
class _FileSystemManagerConfig():
    '''
    用于文件系统数据管理器的配置
    '''
    CHUNK_SIZE = 1000

    def __init__(self, config):
        self.ResponseFileSystemManagerConfig: _ResponseFileSystemManagerConfig = _ResponseFileSystemManagerConfig(config)
        self.TableFileSystemManagerConfig: _TableFileSystemManagerConfig = _TableFileSystemManagerConfig(config)

        self.DATA_ROOT_DIR_PATH = os.path.join(config.PROJECT_PATH, "data")


@ utils.singleton
class _TableFileSystemManagerConfig():
    '''
    用于响应文件系统数据管理器的配置
    '''
    def __init__(self, config):
        self.DATA_ROOT_DIR_PATH = os.path.join(config.PROJECT_PATH, "data")
        self.TABLE_COLUMNS = config.DataPackageConfig.TABLE_COLUMNS
        self.TABLE_MAIN_KEY_LIST = [config.DataPackageConfig.DATE_KEY, config.DataPackageConfig.VARIETY_KEY]
        self.TABLE_FILE_PATH = os.path.join(self.DATA_ROOT_DIR_PATH, "table.csv")

@ utils.singleton
class _DataPackageConfig():
    '''
    用于各类数据包的配置
    '''
    MAX_LENGTH_FOR_EVERY_STR = 43
    ###### 标签表 ######
    CHILD_NOT_INIT_TAG = 'CHILD_NOT_INIT_TAG'  # 表示子类没有初始化该标签


    ### 控制各类爬虫获取响应的原始文本
    VEGETABLE_PRICE_RESPONSE_GET_TAG = 'VEGETABLE_PRICE_RESPONSE_GET_TAG'  # 控制蔬菜价格爬虫爬取数据并将响应以 响应df 的形式放入数据包
    DAILY_WEATHER_RESPONSE_GET_TAG = 'DAILY_WEATHER_RESPONSE_GET_TAG'  # 控制天气爬虫爬取数据并将响应以 响应df 的形式放入数据包
    EXTREME_WEATHER_RESPONSE_GET_TAG = 'EXTREME_WEATHER_RESPONSE_GET_TAG'  # 控制天气爬虫爬取数据并将响应以 响应df 的形式放入数据包
    ECONOMY_RESPONSE_GET_TAG = 'ECONOMY_RESPONSE_GET_TAG'  # 控制经济爬虫爬取数据并将响应以 响应df 的形式放入数据包
    OIL_PRICE_RESPONSE_GET_TAG = 'OIL_PRICE_RESPONSE_GET_TAG'  # 控制油价爬虫爬取数据并将响应以 响应df 的形式放入数据包
    ### 爬虫请求失败
    VEGETABLE_PRICE_CRAWL_FAILED_TAG = 'VEGETABLE_PRICE_CRAWL_FAILED_TAG'  # 告诉爬虫数据包生成器这次爬取任务失败
    DAILY_WEATHER_CRAWL_FAILED_TAG = 'DAILY_WEATHER_CRAWL_FAILED_TAG'
    EXTREME_WEATHER_CRAWL_FAILED_TAG = 'EXTREME_WEATHER_CRAWL_FAILED_TAG'
    ECONOMY_CRAWL_FAILED_TAG = 'ECONOMY_CRAWL_FAILED_TAG'
    OIL_PRICE_CRAWL_FAILED_TAG = 'OIL_PRICE_CRAWL_FAILED_TAG'
    ### 控制清洗器将原始文本转换为结构化数据
    VEGETABLE_PRICE_RESPONSE_TO_TABLE_TAG = 'VEGETABLE_PRICE_RESPONSE_TO_TABLE_TAG'  # 控制数据清洗器将响应格式变成 结构化df
    DAILY_WEATHER_RESPONSE__TO_TABLE_TAG = 'DAILY_WEATHER_RESPONSE__TO_TABLE_TAG'
    EXTREME_WEATHER_RESPONSE__TO_TABLE_TAG = 'EXTREME_WEATHER_RESPONSE__TO_TABLE_TAG'
    ECONOMY_RESPONSE_TO_TABLE_TAG = 'ECONOMY_RESPONSE_TO_TABLE_TAG'
    OIL_PRICE_RESPONSE_TO_TABLE_TAG = 'OIL_PRICE_RESPONSE_TO_TABLE_TAG'
    ### 控制数据管理器存储原始响应
    VEGETABLE_PRICE_RESPONSE_MERGE_TAG = 'VEGETABLE_PRICE_RESPONSE_MERGE_TAG'  # 控制数据管理器将数据包内的 响应df 合并到响应表
    DAILY_WEATHER_RESPONSE_MERGE_TAG = 'DAILY_WEATHER_RESPONSE_MERGE_TAG'
    EXTREME_WEATHER_RESPONSE_MERGE_TAG = 'EXTREME_WEATHER_RESPONSE_MERGE_TAG'
    ### 控制数据管理器存储结构化数据
    VEGETABLE_TABLE_MERGE_TAG = 'VEGETABLE_TABLE_MERGE_TAG'  # 控制数据管理器将数据包内的 结构化df 合并到响应表


    ###### 初始标签栈 ######
    # @property
    # def UPDATE_VEGETABLE_RESPONSE_TAG_STACK(self):
    #     return list(reversed([
    #         self.VEGETABLE_PRICE_RESPONSE_GET_TAG,
    #         self.VEGETABLE_RESPONSE_MERGE_TAG
    #     ]))

    @property
    def UPDATE_VEGETABLE_TABLE_TAG_STACK(self):
        return list(reversed([
            self.VEGETABLE_PRICE_RESPONSE_GET_TAG,
            self.VEGETABLE_PRICE_RESPONSE_MERGE_TAG,
            self.VEGETABLE_PRICE_RESPONSE_TO_TABLE_TAG,
            self.VEGETABLE_TABLE_MERGE_TAG
        ]))

    @property
    def UPDATE_DAILY_WEATHER_RESPONSE_TAG_STACK(self):
        return list(reversed([
            self.DAILY_WEATHER_RESPONSE_GET_TAG,
            self.DAILY_WEATHER_RESPONSE_MERGE_TAG
        ]))

    # @property
    # def UPDATE_WEATHER_TABLE_TAG_STACK(self):
    #     return list(reversed([
    #         self.WEATHER_RESPONSE_GET_TAG,
    #         self.VEGETABLE_RESPONSE_TO_TABLE_TAG,
    #         self.VEGETABLE_TABLE_CLEAN_TAG,
    #         self.VEGETABLE_TABLE_MERGE_TAG
    #     ]))

    # @property
    # def UPDATE_ECONOMY_RESPONSE_TAG_STACK(self):
    #     return list(reversed([
    #         self.ECONOMY_RESPONSE_GET_TAG,
    #         self.VEGETABLE_RESPONSE_MERGE_TAG
    #     ]))

    # @property
    # def UPDATE_ECONOMY_TABLE_TAG_STACK(self):
    #     return list(reversed([
    #         self.ECONOMY_RESPONSE_GET_TAG,
    #         self.VEGETABLE_RESPONSE_TO_TABLE_TAG,
    #         self.VEGETABLE_TABLE_CLEAN_TAG,
    #         self.VEGETABLE_TABLE_MERGE_TAG
    #     ]))

    # @property
    # def UPDATE_OIL_PRICE_RESPONSE_TAG_STACK(self):
    #     return list(reversed([
    #         self.OIL_PRICE_RESPONSE_GET_TAG,
    #         self.VEGETABLE_RESPONSE_MERGE_TAG
    #     ]))

    # @property
    # def UPDATE_OIL_PRICE_TABLE_TAG_STACK(self):
    #     return list(reversed([
    #         self.OIL_PRICE_RESPONSE_GET_TAG,
    #         self.VEGETABLE_RESPONSE_TO_TABLE_TAG,
    #         self.VEGETABLE_TABLE_CLEAN_TAG,
    #         self.VEGETABLE_TABLE_MERGE_TAG
    #     ]))


    ###### 类型判断函数 ######
    IS_FORMATTED_DATETIME_STR_FUNC = \
        lambda series:  \
            pd.api.types.is_string_dtype(series) and \
            all(pd.to_datetime(series, format='%Y-%m-%d', errors='coerce').notnull()) and \
            not any(pd.isnull(series))  # 规定了日期列不能为空值,且必须符合'年-月-日'的格式

    IS_STR_FUNC = pd.api.types.is_string_dtype
    IS_INT_FUNC = pd.api.types.is_integer_dtype
    IS_FLOAT_FUNC = pd.api.types.is_float_dtype
    IS_OBJECT_FUNC = pd.api.types.is_object_dtype
    ###### 字段类型字典 ######
    NOT_INIT_COLUMNS = {
                        "未初始化":              [IS_OBJECT_FUNC],
                        }

    RESPONSE_COLUMNS = {
                        '爬取时间':              [IS_STR_FUNC],
                        '响应文本':              [IS_STR_FUNC]
                        }

    VEGETABLE_PRICE_COLUMNS = {
                        '日期':                [                                                IS_FORMATTED_DATETIME_STR_FUNC],
                        '蔬菜种类':             [IS_STR_FUNC],
                        '批发价最高价':         [               IS_INT_FUNC,   IS_FLOAT_FUNC],
                        '批发价均价':           [               IS_INT_FUNC,   IS_FLOAT_FUNC],
                        '批发价最低价':         [               IS_INT_FUNC,   IS_FLOAT_FUNC],
                        '零售价':               [               IS_INT_FUNC,   IS_FLOAT_FUNC],
                        '田头价':              [                IS_INT_FUNC,   IS_FLOAT_FUNC],
                        '价格单位':             [IS_STR_FUNC],
    }

    WEATHER_COLUMNS = {
                        '上海-'
    }

    TABLE_COLUMNS = {}
    TABLE_COLUMNS.update(VEGETABLE_PRICE_COLUMNS)  # 将蔬菜价格字典拼接到总表字典

    ###### 合并主键 ######
    RESPONSE_MAIN_KEY = '响应文本'
    DATE_KEY = '日期'
    VARIETY_KEY = '蔬菜种类'

    def __init__(self, config):
        pass


@ utils.singleton
class _Config():
    '''
    使用该类可在全局访问所有的配置, 它是从全局调用配置的入口
    '''

    PROJECT_DIR_NAME = "vegatable-price-trends"
    PROJECT_PATH = utils.get_root_path(PROJECT_DIR_NAME)
    LOG_FILE_PATH = os.path.join(PROJECT_PATH, 'logs', 'log1.txt')

    def __init__(self):
        self.DataPackageConfig: _DataPackageConfig = _DataPackageConfig(self)
        self.CrawlerConfig: _CrawlerConfig = _CrawlerConfig(self)
        self.DataManagerConfig: _DataManagerConfig = _DataManagerConfig(self)
        self.CleanerConfig: _CleanerConfig = _CleanerConfig(self)


Config = _Config()  # 实例化出配置类,并且保存到变量Config