forge/backend/debug_topaz_real.py

135 lines
4.2 KiB
Python

import asyncio
import httpx
import os
import json
TOPAZ_API_KEY = "5af61151-913b-4c58-b842-dc52e2913800"
async def test_image_upscale():
print("\n--- Testing Image Upscale ---")
# Payload matching PHP exactly
# PHP:
# 'output_height' => $output_height,
# 'output_format' => 'jpeg',
# 'crop_to_fill' => 'false',
# 'face_enhancement' => 'false' (if off)
# 'model' => send only if not Auto.
params = {
"output_height": "1000",
"output_format": "jpeg",
"crop_to_fill": "false",
"face_enhancement": "false"
}
# In PHP: curl_setopt($ch, CURLOPT_POSTFIELDS, $postFields) where $postFields has file object.
files = {
"image": ("test_image.png", open("test_image.png", "rb"), "image/png")
}
async with httpx.AsyncClient() as client:
try:
print(f"Sending POST to https://api.topazlabs.com/image/v1/enhance/async")
print(f"Params: {params}")
response = await client.post(
"https://api.topazlabs.com/image/v1/enhance/async",
headers={
"X-API-Key": TOPAZ_API_KEY,
"Accept": "application/json"
},
files=files,
data=params
)
print(f"Status: {response.status_code}")
print(f"Response: {response.text}")
except Exception as e:
print(f"Error: {e}")
async def test_video_upscale():
print("\n--- Testing Video Upscale (Creation Request) ---")
# PHP Payload:
# {
# "source": { ... },
# "filters": [ { "model": "prob-4", "videoType": "Progressive", "auto": "Auto" } ],
# "output": { ... }
# }
payload = {
"source": {
"container": "mp4",
"size": 1024,
"duration": 10,
"frameCount": 300,
"frameRate": 30,
"resolution": {
"width": 1920,
"height": 1080
}
},
"filters": [
{
"model": "prob-4",
"videoType": "Progressive",
"auto": "Auto"
}
],
"output": {
"resolution": {
"width": 3840,
"height": 2160
},
"frameRate": 30,
"audioCodec": "AAC",
"audioTransfer": "Copy",
"container": "mp4",
"videoBitrate": "6000k"
}
}
async with httpx.AsyncClient() as client:
try:
# Trying the URL we suspect is correct: /video/
url = "https://api.topazlabs.com/video/"
print(f"Sending POST to {url}")
print(f"Payload: {json.dumps(payload, indent=2)}")
response = await client.post(
url,
headers={
"X-API-Key": TOPAZ_API_KEY,
"Content-Type": "application/json",
"Accept": "application/json"
},
json=payload
)
print(f"Status: {response.status_code}")
print(f"Response: {response.text}")
if response.status_code == 201 or response.status_code == 200:
data = response.json()
# Check for various ID keys
request_id = data.get("id") or data.get("requestId") or data.get("process_id")
print(f"Got Request ID: {request_id}")
if request_id:
# Test ACCEPT endpoint with v1
accept_url = f"https://api.topazlabs.com/video/{request_id}/accept"
print(f"Testing ACCEPT at: {accept_url}")
accept_response = await client.patch(
accept_url,
headers={"X-API-Key": TOPAZ_API_KEY}
)
print(f"Accept Status: {accept_response.status_code}")
print(f"Accept Response: {accept_response.text}")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
asyncio.run(test_image_upscale())
asyncio.run(test_video_upscale())