# Copyright (c) 2025 Patrick Motsch # All rights reserved. import logging from datetime import datetime, UTC from modules.workflows.methods.methodBase import MethodBase from modules.datamodels.datamodelWorkflowActions import WorkflowActionDefinition, WorkflowActionParameter from modules.shared.frontendTypes import FrontendType # Import helpers from .helpers.csvProcessing import CsvProcessingHelper # Import actions from .actions.process import process from .actions.webResearch import webResearch from .actions.summarizeDocument import summarizeDocument from .actions.translateDocument import translateDocument from .actions.convertDocument import convertDocument from .actions.generateDocument import generateDocument from .actions.generateCode import generateCode logger = logging.getLogger(__name__) class MethodAi(MethodBase): """AI processing methods.""" def __init__(self, services): super().__init__(services) self.name = "ai" self.description = "AI processing methods" # Initialize helper modules self.csvProcessing = CsvProcessingHelper(self) # RBAC-Integration: Action-Definitionen mit actionId self._actions = { "process": WorkflowActionDefinition( actionId="ai.process", description="Universal AI document processing action - accepts multiple input documents in any format and processes them together with a prompt. If the prompt specifies document formats to deliver, include them in the prompt", dynamicMode=True, parameters={ "aiPrompt": WorkflowActionParameter( name="aiPrompt", type="str", frontendType=FrontendType.TEXTAREA, required=True, description="Instruction for the AI describing what processing to perform" ), "documentList": WorkflowActionParameter( name="documentList", type="List[str]", frontendType=FrontendType.DOCUMENT_REFERENCE, required=False, description="Document reference(s) in any format to use as input/context" ), "resultType": WorkflowActionParameter( name="resultType", type="str", frontendType=FrontendType.SELECT, frontendOptions=["txt", "json", "md", "csv", "xml", "html", "pdf", "docx", "xlsx", "pptx", "png", "jpg"], required=False, default="txt", description="Output file extension. All output documents will use this format" ), "simpleMode": WorkflowActionParameter( name="simpleMode", type="bool", frontendType=FrontendType.CHECKBOX, required=False, default=False, description="If true, uses fast simple AI call without document generation pipeline. Use for chatbot responses and simple text generation." ) }, execute=process.__get__(self, self.__class__) ), "webResearch": WorkflowActionDefinition( actionId="ai.webResearch", description="Web research with two-step process: search for URLs, then crawl content", dynamicMode=True, parameters={ "prompt": WorkflowActionParameter( name="prompt", type="str", frontendType=FrontendType.TEXTAREA, required=True, description="Natural language research instruction" ), "urlList": WorkflowActionParameter( name="urlList", type="List[str]", frontendType=FrontendType.MULTISELECT, required=False, description="Specific URLs to crawl, if needed" ), "country": WorkflowActionParameter( name="country", type="str", frontendType=FrontendType.TEXT, required=False, description="Two-digit country code (lowercase, e.g., ch, us, de)" ), "language": WorkflowActionParameter( name="language", type="str", frontendType=FrontendType.SELECT, frontendOptions=["de", "en", "fr", "it", "es"], required=False, description="Language code (lowercase, e.g., de, en, fr)" ), "researchDepth": WorkflowActionParameter( name="researchDepth", type="str", frontendType=FrontendType.SELECT, frontendOptions=["fast", "general", "deep"], required=False, default="general", description="Research depth" ) }, execute=webResearch.__get__(self, self.__class__) ), "summarizeDocument": WorkflowActionDefinition( actionId="ai.summarizeDocument", description="Summarize one or more documents, extracting key points and main ideas. If the prompt specifies document formats to deliver, include them in the prompt", dynamicMode=True, parameters={ "documentList": WorkflowActionParameter( name="documentList", type="List[str]", frontendType=FrontendType.DOCUMENT_REFERENCE, required=True, description="Document reference(s) to summarize" ), "summaryLength": WorkflowActionParameter( name="summaryLength", type="str", frontendType=FrontendType.SELECT, frontendOptions=["brief", "medium", "detailed"], required=False, default="medium", description="Desired summary length" ), "focus": WorkflowActionParameter( name="focus", type="str", frontendType=FrontendType.TEXT, required=False, description="Specific aspect to focus on in the summary (e.g., financial data, key decisions)" ), "resultType": WorkflowActionParameter( name="resultType", type="str", frontendType=FrontendType.SELECT, frontendOptions=["txt", "md", "docx"], required=False, default="txt", description="Output file extension" ) }, execute=summarizeDocument.__get__(self, self.__class__) ), "translateDocument": WorkflowActionDefinition( actionId="ai.translateDocument", description="Translate documents to a target language while preserving formatting and structure", dynamicMode=True, parameters={ "documentList": WorkflowActionParameter( name="documentList", type="List[str]", frontendType=FrontendType.DOCUMENT_REFERENCE, required=True, description="Document reference(s) to translate" ), "targetLanguage": WorkflowActionParameter( name="targetLanguage", type="str", frontendType=FrontendType.TEXT, required=True, description="Target language code or name (e.g., de, German, French, es)" ), "sourceLanguage": WorkflowActionParameter( name="sourceLanguage", type="str", frontendType=FrontendType.TEXT, required=False, description="Source language if known (e.g., en, English). If not provided, AI will detect" ), "preserveFormatting": WorkflowActionParameter( name="preserveFormatting", type="bool", frontendType=FrontendType.CHECKBOX, required=False, default=True, description="Whether to preserve original formatting" ), "resultType": WorkflowActionParameter( name="resultType", type="str", frontendType=FrontendType.TEXT, required=False, description="Output file extension. If not specified, uses same format as input" ) }, execute=translateDocument.__get__(self, self.__class__) ), "convertDocument": WorkflowActionDefinition( actionId="ai.convertDocument", description="Convert documents between different formats (PDF→Word, Excel→CSV, etc.)", dynamicMode=True, parameters={ "documentList": WorkflowActionParameter( name="documentList", type="List[str]", frontendType=FrontendType.DOCUMENT_REFERENCE, required=True, description="Document reference(s) to convert" ), "targetFormat": WorkflowActionParameter( name="targetFormat", type="str", frontendType=FrontendType.SELECT, frontendOptions=["docx", "pdf", "xlsx", "csv", "txt", "html", "json", "md"], required=True, description="Target format extension" ), "preserveStructure": WorkflowActionParameter( name="preserveStructure", type="bool", frontendType=FrontendType.CHECKBOX, required=False, default=True, description="Whether to preserve document structure (headings, tables, etc.)" ) }, execute=convertDocument.__get__(self, self.__class__) ), "generateDocument": WorkflowActionDefinition( actionId="ai.generateDocument", description="Generate documents from scratch or based on templates/inputs. If the prompt specifies document formats to deliver, include them in the prompt", dynamicMode=True, parameters={ "prompt": WorkflowActionParameter( name="prompt", type="str", frontendType=FrontendType.TEXTAREA, required=True, description="Description of the document to generate" ), "documentList": WorkflowActionParameter( name="documentList", type="List[str]", frontendType=FrontendType.DOCUMENT_REFERENCE, required=False, description="Template documents or reference documents to use as a guide" ), "documentType": WorkflowActionParameter( name="documentType", type="str", frontendType=FrontendType.SELECT, frontendOptions=["letter", "memo", "proposal", "contract", "report", "email"], required=False, description="Type of document" ), "resultType": WorkflowActionParameter( name="resultType", type="str", frontendType=FrontendType.TEXT, required=False, default="txt", description="Output format (e.g., txt, html, pdf, docx, md, json, csv, xlsx, pptx, png, jpg). Optional: if omitted, formats are determined from prompt by AI. Default \"txt\" is validation fallback only. With per-document format determination, AI can determine different formats for different documents based on prompt." ) }, execute=generateDocument.__get__(self, self.__class__) ), "generateCode": WorkflowActionDefinition( actionId="ai.generateCode", description="Generate code files - explicitly sets intent to 'code'. If the prompt specifies file formats to deliver, include them in the prompt", dynamicMode=True, parameters={ "prompt": WorkflowActionParameter( name="prompt", type="str", frontendType=FrontendType.TEXTAREA, required=True, description="Description of code to generate" ), "documentList": WorkflowActionParameter( name="documentList", type="List[str]", frontendType=FrontendType.DOCUMENT_REFERENCE, required=False, description="Reference documents" ), "resultType": WorkflowActionParameter( name="resultType", type="str", frontendType=FrontendType.SELECT, frontendOptions=["py", "js", "ts", "html", "java", "cpp", "txt"], required=False, description="Output format (html, js, py, etc.). Optional: if omitted, formats are determined from prompt by AI. With per-document format determination, AI can determine different formats for different documents based on prompt." ) }, execute=generateCode.__get__(self, self.__class__) ) } # Validate actions after definition self._validateActions() # Register actions as methods (optional, für direkten Zugriff) self.process = process.__get__(self, self.__class__) self.webResearch = webResearch.__get__(self, self.__class__) self.summarizeDocument = summarizeDocument.__get__(self, self.__class__) self.translateDocument = translateDocument.__get__(self, self.__class__) self.convertDocument = convertDocument.__get__(self, self.__class__) self.generateDocument = generateDocument.__get__(self, self.__class__) self.generateCode = generateCode.__get__(self, self.__class__) def _format_timestamp_for_filename(self) -> str: """Format current timestamp as YYYYMMDD-hhmmss for filenames.""" return datetime.now(UTC).strftime("%Y%m%d-%H%M%S")