133 lines
5.7 KiB
Python
133 lines
5.7 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
|
|
import logging
|
|
import time
|
|
from typing import Dict, Any, Optional, List
|
|
from modules.datamodels.datamodelChat import ActionResult, ActionDocument
|
|
from modules.datamodels.datamodelExtraction import ContentPart
|
|
from modules.datamodels.datamodelAi import AiCallOptions, OperationTypeEnum, PriorityEnum, ProcessingModeEnum
|
|
from modules.datamodels.datamodelWorkflow import AiResponse, DocumentData
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
async def generateDocument(self, parameters: Dict[str, Any]) -> ActionResult:
|
|
prompt = parameters.get("prompt")
|
|
if not prompt:
|
|
return ActionResult.isFailure(error="prompt is required")
|
|
|
|
documentList = parameters.get("documentList", [])
|
|
documentType = parameters.get("documentType")
|
|
resultType = parameters.get("resultType", "txt")
|
|
|
|
# Auto-detect format from prompt if not explicitly provided
|
|
if resultType == "txt" and prompt:
|
|
promptLower = prompt.lower()
|
|
if "html" in promptLower or "html5" in promptLower:
|
|
resultType = "html"
|
|
logger.info(f"Auto-detected HTML format from prompt")
|
|
elif "pdf" in promptLower:
|
|
resultType = "pdf"
|
|
logger.info(f"Auto-detected PDF format from prompt")
|
|
elif "markdown" in promptLower or " md " in promptLower or promptLower.endswith(" md"):
|
|
resultType = "md"
|
|
logger.info(f"Auto-detected Markdown format from prompt")
|
|
elif ("text" in promptLower or "txt" in promptLower) and "html" not in promptLower:
|
|
resultType = "txt"
|
|
logger.info(f"Auto-detected Text format from prompt")
|
|
|
|
# Create operation ID for progress tracking
|
|
workflowId = self.services.workflow.id if self.services.workflow else f"no-workflow-{int(time.time())}"
|
|
operationId = f"doc_gen_{workflowId}_{int(time.time())}"
|
|
parentOperationId = parameters.get('parentOperationId')
|
|
|
|
try:
|
|
# Convert documentList to DocumentReferenceList if needed
|
|
docRefList = None
|
|
if documentList:
|
|
from modules.datamodels.datamodelDocref import DocumentReferenceList
|
|
|
|
if isinstance(documentList, DocumentReferenceList):
|
|
docRefList = documentList
|
|
elif isinstance(documentList, str):
|
|
docRefList = DocumentReferenceList.from_string_list([documentList])
|
|
elif isinstance(documentList, list):
|
|
docRefList = DocumentReferenceList.from_string_list(documentList)
|
|
else:
|
|
docRefList = DocumentReferenceList(references=[])
|
|
|
|
# Prepare title
|
|
title = parameters.get("documentType") or "Generated Document"
|
|
|
|
# Call AI service for document generation
|
|
# callAiContent handles documentList internally via Phases 5A-5E
|
|
options = AiCallOptions(
|
|
operationType=OperationTypeEnum.DATA_GENERATE,
|
|
priority=PriorityEnum.BALANCED,
|
|
processingMode=ProcessingModeEnum.DETAILED,
|
|
compressPrompt=False,
|
|
compressContext=False
|
|
)
|
|
|
|
aiResponse: AiResponse = await self.services.ai.callAiContent(
|
|
prompt=prompt,
|
|
options=options,
|
|
documentList=docRefList, # Übergebe documentList direkt - callAiContent macht Phasen 5A-5E
|
|
outputFormat=resultType,
|
|
title=title,
|
|
parentOperationId=parentOperationId,
|
|
generationIntent="document" # NEW: Explicit intent, skips detection
|
|
)
|
|
|
|
# Convert AiResponse to ActionResult
|
|
documents = []
|
|
|
|
# Convert DocumentData to ActionDocument
|
|
if aiResponse.documents:
|
|
for docData in aiResponse.documents:
|
|
documents.append(ActionDocument(
|
|
documentName=docData.documentName,
|
|
documentData=docData.documentData,
|
|
mimeType=docData.mimeType,
|
|
sourceJson=docData.sourceJson if hasattr(docData, 'sourceJson') else None
|
|
))
|
|
|
|
# If no documents but content exists, create a document from content
|
|
if not documents and aiResponse.content:
|
|
# Determine document name from metadata
|
|
docName = f"document.{resultType}"
|
|
if aiResponse.metadata and aiResponse.metadata.filename:
|
|
docName = aiResponse.metadata.filename
|
|
elif aiResponse.metadata and aiResponse.metadata.title:
|
|
import re
|
|
sanitized = re.sub(r"[^a-zA-Z0-9._-]", "_", aiResponse.metadata.title)
|
|
sanitized = re.sub(r"_+", "_", sanitized).strip("_")
|
|
if sanitized:
|
|
if not sanitized.lower().endswith(f".{resultType}"):
|
|
docName = f"{sanitized}.{resultType}"
|
|
else:
|
|
docName = sanitized
|
|
|
|
# Determine mime type
|
|
mimeType = "text/plain"
|
|
if resultType == "html":
|
|
mimeType = "text/html"
|
|
elif resultType == "json":
|
|
mimeType = "application/json"
|
|
elif resultType == "pdf":
|
|
mimeType = "application/pdf"
|
|
elif resultType == "md":
|
|
mimeType = "text/markdown"
|
|
|
|
documents.append(ActionDocument(
|
|
documentName=docName,
|
|
documentData=aiResponse.content.encode('utf-8') if isinstance(aiResponse.content, str) else aiResponse.content,
|
|
mimeType=mimeType
|
|
))
|
|
|
|
return ActionResult.isSuccess(documents=documents)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in document generation: {str(e)}")
|
|
return ActionResult.isFailure(error=str(e))
|
|
|