🔍 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 confirmation 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 the 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'

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"Status changed from {result['previous_status']} to {result['new_status']}")
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 PDF generation
  • The function requires a helper function _get_filecloud_integration() to be available in the same module scope
  • The function requires get_document_audit_trail() to be available for retrieving audit data
  • Signature directory must exist and contain valid signature files for PDF generation to succeed
  • FileCloud credentials must be properly configured in settings before calling this function
  • The function sends notifications automatically; ensure notification system is configured to avoid errors
  • Effective date defaults to current time if not provided; explicitly set it for scheduled publications
  • The function logs all actions to audit trail; ensure audit_trail module is properly configured
  • Handle all four exception types (ResourceNotFoundError, ValidationError, PermissionError, BusinessRuleError) in calling code
  • The PDF path returned is the FileCloud remote path, not a local filesystem path
  • Document status transitions are validated; only APPROVED documents can be published

Similar Components

AI-powered semantic similarity - components with related functionality:

  • function publish_document_v1 92.5% similar

    Publishes a controlled document by incrementing its version to the next major revision, converting it to a signed PDF, and updating its status to PUBLISHED.

    From: /tf/active/vicechatdev/CDocs/controllers/document_controller.py
  • function archive_document 72.5% similar

    Archives a controlled document by changing its status and all versions to ARCHIVED, moving the published PDF to an archived location with metadata, and logging the archival event.

    From: /tf/active/vicechatdev/CDocs/controllers/document_controller.py
  • function archive_document_v1 70.7% 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 upload_document_to_filecloud 69.3% similar

    Uploads a document version to FileCloud storage system with metadata, handling file creation, folder structure, and audit logging.

    From: /tf/active/vicechatdev/CDocs/controllers/filecloud_controller.py
  • function update_document_v1 68.5% similar

    Updates a controlled document's properties (title, description, status, owner, metadata) with validation, audit logging, and special handling for status transitions that require PDF publishing or training resets.

    From: /tf/active/vicechatdev/CDocs/controllers/document_controller.py
← Back to Browse