feat: bzzoiro 多 API Key 轮换,遇限流自动切换
新增 KeyRing 轮换环(src/data/key_ring.py): - 支持逗号/分号/换行分隔多个 key,单 key 场景零开销 - 遇到 429 自动标记当前 key 冷却(默认 60s)并立即切换到下一个 key - 全部 key 冷却时等待最早恢复的 key,避免无谓重试 - 热更新 key 列表(增删 key 无需重启) - 进程级单例,按 base URL 隔离 后端集成: - bzzoiro._fetch_json_async 接入 KeyRing,429 立即轮换(不等待) - 新增 /admin/keyring/status 端点展示 key 环状态 - 新增 /admin/keyring/cooldown/reset 紧急重置冷却 - BZZOIRO_KEY 配置描述提示多 key 支持 - mask_value 多 key 显示数量(如 "3 个 key(末段 …XXXX)") - 清理已移除 injuries 的 API_FOOTBALL_KEY 配置项 前端: - 数据源页新增 Key Ring 状态面板(每个 key 可用/冷却状态 + 重置按钮) - 配置项脱敏展示支持多 key 计数 测试: 新增 24 个 KeyRing 单元测试 + 2 个 bzzoiro 轮换集成测试,全部通过。 Co-Authored-By: new-provider/LongCat-2.0 <<EMAIL>>
This commit is contained in:
co-authored by
new-provider/LongCat-2.0 <
parent
f05dc1ae15
commit
f52ec8b963
@@ -381,3 +381,28 @@ export function fetchMatchContext(id: number): Promise<MatchContextOut> {
|
|||||||
export function fetchAdminStats(): Promise<AdminStats> {
|
export function fetchAdminStats(): Promise<AdminStats> {
|
||||||
return api.get<AdminStats>(`${API_BASE}/admin/stats`)
|
return api.get<AdminStats>(`${API_BASE}/admin/stats`)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── API Key 轮换环 ──────────────────────────────────────────────
|
||||||
|
|
||||||
|
export interface KeyRingKeyStatus {
|
||||||
|
masked: string
|
||||||
|
blocked_remaining: number
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface KeyRingStatusResponse {
|
||||||
|
base_url: string
|
||||||
|
total: number
|
||||||
|
has_multiple: boolean
|
||||||
|
cooldown_seconds: number
|
||||||
|
active_index: number
|
||||||
|
active_key: string | null
|
||||||
|
keys: KeyRingKeyStatus[]
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function fetchKeyRingStatus(): Promise<KeyRingStatusResponse> {
|
||||||
|
return api.get<KeyRingStatusResponse>(`${API_BASE}/admin/keyring/status`)
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function resetKeyRingCooldown(): Promise<{ ok: boolean; message: string; stats: KeyRingStatusResponse }> {
|
||||||
|
return api.post<{ ok: boolean; message: string; stats: KeyRingStatusResponse }>(`${API_BASE}/admin/keyring/cooldown/reset`)
|
||||||
|
}
|
||||||
|
|||||||
@@ -3,11 +3,14 @@ import {
|
|||||||
fetchDataSourceStatuses,
|
fetchDataSourceStatuses,
|
||||||
fetchIngestStatus,
|
fetchIngestStatus,
|
||||||
fetchAdminStats,
|
fetchAdminStats,
|
||||||
|
fetchKeyRingStatus,
|
||||||
|
resetKeyRingCooldown,
|
||||||
updateSetting,
|
updateSetting,
|
||||||
clearSetting,
|
clearSetting,
|
||||||
testDataSourceConnection,
|
testDataSourceConnection,
|
||||||
} from '../dal'
|
} from '../dal'
|
||||||
import type { DataSourceStatus, DataSourceTestResult, IngestSourceStatus, AdminStats } from '../types'
|
import type { DataSourceStatus, DataSourceTestResult, IngestSourceStatus, AdminStats } from '../types'
|
||||||
|
import type { KeyRingStatusResponse } from '../dal'
|
||||||
import SettingRow from '../SettingRow'
|
import SettingRow from '../SettingRow'
|
||||||
import { Card, CardBody, CardHeader, Badge, SectionHeader, Alert, Spinner, SkeletonBlock } from '../components'
|
import { Card, CardBody, CardHeader, Badge, SectionHeader, Alert, Spinner, SkeletonBlock } from '../components'
|
||||||
|
|
||||||
@@ -34,6 +37,8 @@ export default function DataSourcesPage() {
|
|||||||
const [editingKey, setEditingKey] = useState<string | null>(null)
|
const [editingKey, setEditingKey] = useState<string | null>(null)
|
||||||
const [busyKey, setBusyKey] = useState<string | null>(null)
|
const [busyKey, setBusyKey] = useState<string | null>(null)
|
||||||
const [rowNotice, setRowNotice] = useState<{ key: string; ok: boolean; text: string } | null>(null)
|
const [rowNotice, setRowNotice] = useState<{ key: string; ok: boolean; text: string } | null>(null)
|
||||||
|
const [keyRing, setKeyRing] = useState<KeyRingStatusResponse | null>(null)
|
||||||
|
const [ringLoading, setRingLoading] = useState(false)
|
||||||
|
|
||||||
const loadSources = useCallback(async () => {
|
const loadSources = useCallback(async () => {
|
||||||
setLoading(true)
|
setLoading(true)
|
||||||
@@ -70,11 +75,24 @@ export default function DataSourcesPage() {
|
|||||||
}
|
}
|
||||||
}, [])
|
}, [])
|
||||||
|
|
||||||
|
// Key Ring 状态(只读)
|
||||||
|
const loadKeyRing = useCallback(async () => {
|
||||||
|
setRingLoading(true)
|
||||||
|
try {
|
||||||
|
setKeyRing(await fetchKeyRingStatus())
|
||||||
|
} catch {
|
||||||
|
/* ignore */
|
||||||
|
} finally {
|
||||||
|
setRingLoading(false)
|
||||||
|
}
|
||||||
|
}, [])
|
||||||
|
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
loadSources()
|
loadSources()
|
||||||
loadIngest()
|
loadIngest()
|
||||||
loadStats()
|
loadStats()
|
||||||
}, [loadSources, loadIngest, loadStats])
|
loadKeyRing()
|
||||||
|
}, [loadSources, loadIngest, loadStats, loadKeyRing])
|
||||||
|
|
||||||
async function handleTest(sourceName: string) {
|
async function handleTest(sourceName: string) {
|
||||||
setTestingSource(sourceName)
|
setTestingSource(sourceName)
|
||||||
@@ -160,6 +178,16 @@ export default function DataSourcesPage() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function handleResetCooldown() {
|
||||||
|
try {
|
||||||
|
const res = await resetKeyRingCooldown()
|
||||||
|
setKeyRing(res.stats)
|
||||||
|
setRowNotice({ key: "__ring", ok: true, text: res.message })
|
||||||
|
} catch (err) {
|
||||||
|
setRowNotice({ key: "__ring", ok: false, text: err instanceof Error ? err.message.split('\n')[0] : '重置失败' })
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
return (
|
return (
|
||||||
<div className="space-y-6">
|
<div className="space-y-6">
|
||||||
@@ -252,6 +280,57 @@ export default function DataSourcesPage() {
|
|||||||
</div>
|
</div>
|
||||||
)}
|
)}
|
||||||
|
|
||||||
|
{/* API Key 轮换环状态 */}
|
||||||
|
<Card>
|
||||||
|
<CardHeader
|
||||||
|
title="API Key 轮换环"
|
||||||
|
description={keyRing?.has_multiple
|
||||||
|
? `已配置 ${keyRing.total} 个 key,遇到限流(429)自动切换;冷却 ${keyRing.cooldown_seconds}s`
|
||||||
|
: '当前仅 1 个 key,无法轮换。建议配置多个 key 以提高限流容忍度'
|
||||||
|
}
|
||||||
|
action={
|
||||||
|
<button
|
||||||
|
onClick={handleResetCooldown}
|
||||||
|
disabled={ringLoading}
|
||||||
|
className="btn-sm btn-outline"
|
||||||
|
>
|
||||||
|
重置冷却
|
||||||
|
</button>
|
||||||
|
}
|
||||||
|
/>
|
||||||
|
<CardBody>
|
||||||
|
{ringLoading && !keyRing ? (
|
||||||
|
<SkeletonBlock className="h-10 w-full" />
|
||||||
|
) : keyRing && keyRing.total > 0 ? (
|
||||||
|
<div className="space-y-2">
|
||||||
|
{keyRing.keys.map((k, i) => {
|
||||||
|
const isBlocked = k.blocked_remaining > 0
|
||||||
|
return (
|
||||||
|
<div key={i} className={`flex items-center justify-between gap-3 border-b border-ink-100 py-2 last:border-b-0 ${isBlocked ? 'opacity-70' : ''}`}>
|
||||||
|
<div className="flex items-center gap-2">
|
||||||
|
<span className={`inline-block h-2 w-2 rounded-full ${isBlocked ? 'bg-amber-500' : 'bg-emerald-500'}`} />
|
||||||
|
<span className="font-mono text-xs text-ink-700">{k.masked}</span>
|
||||||
|
{i === keyRing.active_index && (
|
||||||
|
<span className="rounded bg-ink-900 px-1.5 py-0.5 text-2xs text-paper-500">当前</span>
|
||||||
|
)}
|
||||||
|
</div>
|
||||||
|
<span className={`text-2xs tabular-nums ${isBlocked ? 'text-amber-600' : 'text-ink-400'}`}>
|
||||||
|
{isBlocked ? `冷却中 ${k.blocked_remaining.toFixed(0)}s` : '可用'}
|
||||||
|
</span>
|
||||||
|
</div>
|
||||||
|
)
|
||||||
|
})}
|
||||||
|
</div>
|
||||||
|
) : (
|
||||||
|
<p className="text-xs text-ink-400">暂无 key 配置</p>
|
||||||
|
)}
|
||||||
|
<p className="mt-3 border-l-2 border-ink-300 pl-3 text-2xs leading-relaxed text-ink-500">
|
||||||
|
在「Bzzoiro」配置项中用<b>逗号 / 分号 / 换行</b>分隔多个 key 即可启用轮换。遇到 429 自动标记当前 key 为冷却并立即切换到下一个 key;
|
||||||
|
全部 key 冷却时等待最早恢复的 key。「重置冷却」可紧急恢复所有 key。
|
||||||
|
</p>
|
||||||
|
</CardBody>
|
||||||
|
</Card>
|
||||||
|
|
||||||
{/* 近期活动统计(只读) */}
|
{/* 近期活动统计(只读) */}
|
||||||
{stats && stats.predictions && (
|
{stats && stats.predictions && (
|
||||||
<Card>
|
<Card>
|
||||||
|
|||||||
@@ -29,6 +29,7 @@ from src.core.runtime_config import (
|
|||||||
)
|
)
|
||||||
from src.db.base import AsyncSession, get_db_read
|
from src.db.base import AsyncSession, get_db_read
|
||||||
from src.db.models import League, Match, MatchStats, Standing
|
from src.db.models import League, Match, MatchStats, Standing
|
||||||
|
from src.data.key_ring import get_key_ring, parse_keys
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -346,6 +347,30 @@ async def ingest_status(db: AsyncSession = Depends(get_db_read)):
|
|||||||
return {"sources": [bzzoiro]}
|
return {"sources": [bzzoiro]}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/keyring/status")
|
||||||
|
async def keyring_status():
|
||||||
|
"""KeyRing 运行状态:当前使用的 key、冷却状态、轮转信息(供管理后台展示)。"""
|
||||||
|
base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/")
|
||||||
|
raw_keys = await get_runtime_value("BZZOIRO_KEY")
|
||||||
|
ring = get_key_ring(base, raw_keys)
|
||||||
|
st = ring.stats()
|
||||||
|
st["base_url"] = base
|
||||||
|
st["cooldown_seconds"] = ring._cooldown
|
||||||
|
st["has_multiple"] = ring.has_multiple
|
||||||
|
st["active_key"] = ring.active_key
|
||||||
|
return st
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/keyring/cooldown/reset")
|
||||||
|
async def keyring_reset_cooldown():
|
||||||
|
"""手动重置所有 key 的冷却状态(用于紧急恢复)。"""
|
||||||
|
base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/")
|
||||||
|
raw_keys = await get_runtime_value("BZZOIRO_KEY")
|
||||||
|
ring = get_key_ring(base, raw_keys)
|
||||||
|
ring._blocked_until.clear()
|
||||||
|
return {"ok": True, "message": "已重置所有 key 冷却状态", "stats": ring.stats()}
|
||||||
|
|
||||||
|
|
||||||
def _last_failure_log(source: str) -> dict | None:
|
def _last_failure_log(source: str) -> dict | None:
|
||||||
"""从系统日志缓冲中查找某数据源的最近一次错误(仅作参考,非专用失败表)。"""
|
"""从系统日志缓冲中查找某数据源的最近一次错误(仅作参考,非专用失败表)。"""
|
||||||
entries = get_entries(min_level="ERROR", keyword=source, limit=5)
|
entries = get_entries(min_level="ERROR", keyword=source, limit=5)
|
||||||
|
|||||||
@@ -31,7 +31,6 @@ class Settings(BaseSettings):
|
|||||||
# --- data sources ---
|
# --- data sources ---
|
||||||
BZZOIRO_KEY: str = ""
|
BZZOIRO_KEY: str = ""
|
||||||
BZZOIRO_BASE: str = "https://sports.bzzoiro.com/api/v2"
|
BZZOIRO_BASE: str = "https://sports.bzzoiro.com/api/v2"
|
||||||
API_FOOTBALL_KEY: str = ""
|
|
||||||
|
|
||||||
# --- 代理头信任 ---
|
# --- 代理头信任 ---
|
||||||
# 为 True 时才解析 X-Forwarded-For,否则只用 request.client.host。
|
# 为 True 时才解析 X-Forwarded-For,否则只用 request.client.host。
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ DB 读取失败时也回落环境变量,保证采集不因管理表故障而中
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
import re
|
||||||
import secrets
|
import secrets
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
@@ -39,14 +40,13 @@ class SettingDef:
|
|||||||
# 允许在后台查看/修改的配置项白名单(之外的 key 一律拒绝读写)
|
# 允许在后台查看/修改的配置项白名单(之外的 key 一律拒绝读写)
|
||||||
SETTING_DEFS: dict[str, SettingDef] = {
|
SETTING_DEFS: dict[str, SettingDef] = {
|
||||||
"BZZOIRO_KEY": SettingDef(
|
"BZZOIRO_KEY": SettingDef(
|
||||||
"BZZOIRO_KEY", "Bzzoiro API Key", "比赛赛程 / 比分数据源凭证", sensitive=True,
|
"BZZOIRO_KEY", "Bzzoiro API Key",
|
||||||
|
"比赛赛程/比分/积分榜/统计的数据源凭证。支持多 key 轮换:用逗号、分号或换行分隔多个 key,遇到限流(429)自动切换",
|
||||||
|
sensitive=True,
|
||||||
),
|
),
|
||||||
"BZZOIRO_BASE": SettingDef(
|
"BZZOIRO_BASE": SettingDef(
|
||||||
"BZZOIRO_BASE", "Bzzoiro API 地址", "Bzzoiro 接口基础地址", sensitive=False,
|
"BZZOIRO_BASE", "Bzzoiro API 地址", "Bzzoiro 接口基础地址", sensitive=False,
|
||||||
),
|
),
|
||||||
"API_FOOTBALL_KEY": SettingDef(
|
|
||||||
"API_FOOTBALL_KEY", "API-Football Key", "伤停数据源凭证(api-sports)", sensitive=True,
|
|
||||||
),
|
|
||||||
"LLM_API_KEY": SettingDef(
|
"LLM_API_KEY": SettingDef(
|
||||||
"LLM_API_KEY", "LLM API Key", "大模型服务凭证(OpenAI 兼容接口)", sensitive=True,
|
"LLM_API_KEY", "LLM API Key", "大模型服务凭证(OpenAI 兼容接口)", sensitive=True,
|
||||||
),
|
),
|
||||||
@@ -83,11 +83,20 @@ for _agent in AGENT_META:
|
|||||||
|
|
||||||
|
|
||||||
def mask_value(value: str, sensitive: bool) -> str:
|
def mask_value(value: str, sensitive: bool) -> str:
|
||||||
"""脱敏展示:敏感值只留末 4 位;非敏感值原样返回。"""
|
"""脱敏展示:敏感值只留末 4 位;非敏感值原样返回。
|
||||||
|
|
||||||
|
多 key(逗号/分号/换行分隔)时显示数量,如 "3 个 key(末段 …XXXX)"。
|
||||||
|
"""
|
||||||
if not value:
|
if not value:
|
||||||
return ""
|
return ""
|
||||||
if not sensitive:
|
if not sensitive:
|
||||||
return value
|
return value
|
||||||
|
# 检测多 key
|
||||||
|
keys = [k.strip() for k in re.split(r"[,;\n]", value) if k.strip()]
|
||||||
|
if len(keys) > 1:
|
||||||
|
last = keys[-1]
|
||||||
|
tail = last[-4:] if len(last) >= 4 else last
|
||||||
|
return f"{len(keys)} 个 key(末段 …{tail})"
|
||||||
return f"****{value[-4:]}" if len(value) >= 8 else "****"
|
return f"****{value[-4:]}" if len(value) >= 8 else "****"
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+31
-5
@@ -22,6 +22,7 @@ import httpx
|
|||||||
from src.core.runtime_config import get_runtime_value
|
from src.core.runtime_config import get_runtime_value
|
||||||
from src.core.http_client import get_client
|
from src.core.http_client import get_client
|
||||||
from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL
|
from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL
|
||||||
|
from src.data.key_ring import get_key_ring
|
||||||
from src.data.normalize import normalize_bzzoiro
|
from src.data.normalize import normalize_bzzoiro
|
||||||
from src.data.team_names_zh import zh_name
|
from src.data.team_names_zh import zh_name
|
||||||
from src.data.sources import register
|
from src.data.sources import register
|
||||||
@@ -61,20 +62,25 @@ def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, i
|
|||||||
|
|
||||||
|
|
||||||
async def _fetch_json_async(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list:
|
async def _fetch_json_async(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list:
|
||||||
"""异步 HTTP(bzzoiro 使用 httpx,不再阻塞事件循环线程池)。"""
|
"""异步 HTTP(bzzoiro 使用 httpx,不再阻塞事件循环线程池)。
|
||||||
|
|
||||||
|
多 key 轮换:遇到 429 自动切换到下一个 key;全部 key 冷却时等待最早恢复。
|
||||||
|
"""
|
||||||
base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/")
|
base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/")
|
||||||
|
raw_keys = await get_runtime_value("BZZOIRO_KEY")
|
||||||
|
ring = get_key_ring(base, raw_keys)
|
||||||
|
|
||||||
url = f"{base}/{path.lstrip('/')}"
|
url = f"{base}/{path.lstrip('/')}"
|
||||||
key = await get_runtime_value("BZZOIRO_KEY")
|
key = ring.get()
|
||||||
if not key:
|
if not key:
|
||||||
raise RuntimeError("BZZOIRO_KEY 未设置")
|
raise RuntimeError("BZZOIRO_KEY 未设置")
|
||||||
|
|
||||||
|
last_exc: Exception | None = None
|
||||||
|
for attempt in range(max_retries):
|
||||||
headers = {
|
headers = {
|
||||||
"Authorization": f"Token {key}",
|
"Authorization": f"Token {key}",
|
||||||
"Accept": "application/json",
|
"Accept": "application/json",
|
||||||
}
|
}
|
||||||
|
|
||||||
last_exc: Exception | None = None
|
|
||||||
for attempt in range(max_retries):
|
|
||||||
try:
|
try:
|
||||||
client = get_client()
|
client = get_client()
|
||||||
# 整请求兜底: httpx 无 total 超时,用 wait_for 防「滴水式」限速挂死
|
# 整请求兜底: httpx 无 total 超时,用 wait_for 防「滴水式」限速挂死
|
||||||
@@ -91,9 +97,22 @@ async def _fetch_json_async(path: str, params: dict | None = None, max_retries:
|
|||||||
last_exc = e
|
last_exc = e
|
||||||
status = getattr(getattr(e, "response", None), "status_code", None)
|
status = getattr(getattr(e, "response", None), "status_code", None)
|
||||||
if status == 429:
|
if status == 429:
|
||||||
|
# 限流:标记当前 key 冷却,切换到下一个
|
||||||
|
new_key = ring.report_rate_limited(key)
|
||||||
|
if new_key and new_key != key:
|
||||||
|
logger.info("bzzoiro 429 → 切换 key: %s → %s,立即重试", _km(key), _km(new_key))
|
||||||
|
key = new_key
|
||||||
|
continue # 立即重试,不等待
|
||||||
|
# 单 key 或全部冷却:等待最早恢复的 key
|
||||||
|
wait = ring.wait_if_all_blocked()
|
||||||
|
if wait > 0:
|
||||||
|
logger.warning("bzzoiro 全部 key 冷却,等待 %.1fs 后重试", wait)
|
||||||
|
await asyncio.sleep(min(wait, 30.0))
|
||||||
|
else:
|
||||||
delay = min(2 ** attempt, 16) + random.uniform(0, 1)
|
delay = min(2 ** attempt, 16) + random.uniform(0, 1)
|
||||||
logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay)
|
logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay)
|
||||||
await asyncio.sleep(delay)
|
await asyncio.sleep(delay)
|
||||||
|
key = ring.get() or key
|
||||||
continue
|
continue
|
||||||
if 500 <= (status or 0) < 600:
|
if 500 <= (status or 0) < 600:
|
||||||
delay = min(2 ** attempt, 16) + random.uniform(0, 1)
|
delay = min(2 ** attempt, 16) + random.uniform(0, 1)
|
||||||
@@ -110,6 +129,13 @@ async def _fetch_json_async(path: str, params: dict | None = None, max_retries:
|
|||||||
raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}")
|
raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}")
|
||||||
|
|
||||||
|
|
||||||
|
def _km(key: str) -> str:
|
||||||
|
"""key 脱敏缩写(用于日志)。"""
|
||||||
|
if len(key) <= 8:
|
||||||
|
return key[:2] + "***"
|
||||||
|
return key[:4] + "..." + key[-4:]
|
||||||
|
|
||||||
|
|
||||||
async def fetch_bzzoiro_events(
|
async def fetch_bzzoiro_events(
|
||||||
league_code: str,
|
league_code: str,
|
||||||
*,
|
*,
|
||||||
|
|||||||
@@ -0,0 +1,145 @@
|
|||||||
|
"""API Key 轮换环:多 key 自动切换,遇到限流(429)自动跳过已冷却 key。
|
||||||
|
|
||||||
|
设计:
|
||||||
|
- 进程内纯内存状态(限速是短时状态,无需持久化)
|
||||||
|
- 单 key 场景零开销:直接透传
|
||||||
|
- 多 key 场景:429 时把当前 key 标记冷却(默认 60s),轮转到下一个可用 key
|
||||||
|
- 全部 key 都在冷却时:使用最早冷却的那个 key 并等待(退化到单 key 重试)
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import time
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
DEFAULT_COOLDOWN = 60.0 # 单个 key 被限流后的冷却时间(秒)
|
||||||
|
|
||||||
|
|
||||||
|
class KeyRing:
|
||||||
|
"""多 key 轮换环。在 async 单线程事件循环下无需加锁。"""
|
||||||
|
|
||||||
|
def __init__(self, keys: list[str], cooldown_seconds: float = DEFAULT_COOLDOWN) -> None:
|
||||||
|
self._keys: list[str] = [k.strip() for k in keys if k and k.strip()]
|
||||||
|
self._cooldown = cooldown_seconds
|
||||||
|
# key → 冷却过期时间戳(时刻);不在表中表示可用
|
||||||
|
self._blocked_until: dict[str, float] = {}
|
||||||
|
self._index = 0 # 当前轮转位置
|
||||||
|
|
||||||
|
@property
|
||||||
|
def has_multiple(self) -> bool:
|
||||||
|
return len(self._keys) > 1
|
||||||
|
|
||||||
|
@property
|
||||||
|
def all_keys(self) -> list[str]:
|
||||||
|
return list(self._keys)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def active_key(self) -> str | None:
|
||||||
|
"""当前指向的 key(即使正在冷却也返回,用于上报)。"""
|
||||||
|
if not self._keys:
|
||||||
|
return None
|
||||||
|
return self._keys[self._index]
|
||||||
|
|
||||||
|
def get(self) -> str | None:
|
||||||
|
"""获取一个可用 key:优先选不在冷却中的;全部冷却则选最早过期的。"""
|
||||||
|
if not self._keys:
|
||||||
|
return None
|
||||||
|
if len(self._keys) == 1:
|
||||||
|
return self._keys[0]
|
||||||
|
|
||||||
|
now = time.monotonic()
|
||||||
|
n = len(self._keys)
|
||||||
|
# 从当前 index 开始找一圈,找一个可用的
|
||||||
|
for offset in range(n):
|
||||||
|
idx = (self._index + offset) % n
|
||||||
|
key = self._keys[idx]
|
||||||
|
expire = self._blocked_until.get(key, 0.0)
|
||||||
|
if now >= expire:
|
||||||
|
# 可用:把指针移到这里
|
||||||
|
self._index = idx
|
||||||
|
# 清理已过期的冷却记录
|
||||||
|
if key in self._blocked_until:
|
||||||
|
del self._blocked_until[key]
|
||||||
|
return key
|
||||||
|
|
||||||
|
# 全部在冷却中:选最早过期的那个,并等待到它过期
|
||||||
|
earliest_key = min(self._keys, key=lambda k: self._blocked_until.get(k, 0.0))
|
||||||
|
self._index = self._keys.index(earliest_key)
|
||||||
|
return earliest_key
|
||||||
|
|
||||||
|
def report_rate_limited(self, key: str | None = None) -> str | None:
|
||||||
|
"""上报某个 key 被限流(429)。默认是当前 key。返回切换后的新 key。"""
|
||||||
|
target = key or self.active_key
|
||||||
|
if target and len(self._keys) > 1:
|
||||||
|
until = time.monotonic() + self._cooldown
|
||||||
|
self._blocked_until[target] = until
|
||||||
|
logger.warning(
|
||||||
|
"bzzoiro key 被限流(429),冷却 %.0fs: %s", self._cooldown, _mask(target),
|
||||||
|
)
|
||||||
|
# 轮转到下一个(即使只有一个 key 也做一次 get,保持行为一致)
|
||||||
|
return self.get()
|
||||||
|
|
||||||
|
def wait_if_all_blocked(self) -> float:
|
||||||
|
"""如果所有 key 都在冷却中,返回需要等待的秒数;否则返回 0。"""
|
||||||
|
if len(self._keys) <= 1:
|
||||||
|
return 0.0
|
||||||
|
now = time.monotonic()
|
||||||
|
remaining = [self._blocked_until.get(k, 0.0) - now for k in self._keys]
|
||||||
|
if all(r > -0.001 for r in remaining) and any(r > 0.001 for r in remaining):
|
||||||
|
# 全部仍在冷却中
|
||||||
|
return max(remaining)
|
||||||
|
return 0.0
|
||||||
|
|
||||||
|
def stats(self) -> dict:
|
||||||
|
"""当前 key 环状态(用于管理后台展示)。"""
|
||||||
|
now = time.monotonic()
|
||||||
|
return {
|
||||||
|
"total": len(self._keys),
|
||||||
|
"keys": [
|
||||||
|
{
|
||||||
|
"masked": _mask(k),
|
||||||
|
"blocked_remaining": max(0.0, round(self._blocked_until.get(k, 0.0) - now, 1)),
|
||||||
|
}
|
||||||
|
for k in self._keys
|
||||||
|
],
|
||||||
|
"active_index": self._index,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _mask(key: str) -> str:
|
||||||
|
"""脱敏:只显示前 4 位和后 4 位。"""
|
||||||
|
if len(key) <= 10:
|
||||||
|
return key[:2] + "***"
|
||||||
|
return key[:4] + "..." + key[-4:]
|
||||||
|
|
||||||
|
|
||||||
|
# ── 全局单例(进程级,按 base URL 隔离) ──────────────────────────
|
||||||
|
_RINGS: dict[str, KeyRing] = {}
|
||||||
|
|
||||||
|
|
||||||
|
def parse_keys(value: str | None) -> list[str]:
|
||||||
|
"""解析 key 配置值:支持逗号、分号、换行分隔的多个 key。"""
|
||||||
|
if not value:
|
||||||
|
return []
|
||||||
|
# 统一替换分隔符为逗号后拆分
|
||||||
|
normalized = value.replace("\n", ",").replace(";", ",")
|
||||||
|
return [k.strip() for k in normalized.split(",") if k.strip()]
|
||||||
|
|
||||||
|
|
||||||
|
def get_key_ring(base: str, raw_keys: str | None, cooldown_seconds: float = DEFAULT_COOLDOWN) -> KeyRing:
|
||||||
|
"""获取(或创建)某 base URL 对应的 KeyRing。"""
|
||||||
|
key = base
|
||||||
|
ring = _RINGS.get(key)
|
||||||
|
parsed = parse_keys(raw_keys)
|
||||||
|
if ring is None:
|
||||||
|
ring = KeyRing(parsed, cooldown_seconds)
|
||||||
|
_RINGS[key] = ring
|
||||||
|
else:
|
||||||
|
# 热更新 key 列表(增删 key 无需重启)
|
||||||
|
if set(ring.all_keys) != set(parsed):
|
||||||
|
ring._keys = parsed
|
||||||
|
ring._blocked_until.clear()
|
||||||
|
ring._index = 0
|
||||||
|
ring._cooldown = cooldown_seconds
|
||||||
|
return ring
|
||||||
@@ -0,0 +1,183 @@
|
|||||||
|
"""KeyRing 多 key 轮换单元测试。"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import time
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from src.data.key_ring import KeyRing, parse_keys, get_key_ring
|
||||||
|
|
||||||
|
|
||||||
|
class TestParseKeys:
|
||||||
|
def test_single_key(self):
|
||||||
|
assert parse_keys("abc123") == ["abc123"]
|
||||||
|
|
||||||
|
def test_comma_separated(self):
|
||||||
|
assert parse_keys("k1, k2,k3") == ["k1", "k2", "k3"]
|
||||||
|
|
||||||
|
def test_semicolon_separated(self):
|
||||||
|
assert parse_keys("k1;k2;k3") == ["k1", "k2", "k3"]
|
||||||
|
|
||||||
|
def test_newline_separated(self):
|
||||||
|
assert parse_keys("k1\nk2\nk3") == ["k1", "k2", "k3"]
|
||||||
|
|
||||||
|
def test_mixed_separators(self):
|
||||||
|
assert parse_keys("k1, k2; k3\nk4") == ["k1", "k2", "k3", "k4"]
|
||||||
|
|
||||||
|
def test_empty(self):
|
||||||
|
assert parse_keys("") == []
|
||||||
|
assert parse_keys(None) == []
|
||||||
|
assert parse_keys(" , ; ") == []
|
||||||
|
|
||||||
|
def test_strips_whitespace(self):
|
||||||
|
assert parse_keys(" a , b ") == ["a", "b"]
|
||||||
|
|
||||||
|
|
||||||
|
class TestKeyRingSingleKey:
|
||||||
|
"""单 key 场景:行为与之前一致。"""
|
||||||
|
|
||||||
|
def test_get_returns_key(self):
|
||||||
|
ring = KeyRing(["only-key"])
|
||||||
|
assert ring.get() == "only-key"
|
||||||
|
assert ring.active_key == "only-key"
|
||||||
|
|
||||||
|
def test_no_rotation(self):
|
||||||
|
ring = KeyRing(["key"])
|
||||||
|
ring.report_rate_limited()
|
||||||
|
# 单 key 切换后仍是自己
|
||||||
|
assert ring.get() == "key"
|
||||||
|
|
||||||
|
def test_empty_keys(self):
|
||||||
|
ring = KeyRing([])
|
||||||
|
assert ring.get() is None
|
||||||
|
assert ring.active_key is None
|
||||||
|
|
||||||
|
def test_has_multiple_false(self):
|
||||||
|
ring = KeyRing(["key"])
|
||||||
|
assert ring.has_multiple is False
|
||||||
|
|
||||||
|
|
||||||
|
class TestKeyRingMultiKey:
|
||||||
|
"""多 key 场景:429 自动轮换。"""
|
||||||
|
|
||||||
|
def test_get_rounds_robin(self):
|
||||||
|
ring = KeyRing(["a", "b", "c"])
|
||||||
|
# 前三次 get 依次返回 a, b, c
|
||||||
|
assert ring.get() == "a"
|
||||||
|
assert ring.get() == "a" # 不报告限流时保持当前 key
|
||||||
|
# 手动推进:通过 report 后 get
|
||||||
|
ring.report_rate_limited("a")
|
||||||
|
# a 被冷却,下一个可用的是 b
|
||||||
|
assert ring.get() == "b"
|
||||||
|
|
||||||
|
def test_rate_limit_skips_key(self):
|
||||||
|
ring = KeyRing(["a", "b", "c"], cooldown_seconds=60.0)
|
||||||
|
key = ring.get()
|
||||||
|
assert key == "a"
|
||||||
|
new_key = ring.report_rate_limited("a")
|
||||||
|
assert new_key == "b"
|
||||||
|
# 再次 get 应继续是 b(可用)
|
||||||
|
assert ring.get() == "b"
|
||||||
|
|
||||||
|
def test_cycle_back_to_first(self):
|
||||||
|
ring = KeyRing(["a", "b"], cooldown_seconds=0.1)
|
||||||
|
ring.report_rate_limited("a")
|
||||||
|
# b 可用
|
||||||
|
assert ring.get() == "b"
|
||||||
|
ring.report_rate_limited("b")
|
||||||
|
# a 仍在冷却,b 也在冷却 → 选最早过期的(可能是 a)
|
||||||
|
key = ring.get()
|
||||||
|
assert key in ("a", "b")
|
||||||
|
|
||||||
|
def test_cooldown_expires(self):
|
||||||
|
ring = KeyRing(["a", "b"], cooldown_seconds=0.05)
|
||||||
|
ring.report_rate_limited("a")
|
||||||
|
assert ring.get() == "b"
|
||||||
|
# 等 a 的冷却过期
|
||||||
|
time.sleep(0.08)
|
||||||
|
# 现在 get 应该能找到可用的 key(b 或 a 都行,取决于指针)
|
||||||
|
key = ring.get()
|
||||||
|
assert key in ("a", "b")
|
||||||
|
|
||||||
|
def test_wait_if_all_blocked(self):
|
||||||
|
ring = KeyRing(["a", "b"], cooldown_seconds=1.0)
|
||||||
|
ring.report_rate_limited("a")
|
||||||
|
ring.report_rate_limited("b")
|
||||||
|
wait = ring.wait_if_all_blocked()
|
||||||
|
assert wait > 0 # 应返回正数等待时间
|
||||||
|
|
||||||
|
def test_wait_if_not_all_blocked(self):
|
||||||
|
ring = KeyRing(["a", "b"], cooldown_seconds=1.0)
|
||||||
|
ring.report_rate_limited("a")
|
||||||
|
# b 仍可用
|
||||||
|
assert ring.wait_if_all_blocked() == 0.0
|
||||||
|
|
||||||
|
def test_stats(self):
|
||||||
|
ring = KeyRing(["a" * 12, "b" * 12], cooldown_seconds=1.0)
|
||||||
|
ring.report_rate_limited("a" * 12)
|
||||||
|
st = ring.stats()
|
||||||
|
assert st["total"] == 2
|
||||||
|
assert st["keys"][0]["blocked_remaining"] > 0
|
||||||
|
assert st["keys"][1]["blocked_remaining"] == 0
|
||||||
|
# 脱敏
|
||||||
|
assert "***" in st["keys"][0]["masked"] or "..." in st["keys"][0]["masked"]
|
||||||
|
|
||||||
|
def test_all_keys_property(self):
|
||||||
|
ring = KeyRing(["x", "y"])
|
||||||
|
assert ring.all_keys == ["x", "y"]
|
||||||
|
|
||||||
|
|
||||||
|
class TestKeyRingHotUpdate:
|
||||||
|
"""热更新 key 列表。"""
|
||||||
|
|
||||||
|
def test_setter_clears_state(self):
|
||||||
|
ring = KeyRing(["a", "b"])
|
||||||
|
ring.report_rate_limited("a")
|
||||||
|
assert ring.get() == "b"
|
||||||
|
# 更新 key 列表
|
||||||
|
ring._keys = ["c", "d"]
|
||||||
|
ring._blocked_until.clear()
|
||||||
|
ring._index = 0
|
||||||
|
assert ring.get() == "c"
|
||||||
|
|
||||||
|
|
||||||
|
class TestGetKeyRing:
|
||||||
|
def test_singleton_per_base(self):
|
||||||
|
r1 = get_key_ring("https://api.test.com", "k1, k2")
|
||||||
|
r2 = get_key_ring("https://api.test.com", "k1, k2")
|
||||||
|
assert r1 is r2
|
||||||
|
|
||||||
|
def test_different_base_isolated(self):
|
||||||
|
r1 = get_key_ring("https://a.com", "k1")
|
||||||
|
r2 = get_key_ring("https://b.com", "k2")
|
||||||
|
assert r1 is not r2
|
||||||
|
assert r1.get() == "k1"
|
||||||
|
assert r2.get() == "k2"
|
||||||
|
|
||||||
|
def test_hot_update_keys(self):
|
||||||
|
ring = get_key_ring("https://hot.com", "k1, k2")
|
||||||
|
assert set(ring.all_keys) == {"k1", "k2"}
|
||||||
|
# 更新(同 base 会命中缓存,触发热更新)
|
||||||
|
ring2 = get_key_ring("https://hot.com", "k3, k4")
|
||||||
|
assert ring2 is ring
|
||||||
|
assert set(ring.all_keys) == {"k3", "k4"}
|
||||||
|
|
||||||
|
|
||||||
|
class TestKeyRingAsyncSafety:
|
||||||
|
"""async 并发场景下单 event loop 不需要锁,但验证交替 429 不会死锁。"""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_concurrent_get(self):
|
||||||
|
ring = KeyRing(["a", "b", "c"])
|
||||||
|
|
||||||
|
async def worker():
|
||||||
|
for _ in range(20):
|
||||||
|
key = ring.get()
|
||||||
|
assert key in ("a", "b", "c")
|
||||||
|
# 模拟偶发 429
|
||||||
|
if hash(key) % 3 == 0:
|
||||||
|
ring.report_rate_limited(key)
|
||||||
|
await asyncio.sleep(0)
|
||||||
|
|
||||||
|
await asyncio.gather(*(worker() for _ in range(5)))
|
||||||
Reference in New Issue
Block a user