根据媒体类型解析下载代理地址。
(
self,
media_type: str,
fingerprint: Optional[Dict[str, Any]] = None,
)
| 57 | return None |
| 58 | |
| 59 | async def _resolve_download_proxy( |
| 60 | self, |
| 61 | media_type: str, |
| 62 | fingerprint: Optional[Dict[str, Any]] = None, |
| 63 | ) -> Optional[str]: |
| 64 | """根据媒体类型解析下载代理地址。""" |
| 65 | if isinstance(fingerprint, dict): |
| 66 | fingerprint_proxy = str(fingerprint.get("proxy_url") or "").strip() |
| 67 | if fingerprint_proxy: |
| 68 | return fingerprint_proxy |
| 69 | |
| 70 | if not self.proxy_manager: |
| 71 | return None |
| 72 | |
| 73 | try: |
| 74 | # 媒体下载（图片/视频）优先使用独立的上传/下载代理 |
| 75 | if media_type in ("image", "video") and hasattr(self.proxy_manager, "get_media_proxy_url"): |
| 76 | return await self.proxy_manager.get_media_proxy_url() |
| 77 | |
| 78 | # 其他下载走请求代理 |
| 79 | if hasattr(self.proxy_manager, "get_request_proxy_url"): |
| 80 | return await self.proxy_manager.get_request_proxy_url() |
| 81 | |
| 82 | # 向后兼容旧实现 |
| 83 | if hasattr(self.proxy_manager, "get_proxy_url"): |
| 84 | return await self.proxy_manager.get_proxy_url() |
| 85 | except Exception as e: |
| 86 | debug_logger.log_warning(f"Resolve download proxy failed: {str(e)}") |
| 87 | |
| 88 | return None |
| 89 | |
| 90 | def _guess_extension(self, url: str, media_type: str) -> str: |
| 91 | """尽量保留原始扩展名，未知时回退到默认值。""" |
no test coverage detected