okooooooo/tavily-key-generator
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove legacy proxy implementation snapshot

Browse Source
This commit is contained in:
skernelx committed 2026-03-17 11:23:32 +08:00
1 parent 1c191a098a
commit 8eaab82129
12 files changed
+12 -3033
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

README.md
+1 -1
View File
@@ -102,7 +102,7 @@ start_auto.bat
- 做真实可用性验证
- 提供聚合 API 上游能力
`proxy/` 目录目前仅保留给旧部署或迁移场景，不再作为主推荐入口。
`proxy/` 目录现在只保留一份迁移说明 README，不再包含本地可运行实现。
## MySearch 的关系
README_EN.md
+2 -2
View File
@@ -104,8 +104,8 @@ This repository stays focused on:
- verifying them with live requests
- serving as the upstream aggregation source
The local `proxy/` directory is now kept only for legacy deployments or
migration scenarios, not as the primary recommended path.
The local `proxy/` directory now keeps only a migration README and no longer
ships the runnable implementation.
## Relationship to MySearch
proxy/.dockerignore
-7
View File
@@ -1,7 +0,0 @@
__pycache__/
*.pyc
*.pyo
*.sqlite
*.sqlite3
data/
.env
proxy/.env.example
-1
View File
@@ -1 +0,0 @@
ADMIN_PASSWORD=change-me
proxy/Dockerfile
-7
View File
@@ -1,7 +0,0 @@
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 9874
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "9874"]
proxy/README.md
+9 -37
View File
@@ -1,6 +1,9 @@
# Legacy Proxy Snapshot
# Legacy Proxy Note
这个 `proxy/` 目录现在是保留在 `tavily-key-generator` 里的历史兼容副本。
`proxy/` 的实现已经迁移出这个仓库。
当前目录刻意只保留这份 `README.md`，用于说明迁移关系，不再包含
`server.py`、`docker-compose.yml`、模板或运行所需文件。
如果你要部署正式对外可用的统一搜索控制台、MCP、Skill 和 Social / X
路由，请直接使用独立仓库：
@@ -53,7 +56,7 @@
- Firecrawl 上传会进入 Firecrawl 池
- 服务端不需要再靠 key 前缀猜测服务
## 新部署建议
## 迁移建议
推荐直接部署 `MySearch-Proxy`：
@@ -65,36 +68,16 @@ docker pull skernelx/mysearch-proxy:latest
- [skernelx/MySearch-Proxy](https://github.com/skernelx/MySearch-Proxy)
## 旧部署如何迁移
如果你已经在跑这个目录下的旧 `proxy/`，更稳的迁移方式是：
如果你之前已经从这个仓库的旧 `proxy/` 部署过服务，更稳的迁移方式是：
1. 保留原来的数据卷目录
2. 改用 `skernelx/mysearch-proxy:latest`
3. 继续挂载到 `/app/data`
4. 使用新的 `proxy/.env` 或容器环境变量补齐配置
一个通用迁移示例：
```bash
docker pull skernelx/mysearch-proxy:latest
docker rm -f mysearch-proxy 2>/dev/null || true
docker run -d \
--name mysearch-proxy \
--restart unless-stopped \
-p 9874:9874 \
-e ADMIN_PASSWORD=your-admin-password \
-v /your/data/path:/app/data \
skernelx/mysearch-proxy:latest
```
只要保留原数据卷，已有 Key、Token 和控制台里的数据就可以继续沿用。
4. 在新仓库里按最新 README 补齐环境变量
## 兼容 API 摘要
当前这套能力对外仍然兼容这些重点接口：
`MySearch-Proxy` 仍然兼容这些重点接口：
- Tavily
- `POST /api/search`
@@ -110,14 +93,3 @@ docker run -d \
- `Authorization: Bearer YOUR_TOKEN`
- body 里传 `api_key`
## 如果你只是要保留本地旧目录
这个目录还可以继续作为历史兼容副本使用：
```bash
cd proxy
docker compose up -d
```
但它不再是对外发布时的主推荐入口。
proxy/database.py
-428
View File
@@ -1,428 +0,0 @@
"""
SQLite 数据库管理
"""
import os
import random
import re
import sqlite3
import string
from datetime import datetime, timezone
DB_PATH = os.path.join(os.path.dirname(__file__), "data", "proxy.db")
SUPPORTED_SERVICES = ("tavily", "firecrawl")
TOKEN_PREFIX = {
"tavily": "tvly-",
"firecrawl": "fctk-",
}
KEY_PATTERNS = {
"tavily": r"(tvly-[A-Za-z0-9\-_]{20,})",
"firecrawl": r"(fc-[A-Za-z0-9\-_]{20,})",
}
KEY_USAGE_COLUMNS = {
"usage_key_used": "INTEGER",
"usage_key_limit": "INTEGER",
"usage_key_remaining": "INTEGER",
"usage_account_plan": "TEXT DEFAULT ''",
"usage_account_used": "INTEGER",
"usage_account_limit": "INTEGER",
"usage_account_remaining": "INTEGER",
"usage_synced_at": "TEXT",
"usage_sync_error": "TEXT DEFAULT ''",
}
def normalize_service(service):
service = (service or "tavily").strip().lower()
if service not in SUPPORTED_SERVICES:
raise ValueError(f"unsupported service: {service}")
return service
def get_conn():
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
return conn
def init_db():
conn = get_conn()
conn.executescript("""
CREATE TABLE IF NOT EXISTS api_keys (
id INTEGER PRIMARY KEY,
service TEXT NOT NULL DEFAULT 'tavily',
key TEXT UNIQUE NOT NULL,
email TEXT,
active INTEGER DEFAULT 1,
total_used INTEGER DEFAULT 0,
total_failed INTEGER DEFAULT 0,
consecutive_fails INTEGER DEFAULT 0,
last_used_at TEXT,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS tokens (
id INTEGER PRIMARY KEY,
service TEXT NOT NULL DEFAULT 'tavily',
token TEXT UNIQUE NOT NULL,
name TEXT DEFAULT '',
hourly_limit INTEGER DEFAULT 0,
daily_limit INTEGER DEFAULT 0,
monthly_limit INTEGER DEFAULT 0,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS usage_logs (
id INTEGER PRIMARY KEY,
service TEXT NOT NULL DEFAULT 'tavily',
token_id INTEGER,
api_key_id INTEGER,
endpoint TEXT,
success INTEGER,
latency_ms INTEGER,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_usage_created ON usage_logs(created_at);
CREATE INDEX IF NOT EXISTS idx_usage_token ON usage_logs(token_id);
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
""")
_ensure_service_columns(conn)
_ensure_usage_columns(conn)
conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_service_created ON usage_logs(service, created_at)")
conn.commit()
conn.close()
def _table_columns(conn, table_name):
rows = conn.execute(f"PRAGMA table_info({table_name})").fetchall()
return {row["name"] for row in rows}
def _ensure_service_columns(conn):
service_columns = {
"api_keys": "TEXT NOT NULL DEFAULT 'tavily'",
"tokens": "TEXT NOT NULL DEFAULT 'tavily'",
"usage_logs": "TEXT NOT NULL DEFAULT 'tavily'",
}
for table_name, definition in service_columns.items():
existing = _table_columns(conn, table_name)
if "service" not in existing:
conn.execute(f"ALTER TABLE {table_name} ADD COLUMN service {definition}")
def _ensure_usage_columns(conn):
existing = _table_columns(conn, "api_keys")
for name, definition in KEY_USAGE_COLUMNS.items():
if name not in existing:
conn.execute(f"ALTER TABLE api_keys ADD COLUMN {name} {definition}")
def _service_where(service):
if not service:
return "", []
return " WHERE service = ?", [normalize_service(service)]
def _query_all(conn, table_name, service=None):
where_sql, params = _service_where(service)
return conn.execute(f"SELECT * FROM {table_name}{where_sql} ORDER BY id", params).fetchall()
# ═══ Settings ═══
def get_setting(key, default=None):
conn = get_conn()
try:
row = conn.execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()
return row["value"] if row else default
finally:
conn.close()
def set_setting(key, value):
conn = get_conn()
try:
conn.execute("INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)", (key, value))
conn.commit()
finally:
conn.close()
# ═══ API Keys ═══
def add_key(key, email="", service="tavily"):
service = normalize_service(service)
conn = get_conn()
try:
conn.execute(
"INSERT OR IGNORE INTO api_keys (service, key, email) VALUES (?, ?, ?)",
(service, key, email),
)
conn.commit()
return conn.execute("SELECT * FROM api_keys WHERE key = ?", (key,)).fetchone()
finally:
conn.close()
def get_all_keys(service=None):
conn = get_conn()
try:
return _query_all(conn, "api_keys", service)
finally:
conn.close()
def get_key_by_id(key_id):
conn = get_conn()
try:
return conn.execute("SELECT * FROM api_keys WHERE id = ?", (key_id,)).fetchone()
finally:
conn.close()
def get_active_keys(service=None):
conn = get_conn()
try:
where_sql, params = _service_where(service)
sql = f"SELECT * FROM api_keys{where_sql}"
if where_sql:
sql += " AND active = 1"
else:
sql += " WHERE active = 1"
sql += " ORDER BY id"
return conn.execute(sql, params).fetchall()
finally:
conn.close()
def update_key_usage(key_id, success):
conn = get_conn()
try:
now = datetime.now(timezone.utc).isoformat()
if success:
conn.execute(
"UPDATE api_keys SET total_used = total_used + 1, consecutive_fails = 0, last_used_at = ? WHERE id = ?",
(now, key_id),
)
else:
conn.execute(
"UPDATE api_keys SET total_failed = total_failed + 1, consecutive_fails = consecutive_fails + 1, last_used_at = ? WHERE id = ?",
(now, key_id),
)
row = conn.execute("SELECT consecutive_fails FROM api_keys WHERE id = ?", (key_id,)).fetchone()
if row and row["consecutive_fails"] >= 3:
conn.execute("UPDATE api_keys SET active = 0 WHERE id = ?", (key_id,))
conn.commit()
finally:
conn.close()
def toggle_key(key_id, active):
conn = get_conn()
try:
conn.execute("UPDATE api_keys SET active = ?, consecutive_fails = 0 WHERE id = ?", (active, key_id))
conn.commit()
finally:
conn.close()
def delete_key(key_id):
conn = get_conn()
try:
conn.execute("DELETE FROM api_keys WHERE id = ?", (key_id,))
conn.commit()
finally:
conn.close()
def import_keys_from_text(text, service="tavily"):
"""从批量文本导入不同服务的 key。"""
service = normalize_service(service)
pattern = KEY_PATTERNS[service]
count = 0
for line in text.strip().splitlines():
line = line.strip()
if not line:
continue
match = re.search(pattern, line)
if not match:
continue
key = match.group(1)
parts = line.split(",")
email = parts[0].strip() if len(parts) >= 3 else ""
add_key(key, email, service=service)
count += 1
return count
def update_key_remote_usage(
key_id,
*,
key_used=None,
key_limit=None,
key_remaining=None,
account_plan="",
account_used=None,
account_limit=None,
account_remaining=None,
synced_at=None,
):
conn = get_conn()
try:
conn.execute(
"""
UPDATE api_keys
SET usage_key_used = ?,
usage_key_limit = ?,
usage_key_remaining = ?,
usage_account_plan = ?,
usage_account_used = ?,
usage_account_limit = ?,
usage_account_remaining = ?,
usage_synced_at = ?,
usage_sync_error = ''
WHERE id = ?
""",
(
key_used,
key_limit,
key_remaining,
account_plan or "",
account_used,
account_limit,
account_remaining,
synced_at or datetime.now(timezone.utc).isoformat(),
key_id,
),
)
conn.commit()
finally:
conn.close()
def update_key_remote_usage_error(key_id, error_message):
conn = get_conn()
try:
conn.execute(
"UPDATE api_keys SET usage_sync_error = ? WHERE id = ?",
((error_message or "").strip(), key_id),
)
conn.commit()
finally:
conn.close()
# ═══ Tokens ═══
def create_token(name="", service="tavily"):
service = normalize_service(service)
token = TOKEN_PREFIX[service] + "".join(random.choices(string.ascii_letters + string.digits, k=32))
conn = get_conn()
try:
conn.execute(
"INSERT INTO tokens (service, token, name) VALUES (?, ?, ?)",
(service, token, name),
)
conn.commit()
return conn.execute("SELECT * FROM tokens WHERE token = ?", (token,)).fetchone()
finally:
conn.close()
def get_all_tokens(service=None):
conn = get_conn()
try:
return _query_all(conn, "tokens", service)
finally:
conn.close()
def get_token_by_value(token_value):
conn = get_conn()
try:
return conn.execute("SELECT * FROM tokens WHERE token = ?", (token_value,)).fetchone()
finally:
conn.close()
def delete_token(token_id):
conn = get_conn()
try:
conn.execute("DELETE FROM tokens WHERE id = ?", (token_id,))
conn.commit()
finally:
conn.close()
# ═══ Usage Logs ═══
def log_usage(token_id, api_key_id, endpoint, success, latency_ms, service="tavily"):
service = normalize_service(service)
conn = get_conn()
try:
conn.execute(
"""
INSERT INTO usage_logs (service, token_id, api_key_id, endpoint, success, latency_ms)
VALUES (?, ?, ?, ?, ?, ?)
""",
(service, token_id, api_key_id, endpoint, success, latency_ms),
)
conn.commit()
finally:
conn.close()
def get_usage_stats(token_id=None, service=None):
"""获取用量统计。"""
conn = get_conn()
try:
now = datetime.now(timezone.utc)
today = now.strftime("%Y-%m-%d")
month = now.strftime("%Y-%m")
hour_ago = now.replace(minute=0, second=0, microsecond=0).isoformat()
filters = []
filter_params = []
if service:
filters.append("service = ?")
filter_params.append(normalize_service(service))
if token_id is not None:
filters.append("token_id = ?")
filter_params.append(token_id)
def count(condition, extra_params=None):
where_parts = [condition] + filters
sql = "SELECT COUNT(*) as c FROM usage_logs WHERE " + " AND ".join(where_parts)
params = list(extra_params or []) + filter_params
row = conn.execute(sql, params).fetchone()
return row["c"]
return {
"today_success": count("success = 1 AND created_at >= ?", [today]),
"today_failed": count("success = 0 AND created_at >= ?", [today]),
"month_success": count("success = 1 AND created_at >= ?", [month]),
"hour_count": count("created_at >= ?", [hour_ago]),
"today_count": count("created_at >= ?", [today]),
"month_count": count("created_at >= ?", [month]),
}
finally:
conn.close()
def check_quota(token_id, hourly_limit, daily_limit, monthly_limit, service=None):
"""检查 token 配额是否超限，返回 (ok, reason)。"""
stats = get_usage_stats(token_id=token_id, service=service)
if hourly_limit and stats["hour_count"] >= hourly_limit:
return False, "hourly quota exceeded"
if daily_limit and stats["today_count"] >= daily_limit:
return False, "daily quota exceeded"
if monthly_limit and stats["month_count"] >= monthly_limit:
return False, "monthly quota exceeded"
return True, ""
proxy/docker-compose.yml
-10
View File
@@ -1,10 +0,0 @@
services:
tavily-proxy:
build: .
ports:
- "9874:9874"
environment:
- ADMIN_PASSWORD=change-me
volumes:
- ./data:/app/data
restart: unless-stopped
proxy/key_pool.py
-53
View File
@@ -1,53 +0,0 @@
"""
按服务维度管理 API Key 轮询池
"""
import threading
from database import SUPPORTED_SERVICES, get_active_keys, normalize_service, update_key_usage
class ServiceKeyPool:
def __init__(self):
self._lock = threading.Lock()
self._keys = {service: [] for service in SUPPORTED_SERVICES}
self._indexes = {service: 0 for service in SUPPORTED_SERVICES}
self._initialized = set()
def reload(self, service=None):
services = [normalize_service(service)] if service else list(SUPPORTED_SERVICES)
with self._lock:
for item in services:
self._keys[item] = [dict(row) for row in get_active_keys(item)]
if self._indexes[item] >= len(self._keys[item]):
self._indexes[item] = 0
self._initialized.add(item)
def get_next_key(self, service="tavily"):
"""Round-robin 返回某个服务下一个可用 key。"""
service = normalize_service(service)
if service not in self._initialized:
self.reload(service)
with self._lock:
keys = self._keys[service]
if not keys:
return None
index = self._indexes[service]
key = keys[index]
self._indexes[service] = (index + 1) % len(keys)
return key
def report_result(self, service, key_id, success):
"""记录使用结果，失败 3 次自动禁用并从对应池中移除。"""
service = normalize_service(service)
update_key_usage(key_id, success)
if not success:
with self._lock:
self._keys[service] = [
key for key in self._keys[service]
if key["id"] != key_id or key.get("consecutive_fails", 0) < 2
]
self.reload(service)
pool = ServiceKeyPool()
proxy/requirements.txt
-4
View File
@@ -1,4 +0,0 @@
fastapi>=0.104.0
uvicorn>=0.24.0
httpx>=0.25.0
jinja2>=3.1.0
proxy/server.py
-594
View File
@@ -1,594 +0,0 @@
"""
多服务 API Proxy — FastAPI 主服务
"""
import asyncio
import json
import os
import time
from datetime import datetime, timezone
import httpx
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, Response
from fastapi.templating import Jinja2Templates
import database as db
from key_pool import pool
ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "admin")
TAVILY_API_BASE = "https://api.tavily.com"
FIRECRAWL_API_BASE = "https://api.firecrawl.dev"
USAGE_SYNC_TTL_SECONDS = int(os.environ.get("USAGE_SYNC_TTL_SECONDS", "300"))
USAGE_SYNC_CONCURRENCY = max(1, int(os.environ.get("USAGE_SYNC_CONCURRENCY", "4")))
SERVICE_LABELS = {
"tavily": "Tavily",
"firecrawl": "Firecrawl",
}
app = FastAPI(title="Tavily / Firecrawl API Proxy")
templates = Jinja2Templates(directory=os.path.join(os.path.dirname(__file__), "templates"))
http_client = httpx.AsyncClient(timeout=60)
def get_admin_password():
return db.get_setting("admin_password", ADMIN_PASSWORD)
def get_service(service_value, default="tavily"):
try:
return db.normalize_service(service_value or default)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
# ═══ Auth helpers ═══
def verify_admin(request: Request):
auth = request.headers.get("Authorization", "")
password = request.headers.get("X-Admin-Password", "")
pwd = get_admin_password()
if auth == f"Bearer {pwd}" or password == pwd:
return True
raise HTTPException(status_code=401, detail="Unauthorized")
def extract_token(request: Request, body: dict = None):
"""从请求中提取代理 token。"""
auth = request.headers.get("Authorization", "")
if auth.startswith("Bearer "):
return auth[7:]
if body and body.get("api_key"):
return body["api_key"]
return None
def get_token_row_or_401(token_value, service):
if not token_value:
raise HTTPException(status_code=401, detail="Missing API token")
token_row = db.get_token_by_value(token_value)
if not token_row or token_row["service"] != service:
raise HTTPException(status_code=401, detail="Invalid token")
return token_row
def parse_usage_number(value):
if value in (None, ""):
return None
try:
return int(value)
except (TypeError, ValueError):
try:
return int(float(value))
except (TypeError, ValueError):
return None
def compute_remaining(limit_value, used_value):
if limit_value is None or used_value is None:
return None
return max(0, limit_value - used_value)
def parse_sync_time(value):
if not value:
return None
try:
return datetime.fromisoformat(value)
except ValueError:
return None
def is_usage_sync_stale(key_row, ttl_seconds=USAGE_SYNC_TTL_SECONDS):
synced_at = parse_sync_time(key_row.get("usage_synced_at"))
if not synced_at:
return True
return (datetime.now(timezone.utc) - synced_at).total_seconds() >= ttl_seconds
async def fetch_remote_usage_tavily(key_value):
resp = await http_client.get(
f"{TAVILY_API_BASE}/usage",
headers={"Authorization": f"Bearer {key_value}"},
)
if resp.status_code != 200:
detail = ""
try:
payload = resp.json()
detail = payload.get("detail") or payload.get("message") or ""
except Exception:
detail = resp.text.strip()
detail = detail[:200] if detail else f"HTTP {resp.status_code}"
raise HTTPException(status_code=resp.status_code, detail=detail)
return resp.json()
async def fetch_remote_usage_firecrawl(key_value):
headers = {"Authorization": f"Bearer {key_value}"}
current_resp, history_resp = await asyncio.gather(
http_client.get(f"{FIRECRAWL_API_BASE}/v2/team/credit-usage", headers=headers),
http_client.get(
f"{FIRECRAWL_API_BASE}/v2/team/credit-usage/historical",
params={"byApiKey": "true"},
headers=headers,
),
)
for resp in (current_resp, history_resp):
if resp.status_code != 200:
detail = resp.text.strip()[:200] or f"HTTP {resp.status_code}"
raise HTTPException(status_code=resp.status_code, detail=detail)
return {
"current": current_resp.json(),
"historical": history_resp.json(),
}
def normalize_usage_payload(service, payload):
if service == "tavily":
key_info = payload.get("key") or {}
account_info = payload.get("account") or {}
key_used = parse_usage_number(key_info.get("usage"))
key_limit = parse_usage_number(key_info.get("limit"))
account_used = parse_usage_number(account_info.get("plan_usage"))
account_limit = parse_usage_number(account_info.get("plan_limit"))
return {
"key_used": key_used,
"key_limit": key_limit,
"key_remaining": compute_remaining(key_limit, key_used),
"account_plan": (account_info.get("current_plan") or "").strip(),
"account_used": account_used,
"account_limit": account_limit,
"account_remaining": compute_remaining(account_limit, account_used),
}
current_data = (payload.get("current") or {}).get("data") or {}
history_periods = (payload.get("historical") or {}).get("periods") or []
if history_periods:
latest_period = max(
history_periods,
key=lambda item: ((item.get("endDate") or ""), (item.get("startDate") or "")),
)
current_period_rows = [
item for item in history_periods
if item.get("startDate") == latest_period.get("startDate")
and item.get("endDate") == latest_period.get("endDate")
]
else:
current_period_rows = []
account_remaining = parse_usage_number(current_data.get("remainingCredits"))
plan_credits = parse_usage_number(current_data.get("planCredits"))
account_used = sum(parse_usage_number(item.get("creditsUsed")) or 0 for item in current_period_rows)
account_limit = None
if account_remaining is not None:
account_limit = account_remaining + account_used
if plan_credits is None:
account_plan = "Firecrawl"
else:
account_plan = f"Plan credits {plan_credits}"
return {
"key_used": None,
"key_limit": None,
"key_remaining": None,
"account_plan": account_plan,
"account_used": account_used,
"account_limit": account_limit,
"account_remaining": account_remaining,
}
async def sync_usage_for_key_row(key_row):
service = key_row.get("service") or "tavily"
try:
if service == "firecrawl":
payload = await fetch_remote_usage_firecrawl(key_row["key"])
else:
payload = await fetch_remote_usage_tavily(key_row["key"])
normalized = normalize_usage_payload(service, payload)
db.update_key_remote_usage(
key_row["id"],
key_used=normalized["key_used"],
key_limit=normalized["key_limit"],
key_remaining=normalized["key_remaining"],
account_plan=normalized["account_plan"],
account_used=normalized["account_used"],
account_limit=normalized["account_limit"],
account_remaining=normalized["account_remaining"],
)
return {"key_id": key_row["id"], "status": "synced"}
except HTTPException as exc:
db.update_key_remote_usage_error(key_row["id"], exc.detail)
return {"key_id": key_row["id"], "status": "error", "detail": exc.detail}
except Exception as exc:
db.update_key_remote_usage_error(key_row["id"], str(exc))
return {"key_id": key_row["id"], "status": "error", "detail": str(exc)}
async def sync_usage_cache(force=False, key_id=None, service=None):
rows = []
if key_id is not None:
row = db.get_key_by_id(key_id)
if row and (service is None or row["service"] == service):
rows = [dict(row)]
else:
rows = [dict(row) for row in db.get_all_keys(service)]
if not rows:
return {"requested": 0, "synced": 0, "skipped": 0, "errors": 0}
to_sync = rows if force else [row for row in rows if is_usage_sync_stale(row)]
if not to_sync:
return {"requested": len(rows), "synced": 0, "skipped": len(rows), "errors": 0}
semaphore = asyncio.Semaphore(USAGE_SYNC_CONCURRENCY)
async def worker(row):
async with semaphore:
return await sync_usage_for_key_row(row)
results = await asyncio.gather(*(worker(row) for row in to_sync))
synced = sum(1 for item in results if item["status"] == "synced")
errors = sum(1 for item in results if item["status"] == "error")
return {
"requested": len(rows),
"synced": synced,
"skipped": len(rows) - len(to_sync),
"errors": errors,
}
def build_real_quota_summary(keys):
synced_keys = [
key for key in keys
if key.get("usage_key_used") is not None or key.get("usage_account_used") is not None
]
total_limit = 0
total_used = 0
total_remaining = 0
key_level_count = 0
account_fallback_count = 0
accounted_groups = set()
latest_sync = None
for key in synced_keys:
key_limit = key.get("usage_key_limit")
key_used = key.get("usage_key_used")
account_limit = key.get("usage_account_limit")
account_used = key.get("usage_account_used")
if key_limit is not None and key_used is not None:
total_limit += key_limit
total_used += key_used
total_remaining += key.get("usage_key_remaining") or compute_remaining(key_limit, key_used) or 0
key_level_count += 1
elif account_limit is not None and account_used is not None:
group_id = (key.get("email") or "").strip().lower() or f"key:{key.get('id')}"
if group_id not in accounted_groups:
accounted_groups.add(group_id)
total_limit += account_limit
total_used += account_used
total_remaining += key.get("usage_account_remaining") or compute_remaining(account_limit, account_used) or 0
account_fallback_count += 1
synced_at = parse_sync_time(key.get("usage_synced_at"))
if synced_at and (latest_sync is None or synced_at > latest_sync):
latest_sync = synced_at
error_count = sum(1 for key in keys if (key.get("usage_sync_error") or "").strip())
return {
"synced_keys": len(synced_keys),
"total_keys": len(keys),
"total_limit": total_limit,
"total_used": total_used,
"total_remaining": total_remaining,
"error_keys": error_count,
"last_synced_at": latest_sync.isoformat() if latest_sync else "",
"key_level_count": key_level_count,
"account_fallback_count": account_fallback_count,
}
def mask_key_rows(keys):
for key in keys:
raw = key["key"]
key["key_masked"] = raw[:8] + "***" + raw[-4:] if len(raw) > 12 else raw
return keys
async def build_service_dashboard(service):
service = get_service(service)
sync_result = await sync_usage_cache(force=False, service=service)
overview = db.get_usage_stats(service=service)
tokens = [dict(token) for token in db.get_all_tokens(service)]
for token in tokens:
token["stats"] = db.get_usage_stats(token_id=token["id"], service=service)
keys = mask_key_rows([dict(key) for key in db.get_all_keys(service)])
active_keys = [key for key in keys if key["active"]]
return {
"service": service,
"label": SERVICE_LABELS[service],
"overview": overview,
"tokens": tokens,
"keys": keys,
"keys_total": len(keys),
"keys_active": len(active_keys),
"real_quota": build_real_quota_summary(active_keys),
"usage_sync": sync_result,
}
def build_forward_headers(request, real_key):
skip_headers = {
"authorization",
"content-length",
"host",
"x-admin-password",
}
headers = {
key: value
for key, value in request.headers.items()
if key.lower() not in skip_headers
}
headers["Authorization"] = f"Bearer {real_key}"
return headers
async def parse_json_body(request):
raw_body = await request.body()
if not raw_body:
return raw_body, None
content_type = request.headers.get("content-type", "").lower()
if "application/json" not in content_type:
return raw_body, None
try:
return raw_body, json.loads(raw_body.decode("utf-8"))
except Exception:
return raw_body, None
def forward_raw_response(resp):
"""尽量保留上游返回格式，避免把非 JSON Firecrawl 响应再包一层。"""
content_type = resp.headers.get("content-type", "")
return Response(
content=resp.content,
status_code=resp.status_code,
media_type=content_type or None,
)
# ═══ 启动 ═══
@app.on_event("startup")
def startup():
db.init_db()
# ═══ Tavily 代理端点 ═══
@app.post("/api/search")
@app.post("/api/extract")
async def proxy_tavily(request: Request):
body = await request.json()
endpoint = request.url.path.replace("/api/", "")
token_value = extract_token(request, body)
token_row = get_token_row_or_401(token_value, "tavily")
ok, reason = db.check_quota(
token_row["id"],
token_row["hourly_limit"],
token_row["daily_limit"],
token_row["monthly_limit"],
service="tavily",
)
if not ok:
raise HTTPException(status_code=429, detail=reason)
key_info = pool.get_next_key("tavily")
if not key_info:
raise HTTPException(status_code=503, detail="No available API keys")
body["api_key"] = key_info["key"]
start = time.time()
try:
resp = await http_client.post(f"{TAVILY_API_BASE}/{endpoint}", json=body)
latency = int((time.time() - start) * 1000)
success = resp.status_code == 200
pool.report_result("tavily", key_info["id"], success)
db.log_usage(token_row["id"], key_info["id"], endpoint, int(success), latency, service="tavily")
return JSONResponse(content=resp.json(), status_code=resp.status_code)
except Exception as exc:
latency = int((time.time() - start) * 1000)
pool.report_result("tavily", key_info["id"], False)
db.log_usage(token_row["id"], key_info["id"], endpoint, 0, latency, service="tavily")
raise HTTPException(status_code=502, detail=str(exc))
# ═══ Firecrawl 代理端点 ═══
@app.api_route("/firecrawl/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"])
async def proxy_firecrawl(path: str, request: Request):
raw_body, body_json = await parse_json_body(request)
token_value = extract_token(request, body_json)
token_row = get_token_row_or_401(token_value, "firecrawl")
ok, reason = db.check_quota(
token_row["id"],
token_row["hourly_limit"],
token_row["daily_limit"],
token_row["monthly_limit"],
service="firecrawl",
)
if not ok:
raise HTTPException(status_code=429, detail=reason)
key_info = pool.get_next_key("firecrawl")
if not key_info:
raise HTTPException(status_code=503, detail="No available API keys")
forward_content = raw_body
if body_json is not None and "api_key" in body_json:
body_json["api_key"] = key_info["key"]
forward_content = json.dumps(body_json).encode("utf-8")
start = time.time()
try:
resp = await http_client.request(
request.method,
f"{FIRECRAWL_API_BASE}/{path}",
params=dict(request.query_params),
content=forward_content if request.method != "GET" else None,
headers=build_forward_headers(request, key_info["key"]),
)
latency = int((time.time() - start) * 1000)
success = resp.status_code < 400
pool.report_result("firecrawl", key_info["id"], success)
db.log_usage(token_row["id"], key_info["id"], path, int(success), latency, service="firecrawl")
content_type = resp.headers.get("content-type", "").lower()
if "application/json" in content_type:
return JSONResponse(content=resp.json(), status_code=resp.status_code)
return forward_raw_response(resp)
except Exception as exc:
latency = int((time.time() - start) * 1000)
pool.report_result("firecrawl", key_info["id"], False)
db.log_usage(token_row["id"], key_info["id"], path, 0, latency, service="firecrawl")
raise HTTPException(status_code=502, detail=str(exc))
# ═══ 控制台 ═══
@app.get("/", response_class=HTMLResponse)
async def console(request: Request):
return templates.TemplateResponse("console.html", {"request": request})
# ═══ 管理 API ═══
@app.get("/api/stats")
async def stats(request: Request, _=Depends(verify_admin)):
tavily_stats, firecrawl_stats = await asyncio.gather(
build_service_dashboard("tavily"),
build_service_dashboard("firecrawl"),
)
return {
"services": {
"tavily": tavily_stats,
"firecrawl": firecrawl_stats,
}
}
@app.get("/api/keys")
async def list_keys(request: Request, _=Depends(verify_admin)):
service = request.query_params.get("service")
keys = mask_key_rows([dict(key) for key in db.get_all_keys(service)])
return {"keys": keys}
@app.post("/api/usage/sync")
async def sync_usage(request: Request, _=Depends(verify_admin)):
body = await request.json() if request.headers.get("content-type", "").startswith("application/json") else {}
service = get_service(body.get("service"), default="tavily")
force = bool(body.get("force", True))
key_id = body.get("key_id")
result = await sync_usage_cache(force=force, key_id=key_id, service=service)
keys = [dict(key) for key in db.get_all_keys(service)]
active_keys = [key for key in keys if key["active"]]
return {
"ok": True,
"service": service,
"result": result,
"real_quota": build_real_quota_summary(active_keys),
}
@app.post("/api/keys")
async def add_keys(request: Request, _=Depends(verify_admin)):
body = await request.json()
service = get_service(body.get("service"), default="tavily")
if "file" in body:
count = db.import_keys_from_text(body["file"], service=service)
pool.reload(service)
return {"imported": count, "service": service}
if "key" in body:
db.add_key(body["key"], body.get("email", ""), service=service)
pool.reload(service)
return {"ok": True, "service": service}
raise HTTPException(status_code=400, detail="Provide 'key' or 'file'")
@app.delete("/api/keys/{key_id}")
async def remove_key(key_id: int, _=Depends(verify_admin)):
key_row = db.get_key_by_id(key_id)
db.delete_key(key_id)
if key_row:
pool.reload(key_row["service"])
return {"ok": True}
@app.put("/api/keys/{key_id}/toggle")
async def toggle_key(key_id: int, request: Request, _=Depends(verify_admin)):
body = await request.json()
db.toggle_key(key_id, body.get("active", 1))
key_row = db.get_key_by_id(key_id)
if key_row:
pool.reload(key_row["service"])
return {"ok": True}
@app.get("/api/tokens")
async def list_tokens(request: Request, _=Depends(verify_admin)):
service = request.query_params.get("service")
tokens = [dict(token) for token in db.get_all_tokens(service)]
for token in tokens:
token["stats"] = db.get_usage_stats(token_id=token["id"], service=token["service"])
return {"tokens": tokens}
@app.post("/api/tokens")
async def create_token(request: Request, _=Depends(verify_admin)):
body = await request.json()
service = get_service(body.get("service"), default="tavily")
token = db.create_token(body.get("name", ""), service=service)
return {"token": dict(token)}
@app.delete("/api/tokens/{token_id}")
async def remove_token(token_id: int, _=Depends(verify_admin)):
db.delete_token(token_id)
return {"ok": True}
@app.put("/api/password")
async def change_password(request: Request, _=Depends(verify_admin)):
body = await request.json()
new_pwd = body.get("password", "").strip()
if not new_pwd or len(new_pwd) < 4:
raise HTTPException(status_code=400, detail="Password too short (min 4)")
db.set_setting("admin_password", new_pwd)
return {"ok": True}
proxy/templates/console.html
-1889
View File
File diff suppressed because it is too large. Load diff