From 608ef009209cfa31f0ed1c6b9777816db30c0d67 Mon Sep 17 00:00:00 2001 From: liumangmang Date: Thu, 2 Jul 2026 09:48:54 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E8=AE=BE=E7=BD=AE?= =?UTF-8?q?=E8=B4=A6=E5=8F=B7=E5=B9=B6=E5=8F=91=E6=95=B0=E5=8A=9F=E8=83=BD?= =?UTF-8?q?=EF=BC=88=E6=94=AF=E6=8C=81=20bulk-update=20=E4=B8=8E=20fallbac?= =?UTF-8?q?k=20=E9=80=90=E4=B8=AA=E6=9B=B4=E6=96=B0=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/routers/websites.py | 171 +++++++++++++++- backend/app/schemas/website.py | 21 +- backend/app/services/website_client.py | 15 ++ backend/test_account_concurrency.py | 212 ++++++++++++++++++++ backend/test_upstream_key_account_import.py | 2 +- frontend/src/api/index.ts | 11 + frontend/src/views/Websites.vue | 143 ++++++++++++- 7 files changed, 570 insertions(+), 5 deletions(-) create mode 100644 backend/test_account_concurrency.py diff --git a/backend/app/routers/websites.py b/backend/app/routers/websites.py index 71dd6b1..b000024 100644 --- a/backend/app/routers/websites.py +++ b/backend/app/routers/websites.py @@ -41,6 +41,9 @@ from app.schemas.website import ( CleanupInvalidAccountsItem, CleanupInvalidAccountsPreviewResponse, CleanupInvalidAccountsExecuteResponse, + SetConcurrencyRequest, + SetConcurrencyItem, + SetConcurrencyResponse, ) 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) def organize_website_groups( wid: int, @@ -1411,7 +1580,7 @@ def organize_website_groups( }, "group_ids": group_ids, "rate_multiplier": 1, - "concurrency": 10, + "concurrency": 100, "priority": priority_val, "notes": f"Imported by SmartUp from upstream key #{row.id}", } diff --git a/backend/app/schemas/website.py b/backend/app/schemas/website.py index 62731d9..3c4af20 100644 --- a/backend/app/schemas/website.py +++ b/backend/app/schemas/website.py @@ -166,7 +166,7 @@ class ImportAccountsRequest(BaseModel): account_name_prefix: str = "SmartUp" default_platform: str = "openai" 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) auto_priority_by_rate: bool = True @@ -268,3 +268,22 @@ class CleanupInvalidAccountsExecuteResponse(BaseModel): success: bool message: str 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] diff --git a/backend/app/services/website_client.py b/backend/app/services/website_client.py index 396fefa..3f3be06 100644 --- a/backend/app/services/website_client.py +++ b/backend/app/services/website_client.py @@ -297,6 +297,21 @@ class Sub2ApiWebsiteClient: data = _unwrap_data(resp) 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 def _unwrap_list(value: dict) -> list | None: """递归展开嵌套的列表包装：data.items、data.data、items、accounts 等。""" diff --git a/backend/test_account_concurrency.py b/backend/test_account_concurrency.py new file mode 100644 index 0000000..7c3c275 --- /dev/null +++ b/backend/test_account_concurrency.py @@ -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 diff --git a/backend/test_upstream_key_account_import.py b/backend/test_upstream_key_account_import.py index 7abb896..9cf8908 100644 --- a/backend/test_upstream_key_account_import.py +++ b/backend/test_upstream_key_account_import.py @@ -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"]["base_url"] == "http://packy.local" assert body["group_ids"] == [7] - assert body["concurrency"] == 10 + assert body["concurrency"] == 100 assert body["priority"] == 1 assert body["credentials"]["base_url"] == "http://packy.local" diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 65e03a9..dee6124 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -343,6 +343,15 @@ export interface CleanupInvalidAccountsItem { 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 = { list: () => api.get('/api/websites'), create: (data: WebsiteForm) => api.post('/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`), cleanupExecute: (id: number) => 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('/api/group-bindings'), createBinding: (data: GroupBindingForm) => api.post('/api/group-bindings', data), updateBinding: (id: number, data: Partial) => api.put(`/api/group-bindings/${id}`, data), diff --git a/frontend/src/views/Websites.vue b/frontend/src/views/Websites.vue index f564d77..475f77b 100644 --- a/frontend/src/views/Websites.vue +++ b/frontend/src/views/Websites.vue @@ -133,6 +133,15 @@ > 清理失效账号 + + 设置账号并发 + 新增绑定
@@ -841,6 +850,87 @@
+ + + +
+ +
+ 该操作将仅修改由 SmartUp 导入并绑定到当前网站的账号并发数，不会影响任何由目标站手工创建的其他账号。
+ 如果目标站接口支持批量更新（bulk-update），系统将一次性批量设置；否则将自动退回至逐个请求更新模式。 +
+ + + + + +
+ 请输入 1 到 1000 之间的并发值，推荐设置为 100。 +
+ + +
+ +
+
+ {{ concurrencyMessage }} +
+ + + + + + + + + + + + + +
+ + + @@ -866,6 +956,7 @@ import { type WebsiteSyncLog, type OrganizeGroupsItem, type CleanupInvalidAccountsItem, + type SetConcurrencyItem, } from '@/api' const websites = ref<(WebsiteData & { _testing?: boolean })[]>([]) @@ -1004,7 +1095,7 @@ const importAccountsForm = ref({ account_name_prefix: 'SmartUp', default_platform: 'openai', platform_mode: 'auto', - concurrency: 10, + concurrency: 100, priority: 1, auto_priority_by_rate: true, }) @@ -1021,6 +1112,13 @@ const cleanupExecuting = ref(false) const cleanupHasExecuted = ref(false) const cleanupItems = ref([]) +const concurrencyDialog = ref(false) +const concurrencyExecuting = ref(false) +const concurrencyHasExecuted = ref(false) +const targetConcurrency = ref(100) +const concurrencyMessage = ref('') +const concurrencyResults = ref([]) + const upstreamGroupOptions = computed(() => { const rows: Array<{ key: string; label: string; rate: string | number; source: BindingSourceGroup }> = [] 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) { try { await websitesApi.updateBinding(row.id, { enabled: row.enabled }) @@ -1797,7 +1936,7 @@ async function openImportAccounts(site?: WebsiteData | null) { account_name_prefix: 'SmartUp', default_platform: 'openai', platform_mode: 'auto', - concurrency: 10, + concurrency: 100, priority: 1, auto_priority_by_rate: true, }