🔍 Code Extractor

function publish_document

Maturity: 81

Publishes an approved controlled document by converting it to PDF with signatures and audit trail, uploading to FileCloud, and updating the document status to PUBLISHED.

File:
/tf/active/vicechatdev/document_controller_backup.py
Lines:
895 - 1103
Complexity:
complex

Purpose

This function handles the complete document publishing workflow in a controlled document management system. It validates that a document is in APPROVED status, downloads the editable version from FileCloud, converts it to a PDF with embedded signatures and audit trail, uploads the PDF back to FileCloud, and updates the document metadata with publication information including effective and expiry dates. This is a critical function for document lifecycle management in regulated environments requiring audit trails and version control.

Source Code

def publish_document(
    user: DocUser,
    document_uid: str,
    effective_date: Optional[datetime] = None,
    expiry_date: Optional[datetime] = None,
    publish_comment: Optional[str] = None
) -> Dict[str, Any]:
    """
    Publish a document, changing its status to PUBLISHED or EFFECTIVE.
    This includes converting the document to PDF format with signatures and audit trail.
    
    Args:
        user: User publishing the document
        document_uid: UID of document to publish
        effective_date: Optional date when document becomes effective
        expiry_date: Optional expiry date
        publish_comment: Optional comment about publishing
        
    Returns:
        Dictionary with publish status
        
    Raises:
        ResourceNotFoundError: If document not found
        ValidationError: If validation fails
        PermissionError: If user doesn't have permission
        BusinessRuleError: If publishing is not allowed
    """
    try:
        # Get document instance
        document = ControlledDocument(uid=document_uid)
        if not document.uid:
            raise ResourceNotFoundError(f"Document not found: {document_uid}")
            
        # Check if document can be published
        if document.status != STATUS_APPROVED:
            raise BusinessRuleError(f"Cannot publish document with status {document.status}. Document must be approved first.")
            
        # Check if document has a current version
        current_version = document.current_version
        if not current_version:
            raise BusinessRuleError("Cannot publish document without a current version")
            
        # Check if the current version has an editable file
        if not current_version.word_file_path:
            raise BusinessRuleError("Current version has no editable document to publish")
            
        # Set effective date if not provided
        if not effective_date:
            effective_date = datetime.now()
            
        # Store previous status for audit
        previous_status = document.status
        
        # Create a temporary directory for processing
        temp_dir = tempfile.mkdtemp()
        
        try:
            # Download the current version's editable file
            editable_file_path = current_version.word_file_path
            
            # Download document from FileCloud using the filecloud_integration utility
            fc_integration = _get_filecloud_integration()
            download_result = fc_integration.download_document(editable_file_path)
            
            if not download_result.get('success'):
                raise BusinessRuleError(f"Failed to download editable document: {download_result.get('message', 'Unknown error')}")
                
            file_content = download_result.get('content')
            if not file_content:
                raise BusinessRuleError("Downloaded file content is empty")
                
            # Save to temp file
            file_ext = os.path.splitext(editable_file_path)[1]
            temp_file_path = os.path.join(temp_dir, f"document{file_ext}")
            
            with open(temp_file_path, 'wb') as f:
                f.write(file_content)
                
            # Get document metadata
            doc_data = {
                'doc_number': document.doc_number,
                'title': document.title,
                'version': current_version.version_number,
                'approved_date': datetime.now().strftime('%Y-%m-%d'),
                'status': STATUS_PUBLISHED
            }
            
            # Get audit trail for approval process
            audit_data = get_document_audit_trail(document_uid)
            
            # Initialize the document converter
            converter = ControlledDocumentConverter(
                signature_dir=os.path.join('/tf/active/document_auditor/signatures')
            )
            
            # Create the PDF with signature page and audit trail
            output_pdf_path = os.path.join(temp_dir, f"{document.doc_number}_v{current_version.version_number}.pdf")
            
            try:
                converter.create_archived_pdf(
                    input_path=temp_file_path,
                    output_path=output_pdf_path,
                    document_data=doc_data,
                    audit_data=audit_data
                )
            except Exception as convert_err:
                raise BusinessRuleError(f"Failed to convert document to PDF: {str(convert_err)}")
                
            # Calculate the FileCloud path for the PDF
            editable_dir = os.path.dirname(editable_file_path)
            pdf_filename = f"{os.path.splitext(os.path.basename(editable_file_path))[0]}.pdf"
            pdf_file_path = os.path.join(editable_dir, pdf_filename)
            
            # Upload PDF to FileCloud
            with open(output_pdf_path, 'rb') as pdf_file:
                upload_result = fc_integration.upload_document(
                    local_file_path=output_pdf_path,
                    remote_folder_path=editable_dir,
                    filename=pdf_filename,
                    metadata={
                        'DocNumber': document.doc_number,
                        'Version': current_version.version_number,
                        'Status': STATUS_PUBLISHED,
                        'PublishedBy': user.username,
                        'PublishedDate': datetime.now().isoformat()
                    }
                )
                
            if not upload_result.get('success'):
                raise BusinessRuleError(f"Failed to upload PDF to FileCloud: {upload_result.get('message', 'Unknown error')}")
                
            # Update document version with PDF path
            current_version.pdf_file_path = pdf_file_path
            
            # Update document status
            document.status = STATUS_PUBLISHED
            
            # Update document metadata in database
            metadata = {
                'published_by': user.uid,
                'published_date': datetime.now().isoformat(),
                'publish_comment': publish_comment,
                'effective_date': effective_date.isoformat() if effective_date else None,
                'expiry_date': expiry_date.isoformat() if expiry_date else None
            }
            
            # Update the document node in the database
            update_data = {
                'status': STATUS_PUBLISHED,
                'metadata': metadata,
                'modifiedDate': datetime.now()
            }
            
            if effective_date:
                update_data['effectiveDate'] = effective_date
                
            if expiry_date:
                update_data['expiryDate'] = expiry_date
                
            db.update_node(document_uid, update_data)
            
            # Log document action
            audit_trail.log_document_lifecycle_event(
                event_type="DOCUMENT_PUBLISHED",
                user=user,
                document_uid=document_uid,
                details={
                    "previous_status": previous_status,
                    "new_status": STATUS_PUBLISHED,
                    "pdf_path": pdf_file_path,
                    "effective_date": effective_date.isoformat() if effective_date else None,
                    "expiry_date": expiry_date.isoformat() if expiry_date else None,
                    "comment": publish_comment
                }
            )
            
            # Send notification
            notifications.notify_document_update(document, "DOCUMENT_STATUS_CHANGED")
            
            return {
                "success": True,
                "document_uid": document_uid,
                "document_number": document.doc_number,
                "title": document.title,
                "previous_status": previous_status,
                "new_status": STATUS_PUBLISHED,
                "pdf_path": pdf_file_path,
                "effective_date": effective_date,
                "expiry_date": expiry_date,
                "message": f"Document {document.doc_number} published successfully"
            }
            
        except Exception as e:
            logger.error(f"Error in document publishing process: {str(e)}")
            raise BusinessRuleError(f"Failed to publish document: {str(e)}")
        finally:
            # Clean up temporary directory
            try:
                if os.path.exists(temp_dir):
                    shutil.rmtree(temp_dir)
            except Exception as cleanup_error:
                logger.warning(f"Failed to remove temporary directory: {temp_dir}: {str(cleanup_error)}")
                
    except (ResourceNotFoundError, ValidationError, PermissionError, BusinessRuleError) as e:
        # Re-raise known errors
        raise
    except Exception as e:
        logger.error(f"Error publishing document {document_uid}: {str(e)}")
        raise BusinessRuleError(f"Failed to publish document: {str(e)}")

Parameters

Name Type Default Kind
user DocUser - positional_or_keyword
document_uid str - positional_or_keyword
effective_date Optional[datetime] None positional_or_keyword
expiry_date Optional[datetime] None positional_or_keyword
publish_comment Optional[str] None positional_or_keyword

Parameter Details

user: DocUser object representing the user performing the publish action. Must have PUBLISH_DOCUMENT permission. Used for audit logging and metadata tracking.

document_uid: String containing the unique identifier (UID) of the document to be published. Must correspond to an existing document in the database.

effective_date: Optional datetime object specifying when the document becomes effective. If not provided, defaults to the current datetime. Used for compliance and version control tracking.

expiry_date: Optional datetime object specifying when the document expires. Can be None for documents without expiration. Used for document lifecycle management.

publish_comment: Optional string containing a comment or note about the publication. Used for audit trail and documentation purposes. Can be None.

Return Value

Type: Dict[str, Any]

Returns a dictionary with keys: 'success' (bool indicating operation success), 'document_uid' (str), 'document_number' (str), 'title' (str), 'previous_status' (str), 'new_status' (str, typically STATUS_PUBLISHED), 'pdf_path' (str path to PDF in FileCloud), 'effective_date' (datetime or None), 'expiry_date' (datetime or None), and 'message' (str with success message). On error, raises one of the documented exceptions instead of returning.

Dependencies

  • logging
  • uuid
  • os
  • tempfile
  • typing
  • datetime
  • io
  • panel
  • shutil
  • traceback
  • CDocs
  • random

Required Imports

import logging
import os
import tempfile
import shutil
from typing import Dict, Any, Optional
from datetime import datetime
from CDocs import db
from CDocs.config import settings
from CDocs.models.document import ControlledDocument
from CDocs.models.user_extensions import DocUser
from CDocs.utils import notifications
from CDocs.utils import audit_trail
from CDocs.controllers import require_permission, log_controller_action, transaction
from CDocs.controllers import ResourceNotFoundError, ValidationError, PermissionError, BusinessRuleError
from CDocs.utils.document_converter import ControlledDocumentConverter
from CDocs.models.document_status import STATUS_APPROVED, STATUS_PUBLISHED
from CDocs.utils.filecloud_integration import FileCloudIntegration

Conditional/Optional Imports

These imports are only needed under specific conditions:

from CDocs.utils.filecloud_integration import FileCloudIntegration

Condition: Required for FileCloud integration; function _get_filecloud_integration() must be available in scope

Required (conditional)
from CDocs.controllers.document_controller import get_document_audit_trail

Condition: Required for retrieving audit trail data; must be imported or available in scope

Required (conditional)

Usage Example

from datetime import datetime, timedelta
from CDocs.models.user_extensions import DocUser
from CDocs.controllers.document_controller import publish_document

# Get user object (assumes authentication is handled)
user = DocUser(uid='user123', username='john.doe')

# Define publication parameters
document_uid = 'doc-12345-abcde'
effective_date = datetime.now()
expiry_date = datetime.now() + timedelta(days=365)
publish_comment = 'Annual review completed, publishing updated version'

# Publish the document
try:
    result = publish_document(
        user=user,
        document_uid=document_uid,
        effective_date=effective_date,
        expiry_date=expiry_date,
        publish_comment=publish_comment
    )
    
    if result['success']:
        print(f"Document {result['document_number']} published successfully")
        print(f"PDF available at: {result['pdf_path']}")
        print(f"Effective date: {result['effective_date']}")
except ResourceNotFoundError as e:
    print(f"Document not found: {e}")
except BusinessRuleError as e:
    print(f"Cannot publish: {e}")
except PermissionError as e:
    print(f"Permission denied: {e}")

Best Practices

  • Always ensure the document is in APPROVED status before calling this function; the function will raise BusinessRuleError otherwise
  • The function is decorated with @transaction, so database operations are atomic; errors will trigger rollback
  • The function is decorated with @require_permission(['PUBLISH_DOCUMENT'], 'document_uid'), so permission checks are automatic
  • Temporary files are automatically cleaned up in the finally block, but ensure sufficient disk space for document processing
  • The function requires a helper function _get_filecloud_integration() to be available in scope for FileCloud operations
  • The function requires get_document_audit_trail() to be available for retrieving audit data
  • Handle all four exception types: ResourceNotFoundError, ValidationError, PermissionError, and BusinessRuleError
  • The PDF conversion requires LibreOffice or similar tool to be installed and accessible to ControlledDocumentConverter
  • Ensure the signature directory (/tf/active/document_auditor/signatures) exists and contains required signature files
  • The function sends notifications via notifications.notify_document_update(); ensure notification system is configured
  • Audit logging is automatic via the @log_controller_action decorator and explicit audit_trail.log_document_lifecycle_event() call
  • The function modifies both the document status and creates a PDF version; both operations must succeed for transaction to commit
  • If effective_date is not provided, it defaults to datetime.now(); consider timezone implications in production
  • The PDF filename is derived from the editable file's basename; ensure consistent naming conventions

Similar Components

AI-powered semantic similarity - components with related functionality:

  • function archive_document 70.8% similar

    Archives a controlled document by changing its status to ARCHIVED, ensuring a PDF version exists and logging the action with audit trail and notifications.

    From: /tf/active/vicechatdev/document_controller_backup.py
  • function convert_document_to_pdf 68.0% similar

    Converts a document version from an editable format (e.g., Word) to PDF without changing the document's status, uploading the result to FileCloud and updating the version record.

    From: /tf/active/vicechatdev/document_controller_backup.py
  • function delete_document 68.0% similar

    Deletes a controlled document from the system with permission checks, status validation, and audit logging.

    From: /tf/active/vicechatdev/document_controller_backup.py
  • function update_document 67.6% similar

    Updates properties of a controlled document including title, description, status, owner, and metadata, with special handling for status transitions that require format conversions or publishing workflows.

    From: /tf/active/vicechatdev/document_controller_backup.py
  • function get_document_download_url 64.2% similar

    Retrieves a download URL for a controlled document, automatically selecting between editable (Word) and PDF formats based on document status or explicit request.

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