Update app_logic.py

app_logic.py

@@ -1,12 +1,12 @@
 import os
 import re
 import tempfile
-import git # Used by Repository
+import git
 from huggingface_hub import (
     create_repo,
     upload_folder,
     list_repo_files,
-    Repository, # Used in update_space_file
+    Repository,
     whoami,
 )
 import logging
@@ -19,50 +19,30 @@ try:
     KEYLOCK_DECODE_AVAILABLE = True
 except ImportError:
     KEYLOCK_DECODE_AVAILABLE = False
-    decode_from_image_pil = None # Placeholder
+    decode_from_image_pil = None
     logging.warning("keylock-decode library not found. Image decoding feature will be disabled.")
 
-
-# Configure logging (ensure this is suitable for your environment, might be set in app.py too)
-# If keylock-decode uses its own logger (e.g., logging.getLogger('keylock_decode')),
-# you might need to adjust its level separately if you want more/less verbose output from it.
 logging.basicConfig(
     level=logging.INFO, 
     format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 )
 logger = logging.getLogger(__name__)
-if KEYLOCK_DECODE_AVAILABLE: # Example: Set keylock_decode's logger to DEBUG if you want its verbose output
-    # logging.getLogger('keylock_decode.decoder').setLevel(logging.DEBUG) # Check actual logger name in keylock-decode
-    pass
 
 
-# --- New Helper Function to Get API Token ---
+# --- Helper Function to Get API Token (Unchanged from previous version) ---
 def _get_api_token(ui_token_from_textbox=None):
-    """
-    Retrieves the Hugging Face API token.
-    Prioritizes HF_TOKEN from environment variables (potentially set by KeyLock-Decode),
-    then falls back to UI input.
-    Returns (token, error_message)
-    """
     env_token = os.getenv('HF_TOKEN')
     if env_token:
         logger.info("Using HF_TOKEN from environment.")
         return env_token, None
-    
     if ui_token_from_textbox:
         logger.info("Using API token from UI textbox.")
         return ui_token_from_textbox, None
-    
     logger.warning("HF API token not found in environment or UI textbox.")
     return None, "Error: Hugging Face API token not provided. Please enter it in the textbox or load it from an image."
 
-
-# --- New Function for KeyLock-Decode Integration ---
+# --- Updated Function for KeyLock-Decode Integration with Debugging ---
 def load_token_from_image_and_set_env(image_pil_object: Image.Image, password: str):
-    """
-    Decodes data from a PIL Image object using keylock-decode, sets environment variables,
-    and returns status messages.
-    """
     if not KEYLOCK_DECODE_AVAILABLE:
         return "Error: KeyLock-Decode library is not installed. This feature is disabled."
 
@@ -73,13 +53,46 @@ def load_token_from_image_and_set_env(image_pil_object: Image.Image, password: s
 
     status_messages_display = []
     
+    # --- DEBUGGING: Check image properties and save it ---
+    debug_image_path_str = "Not saved (error or no image)."
+    try:
+        if image_pil_object:
+            # Log original properties
+            original_mode = getattr(image_pil_object, 'mode', 'N/A')
+            original_format = getattr(image_pil_object, 'format', 'N/A') # Format might be None for PIL object not read from file
+            original_size = getattr(image_pil_object, 'size', 'N/A')
+            logger.info(f"Received PIL Image from Gradio. Mode: {original_mode}, Original Format: {original_format}, Size: {original_size}")
+            status_messages_display.append(f"[DEBUG] Gradio provided PIL Image - Mode: {original_mode}, Original Format: {original_format}, Size: {original_size}")
+
+            # Save the image received from Gradio to a temporary file for external checking
+            # Ensure the temp directory is writable in the Hugging Face Space environment
+            # /tmp/ is usually writable.
+            temp_dir = Path(tempfile.gettempdir()) / "gradio_image_debug"
+            temp_dir.mkdir(parents=True, exist_ok=True)
+            debug_image_path = temp_dir / "image_from_gradio.png"
+            
+            # Save the image. If it's RGBA, try to keep it RGBA.
+            # PNG is lossless, so this should be okay if the mode is preserved.
+            image_pil_object.save(debug_image_path, "PNG") 
+            debug_image_path_str = str(debug_image_path)
+            status_messages_display.append(f"**[DEBUG ACTION REQUIRED]** Image received from Gradio has been saved to: `{debug_image_path_str}` in the Space's temporary filesystem. Please try decoding this *specific file* using the standalone `keylock-decode` CLI to verify its integrity. E.g., `keylock-decode {debug_image_path_str} \"YOUR_PASSWORD\"`")
+            logger.info(f"Debug image from Gradio saved to: {debug_image_path_str}")
+        else:
+            logger.warning("image_pil_object is None when trying to debug-save.")
+            status_messages_display.append("[DEBUG] No PIL image object received to save for debugging.")
+
+    except Exception as save_exc:
+        logger.exception("Error during debug-saving of image from Gradio:")
+        status_messages_display.append(f"[DEBUG] Error saving image for debugging: {str(save_exc)}. Path was: {debug_image_path_str}")
+    # --- END DEBUGGING ---
+
     try:
         logger.info(f"Attempting to decode from image using KeyLock-Decode...")
-        # decode_from_image_pil expects a PIL Image object
+        # Pass the original PIL object
         decoded_data, status_msgs_from_lib = decode_from_image_pil(
-            image_pil_object,
+            image_pil_object, # Use the PIL object directly
             password,
-            set_environment_variables=True # CRITICAL: This sets os.environ
+            set_environment_variables=True
         )
 
         status_messages_display.extend(status_msgs_from_lib)
@@ -92,26 +105,26 @@ def load_token_from_image_and_set_env(image_pil_object: Image.Image, password: s
             
             if os.getenv('HF_TOKEN'):
                 status_messages_display.append(f"\n**SUCCESS: HF_TOKEN was found and has been set in the current process environment.** It will be prioritized for operations.")
-            elif 'HF_TOKEN' in decoded_data: # Should be rare if set_env_vars=True
+            elif 'HF_TOKEN' in decoded_data:
                  status_messages_display.append(f"\nWarning: HF_TOKEN was in decoded data but os.getenv('HF_TOKEN') is not picking it up. This is unexpected if 'set_environment_variables=True' worked.")
             else:
                 status_messages_display.append("\nNote: HF_TOKEN was not specifically found in the decoded data from the image.")
-        else:
-            # This case is often covered by status_msgs_from_lib, but good to have a fallback.
-            if not any("No payload data found" in msg for msg in status_msgs_from_lib):
-                 status_messages_display.append("No data was decoded, or the decoded data was empty.")
+        # else:
+        #     if not any("No payload data found" in msg for msg in status_msgs_from_lib) and not any("Failed to decrypt" in msg for msg in status_msgs_from_lib): # Avoid redundant messages
+        #          status_messages_display.append("No data was decoded, or the decoded data was empty (and no specific error from library).")
+
 
-    except ValueError as e: # Specific errors from keylock-decode (e.g., bad password, corrupted data)
+    except ValueError as e:
         logger.error(f"KeyLock-Decode ValueError: {e}")
-        status_messages_display.append(f"**Decoding Error:** {e}")
+        status_messages_display.append(f"**Decoding Error (e.g., bad password, corrupted data):** {e}")
     except Exception as e:
         logger.exception("An unexpected error occurred during image decoding with KeyLock-Decode:")
-        status_messages_display.append(f"**An unexpected error occurred:** {str(e)}")
+        status_messages_display.append(f"**An unexpected error occurred during decoding:** {str(e)}")
     
     return "\n".join(status_messages_display)
 
 
-# --- Original `parse_markdown` function (unchanged) ---
+# --- `parse_markdown` (Unchanged) ---
 def parse_markdown(markdown_input):
     """Parse markdown input to extract space details and file structure."""
     space_info = {"repo_name_md": "", "owner_md": "", "files": []}
@@ -181,161 +194,95 @@ def parse_markdown(markdown_input):
     space_info["files"] = [f for f in space_info["files"] if f.get("path")]
     return space_info
 
-
-# --- Updated `_determine_repo_id` to use `_get_api_token` ---
+# --- `_determine_repo_id` (Unchanged) ---
 def _determine_repo_id(ui_api_token_from_textbox, space_name_ui, owner_ui):
     if not space_name_ui:
         return None, "Error: Space Name cannot be empty."
     if "/" in space_name_ui:
         return None, "Error: Space Name should not contain '/'. Please use the Owner field for the namespace."
 
-    final_owner = owner_ui
-    error_message = None
-
+    final_owner = owner_ui; error_message = None
     if not final_owner:
-        # Get the actual token to use for whoami
         resolved_api_token, token_err = _get_api_token(ui_api_token_from_textbox)
-        if token_err:
-            return None, token_err 
-        # If resolved_api_token is None here but token_err is None, it means _get_api_token had an issue not returning error string
-        # which _get_api_token is designed not to do, it should always return error if no token.
-        if not resolved_api_token: 
-             return None, "Error: API token is required to automatically determine owner when Owner field is empty (internal check failed)."
-
+        if token_err: return None, token_err 
+        if not resolved_api_token: return None, "Error: API token required for auto owner determination (internal check failed)."
         try:
             user_info = whoami(token=resolved_api_token)
             if user_info and 'name' in user_info:
                 final_owner = user_info['name']
                 logger.info(f"Determined owner: {final_owner} using API token.")
             else:
-                logger.error(f"whoami(token=...) returned: {user_info} - 'name' field missing or user_info is None.")
-                error_message = "Error: Could not retrieve username from API token. Ensure token is valid and has 'Read profile' permissions. Or, specify Owner manually."
+                error_message = "Error: Could not retrieve username from API token. Check permissions or specify Owner."
         except Exception as e:
-            logger.error(f"Error calling whoami for owner: {str(e)}")
-            error_message = f"Error retrieving username from API token: {str(e)}. Please specify Owner manually."
-        
-        if error_message:
-            return None, error_message
-
-    if not final_owner:
-        return None, "Error: Owner could not be determined. Please provide an owner or ensure your API token is valid."
-
+            error_message = f"Error retrieving username from API token: {str(e)}. Specify Owner manually."
+        if error_message: return None, error_message
+    if not final_owner: return None, "Error: Owner could not be determined."
     return f"{final_owner}/{space_name_ui}", None
 
-
-# --- Updated Core Functions to use `_get_api_token` ---
-
+# --- Core Functions: `create_space`, `view_space_files`, `update_space_file` (Unchanged from previous correct version) ---
 def create_space(ui_api_token_from_textbox, space_name_ui, owner_ui, sdk_ui, markdown_input):
     repo_id_for_error_logging = f"{owner_ui}/{space_name_ui}" if owner_ui else space_name_ui
     try:
         resolved_api_token, token_err = _get_api_token(ui_api_token_from_textbox)
-        if token_err:
-            return token_err
-
-        # _determine_repo_id will also call _get_api_token internally if owner_ui is blank
+        if token_err: return token_err
         repo_id, err = _determine_repo_id(ui_api_token_from_textbox, space_name_ui, owner_ui)
-        if err:
-            return err
-        repo_id_for_error_logging = repo_id # Update for more accurate logging if determination succeeds
-        
+        if err: return err
+        repo_id_for_error_logging = repo_id
         space_info = parse_markdown(markdown_input)
-        if not space_info["files"]:
-            return "Error: No files found in the markdown input. Ensure '### File: path/to/file.ext' markers are used correctly."
-
+        if not space_info["files"]: return "Error: No files found in markdown. Use '### File: path/to/file.ext'."
         with tempfile.TemporaryDirectory() as temp_dir:
             repo_local_path = Path(temp_dir) / "repo_upload_content"
             repo_local_path.mkdir(exist_ok=True)
-
             for file_info in space_info["files"]:
                 if not file_info.get("path"): continue
                 file_path_abs = repo_local_path / file_info["path"]
                 file_path_abs.parent.mkdir(parents=True, exist_ok=True)
                 with open(file_path_abs, "w", encoding="utf-8") as f: f.write(file_info["content"])
-                logger.info(f"Wrote file: {file_path_abs}")
-
             try:
-                create_repo(
-                    repo_id=repo_id, token=resolved_api_token, repo_type="space",
-                    space_sdk=sdk_ui, private=False,
-                )
-                logger.info(f"Created Space repo: {repo_id}")
+                create_repo(repo_id=repo_id, token=resolved_api_token, repo_type="space", space_sdk=sdk_ui, private=False)
             except Exception as e:
                 err_str = str(e).lower()
-                if "already exists" in err_str or "you already created this repo" in err_str or "exists" in err_str:
-                    logger.info(f"Space {repo_id} already exists. Proceeding to upload/update files.")
-                else:
+                if not ("already exists" in err_str or "you already created this repo" in err_str or "exists" in err_str):
                     return f"Error creating Space '{repo_id}': {str(e)}"
-            
-            upload_folder(
-                repo_id=repo_id, folder_path=str(repo_local_path), path_in_repo=".",
-                token=resolved_api_token, repo_type="space",
-                commit_message=f"Initial Space setup of {repo_id} via Builder",
-            )
-            logger.info(f"Uploaded files to Space: {repo_id}")
+            upload_folder(repo_id=repo_id, folder_path=str(repo_local_path), path_in_repo=".", token=resolved_api_token, repo_type="space", commit_message=f"Initial Space setup of {repo_id} via Builder")
             return f"Successfully created/updated Space: [{repo_id}](https://huggingface.co/spaces/{repo_id})"
-
     except Exception as e:
         logger.exception(f"Error in create_space for {repo_id_for_error_logging}:")
         return f"Error during Space creation/update: {str(e)}"
 
-
 def view_space_files(ui_api_token_from_textbox, space_name_ui, owner_ui):
     repo_id_for_error_logging = f"{owner_ui}/{space_name_ui}" if owner_ui else space_name_ui
     try:
         resolved_api_token, token_err = _get_api_token(ui_api_token_from_textbox)
-        if token_err:
-            return token_err
-        
+        if token_err: return token_err
         repo_id, err = _determine_repo_id(ui_api_token_from_textbox, space_name_ui, owner_ui)
-        if err:
-            return err
+        if err: return err
         repo_id_for_error_logging = repo_id
-            
         files = list_repo_files(repo_id=repo_id, token=resolved_api_token, repo_type="space")
-        if files:
-            return f"Files in `{repo_id}`:\n\n" + "\n".join([f"- `{f}`" for f in files])
-        return f"No files found in the Space `{repo_id}`."
+        if files: return f"Files in `{repo_id}`:\n\n" + "\n".join([f"- `{f}`" for f in files])
+        return f"No files found in Space `{repo_id}`."
     except Exception as e:
         logger.exception(f"Error in view_space_files for {repo_id_for_error_logging}:")
         return f"Error listing files for `{repo_id_for_error_logging}`: {str(e)}"
 
-
 def update_space_file(ui_api_token_from_textbox, space_name_ui, owner_ui, file_path_in_repo, file_content, commit_message_ui):
     repo_id_for_error_logging = f"{owner_ui}/{space_name_ui}" if owner_ui else space_name_ui
     try:
         resolved_api_token, token_err = _get_api_token(ui_api_token_from_textbox)
-        if token_err:
-            return token_err
-
+        if token_err: return token_err
         repo_id, err = _determine_repo_id(ui_api_token_from_textbox, space_name_ui, owner_ui)
-        if err:
-            return err
+        if err: return err
         repo_id_for_error_logging = repo_id
-        
         if not file_path_in_repo: return "Error: File Path cannot be empty."
         commit_message_ui = commit_message_ui or f"Update {file_path_in_repo} via Space Builder"
-
         with tempfile.TemporaryDirectory() as temp_dir:
             repo_local_clone_path = Path(temp_dir) / "cloned_space_repo_for_update"
-            
-            cloned_repo = Repository(
-                local_dir=str(repo_local_clone_path),
-                clone_from=f"https://huggingface.co/spaces/{repo_id}",
-                repo_type="space", use_auth_token=resolved_api_token,
-                git_user="Space Builder Bot", git_email="space-builder@huggingface.co"
-            )
-            logger.info(f"Cloned Space {repo_id} to {repo_local_clone_path}")
-
+            cloned_repo = Repository(local_dir=str(repo_local_clone_path), clone_from=f"https://huggingface.co/spaces/{repo_id}", repo_type="space", use_auth_token=resolved_api_token, git_user="Space Builder Bot", git_email="space-builder@huggingface.co")
             full_local_file_path = Path(cloned_repo.local_dir) / file_path_in_repo
             full_local_file_path.parent.mkdir(parents=True, exist_ok=True)
             with open(full_local_file_path, "w", encoding="utf-8") as f: f.write(file_content)
-            logger.info(f"Wrote updated file {file_path_in_repo} locally.")
-
             cloned_repo.push_to_hub(commit_message=commit_message_ui)
-            logger.info(f"Pushed changes for {file_path_in_repo} to {repo_id}")
-            
             return f"Successfully updated `{file_path_in_repo}` in Space [{repo_id}](https://huggingface.co/spaces/{repo_id})"
-
     except Exception as e:
         logger.exception(f"Error in update_space_file for {repo_id_for_error_logging}:")
         return f"Error updating file for `{repo_id_for_error_logging}`: {str(e)}"