110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Utility functions for automation feature.
|
|
|
|
Moved from interfaces/interfaceDbChat.py.
|
|
"""
|
|
|
|
import json
|
|
from typing import Dict, Any
|
|
from datetime import datetime, UTC
|
|
|
|
|
|
def parseScheduleToCron(schedule: str) -> Dict[str, Any]:
|
|
"""Parse schedule string to cron kwargs for APScheduler"""
|
|
parts = schedule.split()
|
|
if len(parts) != 5:
|
|
raise ValueError(f"Invalid schedule format: {schedule}")
|
|
|
|
return {
|
|
"minute": parts[0],
|
|
"hour": parts[1],
|
|
"day": parts[2],
|
|
"month": parts[3],
|
|
"day_of_week": parts[4]
|
|
}
|
|
|
|
|
|
def planToPrompt(plan: Dict) -> str:
|
|
"""Convert plan structure to prompt string for workflow execution"""
|
|
return plan.get("userMessage", plan.get("overview", "Execute automation workflow"))
|
|
|
|
|
|
def replacePlaceholders(template: str, placeholders: Dict[str, str]) -> str:
|
|
"""Replace placeholders in template with actual values. Placeholder format: {{KEY:PLACEHOLDER_NAME}} or {{TIMESTAMP}}"""
|
|
result = template
|
|
|
|
# Replace TIMESTAMP placeholder first (calculated placeholder, not from parameters)
|
|
timestampPattern = "{{TIMESTAMP}}"
|
|
if timestampPattern in result:
|
|
timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
|
|
result = result.replace(timestampPattern, timestamp)
|
|
|
|
for placeholderName, value in placeholders.items():
|
|
pattern = f"{{{{KEY:{placeholderName}}}}}"
|
|
|
|
# Check if placeholder is in an array context like ["{{KEY:...}}"]
|
|
# If value is a JSON array/dict, we should replace the entire ["{{KEY:...}}"] with the array
|
|
arrayPattern = f'["{pattern}"]'
|
|
if arrayPattern in result:
|
|
# Check if value is a JSON array/dict
|
|
isArrayValue = False
|
|
arrayValue = None
|
|
|
|
if isinstance(value, (list, dict)):
|
|
isArrayValue = True
|
|
arrayValue = json.dumps(value)
|
|
elif isinstance(value, str):
|
|
try:
|
|
parsed = json.loads(value)
|
|
if isinstance(parsed, (list, dict)):
|
|
isArrayValue = True
|
|
arrayValue = value # Already valid JSON string
|
|
except (json.JSONDecodeError, ValueError):
|
|
pass
|
|
|
|
if isArrayValue:
|
|
# Replace ["{{KEY:...}}"] with the array value
|
|
result = result.replace(arrayPattern, arrayValue)
|
|
continue # Skip the regular replacement below
|
|
|
|
# Replace occurrences one-by-one to handle mixed contexts
|
|
while pattern in result:
|
|
patternStart = result.find(pattern)
|
|
isQuoted = False
|
|
if patternStart > 0:
|
|
charBefore = result[patternStart - 1]
|
|
patternEnd = patternStart + len(pattern)
|
|
charAfter = result[patternEnd] if patternEnd < len(result) else None
|
|
if charBefore == '"' and charAfter == '"':
|
|
isQuoted = True
|
|
|
|
if isinstance(value, (list, dict)):
|
|
replacement = json.dumps(value)
|
|
elif isinstance(value, str):
|
|
try:
|
|
parsed = json.loads(value)
|
|
if isinstance(parsed, (list, dict)):
|
|
if isQuoted:
|
|
escaped = json.dumps(value)
|
|
replacement = escaped[1:-1]
|
|
else:
|
|
replacement = value
|
|
else:
|
|
if isQuoted:
|
|
escaped = json.dumps(value)
|
|
replacement = escaped[1:-1]
|
|
else:
|
|
replacement = value
|
|
except (json.JSONDecodeError, ValueError):
|
|
if isQuoted:
|
|
escaped = json.dumps(value)
|
|
replacement = escaped[1:-1]
|
|
else:
|
|
replacement = value
|
|
else:
|
|
replacement = str(value)
|
|
result = result[:patternStart] + replacement + result[patternStart + len(pattern):]
|
|
return result
|
|
|