Update app.py

app.py

@@ -12,51 +12,38 @@ from huggingface_hub import login
 
 # --- 1. Configuration and Authentication ---
 
-# <<< START: CODE UPDATE >>>
-# Load configuration from Hugging Face Secrets instead of hardcoding.
 GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID")
 GCP_LOCATION = os.environ.get("GCP_LOCATION")
-# <<< END: CODE UPDATE >>>
 
 # --- Authentication and Sanity Checks Block ---
 
-# Part A: Hugging Face Hub Authentication
 hf_token = os.environ.get("HF_TOKEN")
 if hf_token:
     print("Hugging Face token found. Logging in.")
     login(token=hf_token)
 else:
-    print("WARNING: Hugging Face token ('HF_TOKEN') not found. Hub-related features may be disabled.")
+    print("WARNING: Hugging Face token ('HF_TOKEN') not found.")
 
-# Part B: Google Cloud Authentication
 creds_json_str = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS_JSON")
 
-# Check if all necessary secrets are loaded
 if not all([GCP_PROJECT_ID, GCP_LOCATION, creds_json_str]):
-    # This block runs if any of the crucial secrets are missing.
-    missing_secrets = []
-    if not GCP_PROJECT_ID: missing_secrets.append("GCP_PROJECT_ID")
-    if not GCP_LOCATION: missing_secrets.append("GCP_LOCATION")
-    if not creds_json_str: missing_secrets.append("GOOGLE_APPLICATION_CREDENTIALS_JSON")
-    
-    error_message = f"FATAL: The following required secrets are missing: {', '.join(missing_secrets)}. Please set them in the Space settings."
+    missing_secrets = [s for s, v in {
+        "GCP_PROJECT_ID": GCP_PROJECT_ID,
+        "GCP_LOCATION": GCP_LOCATION,
+        "GOOGLE_APPLICATION_CREDENTIALS_JSON": creds_json_str
+    }.items() if not v]
+    error_message = f"FATAL: Missing required secrets: {', '.join(missing_secrets)}."
     print(error_message)
-    # Define a dummy function to show a clear error in the UI.
     def generate_video(prompt):
         raise gr.Error(error_message)
 else:
-    # This block runs only if all secrets are present.
-    print("All required secrets (GCP Project, Location, Credentials, HF Token) are loaded.")
-    
-    # Construct API URLs now that we have the project and location
+    print("All required secrets are loaded.")
     MODEL_ID = "veo-3.0-generate-preview"
     API_ENDPOINT = f"{GCP_LOCATION}-aiplatform.googleapis.com"
     PREDICT_URL = f"https://{API_ENDPOINT}/v1/projects/{GCP_PROJECT_ID}/locations/{GCP_LOCATION}/publishers/google/models/{MODEL_ID}:predictLongRunning"
     FETCH_URL = f"https://{API_ENDPOINT}/v1/projects/{GCP_PROJECT_ID}/locations/{GCP_LOCATION}/publishers/google/models/{MODEL_ID}:fetchPredictOperation"
 
-    with open("gcp_creds.json", "w") as f:
-        f.write(creds_json_str)
-    
+    with open("gcp_creds.json", "w") as f: f.write(creds_json_str)
     SCOPES = ["https://www.googleapis.com/auth/cloud-platform"]
     credentials, _ = google.auth.load_credentials_from_file("gcp_creds.json", scopes=SCOPES)
 
@@ -69,51 +56,48 @@ else:
     def generate_video(prompt: str):
         if not prompt:
             raise gr.Error("Prompt cannot be empty.")
-            
-        yield "Status: Authenticating and submitting job...", None
-
+        yield "Status: Submitting job...", None
         try:
-            access_token = get_access_token()
-            headers = {
-                "Authorization": f"Bearer {access_token}",
-                "Content-Type": "application/json",
-            }
-            payload = {
-                "instances": [{"prompt": prompt}],
-                "parameters": {
-                    "aspectRatio": "16:9", "sampleCount": 1, "durationSeconds": 8,
-                    "personGeneration": "allow_all", "addWatermark": True,
-                    "includeRaiReason": True, "generateAudio": True,
-                }
-            }
+            headers = {"Authorization": f"Bearer {get_access_token()}", "Content-Type": "application/json"}
+            payload = {"instances": [{"prompt": prompt}], "parameters": {"aspectRatio": "16:9", "sampleCount": 1, "durationSeconds": 8, "personGeneration": "allow_all", "addWatermark": True, "includeRaiReason": True, "generateAudio": True}}
             response = requests.post(PREDICT_URL, headers=headers, json=payload)
-            response.raise_for_status() 
+            response.raise_for_status()
             operation_name = response.json()["name"]
             print(f"Successfully submitted job. Operation Name: {operation_name}")
-
-            MAX_POLL_ATTEMPTS = 60 
+            MAX_POLL_ATTEMPTS = 60
             for i in range(MAX_POLL_ATTEMPTS):
-                yield f"Status: Polling for result (Attempt {i+1}/{MAX_POLL_ATTEMPTS})...", None
-                access_token = get_access_token()
-                headers["Authorization"] = f"Bearer {access_token}"
+                yield f"Status: Polling (Attempt {i+1}/{MAX_POLL_ATTEMPTS})...", None
+                headers["Authorization"] = f"Bearer {get_access_token()}"
                 fetch_payload = {"operationName": operation_name}
                 poll_response = requests.post(FETCH_URL, headers=headers, json=fetch_payload)
                 poll_response.raise_for_status()
                 poll_result = poll_response.json()
-
                 if poll_result.get("done"):
-                    print("Job finished successfully.")
+                    print("Job finished.")
+                    print(f"Full response payload: {json.dumps(poll_result, indent=2)}") # For debugging
                     response_data = poll_result.get("response", {})
                     if "videos" in response_data and response_data["videos"]:
                         video_base64 = response_data["videos"][0]["bytesBase64Encoded"]
                         video_bytes = base64.b64decode(video_base64)
-                        temp_video_path = "generated_video.mp4"
-                        with open(temp_video_path, "wb") as f:
-                            f.write(video_bytes)
-                        yield "Status: Done! Video generated.", temp_video_path
+                        with open("generated_video.mp4", "wb") as f: f.write(video_bytes)
+                        yield "Status: Done!", "generated_video.mp4"
                         return
                     else:
-                        raise gr.Error("Video generation finished, but content was blocked or another issue occurred.")
+                        # <<< START: IMPROVED ERROR HANDLING >>>
+                        error_message = "Video generation failed."
+                        # Check for a specific error message in the operation response
+                        if "error" in poll_result:
+                            error_details = poll_result["error"].get("message", "No details provided.")
+                            error_message += f"\nAPI Error: {error_details}"
+                        # Check for a specific RAI reason
+                        elif "raiResult" in response_data:
+                            rai_reason = response_data.get("raiMediaFilteredReason", "Unknown reason.")
+                            error_message += f"\nReason: Content was blocked by safety filters ({rai_reason})."
+                        else:
+                            error_message += "\nReason: The API did not return a video or a specific error."
+                        
+                        raise gr.Error(error_message)
+                        # <<< END: IMPROVED ERROR HANDLING >>>
                 time.sleep(10)
             raise gr.Error("Operation timed out.")
         except Exception as e:
@@ -131,14 +115,7 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
         with gr.Column(scale=1):
             status_output = gr.Markdown("Status: Ready")
             video_output = gr.Video(label="Generated Video", interactive=False)
-    gr.Examples(
-        ["A high-speed drone shot flying through a futuristic city with flying vehicles."],
-        inputs=prompt_input,
-    )
-    submit_button.click(
-        fn=generate_video,
-        inputs=prompt_input,
-        outputs=[status_output, video_output]
-    )
+    gr.Examples(["A high-speed drone shot flying through a futuristic city with flying vehicles."], inputs=prompt_input)
+    submit_button.click(fn=generate_video, inputs=prompt_input, outputs=[status_output, video_output])
 
 demo.launch()