🔍 Code Extractor

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

Purpose

FileCloudClient provides a comprehensive interface for managing files and folders on a FileCloud server. It handles authentication, maintains session state, and provides methods for uploading files, retrieving file information, creating folder structures, and comparing file modification dates. The class automatically authenticates upon instantiation and manages cookies for subsequent API calls. It's designed for synchronization workflows where files need to be uploaded to FileCloud with proper metadata preservation.

Source Code

class FileCloudClient:
    """
    FileCloud client for uploading and managing files.
    """
    
    def __init__(self, server_url: str, username: str, password: str):
        """
        Initialize FileCloud client.
        
        Args:
            server_url: The URL of the FileCloud server
            username: Username for authentication
            password: Password for authentication
        """
        self.server_url = server_url.rstrip('/')
        self.username = username
        self.password = password
        self.session = requests.session()
        self.authenticated = False
        self.headers = {'Accept': 'application/json'}
        
        # Setup logging
        self.logger = logging.getLogger(__name__)
        
        # Authenticate
        self.login()
    
    def login(self) -> bool:
        """
        Authenticate with the FileCloud server.
        
        Returns:
            bool: True if authentication is successful, False otherwise.
        """
        login_endpoint = '/core/loginguest'
        credentials = {'userid': self.username, 'password': self.password}
        
        try:
            response = self.session.post(
                f"{self.server_url}{login_endpoint}", 
                data=credentials, 
                headers=self.headers
            )
            
            login_result = response.json()
            
            if login_result['command'][0]['result'] == 1:
                self.authenticated = True
                self.logger.info("Successfully authenticated with FileCloud")
                return True
            else:
                self.authenticated = False
                error_message = login_result['command'][0].get('message', 'Unknown error')
                self.logger.error(f"FileCloud login failed: {error_message}")
                return False
        except Exception as e:
            self.logger.error(f"FileCloud login error: {str(e)}")
            self.authenticated = False
            return False
    
    def get_file_info(self, file_path: str) -> Optional[Dict]:
        """
        Get information about a file in FileCloud.
        
        Args:
            file_path: Path to file in FileCloud
            
        Returns:
            File information dictionary or None if file doesn't exist
        """
        if not self.authenticated:
            self.logger.error("Not authenticated with FileCloud")
            return None
        
        info_endpoint = '/core/fileinfo'
        params = {'file': file_path}
        
        try:
            response = self.session.post(
                f"{self.server_url}{info_endpoint}",
                params=params,
                cookies=self.session.cookies
            )
            
            doc = xmltodict.parse(response.text)
            
            if doc['fileinfo'] is None:
                return None  # File doesn't exist
            
            file_info = doc['fileinfo']['entry']
            
            # Parse size field - FileCloud returns formatted strings like "6.08 MB"
            size_str = file_info.get('size', '0')
            size_bytes = self._parse_size_string(size_str)
            
            return {
                'name': file_info['name'],
                'path': file_path,
                'size': size_bytes,
                'size_formatted': size_str,  # Keep original formatted string
                'modified': file_info.get('modifiediso'),
                'lastmodified': file_info.get('modifiediso'),  # Alias for compatibility
                'created': file_info.get('creatediso'),
                'type': file_info.get('type')
            }
            
        except Exception as e:
            self.logger.error(f"Error getting file info for {file_path}: {e}")
            return None
    
    def _parse_size_string(self, size_str: str) -> int:
        """
        Parse FileCloud size string (e.g., "6.08 MB") to bytes.
        
        Args:
            size_str: Size string from FileCloud API
            
        Returns:
            Size in bytes as integer
        """
        if not size_str or size_str == '0':
            return 0
        
        try:
            # If it's already a number, return it
            if size_str.isdigit():
                return int(size_str)
            
            # Parse formatted size strings like "6.08 MB"
            size_str = size_str.strip().upper()
            
            # Extract number and unit
            parts = size_str.split()
            if len(parts) != 2:
                # Try to extract number from start of string
                import re
                match = re.match(r'^([\d.]+)', size_str)
                if match:
                    number = float(match.group(1))
                    unit = size_str[len(match.group(1)):].strip()
                else:
                    return 0
            else:
                number = float(parts[0])
                unit = parts[1]
            
            # Convert to bytes
            multipliers = {
                'B': 1,
                'BYTES': 1,
                'KB': 1024,
                'MB': 1024 * 1024,
                'GB': 1024 * 1024 * 1024,
                'TB': 1024 * 1024 * 1024 * 1024
            }
            
            multiplier = multipliers.get(unit, 1)
            return int(number * multiplier)
            
        except Exception as e:
            self.logger.warning(f"Could not parse size string '{size_str}': {e}")
            return 0
    
    def create_folder(self, folder_path: str) -> bool:
        """
        Create a folder structure in FileCloud.
        
        Args:
            folder_path: Path to create
            
        Returns:
            True if successful, False otherwise
        """
        if not self.authenticated:
            self.logger.error("Not authenticated with FileCloud")
            return False
        
        path_elements = folder_path.strip('/').split('/')
        
        # Build path incrementally
        for i in range(len(path_elements)):
            current_path = '/' + '/'.join(path_elements[:i+1])
            
            # Check if folder exists
            if self._folder_exists(current_path):
                continue
            
            # Create folder
            parent_path = '/' + '/'.join(path_elements[:i]) if i > 0 else '/'
            folder_name = path_elements[i]
            
            create_endpoint = '/core/createfolder'
            params = {'path': parent_path, 'name': folder_name}
            
            try:
                response = self.session.post(
                    f"{self.server_url}{create_endpoint}",
                    params=params,
                    cookies=self.session.cookies
                )
                
                if response.text == 'OK':
                    self.logger.info(f"Created folder: {current_path}")
                else:
                    self.logger.error(f"Failed to create folder {current_path}: {response.text}")
                    return False
                    
            except Exception as e:
                self.logger.error(f"Error creating folder {current_path}: {e}")
                return False
        
        return True
    
    def _folder_exists(self, folder_path: str) -> bool:
        """
        Check if a folder exists in FileCloud.
        
        Args:
            folder_path: Path to check
            
        Returns:
            True if folder exists, False otherwise
        """
        info_endpoint = '/core/getfilelist'
        parent_path = '/'.join(folder_path.rstrip('/').split('/')[:-1]) or '/'
        folder_name = folder_path.rstrip('/').split('/')[-1]
        
        params = {'path': parent_path}
        
        try:
            response = self.session.post(
                f"{self.server_url}{info_endpoint}",
                params=params,
                cookies=self.session.cookies
            )
            
            doc = xmltodict.parse(response.text)
            
            if 'entries' in doc and 'entry' in doc['entries']:
                entries = doc['entries']['entry']
                if isinstance(entries, dict):
                    entries = [entries]
                
                for entry in entries:
                    if entry.get('name') == folder_name and entry.get('type') == 'dir':
                        return True
            
            return False
            
        except Exception as e:
            self.logger.error(f"Error checking folder existence {folder_path}: {e}")
            return False
    
    def create_folder(self, folder_path: str) -> bool:
        """
        Create folder structure recursively in FileCloud.
        Based on the working implementation from filecloud_wuxi_sync.py
        """
        try:
            if not folder_path.startswith('/'):
                folder_path = '/' + folder_path
            
            path_elements = folder_path[1:].split('/')
            walking = True
            
            for i, p in enumerate(path_elements[3:]):
                if walking:
                    # Check if this level exists
                    info_endpoint = 'core/getfilelist'
                    api_params = {'path': '/' + '/'.join(path_elements[:i+3])}
                    
                    response = requests.post(
                        f"{self.server_url}/{info_endpoint}",
                        params=api_params,
                        cookies=self.session.cookies,
                        timeout=30
                    )
                    
                    if response.status_code != 200:
                        self.logger.error(f"Failed to check folder existence: {response.status_code}")
                        return False
                    
                    doc = xmltodict.parse(response.text)
                    found = False
                    
                    try:
                        entries = doc.get('entries', {}).get('entry', [])
                        if not isinstance(entries, list):
                            entries = [entries] if entries else []
                        
                        for entry in entries:
                            if entry.get('name') == p:
                                found = True
                                break
                    except Exception as e:
                        self.logger.debug(f"No entries found at level {i}: {e}")
                    
                    if found:
                        # Directory exists, continue to next level
                        continue
                    else:
                        # Need to create this directory and all remaining subdirectories
                        create_endpoint = 'core/createfolder'
                        api_params = {
                            'path': '/' + '/'.join(path_elements[:i+3]),
                            'subpath': "/".join(path_elements[i+3:])
                        }
                        
                        self.logger.info(f"Creating folder structure: {api_params}")
                        
                        create_response = requests.post(
                            f"{self.server_url}/{create_endpoint}",
                            params=api_params,
                            cookies=self.session.cookies,
                            timeout=30
                        )
                        
                        # Check for success - FileCloud returns either 'OK' or XML with success message
                        response_text = create_response.text.strip()
                        if (response_text == 'OK' or 
                            'Folder Created Successfully' in response_text or
                            '<result>1</result>' in response_text):
                            self.logger.info(f"Folder creation successful: {folder_path}")
                            return True
                        else:
                            self.logger.error(f"Folder creation failed: {response_text}")
                            return False
                        
                        walking = False
            
            return True
            
        except Exception as e:
            self.logger.error(f"Error creating folder {folder_path}: {e}")
            return False

    def upload_file(self, file_content: bytes, file_path: str, modified_date: datetime) -> bool:
        """
        Upload a file to FileCloud.
        
        Args:
            file_content: File content as bytes
            file_path: Destination path in FileCloud
            modified_date: Modified date to set for the file
            
        Returns:
            True if successful, False otherwise
        """
        if not self.authenticated:
            self.logger.error("Not authenticated with FileCloud")
            return False
        
        # Ensure parent folder exists
        parent_path = '/'.join(file_path.split('/')[:-1])
        if not self.create_folder(parent_path):
            self.logger.error(f"Failed to create parent folder for {file_path}")
            return False
        
        upload_endpoint = '/core/upload'
        
        # Prepare upload parameters
        params = {
            'appname': 'explorer',
            'path': parent_path,
            'date': modified_date.isoformat(timespec='seconds'),
            'offset': 0
        }
        
        # Prepare file for upload
        file_name = file_path.split('/')[-1]
        files = {'file': (file_name, io.BytesIO(file_content))}
        
        try:
            response = self.session.post(
                f"{self.server_url}{upload_endpoint}",
                params=params,
                files=files,
                cookies=self.session.cookies
            )
            
            if response.text == 'OK':
                self.logger.info(f"Successfully uploaded: {file_path}")
                return True
            else:
                self.logger.error(f"Upload failed for {file_path}: {response.text}")
                return False
                
        except Exception as e:
            self.logger.error(f"Error uploading file {file_path}: {e}")
            return False
    
    def file_needs_update(self, local_modified: datetime, remote_file_info: Dict) -> bool:
        """
        Check if a file needs to be updated based on modification dates.
        
        Args:
            local_modified: Local file modification date (should be UTC)
            remote_file_info: Remote file information
            
        Returns:
            True if file needs update, False otherwise
        """
        if not remote_file_info or not remote_file_info.get('modified'):
            self.logger.debug("File needs update: no remote file info or modified date")
            return True  # File doesn't exist remotely or no date info
        
        try:
            from datetime import timezone, timedelta
            
            # Parse remote modification date - FileCloud returns ISO format
            remote_modified_str = remote_file_info['modified']
            
            # Handle different timezone formats from FileCloud
            if '+' in remote_modified_str or 'Z' in remote_modified_str:
                if remote_modified_str.endswith('Z'):
                    remote_modified_str = remote_modified_str.replace('Z', '+00:00')
                remote_modified = datetime.fromisoformat(remote_modified_str)
            else:
                # No timezone info - assume it's in server local time (CET/CEST)
                remote_modified_naive = datetime.fromisoformat(remote_modified_str)
                # Assume CET (UTC+1) - adjust for CEST if needed
                cet_tz = timezone(timedelta(hours=1))
                remote_modified = remote_modified_naive.replace(tzinfo=cet_tz)
            
            # Ensure local_modified is timezone-aware (should be UTC)
            if local_modified.tzinfo is None:
                local_modified = local_modified.replace(tzinfo=timezone.utc)
            
            # Convert remote time to UTC for comparison
            remote_modified_utc = remote_modified.astimezone(timezone.utc)
            local_modified_utc = local_modified.astimezone(timezone.utc)
            
            self.logger.debug(f"Date comparison (UTC):")
            self.logger.debug(f"  Local (SharePoint): {local_modified_utc}")
            self.logger.debug(f"  Remote (FileCloud): {remote_modified_utc}")
            
            # Compare dates with tolerance for precision differences and clock skew
            # Use configurable tolerance to account for:
            # - Rounding differences between systems
            # - Clock synchronization issues
            # - Processing delays
            from config import Config
            time_diff = abs((local_modified_utc - remote_modified_utc).total_seconds())
            tolerance_seconds = Config.DATE_COMPARISON_TOLERANCE_SECONDS
            needs_update = time_diff > tolerance_seconds
            
            self.logger.debug(f"  Time difference: {time_diff:.2f} seconds")
            self.logger.debug(f"  Tolerance: {tolerance_seconds} seconds")
            self.logger.debug(f"  Needs update: {needs_update}")
            
            # If difference is more than 1 second, consider it different
            return needs_update
            
        except Exception as e:
            self.logger.error(f"Error comparing dates: {e}")
            return True  # Default to update if we can't compare

Parameters

Parameter Details

server_url: The base URL of the FileCloud server (e.g., 'https://filecloud.example.com'). Trailing slashes are automatically removed. This is the root endpoint for all API calls.

username: Username credential for FileCloud authentication. Used in the login process to establish an authenticated session.

password: Password credential for FileCloud authentication. Used together with username to authenticate with the FileCloud server.

Return Value

The constructor returns an initialized FileCloudClient instance with an authenticated session (if login succeeds). Key method returns: login() returns bool indicating authentication success; get_file_info() returns Optional[Dict] with file metadata or None if file doesn't exist; create_folder() returns bool indicating folder creation success; upload_file() returns bool indicating upload success; file_needs_update() returns bool indicating whether local file is newer than remote.

Class Interface

Methods

Attributes

Dependencies

• requests
• xmltodict
• io
• typing
• datetime
• logging
• config
• re

Required Imports

import requests
import xmltodict
import io
from typing import Dict, List, Union, Optional, BinaryIO, Any
from datetime import datetime, timezone, timedelta
import logging

Conditional/Optional Imports

These imports are only needed under specific conditions:

from config import Config

import re

Usage Example

from datetime import datetime, timezone
import logging

# Setup logging
logging.basicConfig(level=logging.INFO)

# Initialize client (automatically authenticates)
client = FileCloudClient(
    server_url='https://filecloud.example.com',
    username='myuser',
    password='mypassword'
)

# Check if authenticated
if client.authenticated:
    # Get file information
    file_info = client.get_file_info('/path/to/file.pdf')
    if file_info:
        print(f"File size: {file_info['size_formatted']}")
        print(f"Modified: {file_info['modified']}")
    
    # Create folder structure
    success = client.create_folder('/documents/2024/reports')
    
    # Upload a file
    with open('local_file.pdf', 'rb') as f:
        file_content = f.read()
    
    modified_date = datetime.now(timezone.utc)
    success = client.upload_file(
        file_content=file_content,
        file_path='/documents/2024/reports/file.pdf',
        modified_date=modified_date
    )
    
    # Check if file needs update
    local_modified = datetime.now(timezone.utc)
    remote_info = client.get_file_info('/documents/2024/reports/file.pdf')
    needs_update = client.file_needs_update(local_modified, remote_info)

Best Practices

• Always check the 'authenticated' attribute before performing operations to ensure the session is valid
• The client automatically authenticates during __init__, so handle potential authentication failures by checking the authenticated attribute immediately after instantiation
• Use timezone-aware datetime objects (with UTC timezone) when calling upload_file() and file_needs_update() to ensure proper date comparisons
• The session is maintained throughout the object's lifetime - reuse the same client instance for multiple operations rather than creating new instances
• File paths in FileCloud should start with '/' - the class handles this in some methods but it's best to provide properly formatted paths
• The create_folder() method creates parent directories recursively, so you don't need to create each level separately
• Error handling is built-in with logging - configure logging appropriately to capture diagnostic information
• The _parse_size_string() and _folder_exists() methods are internal helpers and should not be called directly
• When comparing file dates with file_needs_update(), ensure local_modified is in UTC timezone for accurate comparisons
• The client uses requests.session() which maintains cookies automatically - don't manually manage session cookies
• Upload operations automatically create parent folders, but explicit folder creation can be more efficient for bulk operations

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class FileCloudClient_v1 90.8% 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 FileCloudAPI 70.9% similar

  Python wrapper for the FileCloud REST API. This class provides methods to interact with FileCloud server APIs, handling authentication, session management, and various file operations.

  From: /tf/active/vicechatdev/FC_api copy.py

• function get_filecloud_client 70.2% similar

  Singleton factory function that returns a globally cached FileCloud API client instance, handling initialization, authentication, and re-authentication as needed.

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

• class FileCloudAPI_v1 70.0% similar

  Python wrapper for the FileCloud REST API. This class provides methods to interact with FileCloud server APIs, handling authentication, session management, and various file operations.

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

• function test_filecloud_connection_v1 67.1% similar

  Tests the connection to a FileCloud server by attempting to instantiate a FileCloudClient with credentials from configuration.

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