Spaces:
Running
Running
File size: 3,680 Bytes
8a03d2c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """API密钥认证模块 - 简化版"""
import os
import threading
from typing import Any
from pathlib import Path
from ..utils.logger import get_logger
# 初始化日志
logger = get_logger(__name__)
class APIKeyManager:
"""简化的API密钥管理器 - 只保留基本验证功能"""
def __init__(self, keys_file: str | None = None):
logger.info("初始化 API 密钥管理器")
self.keys_file: str = keys_file or str(Path(__file__).parent.parent.parent / "config" / "api_keys.txt")
self.api_keys: set[str] = set()
self.key_names: dict[str, str] = {} # api_key -> name
self._lock: threading.Lock = threading.Lock()
logger.debug(f"API 密钥文件路径: {self.keys_file}")
def load_keys(self) -> bool:
"""从配置文件加载API密钥"""
logger.info("开始加载 API 密钥")
try:
if not os.path.exists(self.keys_file):
logger.warning(f"API 密钥文件不存在: {self.keys_file}")
return False
with self._lock:
self.api_keys.clear()
self.key_names.clear()
valid_count = 0
error_count = 0
logger.debug(f"读取密钥文件: {self.keys_file}")
with open(self.keys_file, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
# 跳过空行和注释行
if not line or line.startswith('#'):
continue
# 解析格式: key_name:api_key:description
parts = line.split(':', 2)
if len(parts) < 2:
logger.warning(f"第 {line_num} 行格式错误，跳过")
error_count += 1
continue
key_name = parts[0].strip()
api_key = parts[1].strip()
# 验证密钥格式
if not api_key.startswith('sk-'):
logger.warning(f"第 {line_num} 行密钥格式无效 ({key_name})，跳过")
error_count += 1
continue
self.api_keys.add(api_key)
self.key_names[api_key] = key_name
valid_count += 1
logger.debug(f"加载密钥: {key_name} ({api_key[:8]}...)")
if valid_count > 0:
logger.success(f"成功加载 {valid_count} 个 API 密钥")
if error_count > 0:
logger.warning(f"跳过 {error_count} 个无效条目")
return True
except Exception as e:
logger.error(f"加载 API 密钥失败: {e}")
return False
def validate_key(self, api_key: str) -> bool:
"""验证API密钥是否有效"""
if not api_key:
logger.debug("API 密钥为空")
return False
is_valid = api_key.strip() in self.api_keys
if is_valid:
key_name = self.key_names.get(api_key.strip(), 'unknown')
logger.debug(f"API 密钥验证成功: {key_name} ({api_key[:8]}...)")
else:
logger.debug(f"API 密钥验证失败: {api_key[:8]}...")
return is_valid
# 全局密钥管理器实例
api_key_manager = APIKeyManager()
|