diff --git a/app/api/v1/admin.py b/app/api/v1/admin.py
index 7cfdded..7ea3bba 100644
--- a/app/api/v1/admin.py
+++ b/app/api/v1/admin.py
@@ -181,16 +181,20 @@ async def enable_nsfw_api(data: dict):
# 去重并保持顺序
unique_tokens = list(dict.fromkeys(tokens))
- # 限制最大数量（防止误操作）
+ # 限制最大数量（超出时截取前 N 个）
max_tokens = get_config("performance.nsfw_max_tokens", 1000)
try:
max_tokens = int(max_tokens)
except Exception:
max_tokens = 1000
+
+ truncated = False
+ original_count = len(unique_tokens)
if len(unique_tokens) > max_tokens:
- raise HTTPException(
- status_code=400,
- detail=f"Too many tokens: {len(unique_tokens)} > {max_tokens}",
+ unique_tokens = unique_tokens[:max_tokens]
+ truncated = True
+ logger.warning(
+ f"NSFW enable: truncated from {original_count} to {max_tokens} tokens"
)
# 批量执行配置
@@ -200,6 +204,9 @@ async def enable_nsfw_api(data: dict):
# 定义 worker
async def _enable(token: str):
result = await nsfw_service.enable(token)
+ # 成功后添加 nsfw tag
+ if result.success:
+ await mgr.add_tag(token, "nsfw")
return {
"success": result.success,
"http_status": result.http_status,
@@ -227,7 +234,7 @@ async def enable_nsfw_api(data: dict):
fail_count += 1
results[masked] = res.get("data") or {"error": res.get("error")}
- return {
+ response = {
"status": "success",
"summary": {
"total": len(unique_tokens),
@@ -237,6 +244,12 @@ async def enable_nsfw_api(data: dict):
"results": results,
}
+ # 添加截断提示
+ if truncated:
+ response["warning"] = f"数量超出限制，仅处理前 {max_tokens} 个（共 {original_count} 个）"
+
+ return response
+
except HTTPException:
raise
except Exception as e:
diff --git a/app/services/grok/nsfw.py b/app/services/grok/nsfw.py
index 507b2af..2f6a625 100644
--- a/app/services/grok/nsfw.py
+++ b/app/services/grok/nsfw.py
@@ -97,8 +97,11 @@ class NSFWService:
grpc_status = get_grpc_status(trailers)
+ # HTTP 200 且无 grpc-status（空响应）或 grpc-status=0 都算成功
+ success = grpc_status.code == -1 or grpc_status.ok
+
return NSFWResult(
- success=grpc_status.ok,
+ success=success,
http_status=response.status_code,
grpc_status=grpc_status.code,
grpc_message=grpc_status.message or None,
diff --git a/app/services/token/manager.py b/app/services/token/manager.py
index bc7df05..4e8dd75 100644
--- a/app/services/token/manager.py
+++ b/app/services/token/manager.py
@@ -6,7 +6,7 @@ from datetime import datetime
from typing import Dict, List, Optional
from app.core.logger import logger
-from app.services.token.models import TokenInfo, EffortType, TokenPoolStats, FAIL_THRESHOLD, TokenStatus
+from app.services.token.models import TokenInfo, EffortType, FAIL_THRESHOLD, TokenStatus
from app.core.storage import get_storage
from app.core.config import get_config
from app.services.token.pool import TokenPool
@@ -19,10 +19,10 @@ REFRESH_CONCURRENCY = 5
class TokenManager:
"""管理 Token 的增删改查和配额同步"""
-
+
_instance: Optional["TokenManager"] = None
_lock = asyncio.Lock()
-
+
def __init__(self):
self.pools: Dict[str, TokenPool] = {}
self.initialized = False
@@ -31,7 +31,7 @@ class TokenManager:
self._save_task: Optional[asyncio.Task] = None
self._save_delay = 0.5
self._last_reload_at = 0.0
-
+
@classmethod
async def get_instance(cls) -> "TokenManager":
"""获取单例实例"""
@@ -48,16 +48,19 @@ class TokenManager:
try:
storage = get_storage()
data = await storage.load_tokens()
-
+
# 如果后端返回 None 或空数据，尝试从本地 data/token.json 初始化后端
if not data:
from app.core.storage import LocalStorage
+
local_storage = LocalStorage()
local_data = await local_storage.load_tokens()
if local_data:
data = local_data
await storage.save_tokens(local_data)
- logger.info(f"Initialized remote token storage ({storage.__class__.__name__}) with local tokens.")
+ logger.info(
+ f"Initialized remote token storage ({storage.__class__.__name__}) with local tokens."
+ )
else:
data = {}
@@ -69,20 +72,26 @@ class TokenManager:
# 统一存储裸 token
if isinstance(token_data, dict):
raw_token = token_data.get("token")
- if isinstance(raw_token, str) and raw_token.startswith("sso="):
+ if isinstance(raw_token, str) and raw_token.startswith(
+ "sso="
+ ):
token_data["token"] = raw_token[4:]
token_info = TokenInfo(**token_data)
pool.add(token_info)
except Exception as e:
- logger.warning(f"Failed to load token in pool '{pool_name}': {e}")
+ logger.warning(
+ f"Failed to load token in pool '{pool_name}': {e}"
+ )
continue
pool._rebuild_index()
self.pools[pool_name] = pool
-
+
self.initialized = True
self._last_reload_at = time.monotonic()
total = sum(p.count() for p in self.pools.values())
- logger.info(f"TokenManager initialized: {len(self.pools)} pools with {total} tokens")
+ logger.info(
+ f"TokenManager initialized: {len(self.pools)} pools with {total} tokens"
+ )
except Exception as e:
logger.error(f"Failed to initialize TokenManager: {e}")
self.pools = {}
@@ -113,10 +122,8 @@ class TokenManager:
try:
data = {}
for pool_name, pool in self.pools.items():
- data[pool_name] = [
- info.model_dump() for info in pool.list()
- ]
-
+ data[pool_name] = [info.model_dump() for info in pool.list()]
+
storage = get_storage()
async with storage.acquire_lock("tokens_save", timeout=10):
await storage.save_tokens(data)
@@ -153,13 +160,14 @@ class TokenManager:
self._save_task = None
if self._dirty:
self._schedule_save()
+
def get_token(self, pool_name: str = "ssoBasic") -> Optional[str]:
"""
获取可用 Token
-
+
Args:
pool_name: Token 池名称
-
+
Returns:
Token 字符串或 None
"""
@@ -167,73 +175,77 @@ class TokenManager:
if not pool:
logger.warning(f"Pool '{pool_name}' not found")
return None
-
+
token_info = pool.select()
if not token_info:
logger.warning(f"No available token in pool '{pool_name}'")
return None
-
+
token = token_info.token
if token.startswith("sso="):
return token[4:]
return token
- async def consume(self, token_str: str, effort: EffortType = EffortType.LOW) -> bool:
+ async def consume(
+ self, token_str: str, effort: EffortType = EffortType.LOW
+ ) -> bool:
"""
消耗配额（本地预估）
-
+
Args:
token_str: Token 字符串
effort: 消耗力度
-
+
Returns:
是否成功
"""
raw_token = token_str.replace("sso=", "")
-
+
for pool in self.pools.values():
token = pool.get(raw_token)
if token:
consumed = token.consume(effort)
- logger.debug(f"Token {raw_token[:10]}...: consumed {consumed} quota, use_count={token.use_count}")
+ logger.debug(
+ f"Token {raw_token[:10]}...: consumed {consumed} quota, use_count={token.use_count}"
+ )
self._schedule_save()
return True
-
+
logger.warning(f"Token {raw_token[:10]}...: not found for consumption")
return False
async def sync_usage(
- self,
- token_str: str,
- model_name: str,
+ self,
+ token_str: str,
+ model_name: str,
fallback_effort: EffortType = EffortType.LOW,
consume_on_fail: bool = True,
- is_usage: bool = True
+ is_usage: bool = True,
) -> bool:
"""
同步 Token 用量
-
+
优先从 API 获取最新配额，失败则降级到本地预估
-
+
Args:
token_str: Token 字符串（可带 sso= 前缀）
model_name: 模型名称（用于 API 查询）
fallback_effort: 降级时的消耗力度
consume_on_fail: 失败时是否降级扣费
is_usage: 是否记录为一次使用（影响 use_count）
-
+
Returns:
是否成功
"""
raw_token = token_str.replace("sso=", "")
-
+
# 查找 Token 对象
target_token: Optional[TokenInfo] = None
for pool in self.pools.values():
target_token = pool.get(raw_token)
if target_token:
break
-
+
if not target_token:
logger.warning(f"Token {raw_token[:10]}...: not found for sync")
return False
@@ -241,51 +253,57 @@ class TokenManager:
# 尝试 API 同步
try:
from app.services.grok.usage import UsageService
-
+
usage_service = UsageService()
result = await usage_service.get(token_str, model_name=model_name)
-
+
if result and "remainingTokens" in result:
old_quota = target_token.quota
new_quota = result["remainingTokens"]
-
+
target_token.update_quota(new_quota)
target_token.record_success(is_usage=is_usage)
-
+
consumed = max(0, old_quota - new_quota)
logger.info(
f"Token {raw_token[:10]}...: synced quota "
f"{old_quota} -> {new_quota} (consumed: {consumed}, use_count: {target_token.use_count})"
)
-
+
self._schedule_save()
return True
-
+
except Exception as e:
- logger.warning(f"Token {raw_token[:10]}...: API sync failed, fallback to local ({e})")
-
+ logger.warning(
+ f"Token {raw_token[:10]}...: API sync failed, fallback to local ({e})"
+ )
+
# 降级：本地预估扣费
if consume_on_fail:
logger.debug(f"Token {raw_token[:10]}...: using local consumption")
return await self.consume(token_str, fallback_effort)
else:
- logger.debug(f"Token {raw_token[:10]}...: sync failed, skipping local consumption")
+ logger.debug(
+ f"Token {raw_token[:10]}...: sync failed, skipping local consumption"
+ )
return False
- async def record_fail(self, token_str: str, status_code: int = 401, reason: str = "") -> bool:
+ async def record_fail(
+ self, token_str: str, status_code: int = 401, reason: str = ""
+ ) -> bool:
"""
记录 Token 失败
-
+
Args:
token_str: Token 字符串
status_code: HTTP 状态码
reason: 失败原因
-
+
Returns:
是否成功
"""
raw_token = token_str.replace("sso=", "")
-
+
for pool in self.pools.values():
token = pool.get(raw_token)
if token:
@@ -301,7 +319,7 @@ class TokenManager:
)
self._schedule_save()
return True
-
+
logger.warning(f"Token {raw_token[:10]}...: not found for failure record")
return False
@@ -310,25 +328,25 @@ class TokenManager:
async def add(self, token: str, pool_name: str = "ssoBasic") -> bool:
"""
添加 Token
-
+
Args:
token: Token 字符串（不含 sso= 前缀）
pool_name: 池名称
-
+
Returns:
是否成功
"""
if pool_name not in self.pools:
self.pools[pool_name] = TokenPool(pool_name)
logger.info(f"Pool '{pool_name}': created")
-
+
pool = self.pools[pool_name]
-
+
token = token[4:] if token.startswith("sso=") else token
if pool.get(token):
logger.warning(f"Pool '{pool_name}': token already exists")
return False
-
+
pool.add(TokenInfo(token=token))
await self._save()
logger.info(f"Pool '{pool_name}': token added")
@@ -345,13 +363,57 @@ class TokenManager:
return True
return False
+ async def add_tag(self, token: str, tag: str) -> bool:
+ """
+ 给 Token 添加标签
+
+ Args:
+ token: Token 字符串
+ tag: 标签名称
+
+ Returns:
+ 是否成功
+ """
+ raw_token = token[4:] if token.startswith("sso=") else token
+ for pool in self.pools.values():
+ info = pool.get(raw_token)
+ if info:
+ if tag not in info.tags:
+ info.tags.append(tag)
+ self._schedule_save()
+ logger.debug(f"Token {raw_token[:10]}...: added tag '{tag}'")
+ return True
+ return False
+
+ async def remove_tag(self, token: str, tag: str) -> bool:
+ """
+ 移除 Token 标签
+
+ Args:
+ token: Token 字符串
+ tag: 标签名称
+
+ Returns:
+ 是否成功
+ """
+ raw_token = token[4:] if token.startswith("sso=") else token
+ for pool in self.pools.values():
+ info = pool.get(raw_token)
+ if info:
+ if tag in info.tags:
+ info.tags.remove(tag)
+ self._schedule_save()
+ logger.debug(f"Token {raw_token[:10]}...: removed tag '{tag}'")
+ return True
+ return False
+
async def remove(self, token: str) -> bool:
"""
删除 Token
-
+
Args:
token: Token 字符串
-
+
Returns:
是否成功
"""
@@ -360,8 +422,8 @@ class TokenManager:
await self._save()
logger.info(f"Pool '{pool_name}': token removed")
return True
-
- logger.warning(f"Token not found for removal")
+
+ logger.warning("Token not found for removal")
return False
async def reset_all(self):
@@ -371,22 +433,22 @@ class TokenManager:
for token in pool:
token.reset()
count += 1
-
+
await self._save()
logger.info(f"Reset all: {count} tokens updated")
async def reset_token(self, token_str: str) -> bool:
"""
重置单个 Token
-
+
Args:
token_str: Token 字符串
-
+
Returns:
是否成功
"""
raw_token = token_str.replace("sso=", "")
-
+
for pool in self.pools.values():
token = pool.get(raw_token)
if token:
@@ -394,7 +456,7 @@ class TokenManager:
await self._save()
logger.info(f"Token {raw_token[:10]}...: reset completed")
return True
-
+
logger.warning(f"Token {raw_token[:10]}...: not found for reset")
return False
@@ -405,14 +467,14 @@ class TokenManager:
pool_stats = pool.get_stats()
stats[name] = pool_stats.model_dump()
return stats
-
+
def get_pool_tokens(self, pool_name: str = "ssoBasic") -> List[TokenInfo]:
"""
获取指定池的所有 Token
-
+
Args:
pool_name: 池名称
-
+
Returns:
Token 列表
"""
@@ -420,72 +482,72 @@ class TokenManager:
if not pool:
return []
return pool.list()
-
+
async def refresh_cooling_tokens(self) -> Dict[str, int]:
"""
批量刷新 cooling 状态的 Token 配额
-
+
Returns:
{"checked": int, "refreshed": int, "recovered": int, "expired": int}
"""
from app.services.grok.usage import UsageService
-
+
# 收集需要刷新的 token
to_refresh: List[TokenInfo] = []
for pool in self.pools.values():
for token in pool:
if token.need_refresh(REFRESH_INTERVAL_HOURS):
to_refresh.append(token)
-
+
if not to_refresh:
logger.debug("Refresh check: no tokens need refresh")
return {"checked": 0, "refreshed": 0, "recovered": 0, "expired": 0}
-
+
logger.info(f"Refresh check: found {len(to_refresh)} cooling tokens to refresh")
-
+
# 批量并发刷新
semaphore = asyncio.Semaphore(REFRESH_CONCURRENCY)
usage_service = UsageService()
refreshed = 0
recovered = 0
expired = 0
-
+
async def _refresh_one(token_info: TokenInfo) -> dict:
"""刷新单个 token"""
async with semaphore:
token_str = token_info.token
if token_str.startswith("sso="):
token_str = token_str[4:]
-
+
# 重试逻辑：最多 2 次重试
for retry in range(3): # 0, 1, 2
try:
result = await usage_service.get(token_str)
-
+
if result and "remainingTokens" in result:
new_quota = result["remainingTokens"]
old_quota = token_info.quota
old_status = token_info.status
-
+
token_info.update_quota(new_quota)
token_info.mark_synced()
-
+
logger.info(
f"Token {token_info.token[:10]}...: refreshed "
f"{old_quota} -> {new_quota}, status: {old_status} -> {token_info.status}"
)
-
+
return {
"recovered": new_quota > 0 and old_quota == 0,
- "expired": False
+ "expired": False,
}
-
+
token_info.mark_synced()
return {"recovered": False, "expired": False}
-
+
except Exception as e:
error_str = str(e)
-
+
# 检查是否为 401 错误
if "401" in error_str or "Unauthorized" in error_str:
if retry < 2:
@@ -510,35 +572,35 @@ class TokenManager:
)
token_info.mark_synced()
return {"recovered": False, "expired": False}
-
+
token_info.mark_synced()
return {"recovered": False, "expired": False}
-
+
# 批量处理
for i in range(0, len(to_refresh), REFRESH_BATCH_SIZE):
- batch = to_refresh[i:i + REFRESH_BATCH_SIZE]
+ batch = to_refresh[i : i + REFRESH_BATCH_SIZE]
results = await asyncio.gather(*[_refresh_one(t) for t in batch])
refreshed += len(batch)
recovered += sum(r["recovered"] for r in results)
expired += sum(r["expired"] for r in results)
-
+
# 批次间延迟
if i + REFRESH_BATCH_SIZE < len(to_refresh):
await asyncio.sleep(1)
-
+
await self._save()
-
+
logger.info(
f"Refresh completed: "
f"checked={len(to_refresh)}, refreshed={refreshed}, "
f"recovered={recovered}, expired={expired}"
)
-
+
return {
"checked": len(to_refresh),
"refreshed": refreshed,
"recovered": recovered,
- "expired": expired
+ "expired": expired,
}
diff --git a/app/static/config/config.html b/app/static/config/config.html
index 55c5575..b8b0f68 100644
--- a/app/static/config/config.html
+++ b/app/static/config/config.html
@@ -50,7 +50,7 @@
-
+