获取一次邮件列表,按新旧接口顺序回退: 1) /api/mails (Bearer jwt) 2) /user_api/mails (x-user-token) 3) /admin/mails (按地址过滤) 4) /admin/mails (不过滤,客户端二次筛选)
(self, email: str, jwt: Optional[str], email_id: Optional[str] = None)
| 345 | return target in text_blob |
| 346 | |
| 347 | def _fetch_mails_once(self, email: str, jwt: Optional[str], email_id: Optional[str] = None) -> List[Dict[str, Any]]: |
| 348 | """ |
| 349 | 获取一次邮件列表,按新旧接口顺序回退: |
| 350 | 1) /api/mails (Bearer jwt) |
| 351 | 2) /user_api/mails (x-user-token) |
| 352 | 3) /admin/mails (按地址过滤) |
| 353 | 4) /admin/mails (不过滤,客户端二次筛选) |
| 354 | """ |
| 355 | attempts: List[Dict[str, Any]] = [] |
| 356 | admin_limit = self._normalize_admin_limit(self.config.get("admin_mails_limit", 50), default=50) |
| 357 | if jwt: |
| 358 | attempts.extend([ |
| 359 | { |
| 360 | "path": "/api/mails", |
| 361 | "params": {"limit": 50, "offset": 0}, |
| 362 | "headers": { |
| 363 | "Authorization": f"Bearer {jwt}", |
| 364 | "Accept": "application/json", |
| 365 | }, |
| 366 | }, |
| 367 | { |
| 368 | "path": "/user_api/mails", |
| 369 | "params": {"limit": 50, "offset": 0}, |
| 370 | "headers": { |
| 371 | "x-user-token": jwt, |
| 372 | "Accept": "application/json", |
| 373 | }, |
| 374 | }, |
| 375 | ]) |
| 376 | |
| 377 | attempts.append({ |
| 378 | "path": "/admin/mails", |
| 379 | "params": {"limit": admin_limit, "offset": 0, "address": email}, |
| 380 | "headers": {"Accept": "application/json"}, |
| 381 | }) |
| 382 | if email_id and email_id != email: |
| 383 | attempts.append({ |
| 384 | "path": "/admin/mails", |
| 385 | "params": {"limit": admin_limit, "offset": 0, "address": email_id}, |
| 386 | "headers": {"Accept": "application/json"}, |
| 387 | }) |
| 388 | attempts.append({ |
| 389 | "path": "/admin/mails", |
| 390 | "params": {"limit": admin_limit, "offset": 0}, |
| 391 | "headers": {"Accept": "application/json"}, |
| 392 | }) |
| 393 | |
| 394 | for attempt in attempts: |
| 395 | path = attempt["path"] |
| 396 | try: |
| 397 | if path == "/admin/mails": |
| 398 | raw_params = dict(attempt["params"] or {}) |
| 399 | preferred_limit = raw_params.pop("limit", admin_limit) |
| 400 | offset = raw_params.pop("offset", 0) |
| 401 | response = self._request_admin_mails_with_limit_fallback( |
| 402 | offset=offset, |
| 403 | extra_params=raw_params, |
| 404 | preferred_limit=preferred_limit, |
no test coverage detected