🔍 Code Extractor

class FileCloudIntegration

Maturity: 26

Manages integration with FileCloud for document storage and metadata

File:
/tf/active/vicechatdev/CDocs/utils/filecloud_integration.py
Lines:
26 - 1313
Complexity:
moderate

Purpose

Manages integration with FileCloud for document storage and metadata

Source Code

class FileCloudIntegration:
    """Manages integration with FileCloud for document storage and metadata"""
    
    def __init__(self, server_url: str, username: str, password: str):
        """
        Initialize FileCloud integration
        
        Parameters
        ----------
        server_url : str
            URL of the FileCloud server
        username : str
            FileCloud username
        password : str
            FileCloud password
        """
        self.client = FileCloudAPI(server_url, username, password)
        self.metadata_catalog = MetadataCatalog(self.client)
        
        logger.debug("Initialized FileCloud integration")
    
    def connect(self) -> bool:
        """
        Establish connection to FileCloud
        
        Returns
        -------
        bool
            True if connection successful, False otherwise
        """
        try:
            return self.client.login()
        except Exception as e:
            logger.error(f"Error connecting to FileCloud: {str(e)}")
            return False
    
    def upload_document(self, local_file_path: str, remote_folder_path: str, 
                       filename: Optional[str] = None, 
                       metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """
        Upload a document to FileCloud
        
        Parameters
        ----------
        local_file_path : str
            Path to the local file
        remote_folder_path : str
            Path to the remote folder
        filename : str, optional
            Name to use for the uploaded file
        metadata : Dict[str, Any], optional
            Metadata to attach to the file
            
        Returns
        -------
        Dict[str, Any]
            Upload result including file path
        """
        try:
            # Ensure folder exists
            self._ensure_folder_exists(remote_folder_path)
            
            # Upload file
            result = self.client.upload_file(
                local_file_path=local_file_path,
                remote_path=remote_folder_path,
                filename=filename,
                overwrite=True
            )
            
            if not result.get('success', False):
                return result
            
            # Get full path to the uploaded file
            if filename:
                file_path = os.path.join(remote_folder_path, filename)
            else:
                file_path = os.path.join(remote_folder_path, os.path.basename(local_file_path))
            
            # Add metadata if provided
            if metadata and file_path:
                self._set_document_metadata(file_path, metadata)
            
            # Add path to result
            result['file_path'] = file_path
            
            return result
            
        except Exception as e:
            logger.error(f"Error uploading document to FileCloud: {str(e)}")
            return {'success': False, 'message': f'Upload error: {str(e)}'}
    
    def download_document(self, file_path: str, local_path: Optional[str] = None) -> Dict[str, Any]:
        """
        Download a document from FileCloud
        
        Parameters
        ----------
        file_path : str
            Path to the file in FileCloud
        local_path : str, optional
            Local path where to save the file
            
        Returns
        -------
        Dict[str, Any]
            Download result with file content if local_path not provided
        """
        try:
            result = self.client.download_file(file_path, local_path)
            
            if local_path and isinstance(result, dict) and result.get('success', False):
                return {
                    'success': True,
                    'local_path': local_path,
                    'file_path': file_path
                }
            
            if not local_path and isinstance(result, bytes):
                return {
                    'success': True,
                    'content': result,
                    'file_path': file_path
                }
            
            return result
            
        except Exception as e:
            logger.error(f"Error downloading document from FileCloud: {str(e)}")
            return {'success': False, 'message': f'Download error: {str(e)}'}
    
    def get_edit_url(self, file_path: str) -> Dict[str, Any]:
        """
        Get URL for online editing
        
        Parameters
        ----------
        file_path : str
            Path to the file in FileCloud
            
        Returns
        -------
        Dict[str, Any]
            Result with edit URL
        """
        try:
            # Check if file exists
            file_exists = self.client.check_file_exists(file_path)
            
            if not file_exists:
                return {'success': False, 'message': f'File not found: {file_path}'}
            
            # Create a temporary share for editing
            share_result = self.client.create_document_share(
                file_path=file_path,
                share_type="collaborate",
                expiry_days=1  # 24-hour expiry
            )
            
            if not share_result.get('success', False):
                return {'success': False, 'message': f'Failed to create edit URL: {share_result.get("message", "Unknown error")}'}
            
            # Return the share URL
            return {
                'success': True,
                'edit_url': share_result.get('share_url'),
                'file_path': file_path,
                'share_id': share_result.get('share_id'),
                'expires': datetime.datetime.now() + datetime.timedelta(days=1)
            }
            
        except Exception as e:
            logger.error(f"Error creating edit URL: {str(e)}")
            return {'success': False, 'message': f'Error creating edit URL: {str(e)}'}
    
    def get_view_url(self, file_path: str) -> Dict[str, Any]:
        """
        Get URL for viewing a document
        
        Parameters
        ----------
        file_path : str
            Path to the file in FileCloud
            
        Returns
        -------
        Dict[str, Any]
            Result with view URL
        """
        try:
            # Check if file exists
            file_exists = self.client.check_file_exists(file_path)
            
            if not file_exists:
                return {'success': False, 'message': f'File not found: {file_path}'}
            
            # Create a temporary share for viewing
            share_result = self.client.create_document_share(
                file_path=file_path,
                share_type="view",
                expiry_days=1  # 24-hour expiry
            )
            
            if not share_result.get('success', False):
                return {'success': False, 'message': f'Failed to create view URL: {share_result.get("message", "Unknown error")}'}
            
            # Return the share URL
            return {
                'success': True,
                'view_url': share_result.get('share_url'),
                'file_path': file_path,
                'share_id': share_result.get('share_id'),
                'expires': datetime.datetime.now() + datetime.timedelta(days=1)
            }
            
        except Exception as e:
            logger.error(f"Error creating view URL: {str(e)}")
            return {'success': False, 'message': f'Error creating view URL: {str(e)}'}
    
    def create_document_version(self, document: ControlledDocument, 
                              local_file_path: str, 
                              version_number: str,
                              created_by: str,
                              version_comment: str = None) -> Optional[DocumentVersion]:
        """
        Create a new version of a controlled document
        
        Parameters
        ----------
        document : ControlledDocument
            The document to create a version for
        local_file_path : str
            Path to the local file
        version_number : str
            Version number (e.g., "1.0")
        created_by : str
            UID of the user creating the version
        version_comment : str, optional
            Comment for the version
            
        Returns
        -------
        Optional[DocumentVersion]
            The new document version or None if creation failed
        """
        try:
            # Get file extension and type
            _, file_ext = os.path.splitext(local_file_path)
            is_word = file_ext.lower() in ['.docx', '.doc', '.rtf']
            is_pdf = file_ext.lower() == '.pdf'
            
            # Get folder path from centralized function
            from CDocs.controllers.filecloud_controller import get_filecloud_document_path
            
            # This will create the path based on document settings (custom or standard)
            target_path = get_filecloud_document_path(document, version_number)
            version_folder = os.path.dirname(target_path)
            
            # Ensure folder structure exists
            self._ensure_folder_exists(version_folder)
            
            # Get filename from path
            filename = os.path.basename(target_path)
            
            # Upload file to FileCloud
            upload_result = self.upload_document(
                local_file_path=local_file_path,
                remote_folder_path=version_folder,
                filename=filename
            )
            
            if not upload_result.get('success', False):
                logger.error(f"Failed to upload file: {upload_result.get('message')}")
                return None
            
            # Prepare file paths
            file_paths = {}
            if is_word:
                file_paths['word'] = upload_result['file_path']
            elif is_pdf:
                file_paths['pdf'] = upload_result['file_path']
                
            # Create document version
            properties = {
                'changeSummary': version_comment if version_comment else '',
                'file_type': file_ext.lstrip('.').lower(),
                'file_name': filename
            }
            
            # Create version in the database
            version = document.create_version(
                version_number=version_number,
                created_by=created_by,
                file_paths=file_paths,
                properties=properties
            )
            
            return version
            
        except Exception as e:
            logger.error(f"Error creating document version: {str(e)}")
            logger.error(traceback.format_exc())
            return None
            
    def upload_new_controlled_document(self, 
                                     local_file_path: str,
                                     title: str,
                                     doc_type: str,
                                     department: str,
                                     owner_uid: str,
                                     initial_version: str = "1.0",
                                     additional_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """
        Upload a new controlled document to FileCloud and create it in the database
        
        Parameters
        ----------
        local_file_path : str
            Path to the local file
        title : str
            Document title
        doc_type : str
            Document type code or name
        department : str
            Department code or name
        owner_uid : str
            UID of the document owner
        initial_version : str, optional
            Initial version number (default: "1.0")
        additional_metadata : Dict[str, Any], optional
            Additional metadata for the document
            
        Returns
        -------
        Dict[str, Any]
            Result with document and version information
        """
        try:
            # Create controlled document in database
            document = ControlledDocument.create(
                title=title,
                doc_type=doc_type,
                department=department,
                owner=owner_uid
            )
            
            if not document:
                return {
                    'success': False,
                    'message': 'Failed to create controlled document in database'
                }
            
            # Create initial version
            version = self.create_document_version(
                document=document,
                local_file_path=local_file_path,
                version_number=initial_version,
                created_by=owner_uid
            )
            
            if not version:
                return {
                    'success': False,
                    'message': 'Failed to create document version in FileCloud'
                }
            
            # Add FileCloud metadata
            doc_metadata = {
                'DocNumber': document.doc_number,
                'Title': title,
                'Department': document.get_department_name(),
                'DocType': document.doc_type_name,
                'Version': initial_version,
                'Status': document.get_status_name()
            }
            
            # Add any additional metadata
            if additional_metadata:
                doc_metadata.update(additional_metadata)
                
            # Set metadata on the file
            self._set_document_metadata(version.word_file_path or version.pdf_file_path, doc_metadata)
            
            return {
                'success': True,
                'document': document.to_dict(),
                'version': version.to_dict(),
                'doc_number': document.doc_number,
                'file_path': version.word_file_path or version.pdf_file_path
            }
            
        except Exception as e:
            logger.error(f"Error uploading new controlled document: {str(e)}")
            return {'success': False, 'message': f'Upload error: {str(e)}'}
    
    def update_document_metadata(self, document_uid: str, metadata: Dict[str, Any]) -> Dict[str, Any]:
        """
        Update metadata for a document both in database and FileCloud
        
        Parameters
        ----------
        document_uid : str
            UID of the document to update
        metadata : Dict[str, Any]
            Metadata to update
            
        Returns
        -------
        Dict[str, Any]
            Result of the update
        """
        try:
            # Get document from database
            document = ControlledDocument(uid=document_uid)
            if not document.uid:
                return {'success': False, 'message': f'Document not found: {document_uid}'}
                
            # Get current version
            current_version = document.current_version
            if not current_version:
                return {'success': False, 'message': f'Document has no current version'}
                
            # Update document properties in database
            properties_to_update = {}
            
            if 'title' in metadata:
                properties_to_update['title'] = metadata['title']
                
            if 'docType' in metadata:
                properties_to_update['docType'] = metadata['docType']
                
            if 'department' in metadata:
                properties_to_update['department'] = metadata['department']
                
            if properties_to_update:
                db.update_node(document.uid, properties_to_update)
                
            # Update FileCloud metadata if we have a file path
            file_path = current_version.word_file_path or current_version.pdf_file_path
            if file_path:
                fc_metadata = {
                    'DocNumber': document.doc_number,
                    'Title': document.title,
                    'Department': document.get_department_name(),
                    'DocType': document.doc_type_name,
                    'Version': current_version.version_number,
                    'Status': document.get_status_name()
                }
                
                self._set_document_metadata(file_path, fc_metadata)
                
            return {
                'success': True,
                'document': document.to_dict(),
                'message': 'Document metadata updated successfully'
            }
            
        except Exception as e:
            logger.error(f"Error updating document metadata: {str(e)}")
            return {'success': False, 'message': f'Update error: {str(e)}'}
    
    def convert_to_pdf(self, document_version: DocumentVersion) -> Dict[str, Any]:
        """
        Convert a document version to PDF using FileCloud's conversion service
        
        Parameters
        ----------
        document_version : DocumentVersion
            The document version to convert
            
        Returns
        -------
        Dict[str, Any]
            Result of the conversion
        """
        try:
            # Use the centralized function to get the Word file path
            from CDocs.controllers.filecloud_controller import get_filecloud_document_path
            
            document = document_version.document
            if not document:
                return {'success': False, 'message': 'Document not available'}
                
            # Get the editable file path
            word_path = get_filecloud_document_path(
                document=document,
                version=document_version.version_number
            )
            
            if not word_path:
                return {'success': False, 'message': 'No Word document to convert'}
                
            # Check if document already has a PDF version
            pdf_path = f"{os.path.splitext(word_path)[0]}.pdf"
            
            # Check if PDF already exists
            if self.check_file_exists(pdf_path):
                return {
                    'success': True, 
                    'message': 'PDF already exists',
                    'pdf_path': pdf_path
                }
            
            # Determine file type
            _, ext = os.path.splitext(word_path)
            ext = ext.lower()
            
            # Only convert supported file types
            if ext not in ['.doc', '.docx', '.rtf', '.odt', '.txt']:
                return {
                    'success': False,
                    'message': f'File type {ext} not supported for conversion'
                }
            
            # Create temporary directory for processing
            temp_dir = tempfile.mkdtemp()
            
            try:
                # Download the document
                download_result = self.download_document(word_path)
                
                if not download_result.get('success', False):
                    return {
                        'success': False,
                        'message': f"Download failed: {download_result.get('message', 'Unknown error')}"
                    }
                    
                # Save to temp file
                temp_doc_path = os.path.join(temp_dir, f"document{ext}")
                with open(temp_doc_path, 'wb') as f:
                    f.write(download_result['content'])
                    
                # Set up output path
                temp_pdf_path = os.path.join(temp_dir, "document.pdf")
                
                # Convert to PDF using LibreOffice
                conversion_command = [
                    'libreoffice',
                    '--headless',
                    '--convert-to', 'pdf',
                    '--outdir', temp_dir,
                    temp_doc_path
                ]
                
                # Run conversion process
                process = subprocess.Popen(
                    conversion_command,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE
                )
                
                stdout, stderr = process.communicate()
                
                if process.returncode != 0:
                    return {
                        'success': False,
                        'message': f'PDF conversion failed: {stderr.decode()}'
                    }
                    
                # Check if PDF was created
                if not os.path.exists(temp_pdf_path):
                    # Check for file with a different name (LibreOffice may use the original filename)
                    base_filename = os.path.splitext(os.path.basename(temp_doc_path))[0]
                    potential_pdf = os.path.join(temp_dir, f"{base_filename}.pdf")
                    
                    if os.path.exists(potential_pdf):
                        temp_pdf_path = potential_pdf
                    else:
                        return {
                            'success': False,
                            'message': 'PDF file not created by conversion process'
                        }
                
                # Upload PDF to FileCloud
                with open(temp_pdf_path, 'rb') as f:
                    pdf_content = f.read()
                    
                # Extract directory path and filename
                pdf_dir = os.path.dirname(pdf_path)
                pdf_filename = os.path.basename(pdf_path)
                
                # Ensure directory exists
                self._ensure_folder_exists(pdf_dir)
                
                # Upload the PDF
                upload_result = self.upload_document(
                    local_file_path=temp_pdf_path,
                    remote_folder_path=pdf_dir,
                    filename=pdf_filename
                )
                
                if not upload_result.get('success', False):
                    return {
                        'success': False,
                        'message': f"Upload failed: {upload_result.get('message', 'Unknown error')}"
                    }
                    
                # Update the document version with PDF path
                document_version.pdf_file_path = pdf_path
                
                # Add metadata about conversion
                metadata = {
                    'converted_on': datetime.now().isoformat(),
                    'source_path': word_path,
                    'pdf_path': pdf_path,
                    'conversion_method': 'libreoffice'
                }
                
                self.add_file_metadata(pdf_path, metadata)
                
                return {
                    'success': True,
                    'message': 'Document successfully converted to PDF',
                    'pdf_path': pdf_path
                }
                
            except Exception as e:
                return {
                    'success': False,
                    'message': f'Error during conversion: {str(e)}'
                }
                
            finally:
                # Clean up temporary directory
                try:
                    if os.path.exists(temp_dir):
                        shutil.rmtree(temp_dir)
                except Exception as cleanup_err:
                    logger.warning(f"Failed to clean up temporary directory: {cleanup_err}")
            
        except Exception as e:
            logger.error(f"Error in convert_to_pdf: {e}")
            logger.error(traceback.format_exc())
            return {
                'success': False,
                'message': f'Conversion error: {str(e)}'
            }
    
    def get_document_metadata(self, file_path: str) -> Dict[str, Any]:
        """
        Get metadata for a document from FileCloud
        
        Parameters
        ----------
        file_path : str
            Path to the file in FileCloud
            
        Returns
        -------
        Dict[str, Any]
            Document metadata
        """
        try:
            # Get metadata from FileCloud
            metadata_result = self.metadata_catalog.get_metadata_values(file_path)
            
            if not metadata_result.get('success', False):
                return {'success': False, 'message': f'Failed to get metadata: {metadata_result.get("message")}'}
                
            # Convert to simple dictionary format
            attribute_values = self.metadata_catalog.get_attribute_values(metadata_result)
            
            return {
                'success': True,
                'metadata': attribute_values,
                'file_path': file_path
            }
            
        except Exception as e:
            logger.error(f"Error getting document metadata: {str(e)}")
            return {'success': False, 'message': f'Metadata retrieval error: {str(e)}'}
    
    def search_documents(self, search_criteria: Dict[str, Any]) -> List[Dict[str, Any]]:
        """
        Search for documents in FileCloud based on metadata
        
        Parameters
        ----------
        search_criteria : Dict[str, Any]
            Search criteria with keys matching metadata attribute names
            
        Returns
        -------
        List[Dict[str, Any]]
            List of matching documents
        """
        try:
            # Format criteria for metadata search
            formatted_criteria = []
            
            for key, value in search_criteria.items():
                if value is not None:
                    formatted_criteria.append({
                        'set_name': 'ControlledDocument',
                        'attribute_name': key,
                        'value': value,
                        'operator': 'equals'
                    })
            
            # Perform search
            search_results = self.metadata_catalog.search_files_by_metadata(
                search_criteria=formatted_criteria,
                search_string="**",  # Match any filename
                search_location="/documents"  # Limit to documents folder
            )
            
            # Get metadata for each result
            documents = []
            for file_path in search_results:
                metadata_result = self.get_document_metadata(file_path)
                
                if metadata_result.get('success', False):
                    documents.append({
                        'file_path': file_path,
                        'metadata': metadata_result.get('metadata', {})
                    })
            
            return documents
            
        except Exception as e:
            logger.error(f"Error searching documents: {str(e)}")
            return []
    
    def _ensure_folder_exists(self, folder_path: str) -> bool:
        """
        Ensure a folder exists in FileCloud, creating it if necessary
        
        Parameters
        ----------
        folder_path : str
            Path to the folder
            
        Returns
        -------
        bool
            True if folder exists or was created, False otherwise
        """
        try:
            # Check if folder exists
            folder_exists = self.client.check_folder_exists(folder_path)
            
            if folder_exists:
                return True
            
            # Create folder with parent folders
            result = self.client.create_directory_tree("", folder_path)
            
            return result.get('success', False)
            
        except Exception as e:
            logger.error(f"Error ensuring folder exists: {str(e)}")
            return False
    
    def _set_document_metadata(self, file_path: str, metadata: Dict[str, Any]) -> Dict[str, Any]:
        """
        Set metadata for a document
        
        Parameters
        ----------
        file_path : str
            Path to the file
        metadata : Dict[str, Any]
            Metadata to set
            
        Returns
        -------
        Dict[str, Any]
            Result of setting metadata
        """
        try:
            # Get document metadata set ID
            doc_set = self.metadata_catalog.get_metadata_set_by_name("ControlledDocument")
            
            if not doc_set:
                return {'success': False, 'message': 'Metadata set for controlled documents not found'}
            
            # Ensure file has the metadata set associated
            add_set_result = self.metadata_catalog.add_set_to_file_object(
                file_path=file_path,
                set_id=doc_set.get('id')
            )
            
            if not add_set_result.get('success', False) and "already" not in add_set_result.get('message', ''):
                return add_set_result
            
            # Save metadata attributes
            result = self.metadata_catalog.save_attribute_values_by_name(
                file_path=file_path,
                set_name="ControlledDocument",
                attributes=metadata
            )
            
            return result
            
        except Exception as e:
            logger.error(f"Error setting document metadata: {str(e)}")
            return {'success': False, 'message': f'Error setting metadata: {str(e)}'}
            
    def _convert_docx_to_pdf(self, docx_path: str, pdf_path: str) -> bool:
        """
        Convert a DOCX file to PDF
        
        Parameters
        ----------
        docx_path : str
            Path to the Word document
        pdf_path : str
            Path where to save the PDF
            
        Returns
        -------
        bool
            True if conversion was successful, False otherwise
        """
        try:
            # This is a placeholder implementation
            # In a real implementation, you would use a library like python-docx-to-pdf,
            # or an external service like LibreOffice, unoconv, or cloud conversion API
            
            # For demonstration, I'll assume the conversion always succeeds
            # but in a real implementation, this would perform the actual conversion
            
            # Example using docx2pdf (you'd need to install it):
            # from docx2pdf import convert
            # convert(docx_path, pdf_path)
            
            # Example using subprocess to call LibreOffice:
            # import subprocess
            # subprocess.run(['libreoffice', '--headless', '--convert-to', 'pdf', 
            #                '--outdir', os.path.dirname(pdf_path), docx_path])
            
            # For this placeholder, just create an empty PDF file
            with open(pdf_path, 'wb') as f:
                f.write(b'%PDF-1.5\n%EOF\n')
                
            return True
            
        except Exception as e:
            logger.error(f"Error converting DOCX to PDF: {str(e)}")
            return False
    
    def manage_document_share(self, document_version: DocumentVersion, 
                             share_name: Optional[str] = None,
                             validity_period: int = 90) -> Dict[str, Any]:
        """
        Create or get an existing share for a document version
        
        Parameters
        ----------
        document_version : DocumentVersion
            The document version to share
        share_name : str, optional
            Name for the share (defaults to document title + version)
        validity_period : int, optional
            Validity period in days (default: 90)
            
        Returns
        -------
        Dict[str, Any]
            Share information including share_id and share_url
        """
        try:
            # Get document from version
            document = document_version.document
            if not document:
                return {'success': False, 'message': 'Document not available'}
            
            # Get file path to share
            file_path = document_version.word_file_path or document_version.pdf_file_path
            if not file_path:
                return {'success': False, 'message': 'No file path available for sharing'}
            
            # Check if file exists
            if not self.check_file_exists(file_path):
                return {'success': False, 'message': f'File not found: {file_path}'}
            
            # If share already exists on the version, return it
            if document_version.share_id:
                # Verify the share still exists
                share_info = self.client.get_share_for_id(document_version.share_id)
                if share_info.get('success', False):
                    return {
                        'success': True,
                        'message': 'Share already exists',
                        'share_id': document_version.share_id,
                        'share_url': document_version.share_url,
                        'is_new': False
                    }
            
            # Check if share exists for this file path
            share_path_result = self.client.get_share_for_path(file_path)
            
            existing_share_id = None
            if share_path_result.get('success', False) and 'data' in share_path_result:
                # Parse the response to extract share ID if a share exists
                if 'shares' in share_path_result['data'] and 'share' in share_path_result['data']['shares']:
                    share_data = share_path_result['data']['shares']['share']
                    if isinstance(share_data, list) and len(share_data) > 0:
                        existing_share_id = share_data[0].get('shareid')
                    elif isinstance(share_data, dict):
                        existing_share_id = share_data.get('shareid')
            
            # If share exists, use it
            if existing_share_id:
                # Get full share info
                share_info = self.client.get_share_for_id(existing_share_id)
                
                if share_info.get('success', False) and 'data' in share_info:
                    # Extract share URL from response
                    share_url = None
                    share_data = None
                    
                    # Handle different response structures
                    if 'shares' in share_info['data'] and 'share' in share_info['data']['shares']:
                        share_data = share_info['data']['shares']['share']
                    elif 'xml' in share_info['data'] and 'share' in share_info['data']['xml']:
                        share_data = share_info['data']['xml']['share']
                    elif 'share' in share_info['data']:
                        share_data = share_info['data']['share']
                    
                    if share_data:
                        if isinstance(share_data, list):
                            share_url = share_data[0].get('shareurl')
                        else:
                            share_url = share_data.get('shareurl')
                    
                    # Update document version with share ID and URL
                    if share_url:
                        document_version.share_id = existing_share_id
                        document_version.share_url = share_url
                        document_version.save()
                        
                        return {
                            'success': True,
                            'message': 'Using existing share',
                            'share_id': existing_share_id,
                            'share_url': share_url,
                            'is_new': False
                        }
            
            # Create a new share if no valid existing share was found
            if not share_name:
                # Use document title and version as share name
                share_name = f"{document.title} - v{document_version.version_number}"
            
            # Create the share
            share_result = self.client.add_share(
                share_location=file_path,
                share_name=share_name,
                validity_period=validity_period,
                max_downloads=1000  # High number for practical purposes
            )
            
            if not share_result.get('success', False):
                return {
                    'success': False, 
                    'message': f"Failed to create share: {share_result.get('message', 'Unknown error')}"
                }
            
            # Extract share ID from the result
            share_id = None
            if 'data' in share_result:
                # Handle different response formats
                if 'xml' in share_result['data'] and 'share' in share_result['data']['xml']:
                    share_data = share_result['data']['xml']['share']
                    share_id = share_data.get('shareid')
                elif 'share' in share_result['data']:
                    share_data = share_result['data']['share']
                    if isinstance(share_data, list):
                        share_id = share_data[0].get('shareid')
                    else:
                        share_id = share_data.get('shareid')
            
            if not share_id:
                return {'success': False, 'message': 'Failed to get share ID from response'}
            
            # Get share info to get the URL
            share_info = self.client.get_share_for_id(share_id)
            
            share_url = None
            if share_info.get('success', False) and 'data' in share_info:
                # Extract share URL from response
                share_data = None
                
                # Handle different response structures
                if 'shares' in share_info['data'] and 'share' in share_info['data']['shares']:
                    share_data = share_info['data']['shares']['share']
                elif 'xml' in share_info['data'] and 'share' in share_info['data']['xml']:
                    share_data = share_info['data']['xml']['share']
                elif 'share' in share_info['data']:
                    share_data = share_info['data']['share']
                
                if share_data:
                    if isinstance(share_data, list):
                        share_url = share_data[0].get('shareurl')
                    else:
                        share_url = share_data.get('shareurl')
            
            # Update document version with share ID and URL
            if share_url:
                document_version.share_id = share_id
                document_version.share_url = share_url
                document_version.save()
                
                return {
                    'success': True,
                    'message': 'Share created successfully',
                    'share_id': share_id,
                    'share_url': share_url,
                    'is_new': True
                }
            else:
                return {'success': False, 'message': 'Failed to get share URL'}
                
        except Exception as e:
            logger.error(f"Error managing document share: {str(e)}")
            logger.error(traceback.format_exc())
            return {'success': False, 'message': f'Share management error: {str(e)}'}
    
    def manage_user_share_access(self, document_version: DocumentVersion, 
                                user_id: str, 
                                grant_write_access: bool = False) -> Dict[str, Any]:
        """
        Manage a user's access to a document share
        
        Parameters
        ----------
        document_version : DocumentVersion
            The document version to manage access for
        user_id : str
            FileCloud username of the user to grant/modify access
        grant_write_access : bool, optional
            Whether to grant write access (default: False, read-only)
            
        Returns
        -------
        Dict[str, Any]
            Result of the operation
        """
        try:
            # Ensure we have a share for this document
            if not document_version.share_id:
                share_result = self.manage_document_share(document_version)
                if not share_result.get('success', False):
                    return share_result
            
            # Check if user already has access to this share
            users_result = self.client.get_users_for_share(share_id=document_version.share_id)
            
            user_exists = False
            if users_result.get('success', False) and 'data' in users_result:
                # Parse the user list to check if user already has access
                user_data = None
                
                # Handle different response structures
                if 'users' in users_result['data']:
                    user_data = users_result['data']['users'].get('user', [])
                elif 'xml' in users_result['data'] and 'users' in users_result['data']['xml']:
                    user_data = users_result['data']['xml']['users'].get('user', [])
                
                if user_data:
                    # Convert to list if it's a single user dict
                    if isinstance(user_data, dict):
                        user_data = [user_data]
                    
                    # Check if user is in the list
                    for user in user_data:
                        if user.get('userid') == user_id:
                            user_exists = True
                            break
            
            # If user doesn't have access yet, add them
            if not user_exists:
                add_result = self.client.add_user_to_share(user_id=user_id, share_id=document_version.share_id)
                if not add_result.get('success', False):
                    return {
                        'success': False,
                        'message': f"Failed to add user to share: {add_result.get('message', 'Unknown error')}"
                    }
            
            # Set appropriate permissions for the user
            perm_result = self.client.set_user_access_for_share(
                share_id=document_version.share_id,
                user_id=user_id,
                download=True,      # Allow downloading
                write=grant_write_access,  # Write access based on parameter
                share=False,        # Don't allow re-sharing
                sync=False          # Don't allow sync
            )
            
            if not perm_result.get('success', False):
                return {
                    'success': False,
                    'message': f"Failed to set user permissions: {perm_result.get('message', 'Unknown error')}"
                }
            
            return {
                'success': True,
                'message': f"Access {'with write permission' if grant_write_access else 'read-only'} granted to {user_id}",
                'user_id': user_id,
                'share_id': document_version.share_id,
                'write_access': grant_write_access
            }
                
        except Exception as e:
            logger.error(f"Error managing user share access: {str(e)}")
            logger.error(traceback.format_exc())
            return {'success': False, 'message': f'Share access management error: {str(e)}'}

    def remove_user_from_share(self, document_version: DocumentVersion, 
                              user_id: str) -> Dict[str, Any]:
        """
        Remove a user's access to a document share
        
        Parameters
        ----------
        document_version : DocumentVersion
            The document version to remove access for
        user_id : str
            FileCloud username of the user to remove
            
        Returns
        -------
        Dict[str, Any]
            Result of the operation
        """
        try:
            # Check if we have a share ID
            if not document_version.share_id:
                return {
                    'success': True,
                    'message': 'No share exists for this document version'
                }
            
            # Remove the user from the share
            remove_result = self.client.delete_user_from_share(
                user_id=user_id, 
                share_id=document_version.share_id
            )
            
            if not remove_result.get('success', False):
                # Check if the error is because the user doesn't have access (already removed)
                if "not found" in remove_result.get('message', '').lower():
                    return {
                        'success': True,
                        'message': f'User {user_id} already has no access to this share',
                        'user_id': user_id,
                        'share_id': document_version.share_id
                    }
                else:
                    return {
                        'success': False,
                        'message': f"Failed to remove user from share: {remove_result.get('message', 'Unknown error')}"
                    }
            
            return {
                'success': True,
                'message': f'User {user_id} removed from share',
                'user_id': user_id,
                'share_id': document_version.share_id
            }
                
        except Exception as e:
            logger.error(f"Error removing user from share: {str(e)}")
            return {'success': False, 'message': f'Share access removal error: {str(e)}'}
    
    def delete_document_share(self, document_version: DocumentVersion) -> Dict[str, Any]:
        """
        Delete a document share
        
        Parameters
        ----------
        document_version : DocumentVersion
            The document version whose share should be deleted
            
        Returns
        -------
        Dict[str, Any]
            Result of the operation
        """
        try:
            # Check if we have a share ID
            if not document_version.share_id:
                return {
                    'success': True,
                    'message': 'No share exists for this document version'
                }
            
            # Delete the share
            delete_result = self.client.delete_share(share_id=document_version.share_id)
            
            if not delete_result.get('success', False):
                return {
                    'success': False,
                    'message': f"Failed to delete share: {delete_result.get('message', 'Unknown error')}"
                }
            
            # Clear share data from document version
            document_version.share_id = None
            document_version.share_url = None
            document_version.save()
            
            return {
                'success': True,
                'message': 'Share deleted successfully'
            }
                
        except Exception as e:
            logger.error(f"Error deleting document share: {str(e)}")
            return {'success': False, 'message': f'Share deletion error: {str(e)}'}
            
    def manage_group_share_access(self, document_version: DocumentVersion, 
                                 group_id: str,
                                 grant_write_access: bool = False) -> Dict[str, Any]:
        """
        Manage a group's access to a document share
        
        Parameters
        ----------
        document_version : DocumentVersion
            The document version to manage access for
        group_id : str
            FileCloud group ID to grant/modify access
        grant_write_access : bool, optional
            Whether to grant write access (default: False, read-only)
            
        Returns
        -------
        Dict[str, Any]
            Result of the operation
        """
        try:
            # Ensure we have a share for this document
            if not document_version.share_id:
                share_result = self.manage_document_share(document_version)
                if not share_result.get('success', False):
                    return share_result
            
            # Add the group to the share or update permissions
            add_result = self.client.add_group_to_share(group_id=group_id, share_id=document_version.share_id)
            
            # Set appropriate permissions for the group
            perm_result = self.client.set_group_access_for_share(
                share_id=document_version.share_id,
                group_id=group_id,
                download=True,      # Allow downloading
                write=grant_write_access,  # Write access based on parameter
                share=False,        # Don't allow re-sharing
                sync=False          # Don't allow sync
            )
            
            if not perm_result.get('success', False):
                return {
                    'success': False,
                    'message': f"Failed to set group permissions: {perm_result.get('message', 'Unknown error')}"
                }
            
            return {
                'success': True,
                'message': f"Access {'with write permission' if grant_write_access else 'read-only'} granted to group {group_id}",
                'group_id': group_id,
                'share_id': document_version.share_id,
                'write_access': grant_write_access
            }
                
        except Exception as e:
            logger.error(f"Error managing group share access: {str(e)}")
            return {'success': False, 'message': f'Share access management error: {str(e)}'}
    
    def check_file_exists(self, file_path: str) -> bool:
        """
        Check if a file exists in FileCloud
        
        Parameters
        ----------
        file_path : str
            Path to the file
            
        Returns
        -------
        bool
            True if file exists, False otherwise
        """
        try:
            return self.client.check_file_exists(file_path)
        except Exception as e:
            logger.error(f"Error checking if file exists: {str(e)}")
            return False

Parameters

Name Type Default Kind
bases - -

Parameter Details

bases: Parameter of type

Return Value

Returns unspecified type

Class Interface

Methods

__init__(self, server_url, username, password)

Purpose: Initialize FileCloud integration Parameters ---------- server_url : str URL of the FileCloud server username : str FileCloud username password : str FileCloud password

Parameters:

  • server_url: Type: str
  • username: Type: str
  • password: Type: str

Returns: None

connect(self) -> bool

Purpose: Establish connection to FileCloud Returns ------- bool True if connection successful, False otherwise

Returns: Returns bool

upload_document(self, local_file_path, remote_folder_path, filename, metadata) -> Dict[str, Any]

Purpose: Upload a document to FileCloud Parameters ---------- local_file_path : str Path to the local file remote_folder_path : str Path to the remote folder filename : str, optional Name to use for the uploaded file metadata : Dict[str, Any], optional Metadata to attach to the file Returns ------- Dict[str, Any] Upload result including file path

Parameters:

  • local_file_path: Type: str
  • remote_folder_path: Type: str
  • filename: Type: Optional[str]
  • metadata: Type: Optional[Dict[str, Any]]

Returns: Returns Dict[str, Any]

download_document(self, file_path, local_path) -> Dict[str, Any]

Purpose: Download a document from FileCloud Parameters ---------- file_path : str Path to the file in FileCloud local_path : str, optional Local path where to save the file Returns ------- Dict[str, Any] Download result with file content if local_path not provided

Parameters:

  • file_path: Type: str
  • local_path: Type: Optional[str]

Returns: Returns Dict[str, Any]

get_edit_url(self, file_path) -> Dict[str, Any]

Purpose: Get URL for online editing Parameters ---------- file_path : str Path to the file in FileCloud Returns ------- Dict[str, Any] Result with edit URL

Parameters:

  • file_path: Type: str

Returns: Returns Dict[str, Any]

get_view_url(self, file_path) -> Dict[str, Any]

Purpose: Get URL for viewing a document Parameters ---------- file_path : str Path to the file in FileCloud Returns ------- Dict[str, Any] Result with view URL

Parameters:

  • file_path: Type: str

Returns: Returns Dict[str, Any]

create_document_version(self, document, local_file_path, version_number, created_by, version_comment) -> Optional[DocumentVersion]

Purpose: Create a new version of a controlled document Parameters ---------- document : ControlledDocument The document to create a version for local_file_path : str Path to the local file version_number : str Version number (e.g., "1.0") created_by : str UID of the user creating the version version_comment : str, optional Comment for the version Returns ------- Optional[DocumentVersion] The new document version or None if creation failed

Parameters:

  • document: Type: ControlledDocument
  • local_file_path: Type: str
  • version_number: Type: str
  • created_by: Type: str
  • version_comment: Type: str

Returns: Returns Optional[DocumentVersion]

upload_new_controlled_document(self, local_file_path, title, doc_type, department, owner_uid, initial_version, additional_metadata) -> Dict[str, Any]

Purpose: Upload a new controlled document to FileCloud and create it in the database Parameters ---------- local_file_path : str Path to the local file title : str Document title doc_type : str Document type code or name department : str Department code or name owner_uid : str UID of the document owner initial_version : str, optional Initial version number (default: "1.0") additional_metadata : Dict[str, Any], optional Additional metadata for the document Returns ------- Dict[str, Any] Result with document and version information

Parameters:

  • local_file_path: Type: str
  • title: Type: str
  • doc_type: Type: str
  • department: Type: str
  • owner_uid: Type: str
  • initial_version: Type: str
  • additional_metadata: Type: Optional[Dict[str, Any]]

Returns: Returns Dict[str, Any]

update_document_metadata(self, document_uid, metadata) -> Dict[str, Any]

Purpose: Update metadata for a document both in database and FileCloud Parameters ---------- document_uid : str UID of the document to update metadata : Dict[str, Any] Metadata to update Returns ------- Dict[str, Any] Result of the update

Parameters:

  • document_uid: Type: str
  • metadata: Type: Dict[str, Any]

Returns: Returns Dict[str, Any]

convert_to_pdf(self, document_version) -> Dict[str, Any]

Purpose: Convert a document version to PDF using FileCloud's conversion service Parameters ---------- document_version : DocumentVersion The document version to convert Returns ------- Dict[str, Any] Result of the conversion

Parameters:

  • document_version: Type: DocumentVersion

Returns: Returns Dict[str, Any]

get_document_metadata(self, file_path) -> Dict[str, Any]

Purpose: Get metadata for a document from FileCloud Parameters ---------- file_path : str Path to the file in FileCloud Returns ------- Dict[str, Any] Document metadata

Parameters:

  • file_path: Type: str

Returns: Returns Dict[str, Any]

search_documents(self, search_criteria) -> List[Dict[str, Any]]

Purpose: Search for documents in FileCloud based on metadata Parameters ---------- search_criteria : Dict[str, Any] Search criteria with keys matching metadata attribute names Returns ------- List[Dict[str, Any]] List of matching documents

Parameters:

  • search_criteria: Type: Dict[str, Any]

Returns: Returns List[Dict[str, Any]]

_ensure_folder_exists(self, folder_path) -> bool

Purpose: Ensure a folder exists in FileCloud, creating it if necessary Parameters ---------- folder_path : str Path to the folder Returns ------- bool True if folder exists or was created, False otherwise

Parameters:

  • folder_path: Type: str

Returns: Returns bool

_set_document_metadata(self, file_path, metadata) -> Dict[str, Any]

Purpose: Set metadata for a document Parameters ---------- file_path : str Path to the file metadata : Dict[str, Any] Metadata to set Returns ------- Dict[str, Any] Result of setting metadata

Parameters:

  • file_path: Type: str
  • metadata: Type: Dict[str, Any]

Returns: Returns Dict[str, Any]

_convert_docx_to_pdf(self, docx_path, pdf_path) -> bool

Purpose: Convert a DOCX file to PDF Parameters ---------- docx_path : str Path to the Word document pdf_path : str Path where to save the PDF Returns ------- bool True if conversion was successful, False otherwise

Parameters:

  • docx_path: Type: str
  • pdf_path: Type: str

Returns: Returns bool

manage_document_share(self, document_version, share_name, validity_period) -> Dict[str, Any]

Purpose: Create or get an existing share for a document version Parameters ---------- document_version : DocumentVersion The document version to share share_name : str, optional Name for the share (defaults to document title + version) validity_period : int, optional Validity period in days (default: 90) Returns ------- Dict[str, Any] Share information including share_id and share_url

Parameters:

  • document_version: Type: DocumentVersion
  • share_name: Type: Optional[str]
  • validity_period: Type: int

Returns: Returns Dict[str, Any]

manage_user_share_access(self, document_version, user_id, grant_write_access) -> Dict[str, Any]

Purpose: Manage a user's access to a document share Parameters ---------- document_version : DocumentVersion The document version to manage access for user_id : str FileCloud username of the user to grant/modify access grant_write_access : bool, optional Whether to grant write access (default: False, read-only) Returns ------- Dict[str, Any] Result of the operation

Parameters:

  • document_version: Type: DocumentVersion
  • user_id: Type: str
  • grant_write_access: Type: bool

Returns: Returns Dict[str, Any]

remove_user_from_share(self, document_version, user_id) -> Dict[str, Any]

Purpose: Remove a user's access to a document share Parameters ---------- document_version : DocumentVersion The document version to remove access for user_id : str FileCloud username of the user to remove Returns ------- Dict[str, Any] Result of the operation

Parameters:

  • document_version: Type: DocumentVersion
  • user_id: Type: str

Returns: Returns Dict[str, Any]

delete_document_share(self, document_version) -> Dict[str, Any]

Purpose: Delete a document share Parameters ---------- document_version : DocumentVersion The document version whose share should be deleted Returns ------- Dict[str, Any] Result of the operation

Parameters:

  • document_version: Type: DocumentVersion

Returns: Returns Dict[str, Any]

manage_group_share_access(self, document_version, group_id, grant_write_access) -> Dict[str, Any]

Purpose: Manage a group's access to a document share Parameters ---------- document_version : DocumentVersion The document version to manage access for group_id : str FileCloud group ID to grant/modify access grant_write_access : bool, optional Whether to grant write access (default: False, read-only) Returns ------- Dict[str, Any] Result of the operation

Parameters:

  • document_version: Type: DocumentVersion
  • group_id: Type: str
  • grant_write_access: Type: bool

Returns: Returns Dict[str, Any]

check_file_exists(self, file_path) -> bool

Purpose: Check if a file exists in FileCloud Parameters ---------- file_path : str Path to the file Returns ------- bool True if file exists, False otherwise

Parameters:

  • file_path: Type: str

Returns: Returns bool

Required Imports

import os
import logging
from typing import Dict
from typing import Any
from typing import Optional

Usage Example

# Example usage:
# result = FileCloudIntegration(bases)

Tags

class filecloudintegration

Similar Components

AI-powered semantic similarity - components with related functionality:

  • class FileCloudClient_v1 65.0% similar

    A client class for interacting with FileCloud storage systems through direct API calls, providing authentication, file search, download, and metadata retrieval capabilities.

    From: /tf/active/vicechatdev/contract_validity_analyzer/utils/filecloud_client.py

  • class MetadataCatalog 64.7% similar

    Helper class to manage FileCloud metadata sets and attributes. This class provides methods to work with FileCloud metadata by providing a more user-friendly interface on top of the raw API.

    From: /tf/active/vicechatdev/metadata_catalog.py

  • class FileCloudClient 64.6% similar

    A client class for interacting with FileCloud server API, providing authentication, file management, folder creation, and file upload capabilities.

    From: /tf/active/vicechatdev/SPFCsync/filecloud_client.py

  • class FileCloudError 64.5% similar

    Custom exception class for handling FileCloud-specific errors in the CDocs document management system.

    From: /tf/active/vicechatdev/CDocs/controllers/filecloud_controller.py

  • function _get_filecloud_integration_v1 63.6% similar

    Factory function that creates and returns a configured FileCloudIntegration instance using credentials from application settings.

    From: /tf/active/vicechatdev/document_controller_backup.py
← Back to Browse