zhang_crawl_bio/ui/backend/utils/config.py

import yaml
from pathlib import Path
from pydantic import BaseModel
from typing import List, Dict, Union,Optional,Any
from utils.pydantic_auto_field import AutoLoadModel
APP_PATH = Path(__file__).parent.parent
OUTPUT_DIR = APP_PATH / "output"
CONFIG_PATH = APP_PATH / "config.yaml"
REPO_BASE_DIR = Path(APP_PATH.parent.parent)
DOWNLOAD_DIR = REPO_BASE_DIR / "download"
PROXY_POLL_DIR = DOWNLOAD_DIR / "proxy_pool"
WORKER_DIR_BASE = APP_PATH.parent.parent

class Proxy(BaseModel):
    name: Optional[str] = ''
    port: Optional[int] = -1
    file_path: Optional[str] = ''
    startup: Optional[bool] = False


class Sub(AutoLoadModel):
    url: Optional[str] = None
    start_port: Optional[int] = 9660  # Changed to int
    redis_url: Optional[str] = 'redis://localhost:6379/8'
    file: Optional[str] = None
    temp_dir: Optional[str] = str(PROXY_POLL_DIR / "temp")  
    auto_start: Optional[bool] = True 
    proxies: Optional[Dict[Union[int,str], Proxy]] = {}
    def __init__(self, **data):
            super().__init__(**data)
            # Convert proxies dictionary values to Proxy objects
            if self.proxies:
                self.proxies = {k: Proxy(**v) if isinstance(v, dict) else v for k, v in self.proxies.items()}

class Browser(BaseModel):
    exe_path: Optional[str] = str(REPO_BASE_DIR / r"download\chrome-win\chrome.exe")
    no_imgs: Optional[bool] = True

class Config(BaseModel):
    sub: Optional[Sub] = Sub()
    select_proxy: Optional[str] = "system"
    mimo_exe: Optional[str] = str(PROXY_POLL_DIR / r"mihomo-windows-amd64-go120.exe")
    redis_exe: Optional[str] = str(REPO_BASE_DIR / r"download\Redis-x64-5.0.14.1\redis-server.exe")
    worker_backend_py: Optional[str] = str(WORKER_DIR_BASE / r"worker\api\worker_server.py")
    sqluri: Optional[str] = r'G:\code\upwork\zhang_crawl_bio\output\temp.db'
    browser: Optional[Browser] = Browser()
    redis_port: Optional[int] = None  # Changed to int
    def save(self):
        config_path = get_config_path()
        with open(config_path, "w", encoding="utf-8") as file:
            yaml.dump(self.model_dump(), file, encoding="utf-8", )
        return self
            
def get_config_path():
    return CONFIG_PATH

def read_config(config_path: Path):
    if not config_path.exists():
        config = Config()
        config.save()
        return config
    with open(config_path, "r", encoding="utf-8") as file:
        config_dict = yaml.safe_load(file)
    config = Config(**config_dict)
    return config
config = read_config(get_config_path())

import os

WORKER_SERVICE_URL = os.getenv("WORKER_SERVICE_URL", "http://localhost:8003")

def main():
    print(config)
    config.browser = Browser()
    config.save(
        
    )
if __name__ == "__main__":
    main()