mirror of
https://github.com/house-of-vanity/OutFleet.git
synced 2025-08-21 14:37:16 +00:00
Added TG bot
This commit is contained in:
Ultradesu
292
telegram_bot/models.py
Normal file
292
telegram_bot/models.py
Normal file
@@ -0,0 +1,292 @@
|
||||
from django.db import models
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.utils import timezone
|
||||
import json
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class BotSettings(models.Model):
|
||||
"""Singleton model for bot settings"""
|
||||
bot_token = models.CharField(
|
||||
max_length=255,
|
||||
help_text="Telegram Bot Token from @BotFather"
|
||||
)
|
||||
enabled = models.BooleanField(
|
||||
default=False,
|
||||
help_text="Enable/Disable the bot"
|
||||
)
|
||||
use_proxy = models.BooleanField(
|
||||
default=False,
|
||||
help_text="Enable proxy for Telegram API connections"
|
||||
)
|
||||
proxy_url = models.URLField(
|
||||
blank=True,
|
||||
help_text="Proxy URL (e.g., http://proxy:8080 or socks5://proxy:1080)"
|
||||
)
|
||||
api_base_url = models.URLField(
|
||||
blank=True,
|
||||
default="https://api.telegram.org",
|
||||
help_text="Telegram API base URL (change for local bot API server)"
|
||||
)
|
||||
connection_timeout = models.IntegerField(
|
||||
default=30,
|
||||
help_text="Connection timeout in seconds"
|
||||
)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
|
||||
class Meta:
|
||||
verbose_name = "Bot Settings"
|
||||
verbose_name_plural = "Bot Settings"
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
# Ensure only one instance exists
|
||||
self.pk = 1
|
||||
super().save(*args, **kwargs)
|
||||
|
||||
def delete(self, *args, **kwargs):
|
||||
# Prevent deletion
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def get_settings(cls):
|
||||
"""Get or create singleton settings"""
|
||||
obj, created = cls.objects.get_or_create(pk=1)
|
||||
return obj
|
||||
|
||||
def __str__(self):
|
||||
return f"Bot Settings ({'Enabled' if self.enabled else 'Disabled'})"
|
||||
|
||||
|
||||
class TelegramMessage(models.Model):
|
||||
"""Store all telegram messages"""
|
||||
DIRECTION_CHOICES = [
|
||||
('incoming', 'Incoming'),
|
||||
('outgoing', 'Outgoing'),
|
||||
]
|
||||
|
||||
direction = models.CharField(
|
||||
max_length=10,
|
||||
choices=DIRECTION_CHOICES,
|
||||
db_index=True
|
||||
)
|
||||
|
||||
# Telegram user info
|
||||
telegram_user_id = models.BigIntegerField(db_index=True)
|
||||
telegram_username = models.CharField(
|
||||
max_length=255,
|
||||
blank=True,
|
||||
null=True,
|
||||
db_index=True
|
||||
)
|
||||
telegram_first_name = models.CharField(
|
||||
max_length=255,
|
||||
blank=True,
|
||||
null=True
|
||||
)
|
||||
telegram_last_name = models.CharField(
|
||||
max_length=255,
|
||||
blank=True,
|
||||
null=True
|
||||
)
|
||||
user_language = models.CharField(
|
||||
max_length=10,
|
||||
default='en',
|
||||
help_text="User's preferred language (en/ru)"
|
||||
)
|
||||
|
||||
# Message info
|
||||
chat_id = models.BigIntegerField(db_index=True)
|
||||
message_id = models.BigIntegerField(null=True, blank=True)
|
||||
message_text = models.TextField(blank=True)
|
||||
|
||||
# Additional data
|
||||
raw_data = models.JSONField(
|
||||
default=dict,
|
||||
blank=True,
|
||||
help_text="Full message data from Telegram"
|
||||
)
|
||||
|
||||
# Timestamps
|
||||
created_at = models.DateTimeField(auto_now_add=True, db_index=True)
|
||||
|
||||
# Optional link to VPN user if identified
|
||||
linked_user = models.ForeignKey(
|
||||
User,
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
related_name='telegram_messages'
|
||||
)
|
||||
|
||||
class Meta:
|
||||
verbose_name = "Telegram Message"
|
||||
verbose_name_plural = "Telegram Messages"
|
||||
ordering = ['-created_at']
|
||||
indexes = [
|
||||
models.Index(fields=['-created_at', 'direction']),
|
||||
models.Index(fields=['telegram_user_id', '-created_at']),
|
||||
]
|
||||
|
||||
def __str__(self):
|
||||
username = self.telegram_username or f"ID:{self.telegram_user_id}"
|
||||
direction_icon = "⬇️" if self.direction == 'incoming' else "⬆️"
|
||||
text_preview = self.message_text[:50] + "..." if len(self.message_text) > 50 else self.message_text
|
||||
return f"{direction_icon} {username}: {text_preview}"
|
||||
|
||||
@property
|
||||
def full_name(self):
|
||||
"""Get full name of telegram user"""
|
||||
parts = []
|
||||
if self.telegram_first_name:
|
||||
parts.append(self.telegram_first_name)
|
||||
if self.telegram_last_name:
|
||||
parts.append(self.telegram_last_name)
|
||||
return " ".join(parts) if parts else f"User {self.telegram_user_id}"
|
||||
|
||||
@property
|
||||
def display_name(self):
|
||||
"""Get best available display name"""
|
||||
if self.telegram_username:
|
||||
return f"@{self.telegram_username}"
|
||||
return self.full_name
|
||||
|
||||
|
||||
|
||||
|
||||
class AccessRequest(models.Model):
|
||||
"""Access requests from Telegram users"""
|
||||
|
||||
# Telegram user information
|
||||
telegram_user_id = models.BigIntegerField(
|
||||
db_index=True,
|
||||
help_text="Telegram user ID who made the request"
|
||||
)
|
||||
telegram_username = models.CharField(
|
||||
max_length=255,
|
||||
blank=True,
|
||||
null=True,
|
||||
help_text="Telegram username (without @)"
|
||||
)
|
||||
telegram_first_name = models.CharField(
|
||||
max_length=255,
|
||||
blank=True,
|
||||
null=True,
|
||||
help_text="First name from Telegram"
|
||||
)
|
||||
telegram_last_name = models.CharField(
|
||||
max_length=255,
|
||||
blank=True,
|
||||
null=True,
|
||||
help_text="Last name from Telegram"
|
||||
)
|
||||
|
||||
# Request details
|
||||
message_text = models.TextField(
|
||||
help_text="The message sent by user when requesting access"
|
||||
)
|
||||
chat_id = models.BigIntegerField(
|
||||
help_text="Telegram chat ID for sending notifications"
|
||||
)
|
||||
|
||||
# Username for VPN user creation
|
||||
desired_username = models.CharField(
|
||||
max_length=150,
|
||||
blank=True,
|
||||
help_text="Desired username for VPN user (defaults to Telegram username)"
|
||||
)
|
||||
|
||||
# User language
|
||||
user_language = models.CharField(
|
||||
max_length=10,
|
||||
default='en',
|
||||
help_text="User's preferred language (en/ru)"
|
||||
)
|
||||
|
||||
# Status and processing
|
||||
approved = models.BooleanField(
|
||||
default=False,
|
||||
db_index=True,
|
||||
help_text="Request approved by administrator"
|
||||
)
|
||||
admin_comment = models.TextField(
|
||||
blank=True,
|
||||
help_text="Admin comment for approval"
|
||||
)
|
||||
|
||||
# Related objects
|
||||
created_user = models.ForeignKey(
|
||||
User,
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
help_text="User created from this request (when approved)"
|
||||
)
|
||||
processed_by = models.ForeignKey(
|
||||
User,
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
related_name='processed_requests',
|
||||
help_text="Admin who processed this request"
|
||||
)
|
||||
first_message = models.ForeignKey(
|
||||
TelegramMessage,
|
||||
null=True,
|
||||
blank=True,
|
||||
on_delete=models.SET_NULL,
|
||||
help_text="First message from this user"
|
||||
)
|
||||
|
||||
# Timestamps
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
processed_at = models.DateTimeField(null=True, blank=True)
|
||||
|
||||
class Meta:
|
||||
verbose_name = "Access Request"
|
||||
verbose_name_plural = "Access Requests"
|
||||
ordering = ['-created_at']
|
||||
indexes = [
|
||||
models.Index(fields=['telegram_user_id']),
|
||||
models.Index(fields=['approved', '-created_at']),
|
||||
models.Index(fields=['-created_at']),
|
||||
]
|
||||
constraints = [
|
||||
models.UniqueConstraint(
|
||||
fields=['telegram_user_id'],
|
||||
name='unique_telegram_user_request'
|
||||
)
|
||||
]
|
||||
|
||||
def __str__(self):
|
||||
username = self.telegram_username or f"ID:{self.telegram_user_id}"
|
||||
status = "Approved" if self.approved else "Pending"
|
||||
return f"Request from @{username} ({status})"
|
||||
|
||||
@property
|
||||
def display_name(self):
|
||||
"""Get best available display name"""
|
||||
if self.telegram_username:
|
||||
return f"@{self.telegram_username}"
|
||||
|
||||
name_parts = []
|
||||
if self.telegram_first_name:
|
||||
name_parts.append(self.telegram_first_name)
|
||||
if self.telegram_last_name:
|
||||
name_parts.append(self.telegram_last_name)
|
||||
|
||||
if name_parts:
|
||||
return " ".join(name_parts)
|
||||
|
||||
return f"User {self.telegram_user_id}"
|
||||
|
||||
@property
|
||||
def full_name(self):
|
||||
"""Get full name of telegram user"""
|
||||
parts = []
|
||||
if self.telegram_first_name:
|
||||
parts.append(self.telegram_first_name)
|
||||
if self.telegram_last_name:
|
||||
parts.append(self.telegram_last_name)
|
||||
return " ".join(parts) if parts else None
|
||||
Reference in New Issue
Block a user