Aktionen und Filter

"""
title: Personal Information Filter
version: 1.0.0
license: MIT
"""

import re
from pydantic import BaseModel, Field
from typing import Optional


class Filter:
    class Valves(BaseModel):
        priority: int = Field(default=0, description="Prioritätslevel für Filter.")
        redact_email: bool = Field(default=True, description="Email-Adressen entfernen")
        redact_ssn: bool = Field(default=True, description="SSNs entfernen")
        redact_credit_card: bool = Field(
            default=True, description="Kreditkartennummern entfernen"
        )
        redact_ip_address: bool = Field(
            default=True, description="IP Adressen entfernen"
        )
        enabled_for_admins: bool = Field(
            default=False,
            description="PI-Filter auch für Admins aktivieren",
        )

    def __init__(self):
        self.file_handler = False
        self.valves = self.Valves()
        self.patterns = {
            "email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"),
            "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
            "credit_card": re.compile(r"\b(?:\d{4}[-\s]?){3}\d{4}\b"),
            "ip_address": re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"),
        }

    def redact_pii(self, text: str) -> str:
        if self.valves.redact_email:
            text = self.patterns["email"].sub("[EMAIL VERSTECKT]", text)
        if self.valves.redact_ssn:
            text = self.patterns["ssn"].sub("[SSN VERSTECKT]", text)
        if self.valves.redact_credit_card:
            text = self.patterns["credit_card"].sub("[KREDITKARTE VERSTECKT]", text)
        if self.valves.redact_ip_address:
            text = self.patterns["ip_address"].sub("[IP ADRESSE VERSTECKT]", text)
        return text

    def inlet(self, body: dict, __user__: Optional[dict] = None) -> dict:
        print(f"inlet:{__name__}")
        print(f"inlet:body:{body}")
        print(f"inlet:user:{__user__}")

        if (
            __user__ is None
            or not __user__.get("role") == "admin"
            or self.valves.enabled_for_admins
        ):
            messages = body.get("messages", [])
            for message in messages:
                if message.get("role") == "user":
                    message["content"] = self.redact_pii(message["content"])

        return body

    def outlet(self, body: dict, __user__: Optional[dict] = None) -> dict:
        print(f"outlet:{__name__}")
        print(f"outlet:body:{body}")
        print(f"outlet:user:{__user__}")

        return body

"""
title: Text als PDF speichern
version: 1.0.0
license: MIT
"""

import requests
from typing import Optional
from pydantic import BaseModel, Field


class Action:
    class Valves(BaseModel):
        API_BASE_URL: str = Field(
            default="https://skills-api.localmind.io",
            description="Base URL for the PDF generation API.",
        )
        API_KEY: str = Field(
            default="ein_api_key_hier_zu_platzieren",
            description="API key for authentication with the PDF generation service.",
        )

    def __init__(self):
        self.type = "action"
        self.name = "Letzte Assistentennachricht als PDF speichern"
        self.description = "Speichert die letzte Assistentennachricht in der Konversation als PDF-Datei."
        self.valves = self.Valves()

    async def action(
        self, body: dict, __user__=None, __event_emitter__=None, __event_call__=None
    ) -> Optional[dict]:
        if not __event_emitter__:
            return {"error": "Event emitter is not available."}

        await __event_emitter__(
            {
                "type": "status",
                "data": {
                    "description": "PDF wird vorbereitet...",
                    "done": False,
                },
            }
        )

        messages = body.get("messages", [])
        last_assistant_message = next(
            (
                msg["content"]
                for msg in reversed(messages)
                if msg["role"] == "assistant"
            ),
            None,
        )

        if not last_assistant_message:
            await __event_emitter__(
                {
                    "type": "status",
                    "data": {
                        "description": "Keine Assistentennachricht gefunden.",
                        "done": True,
                    },
                }
            )
            return {
                "error": "Keine Assistentennachricht gefunden, um ein PDF zu erstellen."
            }

        sanitized_text = last_assistant_message.replace("\x00", "").strip()

        url = f"{self.valves.API_BASE_URL}/tools/save-as-pdf"
        headers = {
            "accept": "application/json",
            "X-API-Key": self.valves.API_KEY,
            "Content-Type": "application/json",
        }
        data = {"text": sanitized_text}

        try:
            await __event_emitter__(
                {
                    "type": "status",
                    "data": {
                        "description": "PDF-Generierungsanfrage wird übermittelt...",
                        "done": False,
                    },
                }
            )

            response = requests.post(url, headers=headers, json=data)
            response.raise_for_status()

            await __event_emitter__(
                {
                    "type": "status",
                    "data": {
                        "description": "API-Antwort wird verarbeitet...",
                        "done": False,
                    },
                }
            )

            result = response.json()
            if "document_url" in result:
                document_url = result["document_url"]
                await __event_emitter__(
                    {
                        "type": "status",
                        "data": {
                            "description": "PDF erfolgreich erstellt.",
                            "done": True,
                        },
                    }
                )
                await __event_emitter__(
                    {
                        "type": "message",
                        "data": {
                            "content": f"\n\n[PDF-Datei downloaden]({document_url})"
                        },
                    }
                )
                return {"success": True, "document_url": document_url}
            else:
                raise ValueError("Die PDF-Generierung hat nicht geklappt.")

        except requests.RequestException as e:
            error_message = f"Fehler bei der Kommunikation mit der PDF-API: {str(e)}"
        except ValueError as e:
            error_message = str(e)
        except Exception as e:
            error_message = f"Unerwarteter Fehler: {str(e)}"

        await __event_emitter__(
            {
                "type": "status",
                "data": {"description": f"Fehler: {error_message}", "done": True},
            }
        )
        return {"error": f"Fehler beim Generieren des PDFs: {error_message}"}