2206 lines
90 KiB
Python
2206 lines
90 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from datetime import datetime, timezone
|
||
from typing import List
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.database import get_db
|
||
from app.models.snapshot import UpstreamRateSnapshot
|
||
from app.models.upstream import Upstream
|
||
from app.models.upstream_key import UpstreamGeneratedKey
|
||
from app.models.website import Website, WebsiteGroupBinding, WebsiteSyncLog
|
||
from app.schemas.website import (
|
||
BindingCreate,
|
||
BindingResponse,
|
||
BindingUpdate,
|
||
ImportAccountItem,
|
||
ImportAccountsRequest,
|
||
ImportAccountsResponse,
|
||
ImportGroupItem,
|
||
ImportGroupsRequest,
|
||
ImportGroupsResponse,
|
||
ReorderPriorityItem,
|
||
ReorderPriorityRequest,
|
||
ReorderPriorityResponse,
|
||
SyncImportStatusRequest,
|
||
TestResult,
|
||
WebsiteCreate,
|
||
WebsiteGroupResponse,
|
||
WebsiteGroupCreate,
|
||
WebsiteGroupUpdate,
|
||
WebsiteResponse,
|
||
WebsiteSyncLogResponse,
|
||
WebsiteUpdate,
|
||
WebsiteBatchSyncResponse,
|
||
OrganizeGroupsItem,
|
||
OrganizeGroupsResponse,
|
||
CleanupInvalidAccountsItem,
|
||
CleanupInvalidAccountsPreviewResponse,
|
||
CleanupInvalidAccountsExecuteResponse,
|
||
SetConcurrencyRequest,
|
||
SetConcurrencyItem,
|
||
SetConcurrencyResponse,
|
||
SyncUpstreamModelsItem,
|
||
SyncUpstreamModelsResponse,
|
||
)
|
||
|
||
from app.services.website_client import Sub2ApiWebsiteClient, WebsiteError, _extract_id
|
||
from app.services.website_sync import (
|
||
binding_sources,
|
||
sync_binding,
|
||
build_target_group_priority_map,
|
||
sync_account_priorities_for_website,
|
||
reconcile_upstream_keys_full,
|
||
sync_account_priorities_for_upstream,
|
||
latest_rate_map,
|
||
build_imported_account_name,
|
||
)
|
||
from app.utils.auth import get_current_user
|
||
from app.utils.number import fixed_decimal_number
|
||
|
||
router = APIRouter(tags=["websites"])
|
||
logger = logging.getLogger(__name__)
|
||
|
||
MASK = "***"
|
||
SECRET_KEYS = {"password", "token", "key", "secret", "api_key"}
|
||
SENSITIVE_CREDENTIAL_KEYS = {
|
||
"access_token", "refresh_token", "id_token",
|
||
"api_key", "session_key", "cookie",
|
||
"aws_secret_access_key", "aws_session_token",
|
||
"service_account_json", "service_account", "private_key",
|
||
}
|
||
ALGORITHMS = {"max_plus_percent", "average_plus_percent", "min_plus_percent", "priority_weighted_plus_percent"}
|
||
|
||
|
||
def _mask(cfg: dict) -> dict:
|
||
masked = {}
|
||
for key, value in cfg.items():
|
||
masked[key] = MASK if key.lower() in SECRET_KEYS and value else value
|
||
return masked
|
||
|
||
|
||
def _website_response(row: Website) -> WebsiteResponse:
|
||
return WebsiteResponse(
|
||
id=row.id,
|
||
name=row.name,
|
||
site_type=row.site_type,
|
||
base_url=row.base_url,
|
||
api_prefix=row.api_prefix,
|
||
auth_type=row.auth_type,
|
||
auth_config_masked=_mask(json.loads(row.auth_config_json or "{}")),
|
||
groups_endpoint=row.groups_endpoint,
|
||
group_update_endpoint=row.group_update_endpoint,
|
||
enabled=row.enabled,
|
||
auto_sync_enabled=row.auto_sync_enabled,
|
||
timeout_seconds=row.timeout_seconds,
|
||
last_status=row.last_status,
|
||
last_checked_at=row.last_checked_at,
|
||
last_error=row.last_error,
|
||
created_at=row.created_at,
|
||
updated_at=row.updated_at,
|
||
)
|
||
|
||
|
||
def _binding_response(db: Session, row: WebsiteGroupBinding) -> BindingResponse:
|
||
website = db.query(Website).filter(Website.id == row.website_id).first()
|
||
return BindingResponse(
|
||
id=row.id,
|
||
website_id=row.website_id,
|
||
website_name=website.name if website else "",
|
||
target_group_id=row.target_group_id,
|
||
target_group_name=row.target_group_name,
|
||
source_groups=binding_sources(row),
|
||
percent=float(row.percent or 0),
|
||
algorithm=row.algorithm,
|
||
enabled=row.enabled,
|
||
created_at=row.created_at,
|
||
updated_at=row.updated_at,
|
||
)
|
||
|
||
|
||
def _log_response(row: WebsiteSyncLog) -> WebsiteSyncLogResponse:
|
||
return WebsiteSyncLogResponse(
|
||
id=row.id,
|
||
website_id=row.website_id,
|
||
binding_id=row.binding_id,
|
||
target_group_id=row.target_group_id,
|
||
target_group_name=row.target_group_name,
|
||
algorithm=row.algorithm,
|
||
percent=float(row.percent or 0),
|
||
source_rates=json.loads(row.source_rates_json or "[]"),
|
||
old_rate=row.old_rate,
|
||
new_rate=row.new_rate,
|
||
status=row.status,
|
||
message=row.message,
|
||
created_at=row.created_at,
|
||
)
|
||
|
||
|
||
def _ensure_unique_target(db: Session, website_id: int, target_group_id: str, exclude_id: int | None = None) -> None:
|
||
q = db.query(WebsiteGroupBinding).filter(
|
||
WebsiteGroupBinding.website_id == website_id,
|
||
WebsiteGroupBinding.target_group_id == target_group_id,
|
||
)
|
||
if exclude_id is not None:
|
||
q = q.filter(WebsiteGroupBinding.id != exclude_id)
|
||
if q.first():
|
||
raise HTTPException(400, "同一目标网站分组只能维护一条绑定记录")
|
||
|
||
|
||
def _client(row: Website) -> Sub2ApiWebsiteClient:
|
||
return Sub2ApiWebsiteClient(
|
||
base_url=row.base_url,
|
||
api_prefix=row.api_prefix,
|
||
auth_type=row.auth_type,
|
||
auth_config=json.loads(row.auth_config_json or "{}"),
|
||
timeout=float(row.timeout_seconds),
|
||
target_id=row.id,
|
||
target_name=row.name,
|
||
)
|
||
|
||
|
||
def _latest_upstream_groups(db: Session, upstream_id: int) -> list[dict]:
|
||
row = (
|
||
db.query(UpstreamRateSnapshot)
|
||
.filter(UpstreamRateSnapshot.upstream_id == upstream_id)
|
||
.order_by(UpstreamRateSnapshot.captured_at.desc())
|
||
.first()
|
||
)
|
||
if not row:
|
||
raise HTTPException(404, "no upstream snapshot found; run upstream check first")
|
||
snapshot = json.loads(row.snapshot_json or "{}")
|
||
groups = snapshot.get("groups") or {}
|
||
if not isinstance(groups, dict):
|
||
return []
|
||
return [item for item in groups.values() if isinstance(item, dict)]
|
||
|
||
|
||
def _source_group_id(group: dict) -> str:
|
||
return str(group.get("group_id") or group.get("id") or group.get("name") or "")
|
||
|
||
|
||
def _source_group_name(group: dict, gid: str) -> str:
|
||
return str(group.get("group_name") or group.get("name") or gid)
|
||
|
||
|
||
def _source_group_rate(group: dict) -> float:
|
||
raw = group.get("rate") or group.get("default_rate") or group.get("rate_multiplier") or 1
|
||
try:
|
||
return float(raw)
|
||
except (TypeError, ValueError):
|
||
return 1.0
|
||
|
||
|
||
def _numeric_group_id(value: str | None) -> int | None:
|
||
if value is None or value == "":
|
||
return None
|
||
try:
|
||
return int(value)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
|
||
|
||
@router.get("/api/websites", response_model=List[WebsiteResponse])
|
||
def list_websites(db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
return [_website_response(row) for row in db.query(Website).order_by(Website.id).all()]
|
||
|
||
|
||
@router.post("/api/websites", response_model=WebsiteResponse, status_code=201)
|
||
def create_website(body: WebsiteCreate, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
if body.site_type != "sub2api":
|
||
raise HTTPException(400, "目前只支持 sub2api")
|
||
row = Website(
|
||
name=body.name,
|
||
site_type=body.site_type,
|
||
base_url=body.base_url.rstrip("/"),
|
||
api_prefix=body.api_prefix,
|
||
auth_type=body.auth_type,
|
||
auth_config_json=json.dumps(body.auth_config, ensure_ascii=False),
|
||
groups_endpoint=body.groups_endpoint,
|
||
group_update_endpoint=body.group_update_endpoint,
|
||
enabled=body.enabled,
|
||
auto_sync_enabled=body.auto_sync_enabled,
|
||
timeout_seconds=body.timeout_seconds,
|
||
)
|
||
db.add(row)
|
||
db.commit()
|
||
db.refresh(row)
|
||
return _website_response(row)
|
||
|
||
|
||
@router.put("/api/websites/{wid}", response_model=WebsiteResponse)
|
||
def update_website(wid: int, body: WebsiteUpdate, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
row = db.query(Website).filter(Website.id == wid).first()
|
||
if not row:
|
||
raise HTTPException(404, "website not found")
|
||
data = body.model_dump(exclude_none=True)
|
||
if "site_type" in data and data["site_type"] != "sub2api":
|
||
raise HTTPException(400, "目前只支持 sub2api")
|
||
if "auth_config" in data:
|
||
existing = json.loads(row.auth_config_json or "{}")
|
||
incoming = data.pop("auth_config")
|
||
for key, value in incoming.items():
|
||
if value != MASK:
|
||
existing[key] = value
|
||
row.auth_config_json = json.dumps(existing, ensure_ascii=False)
|
||
if "base_url" in data:
|
||
data["base_url"] = data["base_url"].rstrip("/")
|
||
for key, value in data.items():
|
||
setattr(row, key, value)
|
||
row.updated_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
db.refresh(row)
|
||
return _website_response(row)
|
||
|
||
|
||
@router.delete("/api/websites/{wid}", status_code=204)
|
||
def delete_website(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
row = db.query(Website).filter(Website.id == wid).first()
|
||
if not row:
|
||
raise HTTPException(404, "website not found")
|
||
db.query(WebsiteSyncLog).filter(WebsiteSyncLog.website_id == wid).delete(synchronize_session=False)
|
||
db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.website_id == wid).delete(synchronize_session=False)
|
||
db.delete(row)
|
||
db.commit()
|
||
|
||
|
||
@router.post("/api/websites/{wid}/test", response_model=TestResult)
|
||
def test_website(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
row = db.query(Website).filter(Website.id == wid).first()
|
||
if not row:
|
||
raise HTTPException(404, "website not found")
|
||
try:
|
||
with _client(row) as c:
|
||
groups = c.get_groups(row.groups_endpoint)
|
||
row.last_status = "healthy"
|
||
row.last_error = None
|
||
row.last_checked_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
return TestResult(success=True, message=f"连接成功，获取到 {len(groups)} 个分组")
|
||
except Exception as exc:
|
||
row.last_status = "unhealthy"
|
||
row.last_error = str(exc)
|
||
row.last_checked_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
return TestResult(success=False, message="连接失败", detail=str(exc))
|
||
|
||
|
||
@router.get("/api/websites/{wid}/groups", response_model=List[WebsiteGroupResponse])
|
||
def list_website_groups(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
row = db.query(Website).filter(Website.id == wid).first()
|
||
if not row:
|
||
raise HTTPException(404, "website not found")
|
||
try:
|
||
with _client(row) as c:
|
||
return c.get_groups(row.groups_endpoint)
|
||
except Exception as exc:
|
||
raise HTTPException(502, str(exc))
|
||
|
||
|
||
@router.post("/api/websites/{wid}/groups", response_model=dict)
|
||
def create_website_group(
|
||
wid: int,
|
||
body: WebsiteGroupCreate,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
row = db.query(Website).filter(Website.id == wid).first()
|
||
if not row:
|
||
raise HTTPException(404, "website not found")
|
||
try:
|
||
with _client(row) as c:
|
||
payload = {
|
||
"name": body.name,
|
||
"description": body.description,
|
||
}
|
||
return c.create_group(payload, row.groups_endpoint)
|
||
except Exception as exc:
|
||
raise HTTPException(502, str(exc))
|
||
|
||
|
||
@router.put("/api/websites/{wid}/groups/{group_id:path}", response_model=dict)
|
||
def update_website_group(
|
||
wid: int,
|
||
group_id: str,
|
||
body: WebsiteGroupUpdate,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
row = db.query(Website).filter(Website.id == wid).first()
|
||
if not row:
|
||
raise HTTPException(404, "website not found")
|
||
try:
|
||
with _client(row) as c:
|
||
payload = {
|
||
"name": body.name,
|
||
"description": body.description,
|
||
}
|
||
resp = c.update_group(row.group_update_endpoint, group_id, payload)
|
||
|
||
# Sync updated group name to local group bindings cache
|
||
db.query(WebsiteGroupBinding).filter(
|
||
WebsiteGroupBinding.website_id == wid,
|
||
WebsiteGroupBinding.target_group_id == group_id
|
||
).update({"target_group_name": body.name})
|
||
db.commit()
|
||
|
||
return resp
|
||
except Exception as exc:
|
||
raise HTTPException(502, str(exc))
|
||
|
||
|
||
@router.post("/api/websites/{wid}/groups/import-from-upstream/{upstream_id}", response_model=ImportGroupsResponse)
|
||
def import_groups_from_upstream(
|
||
wid: int,
|
||
upstream_id: int,
|
||
body: ImportGroupsRequest,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
if website.site_type != "sub2api":
|
||
raise HTTPException(400, "目前只支持 sub2api")
|
||
upstream = db.query(Upstream).filter(Upstream.id == upstream_id).first()
|
||
if not upstream:
|
||
raise HTTPException(404, "upstream not found")
|
||
|
||
selected = set(body.group_ids)
|
||
groups = _latest_upstream_groups(db, upstream_id)
|
||
# 拉取目标网站已有分组，同名则跳过
|
||
try:
|
||
existing_names = set()
|
||
with _client(website) as c:
|
||
for eg in c.get_groups(website.groups_endpoint):
|
||
gname = eg.get("name") or eg.get("group_name") or ""
|
||
if gname:
|
||
existing_names.add(gname)
|
||
except Exception:
|
||
existing_names = set()
|
||
|
||
items: list[ImportGroupItem] = []
|
||
with _client(website) as c:
|
||
for group in groups:
|
||
source_gid = _source_group_id(group)
|
||
if not source_gid or (selected and source_gid not in selected):
|
||
continue
|
||
source_name = _source_group_name(group, source_gid)
|
||
target_name = f"{body.name_prefix}{source_name}" if body.name_prefix else source_name
|
||
|
||
# 检查是否已存在同名分组
|
||
if target_name in existing_names:
|
||
items.append(ImportGroupItem(
|
||
source_group_id=source_gid,
|
||
source_group_name=source_name,
|
||
target_group_name=target_name,
|
||
status="exists",
|
||
message="目标分组已存在，已跳过",
|
||
))
|
||
continue
|
||
|
||
create_body = {
|
||
"name": target_name,
|
||
"description": group.get("description") or f"Imported from {upstream.name} / {source_name}",
|
||
"platform": group.get("platform") or "openai",
|
||
"rate_multiplier": fixed_decimal_number(_source_group_rate(group), 2),
|
||
}
|
||
if group.get("rpm_limit") is not None:
|
||
create_body["rpm_limit"] = group.get("rpm_limit")
|
||
try:
|
||
created = c.create_group(create_body)
|
||
target_id = c.extract_id(created)
|
||
items.append(ImportGroupItem(
|
||
source_group_id=source_gid,
|
||
source_group_name=source_name,
|
||
target_group_id=target_id or None,
|
||
target_group_name=str(created.get("name") or target_name),
|
||
status="created",
|
||
message="已创建",
|
||
raw=created,
|
||
))
|
||
except Exception as exc:
|
||
msg = str(exc)
|
||
# 捕获 409 等已存在错误
|
||
if "已存在" in msg or "already exists" in msg.lower() or "409" in msg or "Conflict" in msg:
|
||
items.append(ImportGroupItem(
|
||
source_group_id=source_gid,
|
||
source_group_name=source_name,
|
||
target_group_name=target_name,
|
||
status="exists",
|
||
message="目标分组已存在（接口返回冲突）",
|
||
))
|
||
else:
|
||
logger.exception("import website group failed website=%s upstream=%s group=%s", wid, upstream_id, source_gid)
|
||
items.append(ImportGroupItem(
|
||
source_group_id=source_gid,
|
||
source_group_name=source_name,
|
||
target_group_name=target_name,
|
||
status="failed",
|
||
message=msg,
|
||
))
|
||
created_count = len([item for item in items if item.status == "created"])
|
||
exists_count = len([item for item in items if item.status == "exists"])
|
||
failed_count = len([item for item in items if item.status == "failed"])
|
||
msg_parts = []
|
||
if created_count:
|
||
msg_parts.append(f"新建 {created_count}")
|
||
if exists_count:
|
||
msg_parts.append(f"已存在 {exists_count}")
|
||
if failed_count:
|
||
msg_parts.append(f"失败 {failed_count}")
|
||
return ImportGroupsResponse(
|
||
success=failed_count == 0,
|
||
message="、".join(msg_parts) + f" / 共 {len(items)} 个" if msg_parts else f"共处理 {len(items)} 个分组",
|
||
items=items,
|
||
)
|
||
|
||
|
||
SUPPORTED_PLATFORMS = {"openai", "anthropic", "gemini", "antigravity"}
|
||
|
||
|
||
def _resolve_platform(text: str, group_snapshot: dict | None, fallback: str = "openai") -> str:
|
||
if group_snapshot and isinstance(group_snapshot, dict):
|
||
platform = group_snapshot.get("platform")
|
||
if platform and str(platform).lower() in SUPPORTED_PLATFORMS:
|
||
return str(platform).lower()
|
||
return _detect_platform(text, fallback)
|
||
|
||
|
||
def _detect_platform(text: str, fallback: str = "openai") -> str:
|
||
"""根据 Key 名或分组名关键词判断平台类型。"""
|
||
lower = text.lower()
|
||
if "claude" in lower or "anthropic" in lower:
|
||
return "anthropic"
|
||
if "gemini" in lower:
|
||
return "gemini"
|
||
if "antigravity" in lower:
|
||
return "antigravity"
|
||
return fallback
|
||
|
||
|
||
@router.post("/api/websites/{wid}/accounts/sync-imported-upstream-keys", response_model=ImportAccountsResponse)
|
||
def sync_imported_upstream_keys(
|
||
wid: int,
|
||
body: SyncImportStatusRequest,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
"""校验已导入的上游 Key 在目标 Sub2API 账号管理中是否仍存在。"""
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
|
||
rows = (
|
||
db.query(UpstreamGeneratedKey)
|
||
.filter(
|
||
UpstreamGeneratedKey.upstream_id == body.upstream_id,
|
||
UpstreamGeneratedKey.imported_website_id == wid,
|
||
UpstreamGeneratedKey.imported_account_id.isnot(None),
|
||
)
|
||
.all()
|
||
)
|
||
items: list[ImportAccountItem] = []
|
||
# 预载最新快照映射，用于结构化 platform 解析
|
||
upstream_ids = {row.upstream_id for row in rows}
|
||
latest_snapshots = {}
|
||
for uid in upstream_ids:
|
||
latest_snapshots[uid] = latest_rate_map(db, uid)
|
||
|
||
with _client(website) as c:
|
||
remote_accounts = None
|
||
try:
|
||
remote_accounts = c.list_accounts()
|
||
except Exception as e:
|
||
logger.warning("failed to list accounts for website %s at sync: %s", wid, e)
|
||
|
||
if remote_accounts is not None:
|
||
remote_ids = set()
|
||
for acc in remote_accounts:
|
||
acc_id = _extract_id(acc)
|
||
if acc_id:
|
||
remote_ids.add(str(acc_id))
|
||
else:
|
||
remote_ids = None
|
||
|
||
for row in rows:
|
||
snap_groups = latest_snapshots.get(row.upstream_id) or {}
|
||
group_snap = snap_groups.get(row.group_id) if isinstance(snap_groups, dict) else None
|
||
platform = _resolve_platform(f"{row.group_name} {row.group_id} {row.key_name}", group_snap, "openai")
|
||
if not row.imported_account_id:
|
||
continue
|
||
old_account_id = row.imported_account_id
|
||
|
||
if remote_ids is None:
|
||
items.append(ImportAccountItem(
|
||
upstream_key_id=row.id, source_group_id=row.group_id,
|
||
source_group_name=row.group_name, account_id=old_account_id,
|
||
platform=platform, status="check_failed",
|
||
message="无法校验目标账号存在性（拉取账号列表失败）",
|
||
))
|
||
elif str(old_account_id) not in remote_ids:
|
||
row.imported_website_id = None
|
||
row.imported_account_id = None
|
||
row.imported_at = None
|
||
row.status = "created"
|
||
items.append(ImportAccountItem(
|
||
upstream_key_id=row.id, source_group_id=row.group_id,
|
||
source_group_name=row.group_name, account_id=old_account_id,
|
||
platform=platform, status="stale_cleared",
|
||
message="目标账号已删除，已清除导入标记",
|
||
))
|
||
else:
|
||
items.append(ImportAccountItem(
|
||
upstream_key_id=row.id, source_group_id=row.group_id,
|
||
source_group_name=row.group_name, account_id=old_account_id,
|
||
platform=platform, status="exists", message="目标账号仍存在",
|
||
))
|
||
db.commit()
|
||
cleared_count = len([i for i in items if i.status == "stale_cleared"])
|
||
check_failed_count = len([i for i in items if i.status == "check_failed"])
|
||
msg_parts = []
|
||
if cleared_count:
|
||
msg_parts.append(f"清除 {cleared_count}")
|
||
if check_failed_count:
|
||
msg_parts.append(f"校验失败 {check_failed_count}")
|
||
return ImportAccountsResponse(
|
||
success=check_failed_count == 0,
|
||
message="、".join(msg_parts) + f" / 共 {len(items)} 个" if msg_parts else f"共校验 {len(items)} 个，无变化",
|
||
items=items,
|
||
)
|
||
|
||
|
||
@router.post("/api/websites/{wid}/accounts/reorder-priority", response_model=ReorderPriorityResponse)
|
||
def reorder_account_priorities(
|
||
wid: int,
|
||
body: ReorderPriorityRequest,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
if website.site_type != "sub2api":
|
||
raise HTTPException(400, "目前只支持 sub2api")
|
||
|
||
upstream = db.query(Upstream).filter(Upstream.id == body.upstream_id).first()
|
||
if not upstream:
|
||
raise HTTPException(404, "upstream not found")
|
||
|
||
results = sync_account_priorities_for_upstream(db, body.upstream_id, website_id=wid)
|
||
failed_count = sum(1 for item in results if item.get("status") == "failed")
|
||
success_count = sum(1 for item in results if item.get("status") == "success")
|
||
skipped_count = sum(1 for item in results if item.get("status") == "skipped")
|
||
stale_count = sum(1 for item in results if item.get("status") == "stale")
|
||
|
||
parts = []
|
||
if success_count:
|
||
parts.append(f"更新 {success_count}")
|
||
if failed_count:
|
||
parts.append(f"失败 {failed_count}")
|
||
if stale_count:
|
||
parts.append(f"清理失效账号 {stale_count}")
|
||
if skipped_count:
|
||
parts.append(f"跳过 {skipped_count}")
|
||
message = "、".join(parts) + f" / 共 {len(results)} 个" if parts else "没有需要重排的账号"
|
||
|
||
return ReorderPriorityResponse(
|
||
success=failed_count == 0,
|
||
message=message,
|
||
items=[ReorderPriorityItem(**item) for item in results],
|
||
)
|
||
|
||
|
||
@router.post("/api/websites/{wid}/accounts/import-upstream-keys", response_model=ImportAccountsResponse)
|
||
def import_upstream_keys_as_accounts(
|
||
wid: int,
|
||
body: ImportAccountsRequest,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
if website.site_type != "sub2api":
|
||
raise HTTPException(400, "目前只支持 sub2api")
|
||
if not body.upstream_key_ids:
|
||
raise HTTPException(400, "请选择要导入的 Key")
|
||
rows = (
|
||
db.query(UpstreamGeneratedKey)
|
||
.filter(UpstreamGeneratedKey.id.in_(body.upstream_key_ids))
|
||
.order_by(UpstreamGeneratedKey.id)
|
||
.all()
|
||
)
|
||
# 导入前尝试对账（失败不阻塞，仅打日志—避免远端不可达时误删本地 Key）
|
||
upstream_ids = {row.upstream_id for row in rows}
|
||
for uid in upstream_ids:
|
||
try:
|
||
reconcile_upstream_keys_full(db, uid)
|
||
except Exception as exc:
|
||
logger.warning("import reconcile failed for upstream %s: %s", uid, exc)
|
||
# 重新查询（对账成功时已清理失效 Key）
|
||
rows = (
|
||
db.query(UpstreamGeneratedKey)
|
||
.filter(UpstreamGeneratedKey.id.in_(body.upstream_key_ids))
|
||
.order_by(UpstreamGeneratedKey.id)
|
||
.all()
|
||
)
|
||
found_ids = {row.id for row in rows}
|
||
missing_ids = [kid for kid in body.upstream_key_ids if kid not in found_ids]
|
||
items: list[ImportAccountItem] = [
|
||
ImportAccountItem(
|
||
upstream_key_id=kid,
|
||
source_group_id="",
|
||
source_group_name="",
|
||
platform=body.default_platform,
|
||
status="failed",
|
||
message="key not found",
|
||
)
|
||
for kid in missing_ids
|
||
]
|
||
# 预载上游以方便获取名称 and base_url
|
||
from app.models.upstream import Upstream as _Up
|
||
upstream_ids = {row.upstream_id for row in rows}
|
||
upstreams_db = db.query(_Up).filter(_Up.id.in_(upstream_ids)).all()
|
||
upstreams_map = {u.id: u for u in upstreams_db}
|
||
|
||
# 预载最新快照映射，用于结构化 platform 解析
|
||
latest_snapshots: dict[int, dict[str, Any]] = {}
|
||
for uid in upstream_ids:
|
||
latest_snapshots[uid] = latest_rate_map(db, uid)
|
||
|
||
# 按倍率自动分配优先级
|
||
rate_priority_map: dict[tuple[str, int, str], int] = {}
|
||
if body.auto_priority_by_rate:
|
||
try:
|
||
target_group_sources = {}
|
||
for row in rows:
|
||
target_group_id = body.target_group_map.get(row.group_id)
|
||
if target_group_id:
|
||
target_group_sources.setdefault(str(target_group_id), []).append((row.upstream_id, row.group_id))
|
||
rate_priority_map = build_target_group_priority_map(db, target_group_sources)
|
||
except Exception:
|
||
# 没有快照时忽略，后续 fallback 到 body.priority
|
||
pass
|
||
|
||
with _client(website) as c:
|
||
target_group_names: dict[str, str] = {}
|
||
try:
|
||
grps = c.get_groups(endpoint=website.groups_endpoint) or []
|
||
for g in grps:
|
||
g_id = g.get("id")
|
||
if g_id is not None:
|
||
target_group_names[str(g_id)] = g.get("name") or str(g_id)
|
||
except Exception as e:
|
||
logger.warning("failed to fetch groups for website %s at import: %s", wid, e)
|
||
|
||
# 预先拉取远端账号列表以避免循环请求，并能拿到已存在账号 of 远端名称
|
||
remote_accounts_map: dict[str, dict[str, Any]] = {}
|
||
remote_accounts_list = None
|
||
has_list_accounts = hasattr(c, "list_accounts")
|
||
if has_list_accounts:
|
||
try:
|
||
remote_accounts_list = c.list_accounts()
|
||
if remote_accounts_list is not None:
|
||
for acc in remote_accounts_list:
|
||
aid = c.extract_id(acc)
|
||
if aid:
|
||
remote_accounts_map[aid] = acc
|
||
except Exception as e:
|
||
logger.warning("failed to fetch remote accounts at import: %s", e)
|
||
|
||
for row in rows:
|
||
# 跳过远端已不存在或导入失败的 Key（对账后标记为 orphaned / import_failed）
|
||
if row.status in ("orphaned", "failed", "import_failed"):
|
||
items.append(ImportAccountItem(
|
||
upstream_key_id=row.id,
|
||
source_group_id=row.group_id,
|
||
source_group_name=row.group_name,
|
||
target_group_id=body.target_group_map.get(row.group_id),
|
||
platform=body.default_platform,
|
||
status="failed",
|
||
message="远端 Key 已不存在，请重新生成",
|
||
))
|
||
continue
|
||
# 先确定平台（失败项也需要记录）
|
||
if body.platform_mode == "auto":
|
||
snap_groups = latest_snapshots.get(row.upstream_id) or {}
|
||
group_snap = snap_groups.get(row.group_id) if isinstance(snap_groups, dict) else None
|
||
platform = _resolve_platform(
|
||
f"{row.group_name} {row.group_id} {row.key_name}",
|
||
group_snap,
|
||
body.default_platform,
|
||
)
|
||
else:
|
||
platform = body.default_platform
|
||
|
||
upstream = upstreams_map.get(row.upstream_id)
|
||
upstream_name = upstream.name if upstream else "Unknown"
|
||
upstream_base_url = upstream.base_url if upstream else ""
|
||
|
||
# 构建新格式的预期账号名称
|
||
expected_new_name = build_imported_account_name(
|
||
upstream_name=upstream_name,
|
||
group_name=row.group_name,
|
||
group_id=row.group_id,
|
||
key_id=row.id,
|
||
)
|
||
|
||
# 幂等校验：已导入过则检查远端账号是否仍存在
|
||
if row.imported_website_id == wid and row.imported_account_id:
|
||
old_account_id = row.imported_account_id
|
||
if has_list_accounts:
|
||
if remote_accounts_list is None:
|
||
# 接口获取失败，保守判定为校验失败
|
||
exists = None
|
||
else:
|
||
exists = old_account_id in remote_accounts_map
|
||
else:
|
||
# 如果 client 没有 list_accounts，则 fallback 回原先的 account_exists 检测以兼容旧 Mock 测试
|
||
try:
|
||
exists = c.account_exists(row.imported_account_id)
|
||
except Exception:
|
||
exists = None
|
||
|
||
if exists is True:
|
||
# 顺手回填 imported_target_group_id（老数据升级后可通过重导自动补齐）
|
||
new_tgid = body.target_group_map.get(row.group_id) or None
|
||
if new_tgid:
|
||
tg_name = target_group_names.get(str(new_tgid))
|
||
if row.imported_target_group_id != new_tgid or row.imported_target_group_name != tg_name:
|
||
row.imported_target_group_id = new_tgid
|
||
row.imported_target_group_name = tg_name
|
||
db.commit()
|
||
|
||
remote_name = ""
|
||
if old_account_id in remote_accounts_map:
|
||
remote_name = remote_accounts_map[old_account_id].get("name") or ""
|
||
items.append(ImportAccountItem(
|
||
upstream_key_id=row.id,
|
||
source_group_id=row.group_id,
|
||
source_group_name=row.group_name,
|
||
target_group_id=body.target_group_map.get(row.group_id),
|
||
account_id=old_account_id,
|
||
account_name=remote_name or expected_new_name,
|
||
platform=platform,
|
||
upstream_base_url=upstream_base_url,
|
||
status="exists",
|
||
message="已导入过，已跳过",
|
||
))
|
||
continue
|
||
elif exists is False:
|
||
# 远端已删除，清空标记后继续创建
|
||
row.imported_website_id = None
|
||
row.imported_account_id = None
|
||
row.imported_at = None
|
||
row.status = "created"
|
||
# 继续往下走（不 continue）
|
||
else:
|
||
# 校验失败，保守跳过
|
||
items.append(ImportAccountItem(
|
||
upstream_key_id=row.id,
|
||
source_group_id=row.group_id,
|
||
source_group_name=row.group_name,
|
||
target_group_id=body.target_group_map.get(row.group_id),
|
||
account_id=old_account_id,
|
||
platform=platform,
|
||
status="check_failed",
|
||
message="无法校验目标账号状态，已保守跳过",
|
||
))
|
||
continue
|
||
|
||
if not row.key_value:
|
||
items.append(ImportAccountItem(
|
||
upstream_key_id=row.id,
|
||
source_group_id=row.group_id,
|
||
source_group_name=row.group_name,
|
||
platform=platform,
|
||
status="failed",
|
||
message="该 Key 无明文值，无法导入（远端已存在 Key 不会保留明文，请重新创建或手动填入）",
|
||
))
|
||
continue
|
||
|
||
target_group_id = body.target_group_map.get(row.group_id)
|
||
group_ids = []
|
||
numeric_target = _numeric_group_id(target_group_id)
|
||
if numeric_target is not None:
|
||
group_ids.append(numeric_target)
|
||
account_name = expected_new_name
|
||
account_body = {
|
||
"name": account_name,
|
||
"platform": platform,
|
||
"type": "apikey",
|
||
"credentials": {
|
||
"api_key": row.key_value,
|
||
"base_url": upstream_base_url,
|
||
},
|
||
"group_ids": group_ids,
|
||
"rate_multiplier": 1,
|
||
"concurrency": body.concurrency,
|
||
"priority": rate_priority_map.get((str(target_group_id), row.upstream_id, row.group_id), body.priority) if body.auto_priority_by_rate else body.priority,
|
||
"notes": f"Imported by SmartUp from upstream key #{row.id}",
|
||
}
|
||
try:
|
||
created = c.create_account(account_body)
|
||
account_id = c.extract_id(created)
|
||
row.imported_website_id = wid
|
||
row.imported_account_id = account_id or None
|
||
row.imported_at = datetime.now(timezone.utc)
|
||
row.imported_target_group_id = target_group_id or None
|
||
row.imported_target_group_name = target_group_names.get(str(target_group_id)) if target_group_id else None
|
||
row.status = "imported"
|
||
row.error = None
|
||
db.commit()
|
||
items.append(ImportAccountItem(
|
||
upstream_key_id=row.id,
|
||
source_group_id=row.group_id,
|
||
source_group_name=row.group_name,
|
||
target_group_id=target_group_id,
|
||
account_id=account_id or None,
|
||
account_name=str(created.get("name") or account_name),
|
||
platform=platform,
|
||
upstream_base_url=upstream_base_url,
|
||
status="created",
|
||
message="已创建账号",
|
||
raw=created,
|
||
))
|
||
except Exception as exc:
|
||
logger.exception("import upstream key as account failed website=%s key=%s", wid, row.id)
|
||
row.status = "import_failed"
|
||
row.error = str(exc)
|
||
db.commit()
|
||
items.append(ImportAccountItem(
|
||
upstream_key_id=row.id,
|
||
source_group_id=row.group_id,
|
||
source_group_name=row.group_name,
|
||
target_group_id=target_group_id,
|
||
account_name=account_name,
|
||
platform=platform,
|
||
upstream_base_url=upstream_base_url,
|
||
status="failed",
|
||
message=str(exc),
|
||
))
|
||
created_count = len([item for item in items if item.status == "created"])
|
||
exists_count = len([item for item in items if item.status == "exists"])
|
||
failed_count = len([item for item in items if item.status == "failed"])
|
||
check_failed_count = len([item for item in items if item.status == "check_failed"])
|
||
msg_parts = []
|
||
if created_count:
|
||
msg_parts.append(f"新建 {created_count}")
|
||
if exists_count:
|
||
msg_parts.append(f"已存在 {exists_count}")
|
||
if check_failed_count:
|
||
msg_parts.append(f"校验失败 {check_failed_count}")
|
||
if failed_count:
|
||
msg_parts.append(f"失败 {failed_count}")
|
||
return ImportAccountsResponse(
|
||
success=failed_count == 0 and check_failed_count == 0,
|
||
message="、".join(msg_parts) + f" / 共 {len(items)} 个" if msg_parts else f"共处理 {len(items)} 个",
|
||
items=items,
|
||
)
|
||
|
||
|
||
def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClient) -> dict[str, dict[str, Any]] | None:
|
||
import re
|
||
# 1. Fetch remote accounts
|
||
remote_accounts = c.list_accounts()
|
||
if remote_accounts is None:
|
||
return None
|
||
|
||
remote_accounts_map = {}
|
||
for acc in remote_accounts:
|
||
aid = c.extract_id(acc)
|
||
if aid:
|
||
remote_accounts_map[aid] = acc
|
||
|
||
candidates = {}
|
||
|
||
# 2. Condition A: Local orphaned keys
|
||
orphaned_keys = db.query(UpstreamGeneratedKey).filter(
|
||
UpstreamGeneratedKey.status == "orphaned",
|
||
UpstreamGeneratedKey.imported_website_id == website.id,
|
||
UpstreamGeneratedKey.imported_account_id.isnot(None),
|
||
).all()
|
||
|
||
for key in orphaned_keys:
|
||
acc_id = key.imported_account_id
|
||
remote_acc = remote_accounts_map.get(acc_id)
|
||
acc_name = remote_acc.get("name") if remote_acc else key.key_name
|
||
upstream = db.query(Upstream).filter(Upstream.id == key.upstream_id).first()
|
||
upstream_name = upstream.name if upstream else None
|
||
|
||
candidates[acc_id] = {
|
||
"account_id": acc_id,
|
||
"account_name": acc_name,
|
||
"upstream_key_id": key.id,
|
||
"upstream_name": upstream_name,
|
||
"source_group_name": key.group_name,
|
||
"reason": "本地上游 Key 已被标记为 orphaned",
|
||
"is_local_orphan": True,
|
||
"db_key": key,
|
||
}
|
||
|
||
# 3. Condition B: Remote SmartUp-marked orphan accounts
|
||
pattern = re.compile(r"Imported by SmartUp from upstream key #(\d+)")
|
||
for aid, acc in remote_accounts_map.items():
|
||
if aid in candidates:
|
||
continue
|
||
|
||
key_id = None
|
||
for field in ("notes", "description", "remark"):
|
||
val = acc.get(field)
|
||
if val and isinstance(val, str):
|
||
match = pattern.search(val)
|
||
if match:
|
||
key_id = int(match.group(1))
|
||
break
|
||
if key_id is None:
|
||
continue
|
||
|
||
key = db.query(UpstreamGeneratedKey).filter(
|
||
UpstreamGeneratedKey.id == key_id
|
||
).first()
|
||
|
||
is_orphan = False
|
||
reason = ""
|
||
if not key:
|
||
is_orphan = True
|
||
reason = "本地找不到对应的上游 Key 记录"
|
||
elif key.imported_website_id != website.id or key.imported_account_id != aid:
|
||
is_orphan = True
|
||
reason = "本地 Key 记录存在，但未关联到当前目标站的当前账号"
|
||
elif key.status == 'orphaned':
|
||
is_orphan = True
|
||
reason = "本地上游 Key 已被标记为 orphaned"
|
||
|
||
if is_orphan:
|
||
upstream_name = None
|
||
source_group_name = None
|
||
if key:
|
||
upstream = db.query(Upstream).filter(Upstream.id == key.upstream_id).first()
|
||
upstream_name = upstream.name if upstream else None
|
||
source_group_name = key.group_name
|
||
|
||
candidates[aid] = {
|
||
"account_id": aid,
|
||
"account_name": acc.get("name") or aid,
|
||
"upstream_key_id": key_id,
|
||
"upstream_name": upstream_name,
|
||
"source_group_name": source_group_name,
|
||
"reason": reason,
|
||
"is_local_orphan": False,
|
||
"db_key": key,
|
||
}
|
||
|
||
return candidates
|
||
|
||
|
||
def _test_candidate_account(c: Sub2ApiWebsiteClient, account_id: str) -> tuple[str, bool, str]:
|
||
"""返回 (test_status, cleanup_allowed, message)"""
|
||
try:
|
||
res = c.test_account(account_id)
|
||
status = res.get("status")
|
||
msg = res.get("message", "")
|
||
if status == "404":
|
||
return "404", True, msg
|
||
elif status == "success":
|
||
return "test_passed", False, msg
|
||
elif status == "failed":
|
||
return "failed", True, msg
|
||
else:
|
||
return "test_unknown", False, msg
|
||
except Exception as e:
|
||
return "test_unknown", False, f"测试时发生异常: {e}"
|
||
|
||
|
||
@router.post("/api/websites/{wid}/accounts/cleanup-invalid/preview", response_model=CleanupInvalidAccountsPreviewResponse)
|
||
def preview_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
|
||
with _client(website) as c:
|
||
candidates = _get_cleanup_candidates(db, website, c)
|
||
if candidates is None:
|
||
return CleanupInvalidAccountsPreviewResponse(
|
||
success=False,
|
||
message="拉取远端账号列表失败，无法执行清理",
|
||
items=[]
|
||
)
|
||
|
||
items = []
|
||
for aid, cand in candidates.items():
|
||
test_status, cleanup_allowed, test_msg = _test_candidate_account(c, aid)
|
||
items.append(CleanupInvalidAccountsItem(
|
||
account_id=cand["account_id"],
|
||
account_name=cand["account_name"],
|
||
upstream_key_id=cand["upstream_key_id"],
|
||
upstream_name=cand["upstream_name"],
|
||
source_group_name=cand["source_group_name"],
|
||
reason=cand["reason"],
|
||
test_status=test_status,
|
||
cleanup_allowed=cleanup_allowed,
|
||
action_status="previewed",
|
||
message=test_msg,
|
||
))
|
||
|
||
return CleanupInvalidAccountsPreviewResponse(
|
||
success=True,
|
||
message=f"已成功加载 {len(items)} 个失效账号候选对象",
|
||
items=items
|
||
)
|
||
|
||
|
||
@router.post("/api/websites/{wid}/accounts/cleanup-invalid/execute", response_model=CleanupInvalidAccountsExecuteResponse)
|
||
def execute_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
|
||
with _client(website) as c:
|
||
candidates = _get_cleanup_candidates(db, website, c)
|
||
if candidates is None:
|
||
return CleanupInvalidAccountsExecuteResponse(
|
||
success=False,
|
||
message="拉取远端账号列表失败，无法执行清理",
|
||
items=[]
|
||
)
|
||
|
||
items = []
|
||
for aid, cand in candidates.items():
|
||
test_status, cleanup_allowed, test_msg = _test_candidate_account(c, aid)
|
||
|
||
action_status = "skipped"
|
||
msg = "测试通过或未知，已跳过删除"
|
||
|
||
if cleanup_allowed:
|
||
try:
|
||
del_res = c.delete_account(aid)
|
||
status = del_res.get("status")
|
||
del_msg = del_res.get("message", "")
|
||
|
||
if status in ("deleted", "already_deleted"):
|
||
action_status = status
|
||
msg = del_msg
|
||
|
||
# Clear database flags if it's a local orphan key
|
||
if cand["is_local_orphan"] and cand["db_key"]:
|
||
key = cand["db_key"]
|
||
key.imported_website_id = None
|
||
key.imported_account_id = None
|
||
key.imported_at = None
|
||
key.imported_target_group_id = None
|
||
key.imported_target_group_name = None
|
||
db.add(key)
|
||
else:
|
||
action_status = "failed"
|
||
msg = f"删除失败: {del_msg}"
|
||
except Exception as e:
|
||
action_status = "failed"
|
||
msg = f"删除失败: {e}"
|
||
else:
|
||
msg = f"账号状态正常，无需清理 ({test_msg})"
|
||
|
||
items.append(CleanupInvalidAccountsItem(
|
||
account_id=cand["account_id"],
|
||
account_name=cand["account_name"],
|
||
upstream_key_id=cand["upstream_key_id"],
|
||
upstream_name=cand["upstream_name"],
|
||
source_group_name=cand["source_group_name"],
|
||
reason=cand["reason"],
|
||
test_status=test_status,
|
||
cleanup_allowed=cleanup_allowed,
|
||
action_status=action_status,
|
||
message=msg,
|
||
))
|
||
|
||
db.commit()
|
||
|
||
deleted_count = sum(1 for item in items if item.action_status in ("deleted", "already_deleted"))
|
||
failed_count = sum(1 for item in items if item.action_status == "failed")
|
||
skipped_count = sum(1 for item in items if item.action_status == "skipped")
|
||
success = (failed_count == 0)
|
||
|
||
msg_parts = []
|
||
if deleted_count:
|
||
msg_parts.append(f"成功清理 {deleted_count} 个")
|
||
if failed_count:
|
||
msg_parts.append(f"失败 {failed_count} 个")
|
||
if skipped_count:
|
||
msg_parts.append(f"跳过 {skipped_count} 个")
|
||
|
||
message = "清理执行完毕：" + "，".join(msg_parts) if msg_parts else "没有需要清理的失效账号"
|
||
|
||
return CleanupInvalidAccountsExecuteResponse(
|
||
success=success,
|
||
message=message,
|
||
items=items
|
||
)
|
||
|
||
|
||
@router.post("/api/websites/{wid}/accounts/set-concurrency", response_model=SetConcurrencyResponse)
|
||
def set_website_accounts_concurrency(
|
||
wid: int,
|
||
body: SetConcurrencyRequest,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
"""批量设置当前网站已导入账号的并发数"""
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
if website.site_type != "sub2api":
|
||
raise HTTPException(400, "only sub2api site supports account concurrency settings")
|
||
|
||
with _client(website) as c:
|
||
try:
|
||
remote_accounts = c.list_accounts()
|
||
except Exception as e:
|
||
return SetConcurrencyResponse(
|
||
success=False,
|
||
message=f"拉取远端账号列表失败: {e}",
|
||
items=[]
|
||
)
|
||
|
||
if remote_accounts is None:
|
||
return SetConcurrencyResponse(
|
||
success=False,
|
||
message="拉取远端账号列表失败，无法设置并发",
|
||
items=[]
|
||
)
|
||
|
||
remote_map = {}
|
||
for acc in remote_accounts:
|
||
aid = c.extract_id(acc)
|
||
if aid:
|
||
remote_map[aid] = acc
|
||
|
||
keys = db.query(UpstreamGeneratedKey).filter(
|
||
UpstreamGeneratedKey.imported_website_id == wid,
|
||
UpstreamGeneratedKey.imported_account_id.isnot(None)
|
||
).all()
|
||
|
||
candidates = {}
|
||
for key in keys:
|
||
aid = key.imported_account_id
|
||
if aid not in candidates:
|
||
candidates[aid] = {
|
||
"account_id": aid,
|
||
"db_key": key
|
||
}
|
||
|
||
if not candidates:
|
||
return SetConcurrencyResponse(
|
||
success=True,
|
||
message="没有找到 SmartUp 导入的有效账号",
|
||
items=[]
|
||
)
|
||
|
||
bulk_candidates = []
|
||
skipped_items = []
|
||
|
||
for aid, cand in candidates.items():
|
||
remote_acc = remote_map.get(aid)
|
||
if not remote_acc:
|
||
skipped_items.append(SetConcurrencyItem(
|
||
account_id=aid,
|
||
account_name=None,
|
||
old_concurrency=None,
|
||
target_concurrency=body.concurrency,
|
||
status="skipped",
|
||
message="账号在远端已被删除或不存在"
|
||
))
|
||
continue
|
||
|
||
acc_name = remote_acc.get("name")
|
||
old_con = remote_acc.get("concurrency")
|
||
|
||
try:
|
||
int(aid)
|
||
bulk_candidates.append({
|
||
"account_id": aid,
|
||
"account_name": acc_name,
|
||
"old_concurrency": old_con
|
||
})
|
||
except ValueError:
|
||
skipped_items.append(SetConcurrencyItem(
|
||
account_id=aid,
|
||
account_name=acc_name,
|
||
old_concurrency=old_con,
|
||
target_concurrency=body.concurrency,
|
||
status="skipped",
|
||
message="账号 ID 无法转换为数字，跳过批量更新"
|
||
))
|
||
|
||
items = []
|
||
items.extend(skipped_items)
|
||
|
||
if not bulk_candidates:
|
||
success = all(item.status == "skipped" for item in items)
|
||
return SetConcurrencyResponse(
|
||
success=success,
|
||
message=f"无可执行批量更新的账号。已跳过 {len(skipped_items)} 个",
|
||
items=items
|
||
)
|
||
|
||
bulk_ids = [item["account_id"] for item in bulk_candidates]
|
||
bulk_success = False
|
||
bulk_error_msg = ""
|
||
|
||
try:
|
||
c.bulk_update_accounts(bulk_ids, {"concurrency": body.concurrency})
|
||
bulk_success = True
|
||
except Exception as e:
|
||
bulk_error_msg = str(e)
|
||
logger.warning("bulk_update_accounts failed, falling back to individual updates: %s", e)
|
||
|
||
if bulk_success:
|
||
for item in bulk_candidates:
|
||
items.append(SetConcurrencyItem(
|
||
account_id=item["account_id"],
|
||
account_name=item["account_name"],
|
||
old_concurrency=item["old_concurrency"],
|
||
target_concurrency=body.concurrency,
|
||
status="success",
|
||
message="成功"
|
||
))
|
||
else:
|
||
for item in bulk_candidates:
|
||
aid = item["account_id"]
|
||
try:
|
||
c.update_account(aid, {"concurrency": body.concurrency})
|
||
items.append(SetConcurrencyItem(
|
||
account_id=aid,
|
||
account_name=item["account_name"],
|
||
old_concurrency=item["old_concurrency"],
|
||
target_concurrency=body.concurrency,
|
||
status="success",
|
||
message="成功 (逐个更新)"
|
||
))
|
||
except Exception as e:
|
||
items.append(SetConcurrencyItem(
|
||
account_id=aid,
|
||
account_name=item["account_name"],
|
||
old_concurrency=item["old_concurrency"],
|
||
target_concurrency=body.concurrency,
|
||
status="failed",
|
||
message=f"更新失败: {e}"
|
||
))
|
||
|
||
success_count = sum(1 for item in items if item.status == "success")
|
||
failed_count = sum(1 for item in items if item.status == "failed")
|
||
skip_count = sum(1 for item in items if item.status == "skipped")
|
||
|
||
success = (failed_count == 0)
|
||
msg_parts = []
|
||
if success_count:
|
||
msg_parts.append(f"成功 {success_count} 个")
|
||
if failed_count:
|
||
msg_parts.append(f"失败 {failed_count} 个")
|
||
if skip_count:
|
||
msg_parts.append(f"跳过 {skip_count} 个")
|
||
|
||
message = "设置账号并发执行完毕：" + "，".join(msg_parts)
|
||
|
||
return SetConcurrencyResponse(
|
||
success=success,
|
||
message=message,
|
||
items=items
|
||
)
|
||
|
||
|
||
def _sync_upstream_models_generator(wid: int, db: Session):
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
yield "error", {"message": "website not found"}
|
||
return
|
||
if website.site_type != "sub2api":
|
||
yield "error", {"message": "only sub2api site supports account model syncing"}
|
||
return
|
||
|
||
with _client(website) as c:
|
||
try:
|
||
remote_accounts = c.list_accounts()
|
||
except Exception as e:
|
||
yield "error", {"message": f"拉取远端账号列表失败: {e}"}
|
||
return
|
||
|
||
if remote_accounts is None:
|
||
yield "error", {"message": "拉取远端账号列表失败，无法同步上游模型"}
|
||
return
|
||
|
||
remote_map = {}
|
||
for acc in remote_accounts:
|
||
aid = c.extract_id(acc)
|
||
if aid:
|
||
remote_map[aid] = acc
|
||
|
||
# 1. 查找候选账号：本地 UpstreamGeneratedKey.imported_website_id == wid 且 imported_account_id 非空
|
||
keys = db.query(UpstreamGeneratedKey).filter(
|
||
UpstreamGeneratedKey.imported_website_id == wid,
|
||
UpstreamGeneratedKey.imported_account_id.isnot(None)
|
||
).all()
|
||
|
||
candidates = {}
|
||
for key in keys:
|
||
aid = key.imported_account_id
|
||
if aid not in candidates:
|
||
candidates[aid] = {
|
||
"account_id": aid,
|
||
"upstream_id": key.upstream_id,
|
||
"db_key": key
|
||
}
|
||
|
||
if not candidates:
|
||
yield "start", {"total_accounts": 0}
|
||
yield "complete", {
|
||
"success": True,
|
||
"message": "没有找到 SmartUp 导入的有效账号",
|
||
"items": []
|
||
}
|
||
return
|
||
|
||
yield "start", {"total_accounts": len(candidates)}
|
||
|
||
upstream_ids = {cand["upstream_id"] for cand in candidates.values()}
|
||
upstreams_map = {up.id: up for up in db.query(Upstream).filter(Upstream.id.in_(upstream_ids)).all()}
|
||
|
||
items = []
|
||
|
||
for aid, cand in candidates.items():
|
||
# 校验是否存在于远端
|
||
remote_acc = remote_map.get(aid)
|
||
if not remote_acc:
|
||
item = SyncUpstreamModelsItem(
|
||
account_id=aid,
|
||
account_name=None,
|
||
model_count=0,
|
||
models=[],
|
||
status="skipped",
|
||
message="账号在远端已被删除或不存在"
|
||
)
|
||
items.append(item)
|
||
yield "item", item.model_dump()
|
||
continue
|
||
|
||
acc_name = remote_acc.get("name")
|
||
|
||
# 校验账号 ID 是否为数字
|
||
try:
|
||
int(aid)
|
||
except ValueError:
|
||
item = SyncUpstreamModelsItem(
|
||
account_id=aid,
|
||
account_name=acc_name,
|
||
model_count=0,
|
||
models=[],
|
||
status="skipped",
|
||
message="账号 ID 非数字，跳过同步"
|
||
)
|
||
items.append(item)
|
||
yield "item", item.model_dump()
|
||
continue
|
||
|
||
upstream = upstreams_map.get(cand["upstream_id"])
|
||
upstream_base_url = upstream.base_url if upstream else None
|
||
|
||
if not upstream_base_url or not upstream_base_url.strip():
|
||
item = SyncUpstreamModelsItem(
|
||
account_id=aid,
|
||
account_name=acc_name,
|
||
model_count=0,
|
||
models=[],
|
||
status="failed",
|
||
message="上游 base_url 为空，跳过处理避免继续写坏账号"
|
||
)
|
||
items.append(item)
|
||
yield "item", item.model_dump()
|
||
continue
|
||
|
||
# 1. 先安全修复 base_url (不依赖同步模型结果)
|
||
raw_creds = remote_acc.get("credentials")
|
||
if not isinstance(raw_creds, dict):
|
||
raw_creds = {}
|
||
current_creds = {
|
||
k: v for k, v in raw_creds.items()
|
||
if k not in SENSITIVE_CREDENTIAL_KEYS
|
||
}
|
||
current_creds["base_url"] = upstream_base_url
|
||
|
||
# 对 SmartUp 导入的 apikey 类型账号，必须从本地 key_value 补回 api_key；
|
||
# 若 key_value 为空则禁止继续写回，避免写坏账号。
|
||
db_key = cand.get("db_key")
|
||
acc_type = remote_acc.get("type", "")
|
||
if acc_type == "apikey":
|
||
local_key_value = (db_key.key_value or "").strip() if db_key else ""
|
||
if not local_key_value:
|
||
item = SyncUpstreamModelsItem(
|
||
account_id=aid,
|
||
account_name=acc_name,
|
||
model_count=0,
|
||
models=[],
|
||
status="failed",
|
||
message="本地 key_value 为空，跳过处理避免写坏账号 api_key"
|
||
)
|
||
items.append(item)
|
||
yield "item", item.model_dump()
|
||
continue
|
||
current_creds["api_key"] = local_key_value
|
||
|
||
try:
|
||
c.update_account(aid, {"credentials": current_creds})
|
||
except Exception as e:
|
||
item = SyncUpstreamModelsItem(
|
||
account_id=aid,
|
||
account_name=acc_name,
|
||
model_count=0,
|
||
models=[],
|
||
status="failed",
|
||
message=f"修复 Base URL 失败: {e}"
|
||
)
|
||
items.append(item)
|
||
yield "item", item.model_dump()
|
||
continue
|
||
|
||
# 2. 调用 sub2api 同步模型
|
||
try:
|
||
raw_models = c.sync_account_upstream_models(aid)
|
||
# 过滤空、去重、排序
|
||
valid_models = sorted(list(set(m.strip() for m in raw_models if m and m.strip())))
|
||
|
||
if not valid_models:
|
||
item = SyncUpstreamModelsItem(
|
||
account_id=aid,
|
||
account_name=acc_name,
|
||
model_count=0,
|
||
models=[],
|
||
status="failed",
|
||
message="已修复 Base URL，但上游同步返回模型列表为空"
|
||
)
|
||
items.append(item)
|
||
yield "item", item.model_dump()
|
||
continue
|
||
|
||
# 3. 模型同步成功后，在 current_creds 基础上替换 model_mapping 并写回
|
||
updated_creds = dict(current_creds)
|
||
updated_creds["model_mapping"] = {m: m for m in valid_models}
|
||
c.update_account(aid, {"credentials": updated_creds})
|
||
|
||
item = SyncUpstreamModelsItem(
|
||
account_id=aid,
|
||
account_name=acc_name,
|
||
model_count=len(valid_models),
|
||
models=valid_models,
|
||
status="success",
|
||
message=f"已修复 Base URL 并同步 {len(valid_models)} 个模型"
|
||
)
|
||
items.append(item)
|
||
yield "item", item.model_dump()
|
||
except Exception as e:
|
||
item = SyncUpstreamModelsItem(
|
||
account_id=aid,
|
||
account_name=acc_name,
|
||
model_count=0,
|
||
models=[],
|
||
status="failed",
|
||
message=f"已修复 Base URL，但模型同步失败: {e}"
|
||
)
|
||
items.append(item)
|
||
yield "item", item.model_dump()
|
||
|
||
success_count = sum(1 for item in items if item.status == "success")
|
||
failed_count = sum(1 for item in items if item.status == "failed")
|
||
skip_count = sum(1 for item in items if item.status == "skipped")
|
||
|
||
success = (failed_count == 0)
|
||
msg_parts = []
|
||
if success_count:
|
||
msg_parts.append(f"成功 {success_count} 个")
|
||
if failed_count:
|
||
msg_parts.append(f"失败 {failed_count} 个")
|
||
if skip_count:
|
||
msg_parts.append(f"跳过 {skip_count} 个")
|
||
|
||
message = "同步上游模型执行完毕：" + "，".join(msg_parts)
|
||
yield "complete", {
|
||
"success": success,
|
||
"message": message,
|
||
"items": [item.model_dump() for item in items]
|
||
}
|
||
|
||
|
||
@router.post("/api/websites/{wid}/accounts/sync-upstream-models", response_model=SyncUpstreamModelsResponse)
|
||
def sync_website_accounts_upstream_models(
|
||
wid: int,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
"""一键同步上游模型"""
|
||
items = []
|
||
success = False
|
||
message = ""
|
||
error_occurred = False
|
||
|
||
for event, data in _sync_upstream_models_generator(wid, db):
|
||
if event == "item":
|
||
items.append(SyncUpstreamModelsItem(**data))
|
||
elif event == "complete":
|
||
success = data["success"]
|
||
message = data["message"]
|
||
elif event == "error":
|
||
error_occurred = True
|
||
message = data["message"]
|
||
|
||
if error_occurred:
|
||
if message == "website not found":
|
||
raise HTTPException(404, "website not found")
|
||
if "only sub2api" in message:
|
||
raise HTTPException(400, message)
|
||
return SyncUpstreamModelsResponse(
|
||
success=False,
|
||
message=message,
|
||
items=[]
|
||
)
|
||
|
||
return SyncUpstreamModelsResponse(
|
||
success=success,
|
||
message=message,
|
||
items=items
|
||
)
|
||
|
||
|
||
@router.post("/api/websites/{wid}/accounts/sync-upstream-models/stream")
|
||
def sync_website_accounts_upstream_models_stream(
|
||
wid: int,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
"""流式一键同步上游模型"""
|
||
from fastapi.responses import StreamingResponse
|
||
|
||
def event_generator():
|
||
for event, data in _sync_upstream_models_generator(wid, db):
|
||
yield json.dumps({"event": event, "data": data}, ensure_ascii=False) + "\n"
|
||
|
||
return StreamingResponse(event_generator(), media_type="application/x-ndjson")
|
||
|
||
|
||
@router.post("/api/websites/{wid}/groups/organize", response_model=OrganizeGroupsResponse)
|
||
def organize_website_groups(
|
||
wid: int,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
"""一键整理分组：按现有绑定关系导入/自愈账号"""
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
if website.site_type != "sub2api":
|
||
raise HTTPException(400, "目前只支持 sub2api")
|
||
|
||
# 1. 读取当前网站的所有启用绑定关系
|
||
bindings = (
|
||
db.query(WebsiteGroupBinding)
|
||
.filter(WebsiteGroupBinding.website_id == wid, WebsiteGroupBinding.enabled == True)
|
||
.all()
|
||
)
|
||
|
||
smartup_managed_group_ids = {str(b.target_group_id) for b in bindings if b.target_group_id}
|
||
|
||
# 合并已导入 Key 上历史记录的 imported_target_group_id，以确保变更绑定后能清理旧分组
|
||
hist_group_ids = (
|
||
db.query(UpstreamGeneratedKey.imported_target_group_id)
|
||
.filter(
|
||
UpstreamGeneratedKey.imported_website_id == wid,
|
||
UpstreamGeneratedKey.imported_target_group_id.isnot(None),
|
||
UpstreamGeneratedKey.imported_target_group_id != "",
|
||
)
|
||
.distinct()
|
||
.all()
|
||
)
|
||
for row in hist_group_ids:
|
||
smartup_managed_group_ids.add(str(row[0]))
|
||
|
||
items: list[OrganizeGroupsItem] = []
|
||
|
||
# 2. 收集所有的 upstream_id 并执行对账，以保证本地 Key 的状态最新
|
||
upstream_ids = set()
|
||
for b in bindings:
|
||
sources = binding_sources(b)
|
||
for src in sources:
|
||
uid = src.get("upstream_id")
|
||
if uid:
|
||
upstream_ids.add(int(uid))
|
||
|
||
for uid in upstream_ids:
|
||
try:
|
||
reconcile_upstream_keys_full(db, uid)
|
||
except Exception as exc:
|
||
logger.warning("organize reconcile failed for upstream %s: %s", uid, exc)
|
||
|
||
# 3. 按倍率自动分配优先级 (分目标分组进行优先级计算)
|
||
rate_priority_map: dict[tuple[str, int, str], int] = {}
|
||
try:
|
||
target_group_sources = {}
|
||
for b in bindings:
|
||
for src in binding_sources(b):
|
||
uid = src.get("upstream_id")
|
||
gid = src.get("group_id")
|
||
if uid and gid:
|
||
target_group_sources.setdefault(str(b.target_group_id), []).append((int(uid), str(gid)))
|
||
rate_priority_map = build_target_group_priority_map(db, target_group_sources)
|
||
except Exception:
|
||
pass
|
||
|
||
with _client(website) as c:
|
||
# 获取远端分组名称列表和账号列表
|
||
target_group_names: dict[str, str] = {}
|
||
remote_accounts: list[dict[str, Any]] = []
|
||
try:
|
||
grps = c.get_groups(endpoint=website.groups_endpoint) or []
|
||
for g in grps:
|
||
g_id = g.get("id")
|
||
if g_id is not None:
|
||
target_group_names[str(g_id)] = g.get("name") or str(g_id)
|
||
except Exception as e:
|
||
logger.warning("failed to fetch groups for website %s at organize: %s", wid, e)
|
||
|
||
remote_account_ids: set[str] | None = None
|
||
try:
|
||
remote_accounts = c.list_accounts()
|
||
if remote_accounts is not None:
|
||
remote_account_ids = set()
|
||
for acc in remote_accounts:
|
||
acc_id = c.extract_id(acc)
|
||
if acc_id:
|
||
remote_account_ids.add(str(acc_id))
|
||
except Exception as e:
|
||
logger.warning("failed to fetch remote accounts for website %s at organize: %s", wid, e)
|
||
|
||
# 构建 remote_account_map 用于快速查询账号状态
|
||
remote_account_map = {}
|
||
if remote_accounts is not None:
|
||
for acc in remote_accounts:
|
||
acc_id = c.extract_id(acc)
|
||
if acc_id:
|
||
remote_account_map[str(acc_id)] = acc
|
||
|
||
# 预载所有相关的上游快照映射
|
||
all_upstream_ids = set()
|
||
for b in bindings:
|
||
for src in binding_sources(b):
|
||
_uid = src.get("upstream_id")
|
||
if _uid:
|
||
all_upstream_ids.add(int(_uid))
|
||
|
||
latest_snapshots: dict[int, dict[str, Any]] = {}
|
||
for _uid in all_upstream_ids:
|
||
latest_snapshots[_uid] = latest_rate_map(db, _uid)
|
||
|
||
# 对每个绑定关系遍历其对应的上游 Key 进行整理
|
||
for b in bindings:
|
||
target_group_id = b.target_group_id
|
||
target_group_name = target_group_names.get(str(target_group_id)) or b.target_group_name or str(target_group_id)
|
||
sources = binding_sources(b)
|
||
|
||
for src in sources:
|
||
uid = src.get("upstream_id")
|
||
gid = src.get("group_id")
|
||
if not uid or not gid:
|
||
continue
|
||
|
||
_u = db.query(Upstream).filter(Upstream.id == uid).first()
|
||
upstream_name = _u.name if _u else f"#{uid}"
|
||
|
||
snapshot_groups = latest_snapshots.get(int(uid)) or {}
|
||
# 来源名称优先取 src.group_name，再 fallback 快照，再 fallback ID
|
||
source_group_name = src.get("group_name") or ""
|
||
if not source_group_name:
|
||
if gid in snapshot_groups:
|
||
g_data = snapshot_groups[gid]
|
||
if isinstance(g_data, dict):
|
||
source_group_name = g_data.get("name") or g_data.get("group_name") or ""
|
||
if not source_group_name:
|
||
source_group_name = gid
|
||
|
||
# 查询该 (upstream_id, group_id) 的非 orphaned Key
|
||
keys = (
|
||
db.query(UpstreamGeneratedKey)
|
||
.filter(
|
||
UpstreamGeneratedKey.upstream_id == uid,
|
||
UpstreamGeneratedKey.group_id == gid,
|
||
UpstreamGeneratedKey.status != "orphaned",
|
||
)
|
||
.all()
|
||
)
|
||
|
||
if not keys:
|
||
items.append(
|
||
OrganizeGroupsItem(
|
||
target_group_id=str(target_group_id),
|
||
target_group_name=target_group_name,
|
||
upstream_name=upstream_name,
|
||
source_group_id=str(gid),
|
||
source_group_name=str(source_group_name),
|
||
key_name="",
|
||
account_id=None,
|
||
account_name=None,
|
||
status="missing_key",
|
||
message="请先生成上游 Key",
|
||
)
|
||
)
|
||
continue
|
||
|
||
# 获取来源上游的 Base URL 用于账号创建
|
||
upstream_base_url = _u.base_url if _u else ""
|
||
|
||
for row in keys:
|
||
# 跳过状态为 failed 的 Key
|
||
if row.status == "failed":
|
||
items.append(
|
||
OrganizeGroupsItem(
|
||
target_group_id=str(target_group_id),
|
||
target_group_name=target_group_name,
|
||
upstream_name=upstream_name,
|
||
source_group_id=str(gid),
|
||
source_group_name=str(source_group_name),
|
||
key_name=row.key_name,
|
||
account_id=None,
|
||
account_name=None,
|
||
status="failed",
|
||
message="上游 Key 生成状态为失败",
|
||
)
|
||
)
|
||
continue
|
||
|
||
# 检测平台类型
|
||
group_snap = snapshot_groups.get(row.group_id) if isinstance(snapshot_groups, dict) else None
|
||
platform = _resolve_platform(
|
||
f"{row.group_name} {row.group_id} {row.key_name}",
|
||
group_snap,
|
||
"openai",
|
||
)
|
||
|
||
# 1. 幂等校验：已导入过则检查远端账号是否仍存在
|
||
if row.imported_website_id == wid and row.imported_account_id:
|
||
old_account_id = row.imported_account_id
|
||
|
||
if remote_account_ids is None:
|
||
# 账号列表获取失败，无法校验状态，保守跳过
|
||
items.append(
|
||
OrganizeGroupsItem(
|
||
target_group_id=str(target_group_id),
|
||
target_group_name=target_group_name,
|
||
upstream_name=upstream_name,
|
||
source_group_id=str(gid),
|
||
source_group_name=str(source_group_name),
|
||
key_name=row.key_name,
|
||
account_id=old_account_id,
|
||
status="failed",
|
||
message="无法校验目标账号状态，已保守跳过",
|
||
)
|
||
)
|
||
continue
|
||
|
||
remote_acc = remote_account_map.get(str(old_account_id))
|
||
|
||
if remote_acc is not None:
|
||
# 目标账号存在：检测并对齐目标分组，移除其余已启用的 SmartUp 托管分组，保留非 SmartUp 托管分组
|
||
try:
|
||
current_group_ids = remote_acc.get("group_ids") or []
|
||
if not isinstance(current_group_ids, list):
|
||
current_group_ids = [current_group_ids]
|
||
|
||
numeric_target = _numeric_group_id(target_group_id)
|
||
target_val = numeric_target if numeric_target is not None else target_group_id
|
||
|
||
# 过滤出非 SmartUp 托管的 group_ids
|
||
new_group_ids = [g for g in current_group_ids if str(g) not in smartup_managed_group_ids]
|
||
# 追加当前目标分组
|
||
new_group_ids.append(target_val)
|
||
|
||
old_set = {str(g) for g in current_group_ids}
|
||
new_set = {str(g) for g in new_group_ids}
|
||
|
||
remote_platform = remote_acc.get("platform")
|
||
if remote_platform is not None:
|
||
platform_mismatch = (str(remote_platform).lower() != str(platform).lower())
|
||
else:
|
||
platform_mismatch = (str(platform).lower() != "openai")
|
||
|
||
update_payload = {}
|
||
if old_set != new_set:
|
||
update_payload["group_ids"] = new_group_ids
|
||
if platform_mismatch:
|
||
update_payload["platform"] = platform
|
||
|
||
if update_payload:
|
||
c.update_account(old_account_id, update_payload)
|
||
if "group_ids" in update_payload:
|
||
remote_acc["group_ids"] = new_group_ids
|
||
if "platform" in update_payload:
|
||
# 主动核对防线：防止子站忽略 platform 修改而发生静默失败
|
||
fresh_accs = c.list_accounts()
|
||
if fresh_accs is None:
|
||
raise WebsiteError("复查失败：拉取远端账号列表返回空，无法核对平台修改是否生效")
|
||
fresh_acc = next((a for a in fresh_accs if str(c.extract_id(a)) == str(old_account_id)), None)
|
||
if not fresh_acc:
|
||
raise WebsiteError(f"复查失败：在远端账号列表中未找到要更新的账号 {old_account_id}")
|
||
|
||
remote_account_map[str(old_account_id)] = fresh_acc
|
||
new_platform = fresh_acc.get("platform")
|
||
if str(new_platform).lower() != str(platform).lower():
|
||
raise WebsiteError(
|
||
f"远端账号不支持修改平台属性，尝试将平台从 {remote_platform} 修改为 {platform} 失败"
|
||
)
|
||
remote_acc["platform"] = platform
|
||
|
||
msg_parts = []
|
||
if old_set != new_set:
|
||
msg_parts.append("已迁移到目标分组")
|
||
if platform_mismatch:
|
||
msg_parts.append(f"平台从 {remote_platform} 修正为 {platform}")
|
||
msg = "已存在，" + "且".join(msg_parts)
|
||
else:
|
||
msg = "已存在且已对齐，已跳过"
|
||
|
||
# 补齐本地 DB 的目标分组标记
|
||
if row.imported_target_group_id != target_group_id:
|
||
row.imported_target_group_id = target_group_id
|
||
row.imported_target_group_name = target_group_names.get(str(target_group_id))
|
||
db.commit()
|
||
|
||
items.append(
|
||
OrganizeGroupsItem(
|
||
target_group_id=str(target_group_id),
|
||
target_group_name=target_group_name,
|
||
upstream_name=upstream_name,
|
||
source_group_id=str(gid),
|
||
source_group_name=str(source_group_name),
|
||
key_name=row.key_name,
|
||
account_id=old_account_id,
|
||
account_name=str(remote_acc.get("name") or ""),
|
||
status="exists",
|
||
message=msg,
|
||
)
|
||
)
|
||
continue
|
||
except Exception as exc:
|
||
items.append(
|
||
OrganizeGroupsItem(
|
||
target_group_id=str(target_group_id),
|
||
target_group_name=target_group_name,
|
||
upstream_name=upstream_name,
|
||
source_group_id=str(gid),
|
||
source_group_name=str(source_group_name),
|
||
key_name=row.key_name,
|
||
account_id=old_account_id,
|
||
status="failed",
|
||
message=f"更新绑定失败: {exc}",
|
||
)
|
||
)
|
||
continue
|
||
else:
|
||
# 远端已删除，清理标记后进行重建
|
||
row.imported_website_id = None
|
||
row.imported_account_id = None
|
||
row.imported_at = None
|
||
row.status = "created"
|
||
db.commit()
|
||
is_recreated = True
|
||
else:
|
||
is_recreated = False
|
||
|
||
# 2. 检查是否有明文 Key
|
||
if not row.key_value:
|
||
items.append(
|
||
OrganizeGroupsItem(
|
||
target_group_id=str(target_group_id),
|
||
target_group_name=target_group_name,
|
||
upstream_name=upstream_name,
|
||
source_group_id=str(gid),
|
||
source_group_name=str(source_group_name),
|
||
key_name=row.key_name,
|
||
status="failed",
|
||
message="该 Key 无明文值，无法导入",
|
||
)
|
||
)
|
||
continue
|
||
|
||
# 3. 创建账号并绑定当前目标分组
|
||
group_ids = []
|
||
numeric_target = _numeric_group_id(target_group_id)
|
||
if numeric_target is not None:
|
||
group_ids.append(numeric_target)
|
||
else:
|
||
group_ids.append(target_group_id)
|
||
|
||
account_name = build_imported_account_name(
|
||
upstream_name=upstream_name,
|
||
group_name=row.group_name,
|
||
group_id=row.group_id,
|
||
key_id=row.id,
|
||
)
|
||
priority_val = rate_priority_map.get((str(target_group_id), row.upstream_id, row.group_id), 1)
|
||
account_body = {
|
||
"name": account_name,
|
||
"platform": platform,
|
||
"type": "apikey",
|
||
"credentials": {
|
||
"api_key": row.key_value,
|
||
"base_url": upstream_base_url,
|
||
},
|
||
"group_ids": group_ids,
|
||
"rate_multiplier": 1,
|
||
"concurrency": 100,
|
||
"priority": priority_val,
|
||
"notes": f"Imported by SmartUp from upstream key #{row.id}",
|
||
}
|
||
try:
|
||
created = c.create_account(account_body)
|
||
account_id = c.extract_id(created)
|
||
row.imported_website_id = wid
|
||
row.imported_account_id = account_id or None
|
||
row.imported_at = datetime.now(timezone.utc)
|
||
row.imported_target_group_id = target_group_id or None
|
||
row.imported_target_group_name = target_group_names.get(str(target_group_id)) if target_group_id else None
|
||
row.status = "imported"
|
||
row.error = None
|
||
db.commit()
|
||
|
||
# 将新创建的账号加入 remote_account_map 缓存，以防同账号的其它绑定关系后续做幂等校验
|
||
if account_id:
|
||
remote_account_map[str(account_id)] = created
|
||
|
||
items.append(
|
||
OrganizeGroupsItem(
|
||
target_group_id=str(target_group_id),
|
||
target_group_name=target_group_name,
|
||
upstream_name=upstream_name,
|
||
source_group_id=str(gid),
|
||
source_group_name=str(source_group_name),
|
||
key_name=row.key_name,
|
||
account_id=account_id or None,
|
||
account_name=str(created.get("name") or account_name),
|
||
status="recreated" if is_recreated else "created",
|
||
message="清理后重建账号" if is_recreated else "已创建账号",
|
||
)
|
||
)
|
||
except Exception as exc:
|
||
logger.exception("organize create account failed website=%s key=%s", wid, row.id)
|
||
row.status = "import_failed"
|
||
row.error = str(exc)
|
||
db.commit()
|
||
items.append(
|
||
OrganizeGroupsItem(
|
||
target_group_id=str(target_group_id),
|
||
target_group_name=target_group_name,
|
||
upstream_name=upstream_name,
|
||
source_group_id=str(gid),
|
||
source_group_name=str(source_group_name),
|
||
key_name=row.key_name,
|
||
status="failed",
|
||
message=str(exc),
|
||
)
|
||
)
|
||
|
||
created_count = sum(1 for item in items if item.status == "created")
|
||
recreated_count = sum(1 for item in items if item.status == "recreated")
|
||
aligned_count = sum(1 for item in items if item.status == "exists" and "已对齐" in item.message)
|
||
migrated_count = sum(1 for item in items if item.status == "exists" and "已迁移" in item.message)
|
||
missing_key_count = sum(1 for item in items if item.status == "missing_key")
|
||
failed_count = sum(1 for item in items if item.status == "failed")
|
||
|
||
parts = []
|
||
if created_count:
|
||
parts.append(f"已创建 {created_count}")
|
||
if recreated_count:
|
||
parts.append(f"已重建 {recreated_count}")
|
||
if aligned_count:
|
||
parts.append(f"已存在且已对齐 {aligned_count}")
|
||
if migrated_count:
|
||
parts.append(f"已存在已迁移 {migrated_count}")
|
||
if missing_key_count:
|
||
parts.append(f"缺少 Key {missing_key_count}")
|
||
if failed_count:
|
||
parts.append(f"失败 {failed_count}")
|
||
|
||
# 整理完成后，调用一次网站级 priority 重排以修正历史和当前账号优先级
|
||
try:
|
||
sync_account_priorities_for_website(db, wid)
|
||
except Exception as exc:
|
||
logger.warning("failed to sync priorities after organize for website %s: %s", wid, exc)
|
||
|
||
message = "整理完成：" + " / ".join(parts) if parts else "整理完成：无任何绑定或数据"
|
||
return OrganizeGroupsResponse(
|
||
success=failed_count == 0,
|
||
message=message,
|
||
items=items,
|
||
)
|
||
|
||
|
||
@router.get("/api/group-bindings", response_model=List[BindingResponse])
|
||
def list_bindings(db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
rows = db.query(WebsiteGroupBinding).order_by(WebsiteGroupBinding.id.desc()).all()
|
||
return [_binding_response(db, row) for row in rows]
|
||
|
||
|
||
@router.post("/api/websites/{wid}/group-bindings/sync-now", response_model=WebsiteBatchSyncResponse)
|
||
def sync_website_group_bindings(
|
||
wid: int,
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
"""一键同步网站下所有分组绑定。"""
|
||
website = db.query(Website).filter(Website.id == wid).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
|
||
bindings = db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.website_id == wid).order_by(WebsiteGroupBinding.id.asc()).all()
|
||
if not bindings:
|
||
return WebsiteBatchSyncResponse(
|
||
total=0, success=0, failed=0, skipped=0,
|
||
message="暂无绑定可同步",
|
||
logs=[],
|
||
)
|
||
|
||
results: list[WebsiteSyncLog] = []
|
||
for binding in bindings:
|
||
try:
|
||
log = sync_binding(db, binding, write=True)
|
||
results.append(log)
|
||
except Exception as exc:
|
||
logger.exception("batch sync failed for binding %s", binding.id)
|
||
# Create and persist a synthetic failed log if sync_binding crashed before creating one
|
||
synthetic_log = WebsiteSyncLog(
|
||
website_id=wid,
|
||
binding_id=binding.id,
|
||
target_group_id=binding.target_group_id,
|
||
target_group_name=binding.target_group_name,
|
||
algorithm=binding.algorithm,
|
||
percent=binding.percent,
|
||
source_rates_json="[]",
|
||
status="failed",
|
||
message=str(exc),
|
||
created_at=datetime.now(timezone.utc),
|
||
)
|
||
db.add(synthetic_log)
|
||
db.commit()
|
||
db.refresh(synthetic_log)
|
||
results.append(synthetic_log)
|
||
|
||
success_count = sum(1 for r in results if r.status == "success")
|
||
failed_count = sum(1 for r in results if r.status == "failed")
|
||
skipped_count = sum(1 for r in results if r.status == "skipped")
|
||
|
||
msg_parts = []
|
||
if success_count: msg_parts.append(f"成功 {success_count}")
|
||
if failed_count: msg_parts.append(f"失败 {failed_count}")
|
||
if skipped_count: msg_parts.append(f"跳过 {skipped_count}")
|
||
|
||
return WebsiteBatchSyncResponse(
|
||
total=len(bindings),
|
||
success=success_count,
|
||
failed=failed_count,
|
||
skipped=skipped_count,
|
||
message=f"同步完成：{'、'.join(msg_parts)}" if msg_parts else "同步完成",
|
||
logs=[_log_response(r) for r in results],
|
||
)
|
||
|
||
|
||
@router.post("/api/group-bindings", response_model=BindingResponse, status_code=201)
|
||
def create_binding(body: BindingCreate, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
website = db.query(Website).filter(Website.id == body.website_id).first()
|
||
if not website:
|
||
raise HTTPException(404, "website not found")
|
||
if body.algorithm not in ALGORITHMS:
|
||
raise HTTPException(400, "不支持的算法")
|
||
_ensure_unique_target(db, body.website_id, body.target_group_id)
|
||
row = WebsiteGroupBinding(
|
||
website_id=body.website_id,
|
||
target_group_id=body.target_group_id,
|
||
target_group_name=body.target_group_name,
|
||
source_groups_json=json.dumps([item.model_dump() for item in body.source_groups], ensure_ascii=False),
|
||
percent=str(body.percent),
|
||
algorithm=body.algorithm,
|
||
enabled=body.enabled,
|
||
)
|
||
db.add(row)
|
||
db.commit()
|
||
db.refresh(row)
|
||
try:
|
||
sync_binding(db, row, write=True)
|
||
except Exception as exc:
|
||
logger.exception("initial website sync failed for binding %s: %s", row.id, exc)
|
||
return _binding_response(db, row)
|
||
|
||
|
||
@router.put("/api/group-bindings/{bid}", response_model=BindingResponse)
|
||
def update_binding(bid: int, body: BindingUpdate, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
row = db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.id == bid).first()
|
||
if not row:
|
||
raise HTTPException(404, "binding not found")
|
||
data = body.model_dump(exclude_none=True)
|
||
if "website_id" in data and not db.query(Website).filter(Website.id == data["website_id"]).first():
|
||
raise HTTPException(404, "website not found")
|
||
if "algorithm" in data and data["algorithm"] not in ALGORITHMS:
|
||
raise HTTPException(400, "不支持的算法")
|
||
next_website_id = int(data.get("website_id", row.website_id))
|
||
next_target_group_id = str(data.get("target_group_id", row.target_group_id))
|
||
_ensure_unique_target(db, next_website_id, next_target_group_id, exclude_id=row.id)
|
||
if "source_groups" in data:
|
||
row.source_groups_json = json.dumps(data.pop("source_groups"), ensure_ascii=False)
|
||
if "percent" in data:
|
||
row.percent = str(data.pop("percent"))
|
||
for key, value in data.items():
|
||
setattr(row, key, value)
|
||
row.updated_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
db.refresh(row)
|
||
try:
|
||
sync_binding(db, row, write=True)
|
||
except Exception as exc:
|
||
logger.exception("sync failed after updating binding %s: %s", row.id, exc)
|
||
return _binding_response(db, row)
|
||
|
||
|
||
@router.delete("/api/group-bindings/{bid}", status_code=204)
|
||
def delete_binding(bid: int, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
row = db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.id == bid).first()
|
||
if not row:
|
||
raise HTTPException(404, "binding not found")
|
||
db.delete(row)
|
||
db.commit()
|
||
|
||
|
||
@router.post("/api/group-bindings/{bid}/sync-now", response_model=WebsiteSyncLogResponse)
|
||
def sync_now(bid: int, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||
row = db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.id == bid).first()
|
||
if not row:
|
||
raise HTTPException(404, "binding not found")
|
||
return _log_response(sync_binding(db, row, write=True))
|
||
|
||
|
||
@router.get("/api/website-sync-logs", response_model=List[WebsiteSyncLogResponse])
|
||
def list_sync_logs(
|
||
website_id: int | None = Query(None),
|
||
binding_id: int | None = Query(None),
|
||
limit: int = Query(50, le=200),
|
||
offset: int = Query(0),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(get_current_user),
|
||
):
|
||
q = db.query(WebsiteSyncLog)
|
||
if website_id:
|
||
q = q.filter(WebsiteSyncLog.website_id == website_id)
|
||
if binding_id:
|
||
q = q.filter(WebsiteSyncLog.binding_id == binding_id)
|
||
rows = q.order_by(WebsiteSyncLog.created_at.desc()).offset(offset).limit(limit).all()
|
||
return [_log_response(row) for row in rows]
|