Coordinate account bootstrap, preview, QR creation, and polling.
| 96 | def health_payload(self) -> dict[str, Any]: |
| 97 | return { |
| 98 | "status": "ok", |
| 99 | "transport": self.bigmodel_client.transport_name, |
| 100 | "ocr": self.ocr_service.status_payload(), |
| 101 | "tdc": self.tdc_service.status_payload(), |
| 102 | } |
| 103 | |
| 104 | def list_accounts(self): |
| 105 | return self.state_service.list_accounts() |
| 106 | |
| 107 | def get_account_detail(self, account_id: str) -> AccountDetailResponse: |
| 108 | return self.state_service.get_account_detail(account_id) |
| 109 | |
| 110 | def import_account(self, request): |
| 111 | return self.state_service.import_account(request) |
| 112 | |
| 113 | def update_preferences(self, account_id: str, request: AccountPreferencesRequest) -> AccountDetailResponse: |
| 114 | return self.state_service.update_preferences(account_id, request) |
| 115 | |
| 116 | def delete_account(self, account_id: str) -> dict[str, Any]: |
| 117 | self.state_service.delete_account(account_id) |
| 118 | return {"account_id": account_id, "deleted": True} |
| 119 | |
| 120 | def bootstrap_account( |
| 121 | self, |
| 122 | account_id: str, |
| 123 | *, |
| 124 | refresh_fingerprint: bool = False, |
| 125 | flow: FlowRun | None = None, |
| 126 | ) -> AccountDetailResponse: |
| 127 | own_flow = flow is None |
| 128 | if own_flow: |
| 129 | flow = self.runtime_logs.start_run( |
| 130 | account_id=account_id, |
| 131 | action="bootstrap_account", |
| 132 | source="refresh_fingerprint" if refresh_fingerprint else "manual", |
| 133 | ) |
| 134 | try: |
| 135 | if refresh_fingerprint: |
| 136 | rotated = self.state_service.rotate_browser_impersonate(account_id) |
| 137 | self.runtime_logs.log_event( |
| 138 | flow, |
| 139 | stage="fingerprint", |
| 140 | status="rotated", |
| 141 | message="账号指纹已刷新", |
| 142 | details={"browser_impersonate": rotated.browser_impersonate}, |
| 143 | ) |
| 144 | account = self.state_service.get_account(account_id) |
| 145 | session = self.state_service.load_session(account_id) |
| 146 | |
| 147 | self.runtime_logs.log_event( |
| 148 | flow, |
| 149 | stage="get_customer_info", |
| 150 | status="started", |
| 151 | message="开始同步账号上下文", |
| 152 | details={"browser_impersonate": account.browser_impersonate}, |
| 153 | ) |
| 154 | user_info_result = self.bigmodel_client.get_customer_info(account, session) |
| 155 | user_info = self._ensure_dict(user_info_result.data, label="getCustomerInfo.data") |
no outgoing calls
no test coverage detected