function publish_document
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.
/tf/active/vicechatdev/document_controller_backup.py
895 - 1103
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
logginguuidostempfiletypingdatetimeiopanelshutiltracebackCDocsrandom
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
Tags
Similar Components
AI-powered semantic similarity - components with related functionality:
-
function archive_document 70.8% similar
-
function convert_document_to_pdf 68.0% similar
-
function delete_document 68.0% similar
-
function update_document 67.6% similar
-
function get_document_download_url 64.2% similar