LiuMangMang/SmartUp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(website): 支持基于现有绑定一键整理分组，支持账号校验补齐、自愈重建与缺失 Key 提示 (含异常保守跳过及名称层级匹配)

Browse Source
This commit is contained in:
liumangmang
2026-07-01 11:21:20 +08:00
parent bf5470caf3
commit 397a14c978
5 changed files with 828 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/routers/websites.py
+375
View File
@@ -34,6 +34,8 @@ from app.schemas.website import (
WebsiteSyncLogResponse,
WebsiteUpdate,
WebsiteBatchSyncResponse,
OrganizeGroupsItem,
OrganizeGroupsResponse,
)
from app.services.website_client import Sub2ApiWebsiteClient, _extract_id
@@ -43,6 +45,7 @@ from app.services.website_sync import (
build_rate_priority_map,
reconcile_upstream_keys_full,
sync_account_priorities_for_upstream,
latest_rate_map,
)
from app.utils.auth import get_current_user
from app.utils.number import fixed_decimal_number
@@ -767,6 +770,378 @@ def import_upstream_keys_as_accounts(
)
@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()
)
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[str, int] = {}
try:
rate_priority_map = _build_rate_priority_map(db, upstream_ids)
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
# 对每个绑定关系遍历其对应的上游 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
upstream_name = db.query(Upstream.name).filter(Upstream.id == uid).scalar() or f"#{uid}"
# 来源名称优先取 src.group_name，再 fallback 快照，再 fallback ID
source_group_name = src.get("group_name") or ""
if not source_group_name:
snapshot_groups = {}
try:
snapshot_groups = latest_rate_map(db, uid)
except Exception:
pass
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 = db.query(Upstream).filter(Upstream.id == uid).first()
if _u:
upstream_base_url = _u.base_url
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
# 检测平台类型
platform = _detect_platform(
f"{row.group_name} {row.group_id} {row.key_name}",
"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:
# 目标账号存在：跳过或补齐目标分组记录
try:
current_group_ids = remote_acc.get("group_ids") or []
if not isinstance(current_group_ids, list):
current_group_ids = [current_group_ids]
str_group_ids = [str(g) for g in current_group_ids]
if str(target_group_id) not in str_group_ids:
# 需要补齐绑定关系
numeric_target = _numeric_group_id(target_group_id)
target_val = numeric_target if numeric_target is not None else target_group_id
new_group_ids = list(current_group_ids)
if target_val not in new_group_ids:
new_group_ids.append(target_val)
c.update_account(old_account_id, {"group_ids": new_group_ids})
# 顺便更新下 remote_account_map 中的数据，防止同个账号后续在其他绑定中重复补齐
remote_acc["group_ids"] = new_group_ids
msg = "账号已存在，已补齐目标分组绑定"
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 = f"SmartUp-{row.group_name or row.group_id}-{row.id}"
priority_val = rate_priority_map.get(f"{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": 10,
"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")
exists_count = sum(1 for item in items if item.status == "exists")
missing_key_count = sum(1 for item in items if item.status == "missing_key")
recreated_count = sum(1 for item in items if item.status == "recreated")
failed_count = sum(1 for item in items if item.status == "failed")
parts = []
total_created = created_count + recreated_count
if total_created:
parts.append(f"创建账号 {total_created}")
if exists_count:
parts.append(f"已存在 {exists_count}")
if missing_key_count:
parts.append(f"缺少 Key {missing_key_count}")
if failed_count:
parts.append(f"失败 {failed_count}")
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()
backend/app/schemas/website.py
+19
View File
@@ -213,3 +213,22 @@ class WebsiteBatchSyncResponse(BaseModel):
skipped: int
message: str
logs: list[WebsiteSyncLogResponse]
class OrganizeGroupsItem(BaseModel):
target_group_id: str
target_group_name: str
upstream_name: str
source_group_id: str
source_group_name: str
key_name: Optional[str] = None
account_id: Optional[str] = None
account_name: Optional[str] = None
status: str
message: str = ""
class OrganizeGroupsResponse(BaseModel):
success: bool
message: str
items: list[OrganizeGroupsItem]