import logging import yaml from pathlib import Path from typing import Dict, Any, Tuple def load_config(config_path: str = "/root/travel_rag/config/config.yaml") -> Dict[str, Any]: """ 从配置文件加载配置 Args: config_path: 配置文件路径,默认为 "/root/travel_rag/config.yaml" Returns: 配置字典 """ config_path = Path(config_path) if not config_path.exists(): raise FileNotFoundError(f"配置文件未找到: {config_path}") with open(config_path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) def setup_logging( log_level: str = "INFO", log_file: str = None ) -> None: """ 设置日志配置 Args: log_level: 日志级别,默认为 "INFO" log_file: 日志文件路径,默认为 None(仅控制台输出) """ # 设置日志格式 log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" # 配置根日志记录器 logging.basicConfig( level=getattr(logging, log_level.upper()), format=log_format ) # 如果指定了日志文件,添加文件处理器 if log_file: file_handler = logging.FileHandler(log_file) file_handler.setFormatter(logging.Formatter(log_format)) logging.getLogger().addHandler(file_handler) def setup_proxy(proxy_config_path: str = "/root/clash/config.yaml") -> Tuple[Dict[str, str], bool]: """ 设置系统代理并返回代理配置和代理可用性状态 Args: proxy_config_path: 代理配置文件路径(应为字符串类型) Returns: Tuple[Dict[str, str], bool]: (代理配置字典, 代理是否可用) """ import os import logging import requests from requests.exceptions import RequestException logger = logging.getLogger(__name__) # 设置默认代理地址 proxy_url = 'http://127.0.0.1:8880' # 如果存在配置文件,从配置文件读取 if os.path.exists(proxy_config_path): try: config = load_config(proxy_config_path) # 根据实际配置文件结构调整 proxy_url = config.get('proxy_url', proxy_url) logger.info(f"已从配置文件加载代理设置: {proxy_url}") except Exception as e: logger.warning(f"加载代理配置失败: {e},使用默认配置") # 设置环境变量 os.environ['HTTP_PROXY'] = proxy_url os.environ['HTTPS_PROXY'] = proxy_url proxies = { 'http': proxy_url, 'https': proxy_url } # 测试代理是否可用 try: response = requests.get('https://www.google.com', proxies=proxies, timeout=5, verify=False) # 添加 verify=False 避免证书问题 proxy_available = response.status_code == 200 if proxy_available: logger.info("代理服务器可用") else: logger.warning(f"代理服务器响应异常,状态码: {response.status_code}") except RequestException as e: logger.warning(f"代理服务器连接失败: {e}") proxy_available = False logger.info(f"代理设置完成: {proxies}, 可用状态: {proxy_available}") return proxies, proxy_available