Spaces:
Build error
Build error
| """Collection management utilities for PDF-TEI-Editor.""" | |
| from pathlib import Path | |
| from typing import List, Optional, Dict, Any | |
| from fastapi_app.lib.utils.data_utils import load_entity_data, save_entity_data, get_data_file_path, load_json_file | |
| def find_collection(collection_id: str, collections_data: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]: | |
| """Finds a collection by ID. | |
| Args: | |
| collection_id: The collection ID to search for | |
| collections_data: List of collection dictionaries | |
| Returns: | |
| Collection dictionary if found, None otherwise | |
| """ | |
| for collection in collections_data: | |
| if collection.get('id') == collection_id: | |
| return collection | |
| return None | |
| def collection_exists(collection_id: str, collections_data: List[Dict[str, Any]]) -> bool: | |
| """Checks if a collection exists. | |
| Args: | |
| collection_id: The collection ID to check | |
| collections_data: List of collection dictionaries | |
| Returns: | |
| True if collection exists, False otherwise | |
| """ | |
| return find_collection(collection_id, collections_data) is not None | |
| def get_available_collections(db_dir: Path) -> Optional[List[str]]: | |
| """Gets the list of available collections from db/collections.json. | |
| Args: | |
| db_dir: Path to the db directory | |
| Returns: | |
| List of collection IDs if successful, None if error | |
| """ | |
| collections_file = get_data_file_path(db_dir, 'collections') | |
| if not collections_file.exists(): | |
| raise FileNotFoundError(f"Collections file not found at {collections_file}") | |
| collections_data = load_json_file(collections_file, create_if_missing=False) | |
| if not isinstance(collections_data, list): | |
| raise ValueError("Invalid collections file format. Expected a list.") | |
| collection_ids: List[str] = [ | |
| collection['id'] # Use direct access since we check 'id' in collection | |
| for collection in collections_data | |
| if isinstance(collection, dict) and 'id' in collection | |
| ] | |
| return collection_ids | |
| def get_collections_with_details(db_dir: Path) -> Optional[List[Dict[str, Any]]]: | |
| """Gets all collections with their details from db/collections.json. | |
| Args: | |
| db_dir: Path to the db directory | |
| Returns: | |
| List of collection dictionaries if successful, None if error | |
| """ | |
| collections_file = get_data_file_path(db_dir, 'collections') | |
| if not collections_file.exists(): | |
| raise FileNotFoundError(f"Collections file not found at {collections_file}") | |
| collections_data = load_json_file(collections_file, create_if_missing=False) | |
| if not isinstance(collections_data, list): | |
| raise ValueError("Invalid collections file format. Expected a list.") | |
| return collections_data | |
| def validate_collection(collection_id: str, db_dir: Path) -> bool: | |
| """Validates if a collection exists in the available collections. | |
| Args: | |
| collection_id: The collection ID to validate | |
| db_dir: Path to the db directory | |
| Returns: | |
| True if collection exists, False otherwise | |
| """ | |
| try: | |
| available_collections = get_available_collections(db_dir) | |
| return collection_id in available_collections if available_collections else False | |
| except (FileNotFoundError, ValueError): | |
| return False | |
| def get_collection_owner(collection: Dict[str, Any]) -> Optional[str]: | |
| """Returns the owner username of a collection, or None if unowned. | |
| Args: | |
| collection: Collection dictionary | |
| Returns: | |
| Owner username string, or None | |
| """ | |
| return collection.get('owner') or None | |
| def is_collection_owner(user: Dict[str, Any], collection: Dict[str, Any]) -> bool: | |
| """Checks if a user is the owner of a collection. | |
| Args: | |
| user: User dictionary with 'username' key | |
| collection: Collection dictionary | |
| Returns: | |
| True if the user's username matches the collection owner | |
| """ | |
| owner = get_collection_owner(collection) | |
| return bool(owner and user.get('username') == owner) | |
| def can_delete_collection(user: Dict[str, Any], collection: Dict[str, Any]) -> bool: | |
| """Checks if a user is allowed to delete a collection. | |
| Deletion is allowed for admins and the collection owner. | |
| Args: | |
| user: User dictionary | |
| collection: Collection dictionary | |
| Returns: | |
| True if the user may delete the collection | |
| """ | |
| from fastapi_app.lib.permissions.acl_utils import user_is_admin | |
| return user_is_admin(user) or is_collection_owner(user, collection) | |
| def add_collection(db_dir: Path, collection_id: str, name: str, description: str = "", owner: Optional[str] = None) -> tuple[bool, str]: | |
| """Adds a new collection to the collections.json file. | |
| Args: | |
| db_dir: Path to the db directory | |
| collection_id: The collection ID | |
| name: The collection name | |
| description: The collection description (optional) | |
| owner: Username of the collection owner (optional) | |
| Returns: | |
| Tuple of (success: bool, message: str) | |
| """ | |
| collections_data = load_entity_data(db_dir, 'collections') | |
| # Check if collection already exists | |
| if collection_exists(collection_id, collections_data): | |
| return False, f"Collection '{collection_id}' already exists." | |
| # Add new collection | |
| new_collection = { | |
| "id": collection_id, | |
| "name": name, | |
| "description": description, | |
| "owner": owner | |
| } | |
| collections_data.append(new_collection) | |
| save_entity_data(db_dir, 'collections', collections_data) | |
| return True, f"Collection '{collection_id}' added successfully." | |
| def grant_user_collection_access( | |
| db_dir: Path, | |
| username: str, | |
| collection_id: str, | |
| logger=None | |
| ) -> tuple[bool, str]: | |
| """ | |
| Grant a user access to a collection by adding it to their personal group. | |
| Creates the user's personal group if it doesn't exist, then adds | |
| the collection to that group. | |
| Args: | |
| db_dir: Path to the db directory | |
| username: The username | |
| collection_id: The collection ID to grant access to | |
| logger: Optional logger for debug output | |
| Returns: | |
| Tuple of (success: bool, message: str) | |
| """ | |
| from fastapi_app.lib.permissions.group_utils import find_group, add_group, add_collection_to_group | |
| from fastapi_app.lib.permissions.user_utils import add_group_to_user | |
| groups_data = load_entity_data(db_dir, 'groups') | |
| user_group_id = username # Personal group has same ID as username | |
| user_group = find_group(user_group_id, groups_data) | |
| # Create personal group if it doesn't exist | |
| if not user_group: | |
| group_success, group_message = add_group( | |
| db_dir, | |
| user_group_id, | |
| f"{username}'s collections", | |
| f"Personal collections for {username}" | |
| ) | |
| if group_success: | |
| if logger: | |
| logger.info(f"Created personal group '{user_group_id}' for user '{username}'") | |
| add_group_to_user(db_dir, username, user_group_id) | |
| else: | |
| return False, f"Failed to create personal group: {group_message}" | |
| # Add collection to user's personal group | |
| add_success, add_message = add_collection_to_group( | |
| db_dir, user_group_id, collection_id | |
| ) | |
| if add_success: | |
| if logger: | |
| logger.info(f"Collection '{collection_id}' added to user group '{user_group_id}'") | |
| return True, f"User '{username}' granted access to collection '{collection_id}'" | |
| else: | |
| # Not an error - might already have access (e.g., wildcard) | |
| return False, add_message | |
| def remove_collection(db_dir: Path, collection_id: str) -> tuple[bool, str, dict]: | |
| """Removes a collection from the collections.json file and cleans up file metadata. | |
| For each file in the collection: | |
| - Removes the collection from the file's doc_collections array | |
| - If the file has no other collections after removal, marks it as deleted | |
| Args: | |
| db_dir: Path to the db directory | |
| collection_id: The collection ID to remove | |
| Returns: | |
| Tuple of (success: bool, message: str, stats: dict) | |
| where stats contains 'files_updated' and 'files_deleted' counts | |
| """ | |
| collections_data = load_entity_data(db_dir, 'collections') | |
| if not collection_exists(collection_id, collections_data): | |
| return False, f"Collection '{collection_id}' not found.", {} | |
| # Import database utilities | |
| from fastapi_app.lib.repository.file_repository import FileRepository | |
| from fastapi_app.lib.models.models import FileUpdate | |
| from fastapi_app.lib.core.dependencies import _DatabaseManagerSingleton | |
| # Initialize database access | |
| db_path = db_dir / "metadata.db" | |
| db_manager = _DatabaseManagerSingleton.get_instance(str(db_path)) | |
| file_repo = FileRepository(db_manager) | |
| files_updated = 0 | |
| files_deleted = 0 | |
| # Get all files that contain this collection | |
| files_with_collection = file_repo.get_files_by_collection(collection_id, include_deleted=False) | |
| # Update each file's collection list | |
| for file_metadata in files_with_collection: | |
| doc_collections = file_metadata.doc_collections.copy() | |
| # Remove the collection from the array | |
| if collection_id in doc_collections: | |
| doc_collections.remove(collection_id) | |
| # Update the file | |
| if len(doc_collections) == 0: | |
| # No collections left - mark as deleted | |
| file_repo.mark_deleted(file_metadata.id) | |
| files_deleted += 1 | |
| else: | |
| # Still has other collections - just update the array | |
| file_repo.update_file(file_metadata.id, FileUpdate(doc_collections=doc_collections)) | |
| files_updated += 1 | |
| # Remove collection from collections.json | |
| collections_data = [collection for collection in collections_data | |
| if collection.get('id') != collection_id] | |
| save_entity_data(db_dir, 'collections', collections_data) | |
| stats = { | |
| 'files_updated': files_updated, | |
| 'files_deleted': files_deleted | |
| } | |
| message = f"Collection '{collection_id}' removed successfully." | |
| if files_updated > 0 or files_deleted > 0: | |
| message += f" {files_updated} files updated, {files_deleted} files marked as deleted." | |
| return True, message, stats | |
| def set_collection_property(db_dir: Path, collection_id: str, property_name: str, value: str) -> tuple[bool, str]: | |
| """Sets a property for a collection. | |
| Args: | |
| db_dir: Path to the db directory | |
| collection_id: The collection ID | |
| property_name: The property to set (name, description) | |
| value: The new value | |
| Returns: | |
| Tuple of (success: bool, message: str) | |
| """ | |
| collections_data = load_entity_data(db_dir, 'collections') | |
| # Check for ID conflicts if changing ID | |
| if property_name == 'id': | |
| if collection_exists(value, collections_data): | |
| return False, f"Collection with ID '{value}' already exists." | |
| collection = find_collection(collection_id, collections_data) | |
| if not collection: | |
| return False, f"Collection '{collection_id}' not found." | |
| collection[property_name] = value | |
| save_entity_data(db_dir, 'collections', collections_data) | |
| if property_name == 'id': | |
| return True, f"Collection '{collection_id}' is now '{value}'." | |
| else: | |
| return True, f"Property '{property_name}' for collection '{collection_id}' set to '{value}'." | |
| def list_collections(db_dir: Path) -> List[Dict[str, Any]]: | |
| """Lists all collections. | |
| Args: | |
| db_dir: Path to the db directory | |
| Returns: | |
| List of collection dictionaries | |
| """ | |
| return load_entity_data(db_dir, 'collections') | |