Update build_logic.py

build_logic.py

@@ -12,7 +12,9 @@ from huggingface_hub import (
     whoami,
     hf_hub_download,
     delete_file as hf_delete_file,
-    HfApi
+    HfApi,
+    duplicate_repo as hf_duplicate_repo,
+    list_repos as hf_list_repos
 )
 from huggingface_hub.hf_api import CommitOperationDelete, CommitOperationAdd, CommitOperation
 from huggingface_hub.utils import HfHubHTTPError
@@ -311,6 +313,47 @@ def apply_staged_changes(ui_api_token_from_textbox, owner_ui, space_name_ui, cha
 
         api = HfApi(token=resolved_api_token)
 
+        # --- Handle Exclusive Actions First ---
+        exclusive_action = next((c for c in changeset if c['type'] in ['DUPLICATE_SPACE', 'DELETE_SPACE']), None)
+
+        if exclusive_action:
+             if exclusive_action['type'] == 'DUPLICATE_SPACE':
+                  # This should be handled in the confirm_changes handler to trigger a space load
+                  # Reaching here means the logic in confirm_changes failed to intercept it
+                  status_messages.append("Internal Error: DUPLICATE_SPACE action should have been handled exclusively.")
+                  logger.error("Internal Error: DUPLICATE_SPACE action was passed to apply_staged_changes unexpectedly.")
+             elif exclusive_action['type'] == 'DELETE_SPACE':
+                  # This should also ideally be handled by confirm_changes for UI state reset, but we can execute it here too
+                  delete_owner = exclusive_action.get('owner') or owner_ui
+                  delete_space = exclusive_action.get('space_name') or space_name_ui
+                  delete_repo_id_target = f"{delete_owner}/{delete_space}" if delete_owner and delete_space else repo_id
+
+                  if not delete_repo_id_target:
+                       status_messages.append("DELETE_SPACE Error: Target repo_id not specified.")
+                  elif delete_repo_id_target != repo_id:
+                       status_messages.append(f"DELETE_SPACE Error: AI requested deletion of '{delete_repo_id_target}', but this action is only permitted for the currently loaded space '{repo_id}'. Action blocked.")
+                       logger.warning(f"Blocked DELETE_SPACE action in apply_staged_changes: requested '{delete_repo_id_target}', current '{repo_id}'.")
+                  else:
+                       logger.warning(f"Attempting DESTRUCTIVE DELETE_SPACE action for {delete_repo_id_target}")
+                       try:
+                            api.delete_repo(repo_id=delete_repo_id_target, repo_type='space')
+                            status_messages.append(f"DELETE_SPACE: Successfully deleted space `{delete_repo_id_target}`.")
+                            logger.warning(f"Successfully deleted space {delete_repo_id_target}.")
+                       except HfHubHTTPError as e_http:
+                            status_messages.append(f"DELETE_SPACE HTTP Error ({e_http.response.status_code if e_http.response else 'N/A'}): {e_http.response.text if e_http.response else str(e_http)}. Check token/permissions.")
+                            logger.error(f"HTTP error deleting space {delete_repo_id_target}: {e_http}")
+                       except Exception as e:
+                            status_messages.append(f"DELETE_SPACE Error: {str(e)}. Check logs.")
+                            logger.exception(f"Error deleting space {delete_repo_id_target}:")
+             # If an exclusive action was found and potentially processed, stop here
+             final_status = " | ".join(status_messages) if status_messages else "Exclusive operation attempted."
+             logger.info(f"Exclusive action processed. Final status: {final_status}")
+             return final_status
+
+
+        # --- Handle Non-Exclusive Actions and File Changes ---
+        # This block is only reached if no exclusive action was found
+
         create_space_op = next((c for c in changeset if c['type'] == 'CREATE_SPACE'), None)
         if create_space_op:
             try:
@@ -450,31 +493,7 @@ def apply_staged_changes(ui_api_token_from_textbox, owner_ui, space_name_ui, cha
                       status_messages.append(f"SET_PRIVACY Error: {str(e)}. Check logs.")
                       logger.exception(f"Error setting privacy for {target_repo_id}:")
 
-            elif change['type'] == 'DELETE_SPACE':
-                 delete_owner = change.get('owner') or owner_ui
-                 delete_space = change.get('space_name') or space_name_ui
-                 delete_repo_id = f"{delete_owner}/{delete_space}" if delete_owner and delete_space else repo_id
-
-                 if not delete_repo_id:
-                      status_messages.append("DELETE_SPACE Error: Target repo_id not specified.")
-                      continue
-
-                 if delete_repo_id != repo_id:
-                      status_messages.append(f"DELETE_SPACE Error: AI requested deletion of '{delete_repo_id}', but this action is only permitted for the currently loaded space '{repo_id}'. Action blocked.")
-                      logger.warning(f"Blocked DELETE_SPACE action: requested '{delete_repo_id}', current '{repo_id}'.")
-                      continue
-
-                 logger.warning(f"Attempting DESTRUCTIVE DELETE_SPACE action for {delete_repo_id}")
-                 try:
-                      api.delete_repo(repo_id=delete_repo_id, repo_type='space')
-                      status_messages.append(f"DELETE_SPACE: Successfully deleted space `{delete_repo_id}`.")
-                      logger.warning(f"Successfully deleted space {delete_repo_id}.")
-                 except HfHubHTTPError as e_http:
-                      status_messages.append(f"DELETE_SPACE HTTP Error ({e_http.response.status_code if e_http.response else 'N/A'}): {e_http.response.text if e_http.response else str(e_http)}. Check token/permissions.")
-                      logger.error(f"HTTP error deleting space {delete_repo_id}: {e_http}")
-                 except Exception as e:
-                      status_messages.append(f"DELETE_SPACE Error: {str(e)}. Check logs.")
-                      logger.exception(f"Error deleting space {delete_repo_id}:")
+            # Note: DELETE_SPACE and DUPLICATE_SPACE are handled as exclusive actions at the top
 
     except HfHubHTTPError as e_http:
         logger.error(f"Top-level HTTP error during apply_staged_changes for {repo_id_for_error_logging or 'unknown repo'}: {e_http}")
@@ -680,3 +699,96 @@ def build_logic_delete_space(hf_api_key, owner, space_name):
     except Exception as e:
         logger.exception(f"Error deleting space {repo_id}:")
         return f"Error deleting space `{repo_id}`: {e}"
+
+def duplicate_space(hf_api_key, source_repo_id, target_repo_id, private: bool = False):
+    """Duplicates a Hugging Face Space."""
+    logger.info(f"Attempting to duplicate '{source_repo_id}' to '{target_repo_id}' (private={private}).")
+    try:
+        token, err = _get_api_token(hf_api_key)
+        if err or not token:
+            logger.error(f"Token error duplicating space: {err or 'Token not found'}")
+            return f"Error getting token: {err or 'Token not found.'}"
+
+        # Validate target_repo_id format if it includes owner, extract owner/name
+        if '/' in target_repo_id:
+             target_owner, target_space_name = target_repo_id.split('/', 1)
+             if not target_owner or not target_space_name or '/' in target_space_name:
+                  return f"Error: Invalid target repository ID format '{target_repo_id}'. Must be '<owner>/<space_name>'."
+        else:
+             # If only space name is provided, try to use the token's user as owner
+             target_space_name = target_repo_id
+             try:
+                 user_info = whoami(token=token)
+                 target_owner = user_info.get('name')
+                 if not target_owner: raise Exception("Could not determine owner from token.")
+                 target_repo_id = f"{target_owner}/{target_space_name}" # Update target_repo_id
+             except Exception as e:
+                 logger.error(f"Could not determine target owner from token: {e}")
+                 return f"Error: Target repository ID '{target_repo_id}' is missing owner, and owner could not be determined from token ({e}). Use '<owner>/<space_name>' format or set the Owner field."
+
+
+        hf_duplicate_repo(
+            from_repo=source_repo_id,
+            to_repo=target_repo_id,
+            repo_type="space",
+            token=token,
+            private=private,
+            # Use exist_ok=True? The UI should probably warn, but the backend can handle overwrite
+            # For now, let's assume overwrite is intended if triggered by AI/manual button after warning
+            exist_ok=True # Allow overwriting existing target space
+        )
+        logger.info(f"Successfully duplicated space from {source_repo_id} to {target_repo_id}.")
+        return f"Successfully duplicated space from `{source_repo_id}` to `{target_repo_id}`."
+    except HfHubHTTPError as e_http:
+        logger.error(f"HTTP error duplicating space from {source_repo_id} to {target_repo_id}: {e_http}")
+        status_code = e_http.response.status_code if e_http.response else 'N/A'
+        return f"HTTP Error ({status_code}) duplicating space: {e_http.response.text if e_http.response else str(e_http)}"
+    except Exception as e:
+        logger.exception(f"Error duplicating space from {source_repo_id} to {target_repo_id}:")
+        return f"Error duplicating space: {e}"
+
+def list_user_spaces(hf_api_key, owner=None):
+    """Lists spaces for the authenticated user or a specific owner/org."""
+    logger.info(f"Attempting to list spaces for owner: {owner or 'authenticated user'}.")
+    try:
+        token, err = _get_api_token(hf_api_key)
+        if err or not token:
+            logger.error(f"Token error listing spaces: {err or 'Token not found'}")
+            return None, f"Error getting token: {err or 'Token not found.'}"
+
+        # If owner is not provided, list spaces for the authenticated user
+        # We need the username for list_repos filter if owner is None in UI
+        effective_owner = owner
+        if not effective_owner:
+             try:
+                 user_info = whoami(token=token)
+                 effective_owner = user_info.get('name')
+                 if not effective_owner: raise Exception("Could not determine owner from token.")
+                 logger.info(f"Listing spaces for auto-detected owner: {effective_owner}")
+             except Exception as e:
+                 logger.error(f"Could not determine owner from token for listing: {e}")
+                 # Continue trying list_repos without username filter? No, list_repos
+                 # typically needs user or org specified for filtering unless it's public repos.
+                 # Let's require owner or a valid token for user listing.
+                 return None, f"Error auto-detecting owner for listing: {e}. Please specify Owner field."
+
+
+        api = HfApi(token=token)
+        spaces = hf_list_repos(author=effective_owner, type="space", token=token, timeout=20)
+        space_ids = [f"{r.author}/{r.id}" for r in spaces]
+
+        logger.info(f"Successfully listed {len(space_ids)} spaces for {effective_owner}.")
+        return space_ids, None
+
+    except HfHubHTTPError as e_http:
+        logger.error(f"HTTP error listing spaces for {owner or 'authenticated user'}: {e_http}")
+        status_code = e_http.response.status_code if e_http.response else 'N/A'
+        if status_code == 404:
+             # 404 could mean the owner doesn't exist or has no public spaces and token doesn't give access
+             return [], f"HTTP Error ({status_code}): Owner '{owner}' not found or has no accessible spaces."
+        if status_code in (401, 403):
+             return [], f"HTTP Error ({status_code}): Access denied or authentication required for listing spaces for '{owner}'. Check token permissions."
+        return None, f"HTTP Error ({status_code}) listing spaces for '{owner or 'authenticated user'}': {e_http.response.text if e_http.response else str(e_http)}"
+    except Exception as e:
+        logger.exception(f"Error listing spaces for {owner or 'authenticated user'}:")
+        return None, f"Error listing spaces: {e}"