miui-auto-tasks/utils/config.py

""""配置文件"""
import os
import platform
from hashlib import md5
from json import JSONDecodeError
from pathlib import Path
from typing import Dict, List, Optional, Union

import yaml
from pydantic import BaseModel
from pydantic import ValidationError, field_validator

from .logger import log

ROOT_PATH = Path(__name__).parent.absolute()

DATA_PATH = ROOT_PATH / "data"
"""数据保存目录"""

CONFIG_PATH = DATA_PATH / "config.yaml" if os.getenv("MIUITASK_CONFIG_PATH") is None else Path(os.getenv("MIUITASK_CONFIG_PATH"))
"""数据文件默认路径"""

os.makedirs(DATA_PATH, exist_ok=True)


def md5_crypto(passwd: str) -> str:
    """MD5加密"""
    return md5(passwd.encode('utf8')).hexdigest().upper()


def cookies_to_dict(cookies):
    """将cookies字符串转换为字典"""
    cookies_dict = {}
    for cookie in cookies.split(';'):
        key, value = cookie.strip().split('=', 1)  # 分割键和值
        cookies_dict[key] = value
    return cookies_dict

def get_platform() -> str:
    """获取当前运行平台"""
    if os.path.exists('/.dockerenv'):
        if os.environ.get('QL_DIR') and os.environ.get('QL_BRANCH'):
            return "qinglong"
        else:
            return "docker"
    return platform.system().lower()


class Account(BaseModel):
    """账号处理器"""
    uid: str = "100000"
    """账户ID 非账户用户名或手机号"""
    password: str = ""
    """账户密码或其MD5哈希"""
    cookies: Union[dict, str] = {}
    """账户登录后的cookies"""
    user_agent: str = 'Mozilla/5.0 (Linux; Android 13) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/116.0.0.0 Safari/537.36'
    """登录社区时所用浏览器的 User-Agent"""

    CheckIn: bool = False
    """社区成长值签到，启用功能意味着你愿意自行承担相关风险"""
    BrowseUserPage: bool = False
    """社区浏览个人主页10秒，启用功能意味着你愿意自行承担相关风险"""
    BrowsePost: bool = False
    """社区浏览帖子10秒，启用功能意味着你愿意自行承担相关风险"""
    ThumbUp: bool = False
    """点赞帖子，启用功能意味着你愿意自行承担相关风险"""
    BrowseSpecialPage: bool = False
    """社区在活动期间可能会出现限时的“浏览指定专题页”任务，启用功能意味着你愿意自行承担相关风险"""
    BoardFollow: bool = False
    """社区可能会出现限时的“加入圈子”任务，启用功能意味着你愿意自行承担相关风险"""
    CarrotPull: bool = False
    """社区拔萝卜，启用功能意味着你愿意自行承担相关风险"""

    @field_validator("password")
    @classmethod
    def _password(cls, value: Optional[str]):  # pylint: disable=no-self-argument
        if len(value) == 32:
            return value
        return md5_crypto(value)

    @field_validator("cookies")
    @classmethod
    def _cookies(cls, value: Union[dict, str]):  # pylint: disable=no-self-argument
        if isinstance(value, str):
            return cookies_to_dict(value)
        return value


class OnePush(BaseModel):
    """推送配置"""
    notifier: Union[str, bool] = ""
    """是否开启消息推送"""
    params: Dict = {
        "title": "",
        "markdown": False,
        "token": "",
        "userid": ""
    }
    """推送参数"""


class Preference(BaseModel):
    """偏好设置"""
    geetest_url: str = ""
    """极验验证URL"""
    geetest_params: Dict = {}
    """极验自定义params参数"""
    geetest_data: Dict = {}
    """极验自定义data参数"""

class Config(BaseModel):
    """插件数据"""
    preference: Preference = Preference()
    """偏好设置"""
    accounts: List[Account] = [Account()]
    """账号设置"""
    ONEPUSH: OnePush = OnePush()
    """消息推送"""

def write_plugin_data(data: Config = None):
    """
    写入插件数据文件

    :param data: 配置对象
    """
    try:
        if data is None:
            data = ConfigManager.data_obj
        try:
            # str_data = orjson.dumps(data.dict(), option=orjson.OPT_PASSTHROUGH_DATETIME | orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_INDENT_2)
            str_data = yaml.dump(data.model_dump(), indent=4, allow_unicode=True, sort_keys=False)
        except (AttributeError, TypeError, ValueError):
            log.exception("数据对象序列化失败，可能是数据类型错误")
            return False
        with open(CONFIG_PATH, "w", encoding="utf-8") as file:
            file.write(str_data)
        return True
    except OSError:
        return False


class ConfigManager:
    """配置管理器"""
    data_obj = Config()
    """加载出的插件数据对象"""
    platform = get_platform()
    """运行平台"""

    @classmethod
    def load_config(cls):
        """
        加载插件数据文件
        """
        if os.path.exists(DATA_PATH) and os.path.isfile(CONFIG_PATH):
            try:
                with open(CONFIG_PATH, 'r', encoding="utf-8") as file:
                    data = yaml.safe_load(file)
                new_model = Config.model_validate(data)
                for attr in new_model.model_fields:
                    # ConfigManager.data_obj.__setattr__(attr, new_model.__getattribute__(attr))
                    setattr(ConfigManager.data_obj, attr, getattr(new_model, attr))
                write_plugin_data(ConfigManager.data_obj)  # 同步配置
            except (ValidationError, JSONDecodeError):
                log.exception(f"读取数据文件失败，请检查数据文件 {CONFIG_PATH} 格式是否正确")
                raise
            except Exception:
                log.exception(
                    f"读取数据文件失败，请检查数据文件 {CONFIG_PATH} 是否存在且有权限读取和写入")
                raise
        else:
            try:
                if not os.path.exists(DATA_PATH):
                    os.mkdir(DATA_PATH)
                write_plugin_data()
            except (AttributeError, TypeError, ValueError, PermissionError):
                log.exception(f"创建数据文件失败，请检查是否有权限读取和写入 {CONFIG_PATH}")
                raise
            log.info(f"数据文件 {CONFIG_PATH} 不存在，已创建默认数据文件。")


ConfigManager.load_config()