local_proxy_pool/backend/utils/sub.py

import httpx
import yaml
from pathlib import Path
from typing import Optional, Dict, Any
from fastapi import HTTPException
def save_yaml_dump(config: dict, save_as: Path) -> Path:
    """保存配置文件"""
    save_as.parent.mkdir(parents=True, exist_ok=True)
    
    with open(save_as, 'w', encoding='utf-8') as f:
        yaml.dump(
            config,
            f,
            Dumper=yaml.SafeDumper,
            allow_unicode=True,
            indent=2,
            sort_keys=False
        )
    return save_as

async def async_get_sub(sub_url: str, save_path: Path, timeout: int = 10) -> Path:
    """获取订阅文件"""
    headers = {'User-Agent': 'clash-verge/v1.7.5'}
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get(sub_url, headers=headers, follow_redirects=True, timeout=timeout)
            resp.raise_for_status()
    except httpx.HTTPError as e:
        raise HTTPException(status_code=500, detail=f"订阅获取失败: {str(e)}")
    
    save_path = Path(save_path)
    save_path.parent.mkdir(parents=True, exist_ok=True)
    
    with open(save_path, 'w', encoding='utf-8') as f:
        f.write(resp.text)
    return save_path

def get_sub(sub_url: str, save_path: Path) -> Path:
    """获取订阅文件"""
    headers = {'User-Agent': 'clash-verge/v1.7.5'}
    try:
        resp = httpx.get(sub_url, headers=headers, follow_redirects=True, timeout=10)
        resp.raise_for_status()
    except httpx.HTTPError as e:
        raise HTTPException(status_code=500, detail=f"订阅获取失败: {str(e)}")
    
    save_path = Path(save_path)
    save_path.parent.mkdir(parents=True, exist_ok=True)
    
    with open(save_path, 'w', encoding='utf-8') as f:
        f.write(resp.text)
    return save_path

def update_config(
    read_path: Path,
    config_update: dict,
    save_as: Optional[Path] = None,
) -> Path:
    """更新配置文件"""
    config: Dict[str, Any] = {}
    if read_path.exists():
        with open(read_path, 'r', encoding='utf-8') as f:
            config = yaml.safe_load(f) or {}
    
    config.update(config_update)
    
    save_as = save_as or read_path
    save_yaml_dump(config, save_as)

def main():
    sub_url = 'https://www.yfjc.xyz/api/v1/client/subscribe?token=b74f2207492053926f7511a8e474048f'
    OUTPUT = Path(__file__).parent.parent.absolute() / "output"
    save_path = get_sub(sub_url, OUTPUT / "config.yaml")
    update_config(
        read_path=save_path,
        config_update={
            "port": 7890,
            "socks-port": 7891,
            "redir-port": 7892,
            "allow-lan": True,
            "mode": "rule",
            "log-level": "silent",
            "external-controller": "127.0.0.1:9090",
            "secret": "",
        }
    )

if __name__ == "__main__":
    main()