import logging import uuid from django.conf import settings from django.contrib.auth.models import AbstractUser from django.db import models from core.lib.notify import raw_sendmsg from core.clients import signalapi logger = logging.getLogger(__name__) SERVICE_CHOICES = ( ("signal", "Signal"), ("instagram", "Instagram"), ) MBTI_CHOICES = ( ("INTJ", "INTJ - Architect"), ("INTP", "INTP - Logician"), ("ENTJ", "ENTJ - Commander"), ("ENTP", "ENTP - Debater"), ("INFJ", "INFJ - Advocate"), ("INFP", "INFP - Mediator"), ("ENFJ", "ENFJ - Protagonist"), ("ENFP", "ENFP - Campaigner"), ("ISTJ", "ISTJ - Logistician"), ("ISFJ", "ISFJ - Defender"), ("ESTJ", "ESTJ - Executive"), ("ESFJ", "ESFJ - Consul"), ("ISTP", "ISTP - Virtuoso"), ("ISFP", "ISFP - Adventurer"), ("ESTP", "ESTP - Entrepreneur"), ("ESFP", "ESFP - Entertainer"), ) MODEL_CHOICES = ( ("gpt-4o-mini", "GPT 4o Mini"), ("gpt-4o", "GPT 4o"), ) class User(AbstractUser): # Stripe customer ID stripe_id = models.CharField(max_length=255, null=True, blank=True) customer_id = models.UUIDField(default=uuid.uuid4, null=True, blank=True) billing_provider_id = models.CharField(max_length=255, null=True, blank=True) email = models.EmailField(unique=True) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._original = self def get_notification_settings(self): return NotificationSettings.objects.get_or_create(user=self)[0] def sendmsg(self, *args, **kwargs): notification_settings = self.get_notification_settings() if notification_settings.ntfy_topic is None: # No topic set, so don't send return else: topic = notification_settings.ntfy_topic raw_sendmsg(*args, **kwargs, url=notification_settings.ntfy_url, topic=topic) class NotificationSettings(models.Model): user = models.OneToOneField(User, on_delete=models.CASCADE) ntfy_topic = models.CharField(max_length=255, null=True, blank=True) ntfy_url = models.CharField(max_length=255, null=True, blank=True) def __str__(self): return f"Notification settings for {self.user}" class Chat(models.Model): source_number = models.CharField(max_length=32, null=True, blank=True) source_uuid = models.CharField(max_length=255, null=True, blank=True) source_name = models.CharField(max_length=255, null=True, blank=True) account = models.CharField(max_length=32, null=True, blank=True) class AI(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) base_url = models.CharField(max_length=255, null=True, blank=True) api_key = models.CharField(max_length=255, null=True, blank=True) model = models.CharField(max_length=255, choices=MODEL_CHOICES) def __str__(self): return f"{self.id} - {self.model}" class Person(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) user = models.ForeignKey(User, on_delete=models.CASCADE) name = models.CharField(max_length=255) summary = models.TextField(blank=True, null=True) profile = models.TextField(blank=True, null=True) revealed = models.TextField(blank=True, null=True) dislikes = models.TextField(blank=True, null=True) likes = models.TextField(blank=True, null=True) # -1 (disliked) to +1 (trusted) sentiment = models.FloatField(default=0.0) timezone = models.CharField(max_length=50, blank=True, null=True) last_interaction = models.DateTimeField(blank=True, null=True) def __str__(self): return self.name class PersonIdentifier(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) identifier = models.CharField(max_length=255) service = models.CharField(choices=SERVICE_CHOICES, max_length=255) person = models.ForeignKey(Person, on_delete=models.CASCADE) def __str__(self): return f"{self.person} ({self.service})" async def send(self, text, attachments=[]): """ Send this contact a text. """ if self.service == "signal": ts = await signalapi.send_message_raw( self.identifier, text, attachments, ) print("SENT") return ts else: raise NotImplementedError(f"Service not implemented: {self.service}") class ChatSession(models.Model): """Represents an ongoing chat session, stores summarized history.""" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) user = models.ForeignKey(User, on_delete=models.CASCADE) identifier = models.ForeignKey(PersonIdentifier, on_delete=models.CASCADE) last_interaction = models.DateTimeField(blank=True, null=True) summary = models.TextField(blank=True, null=True) def __str__(self): return f"{self.identifier.person.name} ({self.identifier.service})" class QueuedMessage(models.Model): """Stores individual messages linked to a ChatSession.""" user = models.ForeignKey(User, on_delete=models.CASCADE) id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) session = models.ForeignKey(ChatSession, on_delete=models.CASCADE) manipulation = models.ForeignKey("core.Manipulation", on_delete=models.CASCADE) ts = models.BigIntegerField() # Use Unix timestamp sender_uuid = models.CharField(max_length=255, blank=True, null=True) # Signal UUID text = models.TextField(blank=True, null=True) custom_author = models.CharField(max_length=255, blank=True, null=True) class Meta: ordering = ["ts"] class Message(models.Model): """Stores individual messages linked to a ChatSession.""" user = models.ForeignKey(User, on_delete=models.CASCADE) id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) session = models.ForeignKey(ChatSession, on_delete=models.CASCADE) ts = models.BigIntegerField() # Use Unix timestamp sender_uuid = models.CharField(max_length=255, blank=True, null=True) # Signal UUID text = models.TextField(blank=True, null=True) custom_author = models.CharField(max_length=255, blank=True, null=True) class Meta: ordering = ["ts"] class Group(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) user = models.ForeignKey(User, on_delete=models.CASCADE) name = models.CharField(max_length=255) people = models.ManyToManyField(Person) def __str__(self): return self.name class Persona(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) user = models.ForeignKey(User, on_delete=models.CASCADE) # Core Identity # id = models.UUIDField(default=uuid.uuid4, editable=False, unique=True) alias = models.CharField(max_length=255, blank=True, null=True) # Preferred name or persona alias mbti = models.CharField(max_length=255, choices=MBTI_CHOICES, blank=True, null=True) # -1: assertive, +1: assertive mbti_identity = models.FloatField(default=0.0) # Key Behavioral Traits for Chat Responses inner_story = models.TextField(blank=True, null=True) # Internal philosophy & worldview core_values = models.TextField(blank=True, null=True) # What drives their decisions & interactions communication_style = models.TextField(blank=True, null=True) # How they speak & interact flirting_style = models.TextField(blank=True, null=True) # How they express attraction humor_style = models.CharField( max_length=50, choices=[ ("dry", "Dry"), ("dark", "Dark"), ("playful", "Playful"), ("teasing", "Teasing"), ("sarcastic", "Sarcastic"), ("intellectual", "Intellectual"), ], blank=True, null=True ) # Defines their approach to humor # Conversational Preferences likes = models.TextField(blank=True, null=True) # Topics they enjoy discussing dislikes = models.TextField(blank=True, null=True) # Topics or behaviors they avoid tone = models.CharField( max_length=50, choices=[ ("formal", "Formal"), ("casual", "Casual"), ("witty", "Witty"), ("serious", "Serious"), ("warm", "Warm"), ("detached", "Detached"), ], blank=True, null=True ) # Defines preferred conversational tone # Emotional & Strategic Interaction response_tactics = models.TextField(blank=True, null=True) # How they handle gaslighting, guilt-tripping, etc. persuasion_tactics = models.TextField(blank=True, null=True) # How they convince others boundaries = models.TextField(blank=True, null=True) # What they refuse to tolerate in conversations trust = models.IntegerField(default=50) # Percentage of initial trust given in interactions adaptability = models.IntegerField(default=70) # How easily they shift tones or styles def __str__(self): return f"{self.alias} ({self.mbti}) [{self.tone} {self.humor_style}]" class Manipulation(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) user = models.ForeignKey(User, on_delete=models.CASCADE) name = models.CharField(max_length=255) group = models.ForeignKey(Group, on_delete=models.CASCADE) #self = models.ForeignKey(Group, on_delete=models.CASCADE) ai = models.ForeignKey(AI, on_delete=models.CASCADE) persona = models.ForeignKey(Persona, on_delete=models.CASCADE) enabled = models.BooleanField(default=False) filter_enabled = models.BooleanField(default=False) mode = models.CharField( max_length=50, choices=[ ("active", "Send replies to messages"), ("instant", "Click link to send reply"), ("prospective", "Click link to open page"), ("notify", "Send notification of ideal reply only"), ("mutate", "Change messages sent on XMPP using the persona"), ("silent", "Do not generate or send replies"), ], blank=True, null=True ) def __str__(self): return f"{self.name} [{self.group}]" # class Perms(models.Model): # class Meta: # permissions = ( # ("bypass_hashing", "Can bypass field hashing"), # # ("bypass_blacklist", "Can bypass the blacklist"), # # ("bypass_encryption", "Can bypass field encryption"), # # ("bypass_obfuscation", "Can bypass field obfuscation"), # # ("bypass_delay", "Can bypass data delay"), # # ("bypass_randomisation", "Can bypass data randomisation"), # # ("post_irc", "Can post to IRC"), # ("post_discord", "Can post to Discord"), # ("query_search", "Can search with query strings"), # # ("use_insights", "Can use the Insights page"), # ("index_int", "Can use the internal index"), # ("index_meta", "Can use the meta index"), # ("restricted_sources", "Can access restricted sources"), # )