Spaces:
Running
Running
| import requests | |
| import time | |
| from io import BytesIO | |
| class ImageProcessor: | |
| def __init__(self, api_key): | |
| self.api_key = api_key | |
| self.submit_url = 'https://test.aitanzou.com/web/api/task/submit' | |
| self.result_url = 'https://test.aitanzou.com/web/api/getResult' | |
| self.headers = { | |
| 'API-Key': self.api_key | |
| } | |
| def submit_images(self, image_bytes_list): | |
| files = [('images', ('image.png', image_bytes, 'image/png')) for image_bytes in image_bytes_list] | |
| response = requests.post(self.submit_url, headers=self.headers, files=files) | |
| if response.status_code == 200: | |
| data = response.json() | |
| if 'data' in data and 'taskId' in data['data']: | |
| task_id = data['data']['taskId'] | |
| return task_id | |
| else: | |
| raise Exception(f'Unexpected response format: {data}') | |
| else: | |
| raise Exception(f'Error: {response.status_code}, {response.text}') | |
| def get_result(self, task_id): | |
| params = {'taskId': task_id} | |
| while True: | |
| result_response = requests.get(self.result_url, params=params) | |
| if result_response.status_code == 200: | |
| result_data = result_response.json() | |
| if 'data' in result_data and 'abcPath' in result_data['data']: | |
| if result_data['data']['abcPath'] is None: | |
| print('Task is still pending...') | |
| time.sleep(10) | |
| else: | |
| url = result_data['data']['abcPath'] | |
| response = requests.get(url) | |
| if response.status_code == 200: | |
| return response.text | |
| else: | |
| raise Exception(f'Error retrieving file content: {response.status_code}, {response.text}') | |
| else: | |
| raise Exception(f'Unexpected result format: {result_data}') | |
| else: | |
| raise Exception(f'Error: {result_response.status_code}, {result_response.text}') | |
| def process_images(self, image_bytes_list): | |
| task_id = self.submit_images(image_bytes_list) | |
| return self.get_result(task_id) | |