trauma/api/message/utils.py

import json
import re

from trauma.api.chat.dto import EntityData
from trauma.api.data.dto import AgeGroup
from trauma.api.data.model import EntityModel
from trauma.api.message.dto import Author
from trauma.api.message.model import MessageModel
from trauma.core.config import settings

def transform_messages_to_openai(messages: list[MessageModel]) -> list[dict]:
    openai_messages = []
    for message in messages:
        content = message.text
        openai_messages.append({
            "role": message.author.value,
            "content": content
        })

    return openai_messages


def retrieve_empty_field_from_entity_data(entity_data: dict) -> str | None:
    for k, v in entity_data.items():
        if k != 'postalCode' and not v:
            return k
    return None

def prepare_message_history_str(messages: list[MessageModel], user_message: str) -> str:
    results = ""
    shorted_message_history = messages[-8:]
    for message in shorted_message_history:
        results += f"[{message.author.value}] {message.text}\n"
    results += f"[User] {user_message}"
    return results


def prepare_user_messages_str(user_message: str, messages: list[MessageModel]) -> str:
    user_message_str = ''
    for message in messages:
        if message.author == Author.User:
            user_message_str += f"- {message.text}\n"
    user_message_str += f'- {user_message}'
    return user_message_str


def prepare_final_entities_str(entities: list[EntityModel]) -> str:
    shortened_entities = entities[:3]
    entities_list = []
    for entity in shortened_entities:
        entities_list.append(entity.model_dump(mode='json', exclude={
            'id', 'contactDetails', "highlightedAgeGroup", "highlightedTreatmentArea", "highlightedTreatmentMethod"
        }))
    return json.dumps({"klinieken": entities_list})


def pick_empty_field_instructions(empty_field: str) -> str:
    if empty_field == "age":
        return "De leeftijd van het kind of de jongere waarvoor hulp wordt gezocht, uitgedrukt in jaren (bijvoorbeeld 4, 10 of 15 jaar)."
    elif empty_field == "treatmentMethod":
        return "De behandelmethode die wordt ingezet of gezocht bij het kind of de jongere, zoals EMDR, speltherapie, cognitieve gedragstherapie of de Slapende Honden-methode."
    elif empty_field == "location":
        return "De gewenste locatie voor de behandeling, bijvoorbeeld een stad, wijk, of regio waar de zorgverlener zich bij voorkeur bevindt."
    elif empty_field == "treatmentArea":
        return "Het type trauma of onderliggende problematiek waarmee het kind of de jongere te maken heeft, zoals hechtingsproblemen, verlieservaringen, mishandeling, seksueel misbruik of verwaarlozing."
    return None


def find_matching_age_group(entity: EntityModel, entity_data: dict) -> AgeGroup:
    age_groups = entity.ageGroups
    best_match = None
    for age_group in age_groups:
        if age_group.ageMin <= entity_data['age'] <= age_group.ageMax:
            return age_group
        if age_group.ageMax < entity_data['age']:
            if best_match is None or age_group.ageMax > best_match.ageMax:
                best_match = age_group
    return best_match

def decode_treatment_letters(user_message: str) -> str:
    replacements = {
        "lvb": "Licht Verstandelijke Beperking",
        "emdr": "Eye Movement Desensitization and Reprocessing",
        "ptss": "Posttraumatische Stress Stoornis",
        "cgt": "Cognitieve Gedragstherapie",
        "ass": "Autisme Spectrum Stoornis",
        "adhd": "Attention Deficit Hyperactivity Disorder",
        "add": "Attention Deficit Disorder",
        "bps": "Borderline Persoonlijkheidsstoornis",
        "gad": "Gegeneraliseerde Angststoornis",
        "ocd": "Obsessieve-Compulsieve Stoornis",
        "an": "Anorexia Nervosa",
        "bed": "Eetbuistoornis",
        "solk": "Somatisch Onvoldoende verklaarde Lichamelijke Klachten",
        "dis": "Dissociatieve Identiteitsstoornis",
        "nah": "Niet Aangeboren Hersenletsel",
        "sas": "Stemmings-, angst- en somatoforme stoornissen",
        "epa": "Ernstige Psychiatrische Aandoeningen",
        "net": "Narratieve Exposure Therapie",
        "bepp": "Beknopte Eclectische Psychotherapie voor PTSS",
        "dbt": "Dialectische Gedragstherapie",
        "act": "Acceptance and Commitment Therapy",
        "iht": "Intensive Home Treatment"
    }
    for short, full in replacements.items():
        user_message = re.sub(rf'\b{short}\b', full, user_message, flags=re.IGNORECASE)
    return user_message


def search_changed_field_inst(entity_data: dict, old_entity_data: EntityData) -> dict[str, str]:
    changed_fields = {}
    instruction_map = {
        "age": {'ageMinSupported': 0, "ageMaxSupported": 23},
        "treatmentMethod": [
    "Hypnose",
    "Hypnotherapie",
    "eclectisch",
    "Schematherapie",
    "BEPP (Beknopte Eclectische Psychotherapie bij PTSS",
    "Muziektherapie, klankbehandeling",
    "FITT, family based intensive trauma treatment",
    "systeemtherapie & schematherapie",
    "Narratieve Exposure Therapie (NET)",
    "Ouder-kindtraumatherapie",
    "Lichaamsgerichte therapie",
    "integratieve therapie",
    "Havening, hypnose",
    "Psychotherapie",
    "TIST",
    "Words and pictures",
    "Integratieve Kindertherapie",
    "Buitenpsychologie",
    "systeemtherapie gericht op scheidingsproblematiek",
    "NET",
    "Eigen methodiek, SSP, Focustherapie",
    "Systeemtherapie",
    "BEPP (beknopte eclectische psychotherapie voor PTSS), BEP-TG (beknopte eclectische psychotherapie voor traumatische rouw), NET (narratieve exposure therapie), schematherapie",
    "Narratieve Exposure therapie",
    "Writejunior",
    "Tem Je Draak",
    "EMDR voor ouders",
    "Methodes vanuit de systeemtherapie",
    "CGT (oa exposure)",
    "Horizonmethodiek",
    "TF-CBT",
    "BEPP protocol",
    "Speltherapie",
    "ouder-kind therapie (vanuit de IMH visie)",
    "IGT-K",
    "medicamenteus",
    "Verwerken en versterken",
    "Pre-verbale EMDR",
    "Kortdurende Intensieve traumabehandeling",
    "aantal bovenstaande nog in ontwikkeling, deze niet aangevinkt nu",
    "Drakentemmers",
    "BEPP",
    "schematherapie",
    "Hypno-regressie therapie",
    "ImRS",
    "Beeldende therapie",
    "sensorimotor psychotherapie",
    "Cref-methode /paarde coaching",
    "KINGS",
    "KINGS in voorbereiding",
    "systeemtherapie",
    "EMDR",
    "Ouder Kind Trauma Therapie",
    "BEPP, NET",
    "IEMT",
    "Bokstherapie",
    "babytherapie",
    "sensorimotorpsychotherapie",
    "Dramatherapie",
    "BEPP / Sensorimotor Psychotherapie",
    "Psychomotorische Therapie",
    "Slapende honden",
    "Nika, psycho-educatie",
    "Basic Trust-methode, schematherapie, behandeling gericht op rouw en verlies",
    "Intensieve Ambulante Gezinstherapie",
    "EFFT / EFIT en ABFT",
    "DDP, ABFT, EFFT, AFFT, Theraplay, schematherapie",
    "Imaginaire Exposure",
    "Bokspsychotherapie",
    "ABFT",
    "Lichaamsgerichte, innerlijk bronnen en systemische traumatherapie",
    "NET (Narratieve Exposure Therapie)"
  ],
        "treatmentArea": [
    "Geen beschikking nodig",
    "Normaal begaafd",
    "langdurig pesten behandelen wij ook, waarbij wel wordt gekeken of het niet gaat om complex trauma, dan wordt veelal verwezen naar de S-GGZ, wij bieden GB-GGZ. Een verstandelijke beperking is niet perse een contra, maar schatten dit per casus in. Wanneer het gaat om een licht verstandelijke beperkt niveau tot zwakbegaafd is het vaak wel mogelijk.",
    "automutilatie, geen acute suicidaliteit/crisis",
    "Trauma in het systeem",
    "Visuele beperking",
    "Vluchtelingenproblematiek",
    "Taalontwikkelingsstoornis",
    "ASS",
    "vaak is het comorbiditeit en niet de hoofddiagnose. Ik doe vaak traumatherapie/combi EMDR bij persoonlijkheidsproblematiek",
    "als er sprake is van een VB dan ook bij ASS, ADHD etc",
    "Geen onderscheid",
    "intergenerationeel trauma",
    "Genderproblematiek",
    "Hoogbegaafd",
    "misbruik, grote somberte, bij suicidaliteit en/of automutilatie altijd contact met arts of psychiater",
    "deelbehandeling in samenwerking met andere behandelsetting",
    "(Lichte) Verstandelijke Beperking",
    "Auditieve beperking",
    "Transcultureel",
    "ouder-kind relatieproblematiek",
    "Verslaving",
    "generationeel trauma ivm systemische problematiek",
    "Meervoudig trauma",
    "loyaliteitsproblematiek/ouderverstoting/oudervervreemding",
    "Als er sprake is van een indicatie, dan pakken wij dit als wijkteam in het voorveld niet op",
    "Bijna alle psychische aandoening waar ambulante behandeling van toepassing is",
    "Sucidaliteit en/of automutilatie",
    "Comorbiditeit is geen contra voor traumabehandeling. Middelengebruik moet gestopt zijn voor de traumabehandeling kan starten.",
    "Eetstoornis",
    "overbrugging wachtlijst",
    "Enkelvoudig trauma"
  ],
        "location": [
  "Aalsmeer",
  "Alkmaar",
  "Amstelveen",
  "Amsterdam",
  "Badhoevedorp",
  "Beverwijk",
  "Bergen NH",
  "Broek op Langedijk",
  "Bussum",
  "Callantsoog",
  "Castricum",
  "Cruquiusweg 32",
  "Den Helder",
  "Diemen",
  "Edam",
  "Enkhuizen",
  "Heemstede",
  "Heerhugowaard",
  "Heiloo",
  "Hilversum",
  "Hoofddorp",
  "Hoorn",
  "Huizen",
  "Janmaat Psychotherapie Haarlem",
  "Krommenie",
  "Laren NH",
  "Monnickendam",
  "Naarden",
  "Noord-Scharwoude",
  "Purmerend",
  "Schagen",
  "Spierdijk",
  "Velsen",
  "Vondelstraat 100",
  "Weesp",
  "Wormerland",
  "Zaanstad",
  "Zaandam",
  "Zuid-Kennemerland/ IJmond"
],
        "postalCode": None,
    }
    old_entity_data = old_entity_data.model_dump(mode='json')
    for key, value in entity_data.items():
        if key in old_entity_data and old_entity_data[key] != value:
            real_key = key if key!="treatmentArea" else "traumaType"
            changed_fields[real_key] = instruction_map[key]
    return changed_fields


def xor_cipher(text: str, key: str) -> str:
    key_bytes = key.encode('utf-8')
    text_bytes = text.encode('utf-8')
    key_len = len(key_bytes)
    encrypted_bytes = bytes([
        text_bytes[i] ^ key_bytes[i % key_len]
        for i in range(len(text_bytes))
    ])
    return encrypted_bytes.hex()

def encrypt_message(text: str, words_to_encrypt: list[str]) -> str:
    result = text
    for word in words_to_encrypt:
        if word in result:
            encrypted = xor_cipher(word, settings.SECRET_KEY)
            result = result.replace(word, f"[{encrypted}]")
    return result

def decrypt_messages(messages: list[MessageModel]) -> list[MessageModel]:
    
    def decrypt_match(match):
        encrypted_hex = match.group(1)
        encrypted_bytes = bytes.fromhex(encrypted_hex)
        key_bytes = settings.SECRET_KEY.encode('utf-8')
        decrypted_bytes = bytes([
            encrypted_bytes[i] ^ key_bytes[i % len(key_bytes)]
            for i in range(len(encrypted_bytes))
        ])
        return decrypted_bytes.decode('utf-8')
    
    pattern = r'\[([\da-fA-F]+)\]'

    for message in messages:
        message.text = re.sub(pattern, decrypt_match, message.text)
    return messages