class FileCloudIntegration_v1
Manages integration with FileCloud for document storage and metadata
/tf/active/vicechatdev/CDocs single class/utils/filecloud_integration.py
23 - 792
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:
# Determine file type
_, file_ext = os.path.splitext(local_file_path)
is_word = file_ext.lower() in ['.docx', '.doc']
is_pdf = file_ext.lower() == '.pdf'
# Create folder structure for document if it doesn't exist
doc_folder = f"/documents/{document.doc_type}/{document.doc_number}"
version_folder = f"{doc_folder}/v{version_number}"
self._ensure_folder_exists(version_folder)
# Upload file to FileCloud
filename = f"{document.doc_number}_v{version_number}{file_ext}"
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)}")
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:
# Check if we have a Word file path
word_path = document_version.word_file_path
if not word_path:
return {'success': False, 'message': 'No Word document to convert'}
# Check if document already has a PDF version
if document_version.pdf_file_path:
return {
'success': True,
'message': 'PDF already exists',
'pdf_path': document_version.pdf_file_path
}
# Parse the path to get folder and base name
folder_path = os.path.dirname(word_path)
base_name = os.path.basename(word_path)
name_without_ext, _ = os.path.splitext(base_name)
pdf_filename = f"{name_without_ext}.pdf"
pdf_path = os.path.join(folder_path, pdf_filename)
# Download the Word document
download_result = self.download_document(word_path)
if not isinstance(download_result, dict) or not download_result.get('success', False):
return {'success': False, 'message': 'Failed to download Word document'}
word_content = download_result.get('content')
# Create a temporary file for the Word document
with tempfile.NamedTemporaryFile(suffix='.docx', delete=False) as temp_word:
temp_word.write(word_content)
temp_word_path = temp_word.name
# Create a temporary file for the PDF
pdf_temp_path = temp_word_path.replace('.docx', '.pdf')
# Convert to PDF (using external library or API)
# Note: This is a placeholder. You would need to implement the actual conversion logic
# using libraries like python-docx-to-pdf, or external services.
# For this example, I'll assume we have a helper method
conversion_success = self._convert_docx_to_pdf(temp_word_path, pdf_temp_path)
if not conversion_success:
# Clean up temp files
if os.path.exists(temp_word_path):
os.unlink(temp_word_path)
return {'success': False, 'message': 'PDF conversion failed'}
# Upload the PDF to FileCloud
with open(pdf_temp_path, 'rb') as pdf_file:
upload_result = self.upload_document(
local_file_path=pdf_temp_path,
remote_folder_path=folder_path,
filename=pdf_filename
)
# Clean up temp files
if os.path.exists(temp_word_path):
os.unlink(temp_word_path)
if os.path.exists(pdf_temp_path):
os.unlink(pdf_temp_path)
if not upload_result.get('success', False):
return {'success': False, 'message': f'Failed to upload PDF: {upload_result.get("message")}'}
# Update document version with PDF path
document_version.pdf_file_path = pdf_path
return {
'success': True,
'message': 'Document converted to PDF successfully',
'pdf_path': pdf_path
}
except Exception as e:
logger.error(f"Error converting document to PDF: {str(e)}")
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 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: strusername: Type: strpassword: 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: strremote_folder_path: Type: strfilename: 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: strlocal_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: ControlledDocumentlocal_file_path: Type: strversion_number: Type: strcreated_by: Type: strversion_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: strtitle: Type: strdoc_type: Type: strdepartment: Type: strowner_uid: Type: strinitial_version: Type: stradditional_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: strmetadata: 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: strmetadata: 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: strpdf_path: Type: str
Returns: Returns bool
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)
Similar Components
AI-powered semantic similarity - components with related functionality:
-
class FileCloudIntegration 97.6% similar
-
class FileCloudClient_v1 66.0% similar
-
function _get_filecloud_integration_v1 63.8% similar
-
function _get_filecloud_integration 63.6% similar
-
class FileCloudClient 63.5% similar