Add a new token and prepare its pooled projects.
(
self,
st: str,
project_id: Optional[str] = None,
project_name: Optional[str] = None,
remark: Optional[str] = None,
image_enabled: bool = True,
video_enabled: bool = True,
image_concurrency: int = -1,
video_concurrency: int = -1,
captcha_proxy_url: Optional[str] = None,
extension_route_key: Optional[str] = None,
protocol_mode: str = "session",
google_cookies: Optional[str] = None,
login_account: Optional[str] = None,
login_password: Optional[str] = None,
proxy_url: Optional[str] = None,
auto_refresh_enabled: bool = True,
refresh_interval_minutes: int = 120,
)
| 269 | # ========== Token添加 (支持Project创建) ========== |
| 270 | |
| 271 | async def add_token( |
| 272 | self, |
| 273 | st: str, |
| 274 | project_id: Optional[str] = None, |
| 275 | project_name: Optional[str] = None, |
| 276 | remark: Optional[str] = None, |
| 277 | image_enabled: bool = True, |
| 278 | video_enabled: bool = True, |
| 279 | image_concurrency: int = -1, |
| 280 | video_concurrency: int = -1, |
| 281 | captcha_proxy_url: Optional[str] = None, |
| 282 | extension_route_key: Optional[str] = None, |
| 283 | protocol_mode: str = "session", |
| 284 | google_cookies: Optional[str] = None, |
| 285 | login_account: Optional[str] = None, |
| 286 | login_password: Optional[str] = None, |
| 287 | proxy_url: Optional[str] = None, |
| 288 | auto_refresh_enabled: bool = True, |
| 289 | refresh_interval_minutes: int = 120, |
| 290 | ) -> Token: |
| 291 | """Add a new token and prepare its pooled projects.""" |
| 292 | existing_token = await self.db.get_token_by_st(st) |
| 293 | if existing_token: |
| 294 | raise ValueError(f"Token ??????: {existing_token.email}?") |
| 295 | |
| 296 | debug_logger.log_info(f"[ADD_TOKEN] Converting ST to AT...") |
| 297 | try: |
| 298 | result = await self.flow_client.st_to_at(st) |
| 299 | at = result["access_token"] |
| 300 | expires = result.get("expires") |
| 301 | user_info = result.get("user", {}) |
| 302 | email = user_info.get("email", "") |
| 303 | name = user_info.get("name", email.split("@")[0] if email else "") |
| 304 | at_expires = None |
| 305 | if expires: |
| 306 | try: |
| 307 | at_expires = datetime.fromisoformat(expires.replace('Z', '+00:00')) |
| 308 | except Exception: |
| 309 | pass |
| 310 | except Exception as e: |
| 311 | raise ValueError(f"ST?AT??: {str(e)}") |
| 312 | |
| 313 | try: |
| 314 | credits_result = await self.flow_client.get_credits(at) |
| 315 | credits = credits_result.get("credits", 0) |
| 316 | user_paygate_tier = credits_result.get("userPaygateTier") |
| 317 | except Exception: |
| 318 | credits = 0 |
| 319 | user_paygate_tier = None |
| 320 | |
| 321 | base_project_name = self._normalize_project_name_base(project_name) |
| 322 | project_pool_size = self._get_project_pool_size() |
| 323 | pooled_projects: List[Project] = [] |
| 324 | |
| 325 | if project_id: |
| 326 | first_project_name = self._build_project_name(1, base_project_name) |
| 327 | debug_logger.log_info(f"[ADD_TOKEN] Using provided project_id as pooled project #1: {project_id}") |
| 328 | pooled_projects.append(Project( |
nothing calls this directly
no test coverage detected