feat: 实现设置账号并发数功能(支持 bulk-update 与 fallback 逐个更新)

This commit is contained in:
liumangmang
2026-07-02 09:48:59 +08:00
parent a296525d4e
commit 608ef00920
7 changed files with 570 additions and 5 deletions
+170 -1
View File
@@ -41,6 +41,9 @@ from app.schemas.website import (
CleanupInvalidAccountsItem, CleanupInvalidAccountsItem,
CleanupInvalidAccountsPreviewResponse, CleanupInvalidAccountsPreviewResponse,
CleanupInvalidAccountsExecuteResponse, CleanupInvalidAccountsExecuteResponse,
SetConcurrencyRequest,
SetConcurrencyItem,
SetConcurrencyResponse,
) )
from app.services.website_client import Sub2ApiWebsiteClient, _extract_id from app.services.website_client import Sub2ApiWebsiteClient, _extract_id
@@ -1106,6 +1109,172 @@ def execute_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _=
) )
@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")
c = _client(website)
try:
remote_accounts = c.list_accounts()
except Exception as e:
return SetConcurrencyResponse(
success=False,
message=f"拉取远端账号列表失败: {e}",
items=[]
)
remote_map = {}
if remote_accounts:
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
)
@router.post("/api/websites/{wid}/groups/organize", response_model=OrganizeGroupsResponse) @router.post("/api/websites/{wid}/groups/organize", response_model=OrganizeGroupsResponse)
def organize_website_groups( def organize_website_groups(
wid: int, wid: int,
@@ -1411,7 +1580,7 @@ def organize_website_groups(
}, },
"group_ids": group_ids, "group_ids": group_ids,
"rate_multiplier": 1, "rate_multiplier": 1,
"concurrency": 10, "concurrency": 100,
"priority": priority_val, "priority": priority_val,
"notes": f"Imported by SmartUp from upstream key #{row.id}", "notes": f"Imported by SmartUp from upstream key #{row.id}",
} }
+20 -1
View File
@@ -166,7 +166,7 @@ class ImportAccountsRequest(BaseModel):
account_name_prefix: str = "SmartUp" account_name_prefix: str = "SmartUp"
default_platform: str = "openai" default_platform: str = "openai"
platform_mode: str = "auto" # "auto" | "manual" platform_mode: str = "auto" # "auto" | "manual"
concurrency: int = Field(default=10, ge=1) concurrency: int = Field(default=100, ge=1, le=1000)
priority: int = Field(default=1, ge=0) priority: int = Field(default=1, ge=0)
auto_priority_by_rate: bool = True auto_priority_by_rate: bool = True
@@ -268,3 +268,22 @@ class CleanupInvalidAccountsExecuteResponse(BaseModel):
success: bool success: bool
message: str message: str
items: list[CleanupInvalidAccountsItem] items: list[CleanupInvalidAccountsItem]
class SetConcurrencyRequest(BaseModel):
concurrency: int = Field(default=100, ge=1, le=1000)
class SetConcurrencyItem(BaseModel):
account_id: str
account_name: Optional[str] = None
old_concurrency: Optional[int] = None
target_concurrency: int
status: str # "success" | "skipped" | "failed"
message: str
class SetConcurrencyResponse(BaseModel):
success: bool
message: str
items: list[SetConcurrencyItem]
+15
View File
@@ -297,6 +297,21 @@ class Sub2ApiWebsiteClient:
data = _unwrap_data(resp) data = _unwrap_data(resp)
return data if isinstance(data, dict) else {"value": data} return data if isinstance(data, dict) else {"value": data}
def bulk_update_accounts(self, account_ids: list[str], body: dict[str, Any], endpoint: str = "/accounts/bulk-update") -> dict[str, Any]:
"""批量更新账号。"""
int_ids = []
for aid in account_ids:
try:
int_ids.append(int(aid))
except ValueError:
pass
resp = self._request("POST", endpoint, {
"account_ids": int_ids,
**body
})
data = _unwrap_data(resp)
return data if isinstance(data, dict) else {"value": data}
@staticmethod @staticmethod
def _unwrap_list(value: dict) -> list | None: def _unwrap_list(value: dict) -> list | None:
"""递归展开嵌套的列表包装:data.items、data.data、items、accounts 等。""" """递归展开嵌套的列表包装:data.items、data.data、items、accounts 等。"""
+212
View File
@@ -0,0 +1,212 @@
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.database import Base
from app.models.upstream import Upstream
from app.models.upstream_key import UpstreamGeneratedKey
from app.models.website import Website
from app.routers.websites import set_website_accounts_concurrency
from app.schemas.website import SetConcurrencyRequest
from app.services.website_client import Sub2ApiWebsiteClient
@pytest.fixture()
def db_session():
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
from app.models import admin_user, upstream, snapshot, webhook_config, notification_log, custom_page, website, revoked_token, upstream_key
Base.metadata.create_all(bind=engine)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
Base.metadata.drop_all(bind=engine)
def test_set_website_accounts_concurrency_workflow(db_session, monkeypatch):
# 1. 创建测试用的网站与上游
w = Website(
id=1,
name="Sub2Api site",
site_type="sub2api",
base_url="http://sub2api",
api_prefix="api/v1",
auth_type="bearer",
auth_config_json='{"token": "tok1"}'
)
db_session.add(w)
up = Upstream(id=10, name="Upstream A", base_url="http://upstream-a")
db_session.add(up)
db_session.commit()
# 2. 插入本地导入的 Key 记录
# k1: 属于本站,远端正常存在的账号
k1 = UpstreamGeneratedKey(
id=101,
upstream_id=up.id,
group_id="g1",
key_name="key-101",
key_value="val-101",
status="active",
imported_website_id=w.id,
imported_account_id="1001",
)
# k2: 属于本站,但账号 ID 无法转为数字 (例如字符串 'abc')
k2 = UpstreamGeneratedKey(
id=102,
upstream_id=up.id,
group_id="g1",
key_name="key-102",
key_value="val-102",
status="active",
imported_website_id=w.id,
imported_account_id="abc",
)
# k3: 属于其他站的账号(不应受本次批量设置影响)
k3 = UpstreamGeneratedKey(
id=103,
upstream_id=up.id,
group_id="g1",
key_name="key-103",
key_value="val-103",
status="active",
imported_website_id=999, # 属于其他站
imported_account_id="1003",
)
# k4: 属于本站,但账号在远端已被删除
k4 = UpstreamGeneratedKey(
id=104,
upstream_id=up.id,
group_id="g1",
key_name="key-104",
key_value="val-104",
status="active",
imported_website_id=w.id,
imported_account_id="1004",
)
db_session.add_all([k1, k2, k3, k4])
db_session.commit()
# Mock Sub2Api 客户端
bulk_calls = []
update_calls = []
list_calls_count = 0
class MockClient:
def list_accounts(self):
nonlocal list_calls_count
list_calls_count += 1
return [
{"id": 1001, "name": "acc-1001", "concurrency": 10},
{"id": "abc", "name": "acc-abc", "concurrency": 10},
]
def extract_id(self, val):
return str(val.get("id"))
def bulk_update_accounts(self, account_ids, body):
bulk_calls.append((account_ids, body))
def update_account(self, account_id, body):
update_calls.append((account_id, body))
monkeypatch.setattr("app.routers.websites._client", lambda row: MockClient())
# ── 1. 执行成功设置并发数,期待调用 bulk_update_accounts ──
req = SetConcurrencyRequest(concurrency=200)
res = set_website_accounts_concurrency(wid=w.id, body=req, db=db_session)
assert res.success is True
assert "成功 1 个" in res.message
assert "跳过 2 个" in res.message
items = res.items
assert len(items) == 3
item_1001 = next(i for i in items if i.account_id == "1001")
assert item_1001.status == "success"
assert item_1001.old_concurrency == 10
assert item_1001.target_concurrency == 200
item_abc = next(i for i in items if i.account_id == "abc")
assert item_abc.status == "skipped"
assert "无法转换为数字" in item_abc.message
item_1004 = next(i for i in items if i.account_id == "1004")
assert item_1004.status == "skipped"
assert "不存在" in item_1004.message
assert len(bulk_calls) == 1
assert bulk_calls[0] == (["1001"], {"concurrency": 200})
assert len(update_calls) == 0
def test_set_website_accounts_concurrency_fallback(db_session, monkeypatch):
w = Website(
id=1,
name="Sub2Api site",
site_type="sub2api",
base_url="http://sub2api",
api_prefix="api/v1",
auth_type="bearer",
auth_config_json='{"token": "tok1"}'
)
db_session.add(w)
up = Upstream(id=10, name="Upstream A", base_url="http://upstream-a")
db_session.add(up)
db_session.commit()
k1 = UpstreamGeneratedKey(
id=101,
upstream_id=up.id,
group_id="g1",
key_name="key-101",
key_value="val-101",
status="active",
imported_website_id=w.id,
imported_account_id="1001",
)
db_session.add(k1)
db_session.commit()
bulk_calls = []
update_calls = []
class MockClientError:
def list_accounts(self):
return [{"id": 1001, "name": "acc-1001", "concurrency": 10}]
def extract_id(self, val):
return str(val.get("id"))
def bulk_update_accounts(self, account_ids, body):
bulk_calls.append((account_ids, body))
raise Exception("Bulk update is not supported by target server (404)")
def update_account(self, account_id, body):
update_calls.append((account_id, body))
monkeypatch.setattr("app.routers.websites._client", lambda row: MockClientError())
req = SetConcurrencyRequest(concurrency=150)
res = set_website_accounts_concurrency(wid=w.id, body=req, db=db_session)
assert res.success is True
assert "成功 1 个" in res.message
assert len(bulk_calls) == 1
assert len(update_calls) == 1
assert update_calls[0] == ("1001", {"concurrency": 150})
items = res.items
assert len(items) == 1
assert items[0].status == "success"
assert "逐个更新" in items[0].message
+1 -1
View File
@@ -121,7 +121,7 @@ def test_import_upstream_key_creates_sub2api_account_management_apikey(monkeypat
assert body["credentials"]["api_key"] == "sk-upstream-generated" assert body["credentials"]["api_key"] == "sk-upstream-generated"
assert body["credentials"]["base_url"] == "http://packy.local" assert body["credentials"]["base_url"] == "http://packy.local"
assert body["group_ids"] == [7] assert body["group_ids"] == [7]
assert body["concurrency"] == 10 assert body["concurrency"] == 100
assert body["priority"] == 1 assert body["priority"] == 1
assert body["credentials"]["base_url"] == "http://packy.local" assert body["credentials"]["base_url"] == "http://packy.local"
+11
View File
@@ -343,6 +343,15 @@ export interface CleanupInvalidAccountsItem {
message: string message: string
} }
export interface SetConcurrencyItem {
account_id: string
account_name: string | null
old_concurrency: number | null
target_concurrency: number
status: string
message: string
}
export const websitesApi = { export const websitesApi = {
list: () => api.get<WebsiteData[]>('/api/websites'), list: () => api.get<WebsiteData[]>('/api/websites'),
create: (data: WebsiteForm) => api.post<WebsiteData>('/api/websites', data), create: (data: WebsiteForm) => api.post<WebsiteData>('/api/websites', data),
@@ -374,6 +383,8 @@ export const websitesApi = {
api.post<{ success: boolean; message: string; items: CleanupInvalidAccountsItem[] }>(`/api/websites/${id}/accounts/cleanup-invalid/preview`), api.post<{ success: boolean; message: string; items: CleanupInvalidAccountsItem[] }>(`/api/websites/${id}/accounts/cleanup-invalid/preview`),
cleanupExecute: (id: number) => cleanupExecute: (id: number) =>
api.post<{ success: boolean; message: string; items: CleanupInvalidAccountsItem[] }>(`/api/websites/${id}/accounts/cleanup-invalid/execute`), api.post<{ success: boolean; message: string; items: CleanupInvalidAccountsItem[] }>(`/api/websites/${id}/accounts/cleanup-invalid/execute`),
setConcurrency: (id: number, data: { concurrency: number }) =>
api.post<{ success: boolean; message: string; items: SetConcurrencyItem[] }>(`/api/websites/${id}/accounts/set-concurrency`, data),
listBindings: () => api.get<GroupBindingData[]>('/api/group-bindings'), listBindings: () => api.get<GroupBindingData[]>('/api/group-bindings'),
createBinding: (data: GroupBindingForm) => api.post<GroupBindingData>('/api/group-bindings', data), createBinding: (data: GroupBindingForm) => api.post<GroupBindingData>('/api/group-bindings', data),
updateBinding: (id: number, data: Partial<GroupBindingForm>) => api.put<GroupBindingData>(`/api/group-bindings/${id}`, data), updateBinding: (id: number, data: Partial<GroupBindingForm>) => api.put<GroupBindingData>(`/api/group-bindings/${id}`, data),
+141 -2
View File
@@ -133,6 +133,15 @@
> >
清理失效账号 清理失效账号
</el-button> </el-button>
<el-button
size="small"
text
:disabled="!selectedWebsite"
@click="openConcurrencyDialog"
title="一键设置当前网站已导入账号的并发数"
>
设置账号并发
</el-button>
<el-button size="small" text :disabled="websites.length === 0" @click="openBindingCreate(selectedWebsite || websites[0])">新增绑定</el-button> <el-button size="small" text :disabled="websites.length === 0" @click="openBindingCreate(selectedWebsite || websites[0])">新增绑定</el-button>
</div> </div>
<div class="binding-list" v-loading="bindingLoading"> <div class="binding-list" v-loading="bindingLoading">
@@ -841,6 +850,87 @@
</div> </div>
</template> </template>
</el-dialog> </el-dialog>
<!-- 设置账号并发数弹窗 -->
<el-dialog
v-model="concurrencyDialog"
title="一键设置账号并发数"
width="850px"
destroy-on-close
>
<div v-if="!concurrencyHasExecuted">
<el-alert
title="功能说明"
type="warning"
show-icon
:closable="false"
style="margin-bottom: 15px;"
>
<div style="font-size: 13px; line-height: 1.5;">
该操作将<strong>仅修改由 SmartUp 导入并绑定到当前网站的账号并发数</strong>不会影响任何由目标站手工创建的其他账号<br />
如果目标站接口支持批量更新bulk-update系统将一次性批量设置否则将自动退回至逐个请求更新模式
</div>
</el-alert>
<el-form label-width="120px" style="margin-top: 20px;">
<el-form-item label="目标并发数">
<el-input-number
v-model="targetConcurrency"
:min="1"
:max="1000"
:step="10"
style="width: 200px;"
/>
<div style="font-size: 12px; color: var(--el-text-color-secondary); margin-top: 5px;">
请输入 1 1000 之间的并发值推荐设置为 100
</div>
</el-form-item>
</el-form>
</div>
<div v-else>
<div style="margin-bottom: 15px; font-weight: bold; color: var(--el-color-primary);">
{{ concurrencyMessage }}
</div>
<el-table :data="concurrencyResults" border stripe size="small" style="width: 100%; max-height: 400px; overflow-y: auto;">
<el-table-column prop="account_id" label="账号 ID" width="150" />
<el-table-column prop="account_name" label="账号名" min-width="180" show-overflow-tooltip />
<el-table-column prop="old_concurrency" label="当前并发" width="110" align="center">
<template #default="{ row }">
<span v-if="row.old_concurrency !== null">{{ row.old_concurrency }}</span>
<span v-else class="text-muted">-</span>
</template>
</el-table-column>
<el-table-column prop="target_concurrency" label="目标并发" width="110" align="center" />
<el-table-column prop="status" label="状态" width="100" align="center">
<template #default="{ row }">
<el-tag v-if="row.status === 'success'" type="success" size="small">成功</el-tag>
<el-tag v-else-if="row.status === 'skipped'" type="warning" size="small">跳过</el-tag>
<el-tag v-else-if="row.status === 'failed'" type="danger" size="small">失败</el-tag>
<el-tag v-else type="info" size="small">{{ row.status }}</el-tag>
</template>
</el-table-column>
<el-table-column prop="message" label="说明/原因" min-width="150" show-overflow-tooltip />
</el-table>
</div>
<template #footer>
<div class="dialog-footer">
<el-button @click="concurrencyDialog = false" :disabled="concurrencyExecuting">
{{ concurrencyHasExecuted ? '关闭' : '取消' }}
</el-button>
<el-button
v-if="!concurrencyHasExecuted"
type="primary"
:loading="concurrencyExecuting"
@click="submitSetConcurrency"
>
确认执行
</el-button>
</div>
</template>
</el-dialog>
</div> </div>
</template> </template>
@@ -866,6 +956,7 @@ import {
type WebsiteSyncLog, type WebsiteSyncLog,
type OrganizeGroupsItem, type OrganizeGroupsItem,
type CleanupInvalidAccountsItem, type CleanupInvalidAccountsItem,
type SetConcurrencyItem,
} from '@/api' } from '@/api'
const websites = ref<(WebsiteData & { _testing?: boolean })[]>([]) const websites = ref<(WebsiteData & { _testing?: boolean })[]>([])
@@ -1004,7 +1095,7 @@ const importAccountsForm = ref({
account_name_prefix: 'SmartUp', account_name_prefix: 'SmartUp',
default_platform: 'openai', default_platform: 'openai',
platform_mode: 'auto', platform_mode: 'auto',
concurrency: 10, concurrency: 100,
priority: 1, priority: 1,
auto_priority_by_rate: true, auto_priority_by_rate: true,
}) })
@@ -1021,6 +1112,13 @@ const cleanupExecuting = ref(false)
const cleanupHasExecuted = ref(false) const cleanupHasExecuted = ref(false)
const cleanupItems = ref<CleanupInvalidAccountsItem[]>([]) const cleanupItems = ref<CleanupInvalidAccountsItem[]>([])
const concurrencyDialog = ref(false)
const concurrencyExecuting = ref(false)
const concurrencyHasExecuted = ref(false)
const targetConcurrency = ref(100)
const concurrencyMessage = ref('')
const concurrencyResults = ref<SetConcurrencyItem[]>([])
const upstreamGroupOptions = computed(() => { const upstreamGroupOptions = computed(() => {
const rows: Array<{ key: string; label: string; rate: string | number; source: BindingSourceGroup }> = [] const rows: Array<{ key: string; label: string; rate: string | number; source: BindingSourceGroup }> = []
for (const upstream of upstreams.value) { for (const upstream of upstreams.value) {
@@ -1727,6 +1825,47 @@ async function executeCleanup() {
} }
} }
function openConcurrencyDialog() {
if (!selectedWebsite.value) return
targetConcurrency.value = 100
concurrencyHasExecuted.value = false
concurrencyResults.value = []
concurrencyMessage.value = ''
concurrencyDialog.value = true
}
async function submitSetConcurrency() {
if (!selectedWebsite.value) return
try {
await ElMessageBox.confirm(
`确认将当前网站中由 SmartUp 导入账号的并发数一键设置为 ${targetConcurrency.value}`,
'确认修改并发数',
{ type: 'warning' }
)
} catch {
return
}
concurrencyExecuting.value = true
try {
const res = await websitesApi.setConcurrency(selectedWebsite.value.id, {
concurrency: targetConcurrency.value
})
concurrencyMessage.value = res.data.message
concurrencyResults.value = res.data.items
concurrencyHasExecuted.value = true
if (res.data.success) {
ElMessage.success('设置完成')
} else {
ElMessage.warning(res.data.message || '部分账号设置并发失败')
}
} catch (e: any) {
ElMessage.error(e.response?.data?.detail || '设置并发失败')
} finally {
concurrencyExecuting.value = false
}
}
async function toggleBinding(row: GroupBindingData) { async function toggleBinding(row: GroupBindingData) {
try { try {
await websitesApi.updateBinding(row.id, { enabled: row.enabled }) await websitesApi.updateBinding(row.id, { enabled: row.enabled })
@@ -1797,7 +1936,7 @@ async function openImportAccounts(site?: WebsiteData | null) {
account_name_prefix: 'SmartUp', account_name_prefix: 'SmartUp',
default_platform: 'openai', default_platform: 'openai',
platform_mode: 'auto', platform_mode: 'auto',
concurrency: 10, concurrency: 100,
priority: 1, priority: 1,
auto_priority_by_rate: true, auto_priority_by_rate: true,
} }