94 lines
3.8 KiB
Python
94 lines
3.8 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
|
|
"""
|
|
Get Document Index action for Context operations.
|
|
Generates a comprehensive index of all documents available in the current workflow.
|
|
"""
|
|
|
|
import logging
|
|
import json
|
|
from typing import Dict, Any
|
|
from modules.workflows.methods.methodBase import action
|
|
from modules.datamodels.datamodelChat import ActionResult, ActionDocument
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@action
|
|
async def getDocumentIndex(self, parameters: Dict[str, Any]) -> ActionResult:
|
|
"""
|
|
GENERAL:
|
|
- Purpose: Generate a comprehensive index of all documents available in the current workflow, including documents from all rounds and tasks.
|
|
- Input requirements: No input documents required. Optional resultType parameter.
|
|
- Output format: Structured document index in JSON format (default) or text format, listing all documents with their references, metadata, and organization by rounds/tasks.
|
|
|
|
Parameters:
|
|
- resultType (str, optional): Output format (json, txt, md). Default: json.
|
|
"""
|
|
try:
|
|
workflow = self.services.workflow
|
|
if not workflow:
|
|
return ActionResult.isFailure(
|
|
error="No workflow available"
|
|
)
|
|
|
|
resultType = parameters.get("resultType", "json").lower().strip().lstrip('.')
|
|
|
|
# Get available documents index from chat service
|
|
documentsIndex = self.services.chat.getAvailableDocuments(workflow)
|
|
|
|
if not documentsIndex or documentsIndex == "No documents available" or documentsIndex == "NO DOCUMENTS AVAILABLE - This workflow has no documents to process.":
|
|
# Return empty index structure
|
|
if resultType == "json":
|
|
indexData = {
|
|
"workflowId": getattr(workflow, 'id', 'unknown'),
|
|
"totalDocuments": 0,
|
|
"rounds": [],
|
|
"documentReferences": []
|
|
}
|
|
indexContent = json.dumps(indexData, indent=2, ensure_ascii=False)
|
|
else:
|
|
indexContent = "Document Index\n==============\n\nNo documents available in this workflow.\n"
|
|
else:
|
|
# Parse the document index string to extract structured information
|
|
indexData = self.documentIndex.parseDocumentIndex(documentsIndex, workflow)
|
|
|
|
if resultType == "json":
|
|
indexContent = json.dumps(indexData, indent=2, ensure_ascii=False)
|
|
elif resultType == "md":
|
|
indexContent = self.formatting.formatAsMarkdown(indexData)
|
|
else: # txt
|
|
indexContent = self.formatting.formatAsText(indexData, documentsIndex)
|
|
|
|
# Generate meaningful filename
|
|
workflowContext = self.services.chat.getWorkflowContext()
|
|
filename = self._generateMeaningfulFileName(
|
|
"document_index",
|
|
resultType if resultType in ["json", "txt", "md"] else "json",
|
|
workflowContext,
|
|
"getDocumentIndex"
|
|
)
|
|
|
|
validationMetadata = {
|
|
"actionType": "context.getDocumentIndex",
|
|
"resultType": resultType,
|
|
"workflowId": getattr(workflow, 'id', 'unknown'),
|
|
"totalDocuments": indexData.get("totalDocuments", 0) if isinstance(indexData, dict) else 0
|
|
}
|
|
|
|
# Create ActionDocument
|
|
document = ActionDocument(
|
|
documentName=filename,
|
|
documentData=indexContent,
|
|
mimeType="application/json" if resultType == "json" else "text/plain",
|
|
validationMetadata=validationMetadata
|
|
)
|
|
|
|
return ActionResult.isSuccess(documents=[document])
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error generating document index: {str(e)}")
|
|
return ActionResult.isFailure(
|
|
error=f"Failed to generate document index: {str(e)}"
|
|
)
|
|
|