dorontools/dorontools.py

import json as _json
import os as _os
import datetime as _datetime
import inspect as _inspect
import random as _random
import requests as _requests
import uuid as _uuid
import telebot as _telebot

def oneTo(a):
    return not _random.randint(0, a)

def listInString(b, string):
    for i in b:
        if i in string:
            return True

def logc(*args):
    args = list(map(str, args))
    caller = _inspect.stack()[1]
    print(f'[{_datetime.datetime.now()}] [{caller.function}] [{caller.lineno}] {' '.join(args)}')

class JsonAsDict():
    def __init__(self, filename):
        if not _os.path.isfile(filename):
            with open(filename, 'w', encoding='utf-8') as file:
                _json.dump({}, file, indent=4, ensure_ascii=False)
        self.filename = filename

    def __setitem__(self, key, item):
        with open(self.filename, 'r', encoding='utf-8') as file:
            data = _json.load(file)

        data[key] = item

        with open(self.filename, 'w', encoding='utf-8') as file:
            _json.dump(data, file, indent=4, ensure_ascii=False)

    def __getitem__(self, key):
        with open(self.filename, 'r', encoding='utf-8') as file:
            data = _json.load(file)
        return data[key]
    
    def __repr__(self):
        with open(self.filename, 'r', encoding='utf-8') as file:
            data = _json.load(file)
        return repr(data)
    
    def __iter__(self):
        with open(self.filename, 'r', encoding='utf-8') as file:
            data = _json.load(file)
        return iter(data)
    
    def __len__(self):
        with open(self.filename, 'r', encoding='utf-8') as file:
            data = _json.load(file)
        return len(data)
    
    def pop(self, key):
        with open(self.filename, 'r', encoding='utf-8') as file:
            data = _json.load(file)

        data.pop(key)

        with open(self.filename, 'w', encoding='utf-8') as file:
            _json.dump(data, file, indent=4, ensure_ascii=False)
    
    def values(self):
        with open(self.filename, 'r', encoding='utf-8') as file:
            data = _json.load(file)
        return data.values()
    
    def has_key(self, key):
        with open(self.filename, 'r', encoding='utf-8') as file:
            data = _json.load(file)
        return key in data

class TelebotLogger():
    def __init__(self, conn):
        self.conn = conn
        self.cursor = conn.cursor()
        self.cursor.execute('''
                            CREATE TABLE IF NOT EXISTS msgs (
                            id SERIAL PRIMARY KEY,
                            datetime TEXT,
                            chatid BIGINT,
                            chattitle TEXT,
                            msgid BIGINT,
                            userid BIGINT,
                            firstname TEXT,
                            lastname TEXT,
                            username TEXT,
                            text TEXT
                            )
                            ''')
        conn.commit() 

    def log(self, message):
        msg = message.json
        self.cursor.execute('INSERT INTO msgs (datetime, chatid, chattitle, msgid, userid, firstname, lastname, username, text) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)', 
                        (msg.get('date'), 
                        msg.get('chat').get('id'), 
                        msg.get('chat').get('title'), 
                        msg.get('message_id'), 
                        msg.get('from').get('id'), 
                        msg.get('from').get('first_name'), 
                        msg.get('from').get('last_name'), 
                        msg.get('from').get('username'), 
                        msg.get('text')))
        self.conn.commit()

# 3X-UI

class VPNClient():
    def __init__(self, obj: dict, ses: _requests.Session, url: str, bot: _telebot.TeleBot):
        self.ses = ses
        self.url = url
        self.email = obj.get('email')
        self.subId = obj.get('subId')
        self.uuid = obj.get('uuid', obj.get('id'))
        self.comment = obj.get('comment')
        self.tgId = obj.get('tgId')
        self.expiryTime = obj.get('expiryTime')
        self.bot = bot

    def __str__(self):
        return str(self.__dict__)

    def update(self, fields: dict):
        tmp = {
            'email': self.email,
            'subId': self.subId,
            'id': self.uuid,
            'expiryTime': self.expiryTime,
            'tgId': self.tgId,
            'comment': self.comment,
            'enable': True
        }
        for i in fields:
            tmp[i] = fields[i]
        
        req = self.ses.post(f'{self.url}/panel/api/clients/update/{self.email}', json=tmp)
        return req.json()

    def adjust(self, days: int, notify = False, reason = ''):
        self.bot.send_message(8782863015, f'{days} to {self.email}')
        tmp = {
            'emails': [self.email],
            'addDays': int(days)
        }

        req = self.ses.post(f'{self.url}/panel/api/clients/bulkAdjust', json=tmp)
    
        if self.bot and notify and self.tgId not in [0, '0', '', None]:
            diff = days
            if diff >= 0:
                action = 'продлён'
            elif diff < 0:
                action = 'оштрафован'
                diff *= -1
            text = f'Ваш ключ *{self.comment}* {action} на *{diff} дней*'
            if reason != '':
                text += f'. Причина: *{reason}*'
            self.bot.send_message(self.tgId, text)
            
        return req.json()
    
    def delete(self):
        req = self.ses.post(f'{self.url}/panel/api/clients/del/{self.email}')
        return req.json()

class VPNPanel():
    def __init__(self, url: str, token: str, bot: _telebot.TeleBot = None):
        self.url = url
        self.ses = _requests.session()
        self.ses.headers = {'Authorization': f'Bearer {token}'}
        self.bot = bot

    @property
    def clients(self) -> list[VPNClient]:
        req = self.ses.get(f'{self.url}/panel/api/clients/list')
        tmp = []
        for i in req.json()['obj']:
            tmp.append(VPNClient(i, self.ses, self.url, self.bot))
        return tmp
    
    def find_clients(self, field: str, value) -> list[VPNClient]:
        tmp = []
        for i in self.clients:
            if i.__dict__.get(field) == value:
                tmp.append(i)
        return tmp
    
    def create_client(self, uuid: str = None, email: str = None, expiryTime: int = 0, tgId: int = 0, comment: str = '', inbounds: list = [17, 18, 19]) -> VPNClient:
        if not uuid:
            uuid = str(_uuid.uuid4())

        if not email:
            email = uuid
        
        tmp = {
            'client': {
                'email': email,
                'subId': uuid,
                'id': uuid,
                'expiryTime': expiryTime,
                'tgId': tgId,
                'comment': comment,
                'enable': True
            },
            'inboundIds': inbounds
        }

        self.ses.post(f'{self.url}/panel/api/clients/add', json=tmp)

        return VPNClient(tmp['client'], self.ses, self.url, self.bot)