feat(proxy): 优化代理配置管理功能

- 更新 proxy_url 属性为 get_proxy_url 方法，支持多级代理优先级
- 实现代理获取三路优先级：动态代理 > 代理池 > 静态代理
- 添加取消默认代理功能和 unset_proxy_default 接口
- 实现批量导入代理功能，支持多种格式解析
- 在前端界面添加批量导入代理按钮和模态框
- 重构代理设置页面的默认代理切换交互
- 更新支付流程中的代理获取方式
- 添加 UUID 依赖并优化支付请求头配置

Author: cnlimiter

src/config/settings.py

@@ -638,25 +638,40 @@ def validate_database_url(cls, v):
     proxy_dynamic_api_key_header: str = "X-API-Key"
     proxy_dynamic_result_field: str = ""
 
-    @property
-    def proxy_url(self) -> Optional[str]:
-        """获取完整的代理 URL"""
-        if not self.proxy_enabled:
-            return None
-
-        if self.proxy_type == "http":
-            scheme = "http"
-        elif self.proxy_type == "socks5":
-            scheme = "socks5"
-        else:
-            return None
-
-        auth = ""
-        if self.proxy_username and self.proxy_password:
-            auth = f"{self.proxy_username}:{self.proxy_password.get_secret_value()}@"
-
-        return f"{scheme}://{auth}{self.proxy_host}:{self.proxy_port}"
-
+    def get_proxy_url(self, db=None) -> Optional[str]:
+        """获取当前可用的代理 URL（三路优先级）
+
+        优先级：动态代理 > 代理池（默认/随机）> 静态代理 > None
+
+        Args:
+            db: 可选的数据库 session，传入时检查代理池；不传则跳过代理池
+        """
+        # 1. 动态代理
+        if self.proxy_dynamic_enabled and self.proxy_dynamic_api_url:
+            return self.proxy_dynamic_api_url
+
+        # 2 & 3. 代理池（优先 is_default，否则随机）
+        if db is not None:
+            from src.database import crud
+            proxy = crud.get_random_proxy(db)
+            if proxy is not None:
+                return proxy.proxy_url
+
+        # 4. 静态代理
+        if self.proxy_enabled:
+            if self.proxy_type == "http":
+                scheme = "http"
+            elif self.proxy_type == "socks5":
+                scheme = "socks5"
+            else:
+                return None
+            auth = ""
+            if self.proxy_username and self.proxy_password:
+                auth = f"{self.proxy_username}:{self.proxy_password.get_secret_value()}@"
+            return f"{scheme}://{auth}{self.proxy_host}:{self.proxy_port}"
+
+        # 5. 无可用代理
+        return None
     # 注册配置
     registration_max_retries: int = 3
     registration_timeout: int = 120

src/core/dynamic_proxy.py

@@ -115,4 +115,4 @@ def get_proxy_url_for_task() -> Optional[str]:
         logger.warning("动态代理获取失败，回退到静态代理")
 
     # 使用静态代理
-    return settings.proxy_url
+    return settings.get_proxy_url()

src/core/openai/payment.py

@@ -5,6 +5,7 @@
 import logging
 import subprocess
 import sys
+import uuid
 from typing import Optional
 
 from curl_cffi import requests as cffi_requests
@@ -154,12 +155,8 @@ def generate_team_link(
         "Authorization": f"Bearer {account.access_token}",
         "Content-Type": "application/json",
         "oai-language": "zh-CN",
+        "oai-device-id": str(uuid.uuid4()),
     }
-    if account.cookies:
-        headers["cookie"] = account.cookies
-        oai_did = _extract_oai_did(account.cookies)
-        if oai_did:
-            headers["oai-device-id"] = oai_did
 
     payload = {
         "plan_name": "chatgptteamplan",
@@ -171,7 +168,7 @@ def generate_team_link(
         "billing_details": {"country": country, "currency": currency},
         "promo_campaign": {
             "promo_campaign_id": "team-1-month-free",
-            "is_coupon_from_query_param": True,
+            "is_coupon_from_query_param": False,
         },
         "cancel_url": "https://chatgpt.com/#pricing",
         "checkout_ui_mode": "custom",
@@ -187,9 +184,35 @@ def generate_team_link(
     )
     resp.raise_for_status()
     data = resp.json()
-    if "checkout_session_id" in data:
-        return TEAM_CHECKOUT_BASE_URL + data["checkout_session_id"]
-    raise ValueError(data.get("detail", "API 未返回 checkout_session_id"))
+    resp2 = cffi_requests.post(
+        "https://api.stripe.com/v1/payment_pages/" + data["checkout_session_id"],
+        headers={
+            "Content-Type": "application/x-www-form-urlencoded",
+            "accept": "application/json",
+            "referer": "https://js.stripe.com/"
+        },
+        data=f"tax_region[country]={country}"
+             "&elements_session_client[client_betas][0]=custom_checkout_server_updates_1"
+             "&elements_session_client[client_betas][1]=custom_checkout_manual_approval_1"
+             "&elements_session_client[elements_init_source]=custom_checkout"
+             "&elements_session_client[referrer_host]=chatgpt.com"
+             "&elements_session_client[session_id]=elements_session_1rr8sS4PKIY"
+             "&elements_session_client[stripe_js_id]=72d6a553-c2fb-4f85-941e-8022c8335a85"
+             "&elements_session_client[locale]=zh"
+             "&elements_session_client[is_aggregation_expected]=false"
+             "&client_attribution_metadata[merchant_integration_additional_elements][0]=payment"
+             "&client_attribution_metadata[merchant_integration_additional_elements][1]=address"
+             f"&key={data["publishable_key"]}"
+        ,
+        proxies=_build_proxies(proxy),
+        timeout=30,
+        impersonate="chrome110",
+    )
+    resp2.raise_for_status()
+    data2 = resp2.json()
+    if "stripe_hosted_url" in data2:
+        return data2["stripe_hosted_url"]
+    raise ValueError(data.get("detail", "API 未返回 stripe_hosted_url"))
 
 
 def open_url_incognito(url: str, cookies_str: Optional[str] = None) -> bool:

src/database/crud.py

@@ -579,6 +579,16 @@ def set_proxy_default(db: Session, proxy_id: int) -> Optional[Proxy]:
     return proxy
 
 
+def unset_proxy_default(db: Session, proxy_id: int) -> Optional[Proxy]:
+    """取消指定代理的默认标记"""
+    proxy = db.query(Proxy).filter(Proxy.id == proxy_id).first()
+    if proxy:
+        proxy.is_default = False
+        db.commit()
+        db.refresh(proxy)
+    return proxy
+
+
 def get_proxies_count(db: Session, enabled: Optional[bool] = None) -> int:
     """获取代理数量"""
     query = db.query(func.count(Proxy.id))

src/web/routes/accounts.py

@@ -135,7 +135,7 @@ def _get_proxy(request_proxy: Optional[str] = None) -> Optional[str]:
     proxy_url = get_proxy_url_for_task()
     if proxy_url:
         return proxy_url
-    return get_settings().proxy_url
+    return get_settings().get_proxy_url()
 
 
 # ============== Pydantic Models ==============

src/web/routes/payment.py

@@ -66,7 +66,7 @@ def generate_payment_link(request: GenerateLinkRequest):
         if not account:
             raise HTTPException(status_code=404, detail="账号不存在")
 
-        proxy = request.proxy or get_settings().proxy_url
+        proxy = request.proxy or get_settings().get_proxy_url(db=db)
 
         try:
             if request.plan_type == "plus":
@@ -125,11 +125,10 @@ def open_browser_incognito(request: OpenIncognitoRequest):
 @router.post("/accounts/batch-check-subscription")
 def batch_check_subscription(request: BatchCheckSubscriptionRequest):
     """批量检测账号订阅状态"""
-    proxy = request.proxy or get_settings().proxy_url
-
     results = {"success_count": 0, "failed_count": 0, "details": []}
 
     with get_db() as db:
+        proxy = request.proxy or get_settings().get_proxy_url(db=db)
         ids = resolve_account_ids(
             db, request.ids, request.select_all,
             request.status_filter, request.email_service_filter, request.search_filter

src/web/routes/registration.py

@@ -53,25 +53,31 @@ def get_proxy_for_registration(
     获取用于注册的代理
 
     策略：
-    1. 优先从代理列表中随机选择一个启用的代理
-    2. 如果代理列表为空且启用了动态代理，调用动态代理 API 获取
-    3. 否则使用系统设置中的静态默认代理
+    1. 优先使用动态代理（若已启用）
+    2. 动态代理不可用时，使用代理池（有默认代理则走默认，否则随机轮询）
+    3. 代理池为空时，使用系统静态代理配置兜底
 
     Returns:
         Tuple[proxy_url, proxy_id]: 代理 URL 和代理 ID（如果来自代理列表）
     """
-    # 先尝试从代理列表中获取
+    from ...core.dynamic_proxy import get_proxy_url_for_task
+
+    settings = get_settings()
+
+    # 1. 优先动态代理
+    if settings.proxy_dynamic_enabled and settings.proxy_dynamic_api_url:
+        proxy_url = get_proxy_url_for_task()
+        if proxy_url:
+            return proxy_url, None
+        logger.warning("动态代理获取失败，回退到代理池")
+
+    # 2. 代理池（内部已实现：有默认走默认，无默认随机轮询）
     proxy = crud.get_random_proxy(db, exclude_ids=exclude_proxy_ids)
     if proxy:
         return proxy.proxy_url, proxy.id
 
-    # 代理列表为空，尝试动态代理或静态代理
-    from ...core.dynamic_proxy import get_proxy_url_for_task
-    proxy_url = get_proxy_url_for_task()
-    if proxy_url:
-        return proxy_url, None
-
-    return None, None
+    # 3. 静态代理兜底
+    return settings.get_proxy_url(), None
 
 
 def update_proxy_usage(db, proxy_id: Optional[int]):

src/web/routes/settings.py

@@ -608,6 +608,94 @@ async def set_proxy_default(proxy_id: int):
         return {"success": True, "proxy": proxy.to_dict()}
 
 
+@router.post("/proxies/{proxy_id}/unset-default")
+async def unset_proxy_default(proxy_id: int):
+    """取消指定代理的默认标记"""
+    with get_db() as db:
+        proxy = crud.unset_proxy_default(db, proxy_id)
+        if not proxy:
+            raise HTTPException(status_code=404, detail="代理不存在")
+        return {"success": True, "proxy": proxy.to_dict()}
+
+
+class ProxyBatchImportRequest(BaseModel):
+    """批量导入代理请求"""
+    lines: str
+
+
+def _parse_proxy_line(line: str):
+    """
+    解析单行代理字符串，支持格式：
+    - host:port
+    - type://host:port
+    - type://user:pass@host:port
+    - 名称|type://user:pass@host:port
+    """
+    from urllib.parse import urlparse
+
+    name = None
+    # 解析可选名称前缀（竖线分隔）
+    if '|' in line:
+        name, line = line.split('|', 1)
+        name = name.strip()
+        line = line.strip()
+
+    # 若没有协议头，默认补 http://
+    if '://' not in line:
+        line = 'http://' + line
+
+    parsed = urlparse(line)
+    proxy_type = (parsed.scheme or 'http').lower()
+    if proxy_type not in ('http', 'https', 'socks5', 'socks4'):
+        proxy_type = 'http'
+
+    host = parsed.hostname
+    port = parsed.port
+    username = parsed.username or None
+    password = parsed.password or None
+
+    if not host or not port:
+        raise ValueError(f"无法解析 host/port")
+
+    if not name:
+        name = f"{proxy_type}://{host}:{port}"
+
+    return name, proxy_type, host, port, username, password
+
+
+@router.post("/proxies/batch-import")
+async def batch_import_proxies(request: ProxyBatchImportRequest):
+    """
+    批量导入代理，每行格式支持：
+    - host:port
+    - type://host:port
+    - type://user:pass@host:port
+    - 名称|type://user:pass@host:port
+    """
+    results = {"success": 0, "failed": 0, "errors": []}
+    with get_db() as db:
+        for raw_line in request.lines.splitlines():
+            line = raw_line.strip()
+            if not line or line.startswith("#"):
+                continue
+            try:
+                name, proxy_type, host, port, username, password = _parse_proxy_line(line)
+                crud.create_proxy(
+                    db,
+                    name=name,
+                    type=proxy_type,
+                    host=host,
+                    port=port,
+                    username=username,
+                    password=password,
+                )
+                results["success"] += 1
+            except Exception as e:
+                results["failed"] += 1
+                results["errors"].append(f"{raw_line}: {e}")
+    return results
+
+
 @router.post("/proxies/{proxy_id}/test")
 async def test_proxy_item(proxy_id: int):
     """测试单个代理"""