🔍 Code Extractor

Publishes a controlled document by converting it to PDF with signatures and audit trail, updating its status to PUBLISHED, and storing it in FileCloud.

Purpose

This function manages the complete document publishing workflow in a controlled document management system. It validates the document is approved, downloads the editable version from FileCloud, converts it to a signed PDF with audit trail, uploads the PDF back to FileCloud, updates the document status and metadata in the database, and sends notifications. 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 and not permissions.user_has_permission(user, "FORCE_PUBLISH_DOCUMENT"):
            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

Parameter Details

user: DocUser object representing the user performing the publish action. Must have PUBLISH_DOCUMENT permission (or FORCE_PUBLISH_DOCUMENT to bypass approval checks). Used for audit logging and metadata.

document_uid: String containing the unique identifier (UID) of the document to be published. Must correspond to an existing ControlledDocument 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 publishing action. Stored in metadata for audit purposes. Can be None.

Return Value

Type: Dict[str, Any]

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

Dependencies

• logging
• uuid
• os
• tempfile
• typing
• datetime
• io
• panel
• shutil
• traceback
• json
• re
• CDocs
• document_auditor

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, permissions
from CDocs.models.document import ControlledDocument
from CDocs.models.user_extensions import DocUser
from CDocs.utils import notifications, audit_trail
from CDocs.controllers import require_permission, log_controller_action, transaction
from CDocs.controllers import PermissionError, ResourceNotFoundError, ValidationError, 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.controllers.filecloud_controller import get_filecloud_client

from document_auditor.src.document_processor import DocumentProcessor

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 performing the action
user = DocUser(uid='user123')

# Set effective and expiry dates
effective_date = datetime.now()
expiry_date = datetime.now() + timedelta(days=365)

# Publish the document
try:
    result = publish_document(
        user=user,
        document_uid='doc-abc-123',
        effective_date=effective_date,
        expiry_date=expiry_date,
        publish_comment='Publishing version 2.0 after final approval'
    )
    
    print(f"Success: {result['message']}")
    print(f"PDF Path: {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 wrap calls in try-except blocks to handle ResourceNotFoundError, BusinessRuleError, ValidationError, and PermissionError
• Ensure the document is in STATUS_APPROVED before calling, unless user has FORCE_PUBLISH_DOCUMENT permission
• Verify the document has a current version with an editable file (word_file_path) before attempting to publish
• The function is decorated with @transaction, so database changes will be rolled back on error
• 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 conversion
• The signature directory (/tf/active/document_auditor/signatures) must exist and be readable
• FileCloud credentials must be properly configured before calling this function
• The function sends notifications automatically; ensure notification system is configured to avoid silent failures
• Audit trail events are logged automatically for compliance tracking
• Consider the time required for PDF conversion and FileCloud upload when calling this function in user-facing workflows
• The effective_date defaults to current time if not provided; explicitly set it if backdating or future-dating is needed

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function publish_document 98.3% similar

  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.

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

• function publish_document_v3 94.6% 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 publish_document_v2 92.5% similar

  Publishes a controlled document by changing its status to PUBLISHED or EFFECTIVE based on the effective date, with audit logging and notifications.

  From: /tf/active/vicechatdev/CDocs copy/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