| import os |
| import requests |
| from typing import Optional |
|
|
|
|
| class ExternalFileUploader: |
| def __init__(self): |
| self.base_url = os.getenv("FILE_UPLOAD_API_URL") |
| self.api_key = os.getenv("FILE_UPLOAD_API_KEY") |
|
|
| def upload_file(self, file_bytes: bytes, file_name: str) -> Optional[str]: |
| try: |
| files = { |
| "files": (file_name, file_bytes, "image/png") |
| } |
|
|
| headers = { |
| "X-API-Key": self.api_key |
| } |
|
|
| response = requests.post( |
| f"{self.base_url}/api/v1/files/bulk-upload", |
| headers=headers, |
| files=files, |
| timeout=30 |
| ) |
|
|
| if response.status_code != 201: |
| print("Upload failed:", response.text) |
| return None |
|
|
| data = response.json() |
|
|
| |
| if not isinstance(data, list) or not data: |
| return None |
|
|
| first = data[0] |
|
|
| if not first.get("success"): |
| return None |
|
|
| return first["data"]["file_url"] |
|
|
| except Exception as e: |
| print("Uploader error:", str(e)) |
| return None |