Update app.py

app.py

@@ -5,172 +5,101 @@ from pathlib import Path
 import jwt
 import time
 
-# Kling AI API configuration (keys hardcoded as requested)
+# === Kling AI API configuration ===
 ACCESS_KEY_ID = "AGBGmadNd9hakFYfahytyQQJtN8CJmDJ"
 ACCESS_KEY_SECRET = "dp3pAe4PpdmnAHCAPgEd3PyLmBQrkMde"
-API_URL = "https://api-singapore.klingai.com/v1/image-to-image"
+API_URL = "https://api-singapore.klingai.com/v1/generate/image"
 
+# === Generate JWT Token ===
 def generate_jwt_token():
-    """Generate JWT token for Kling AI API authentication."""
     headers = {
         "alg": "HS256",
         "typ": "JWT"
     }
+    now = int(time.time())
     payload = {
         "iss": ACCESS_KEY_ID,
-        "exp": int(time.time()) + 1800,  # Token expires in 30 minutes
-        "nbf": int(time.time()) - 5  # Effective 5 seconds before current time
+        "exp": now + 1800,  # valid for 30 mins
+        "nbf": now - 5
     }
-    token = jwt.encode(payload, ACCESS_KEY_SECRET, headers=headers)
+    token = jwt.encode(payload, ACCESS_KEY_SECRET, algorithm="HS256", headers=headers)
     return token
 
-def generate_image(image, prompt=""):
-    """
-    Call Kling AI API for single-reference face generation.
-    
-    Args:
-        image: Uploaded image file (from Gradio, face image)
-        prompt (str): Optional text prompt to guide transformation
-    
-    Returns:
-        str: Path to the generated image or error message
-    """
-    if not image:
-        return "Error: Please upload a valid face image (PNG/JPEG, <10 MB, ≥512x512 pixels)."
-    
-    # Convert image to base64
+# === Image to Image Generation (Single Reference Face Mode) ===
+def generate_image(reference_image, prompt=""):
+    if not reference_image:
+        return None
+
     try:
-        with open(image, "rb") as img_file:
-            image_base64 = base64.b64encode(img_file.read()).decode("utf-8")
+        with open(reference_image, "rb") as img_file:
+            reference_base64 = base64.b64encode(img_file.read()).decode("utf-8")
     except Exception as e:
-        return f"Error: Failed to process image. Ensure it’s a valid PNG/JPEG. Details: {str(e)}"
-    
+        return f"Error reading reference image: {str(e)}"
+
     headers = {
         "Authorization": f"Bearer {generate_jwt_token()}",
         "Content-Type": "application/json"
     }
-    
+
     payload = {
-        "image": image_base64,
-        "prompt": prompt or "Transform the face into a cartoon style while preserving identity",
-        "strength": 0.97,  # High reference strength for face preservation
+        "reference_image": reference_base64,
+        "mode": "face",  # face reference mode
+        "prompt": prompt or "Match face with high fidelity",
+        "strength": 0.97,
         "output_format": "png"
     }
-    
+
     try:
         response = requests.post(API_URL, json=payload, headers=headers, timeout=30)
         response.raise_for_status()
-        
         data = response.json()
-        task_id = data.get("task_id") or data.get("taskId") or data.get("id")
+        task_id = data.get("task_id") or data.get("id")
         if not task_id:
-            return "Error: No task ID returned. Check API endpoint or keys at https://klingai.com/global/dev."
-        
-        # Poll for task completion
+            return f"Error: No task ID returned: {data}"
+
         status_url = f"https://api-singapore.klingai.com/v1/predictions/{task_id}"
-        for _ in range(60):  # Poll for up to 5 minutes
+        for _ in range(60):  # poll up to 5 minutes
             status_response = requests.get(status_url, headers=headers, timeout=30)
             status_response.raise_for_status()
             status_data = status_response.json()
             status = status_data.get("status")
+
             if status == "succeeded":
-                image_url = status_data.get("image_url") or status_data.get("result", {}).get("image_url") or status_data.get("output")
+                image_url = status_data.get("image_url") or status_data.get("output")
                 if not image_url:
-                    return "Error: No image URL in API response. Check API documentation."
-                # Download the image
-                image_response = requests.get(image_url, timeout=30)
-                image_response.raise_for_status()
-                output_path = Path("output_image.png")
+                    return f"Error: No image URL found: {status_data}"
+
+                img_resp = requests.get(image_url, timeout=30)
+                img_resp.raise_for_status()
+                output_path = Path("output.png")
                 with open(output_path, "wb") as f:
-                    f.write(image_response.content)
+                    f.write(img_resp.content)
                 return str(output_path)
+
             elif status == "failed":
-                return "Error: Image generation failed. Ensure the image contains a clear face and avoid NSFW or sensitive content."
-            elif status == "processing" and status_data.get("progress", 0) >= 0.99:
-                return "Error: Generation stuck at 99%. Check account credits or upgrade to a paid plan at https://klingai.com."
+                return "Error: Generation failed."
+
             time.sleep(5)
-        
-        return "Error: Image generation timed out. Try during off-peak hours or check account status."
-    
-    except requests.exceptions.HTTPError as e:
-        status_code = e.response.status_code if e.response else None
-        error_data = e.response.json() if e.response and e.response.text else {}
-        service_code = error_data.get("code", 0)
-        
-        if status_code == 401:
-            if service_code == 1000:
-                return "Error: Authentication failed (1000). Verify API keys are correct."
-            elif service_code == 1001:
-                return "Error: Authorization empty (1001). Ensure JWT token is included."
-            elif service_code == 1002:
-                return "Error: Authorization invalid (1002). Check key format."
-            elif service_code == 1003:
-                return "Error: Authorization not yet valid (1003). Adjust token start time (nbf)."
-            elif service_code == 1004:
-                return "Error: Authorization expired (1004). Generate a new token."
-        elif status_code == 429:
-            if service_code == 1100:
-                return "Error: Account exception (1100). Verify account configuration at https://klingai.com."
-            elif service_code == 1101:
-                return "Error: Account in arrears (1101). Recharge your account."
-            elif service_code == 1102:
-                return "Error: Resource pack depleted or expired (1102). Purchase additional resources."
-            elif service_code in (1302, 1303):
-                return "Error: Rate limit exceeded (1302/1303). Reduce request frequency or contact support."
-            elif service_code == 1304:
-                return "Error: IP whitelisting issue (1304). Contact Kling AI support."
-        elif status_code == 403 and service_code == 1103:
-            return "Error: Unauthorized access (1103). Verify account permissions."
-        elif status_code == 400:
-            if service_code == 1200:
-                return "Error: Invalid request parameters (1200). Check payload format."
-            elif service_code == 1201:
-                return "Error: Invalid parameters (1201). Use PNG/JPEG (<10 MB, ≥512x512) with a clear face."
-            elif service_code == 1300:
-                return "Error: Platform policy triggered (1300). Check input content."
-            elif service_code == 1301:
-                return "Error: Content security policy triggered (1301). Ensure the image contains a clear face and avoid NSFW content."
-        elif status_code == 404:
-            if service_code == 1202:
-                return "Error: Invalid request method (1202). Check https://klingai.com/global/dev for correct endpoint."
-            elif service_code == 1203:
-                return "Error: Resource does not exist (1203). Verify model or endpoint."
-        elif status_code in (500, 503, 504):
-            if service_code == 5000:
-                return "Error: Server internal error (5000). Try again later."
-            elif service_code == 5001:
-                return "Error: Server unavailable due to maintenance (5001). Try again later."
-            elif service_code == 5002:
-                return "Error: Server timeout (5002). Try during off-peak hours."
-        return f"Error: API request failed. HTTP {status_code}, Service Code {service_code}. Details: {str(e)}"
+
+        return "Error: Timed out."
+
     except requests.exceptions.RequestException as e:
-        return f"Error: Network issue. Ensure stable internet, disable VPN, and try again. Details: {str(e)}"
-
-def chatbot_interface(image, prompt):
-    """
-    Gradio interface for single-reference face generation.
-    
-    Args:
-        image: Uploaded image file (containing a face)
-        prompt (str): Optional text prompt
-    
-    Returns:
-        Image file path or error message
-    """
-    return generate_image(image, prompt)
-
-# Define Gradio interface
+        return f"Request error: {str(e)}"
+
+# === Gradio Interface ===
+def process(reference_image, prompt):
+    return generate_image(reference_image, prompt)
+
 iface = gr.Interface(
-    fn=chatbot_interface,
+    fn=process,
     inputs=[
-        gr.Image(type="filepath", label="Upload Face Image (PNG/JPEG, <10 MB, ≥512x512)"),
-        gr.Textbox(lines=2, placeholder="Enter an optional prompt (e.g., 'Turn this face into a cartoon')", label="Prompt")
+        gr.Image(type="filepath", label="Upload Reference Face Image"),
+        gr.Textbox(lines=2, placeholder="Optional prompt", label="Prompt")
     ],
-    outputs=gr.Image(label="Generated Face Image"),
-    title="Kling AI Single-Reference Face Generator",
-    description="Upload a PNG/JPEG image (<10 MB, ≥512x512 pixels) with a clear face to generate a transformed image using Kling AI API (reference strength 0.97). Avoid NSFW or sensitive content."
+    outputs=gr.Image(label="Generated Image"),
+    title="Kling AI - Single Reference Face Generator",
+    description="Upload a reference face image. The model will generate an image based on that face with 97% strength."
 )
 
-# Launch the interface
 if __name__ == "__main__":
-    iface.launch(server_name="0.0.0.0", server_port=7860)
+    iface.launch(server_name="0.0.0.0", server_port=7860)