Format code to PEP8 & other small changes
This commit is contained in:
parent
e1d362d759
commit
e316279f38
|
@ -6,8 +6,8 @@ api_hash = ""
|
||||||
group_id = ""
|
group_id = ""
|
||||||
second_group_id = ""
|
second_group_id = ""
|
||||||
|
|
||||||
telegram_api_server = "127.0.0.1:5326"
|
telegram_api_server = "http://127.0.0.1:5326"
|
||||||
db_url = "sqlite://db.db"
|
db_uri = "sqlite://db.db"
|
||||||
|
|
||||||
limit_of_warns = 5
|
limit_of_warns = 5
|
||||||
|
|
||||||
|
|
|
@ -23,7 +23,7 @@ This bot is designed to simplify the moderation and management of Telegram group
|
||||||
| `group_id` | group id |
|
| `group_id` | group id |
|
||||||
| `second_group_id` | seconds group for admins |
|
| `second_group_id` | seconds group for admins |
|
||||||
| `telegram_bot_api_server` | telegram bot api server |
|
| `telegram_bot_api_server` | telegram bot api server |
|
||||||
| `db_url` | connection info to database |
|
| `db_uri` | connection info to database |
|
||||||
| `limit_of_warns` | limit user warnings |
|
| `limit_of_warns` | limit user warnings |
|
||||||
| `update_interval` | interval for update of user data |
|
| `update_interval` | interval for update of user data |
|
||||||
|
|
||||||
|
|
2
app.py
2
app.py
|
@ -22,7 +22,7 @@ WEBAPP_PORT = 3001
|
||||||
|
|
||||||
# Don`t touch anything!
|
# Don`t touch anything!
|
||||||
WEBHOOK_HOST = f'http://{WEBAPP_HOST}:{WEBAPP_PORT}'
|
WEBHOOK_HOST = f'http://{WEBAPP_HOST}:{WEBAPP_PORT}'
|
||||||
WEBHOOK_PATH = f'/bot{config.token}/'
|
WEBHOOK_PATH = f'/bot{config.bot_token}/'
|
||||||
WEBHOOK_URL = f"{WEBHOOK_HOST}{WEBHOOK_PATH}"
|
WEBHOOK_URL = f"{WEBHOOK_HOST}{WEBHOOK_PATH}"
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -8,7 +8,7 @@ env.read_env()
|
||||||
USE_WEBHOOK = True
|
USE_WEBHOOK = True
|
||||||
|
|
||||||
# bot token
|
# bot token
|
||||||
token = env.str("bot_token")
|
bot_token = env.str("bot_token")
|
||||||
|
|
||||||
group_id = env.int("group_id")
|
group_id = env.int("group_id")
|
||||||
second_group_id = env.int("second_group_id")
|
second_group_id = env.int("second_group_id")
|
||||||
|
@ -33,7 +33,6 @@ group_permissions = {
|
||||||
"can_pin_messages": False
|
"can_pin_messages": False
|
||||||
}
|
}
|
||||||
|
|
||||||
db_url = env.str("db_url")
|
db_uri = env.str("db_uri")
|
||||||
|
|
||||||
telegram_api_server = env.str("telegram_api_server").split(":")
|
telegram_api_server = env.str("telegram_api_server")
|
||||||
telegram_api_server = f"http://{telegram_api_server[0]}:{telegram_api_server[1]}"
|
|
||||||
|
|
12
database.py
12
database.py
|
@ -1,20 +1,22 @@
|
||||||
from peewee import Field, Model, BigIntegerField, CharField, DateField, DateTimeField, ForeignKeyField
|
from peewee import CharField, BigIntegerField, DateField, DateTimeField, ForeignKeyField
|
||||||
|
|
||||||
|
from enum import Enum
|
||||||
import config
|
import config
|
||||||
from playhouse.db_url import connect
|
from playhouse.db_url import connect
|
||||||
|
|
||||||
from datetime import datetime, date
|
from datetime import datetime, date
|
||||||
|
from peewee import Model
|
||||||
|
|
||||||
|
db = connect(config.db_uri)
|
||||||
|
|
||||||
|
|
||||||
db = connect(config.db_url)
|
|
||||||
|
|
||||||
from enum import Enum
|
|
||||||
class MemberRoles(Enum):
|
class MemberRoles(Enum):
|
||||||
OWNER = "owner"
|
OWNER = "owner"
|
||||||
ADMIN = "admin"
|
ADMIN = "admin"
|
||||||
HELPER = "helper"
|
HELPER = "helper"
|
||||||
MEMBER = "member"
|
MEMBER = "member"
|
||||||
|
|
||||||
|
|
||||||
class Member(Model):
|
class Member(Model):
|
||||||
user_id = BigIntegerField()
|
user_id = BigIntegerField()
|
||||||
first_name = CharField()
|
first_name = CharField()
|
||||||
|
@ -26,11 +28,11 @@ class Member(Model):
|
||||||
|
|
||||||
joined = DateField(default=date.today())
|
joined = DateField(default=date.today())
|
||||||
|
|
||||||
|
|
||||||
class Meta:
|
class Meta:
|
||||||
db_table = "members"
|
db_table = "members"
|
||||||
database = db
|
database = db
|
||||||
|
|
||||||
|
|
||||||
class Restriction(Model):
|
class Restriction(Model):
|
||||||
from_user = ForeignKeyField(Member, lazy_load=True)
|
from_user = ForeignKeyField(Member, lazy_load=True)
|
||||||
to_user = ForeignKeyField(Member, lazy_load=True)
|
to_user = ForeignKeyField(Member, lazy_load=True)
|
||||||
|
|
4
load.py
4
load.py
|
@ -13,7 +13,7 @@ import utils
|
||||||
storage = MemoryStorage()
|
storage = MemoryStorage()
|
||||||
|
|
||||||
# Create client connection
|
# Create client connection
|
||||||
tgc = utils.TelegramClient(config.api_id, config.api_hash, config.token)
|
tgc = utils.TelegramClient(config.api_id, config.api_hash, config.bot_token)
|
||||||
|
|
||||||
scheduler = AsyncIOScheduler()
|
scheduler = AsyncIOScheduler()
|
||||||
|
|
||||||
|
@ -22,7 +22,7 @@ session = AiohttpSession(
|
||||||
)
|
)
|
||||||
|
|
||||||
bot = Bot(
|
bot = Bot(
|
||||||
token=config.token,
|
token=config.bot_token,
|
||||||
session=session,
|
session=session,
|
||||||
parse_mode="Markdown"
|
parse_mode="Markdown"
|
||||||
)
|
)
|
||||||
|
|
|
@ -1,11 +1,19 @@
|
||||||
import typing
|
from typing import Optional
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from database import Member
|
from database import Member
|
||||||
|
|
||||||
from load import types
|
from load import types
|
||||||
|
|
||||||
|
|
||||||
def get_argument(arguments:list,index:int=0) -> typing.Optional[str]:
|
@dataclass
|
||||||
|
class CommandArguments:
|
||||||
|
to_user: Optional[Member]
|
||||||
|
from_user: Optional[Member]
|
||||||
|
arguments: list
|
||||||
|
is_silent: bool
|
||||||
|
|
||||||
|
|
||||||
|
def get_argument(arguments: list, index: int = 0) -> Optional[str]:
|
||||||
""" Get element from a list.If element not exist return None """
|
""" Get element from a list.If element not exist return None """
|
||||||
if not (arguments):
|
if not (arguments):
|
||||||
return None
|
return None
|
||||||
|
@ -15,12 +23,7 @@ def get_argument(arguments:list,index:int=0) -> typing.Optional[str]:
|
||||||
else:
|
else:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class CommandArguments:
|
|
||||||
to_user:Member | None
|
|
||||||
from_user:Member | None
|
|
||||||
arguments:list
|
|
||||||
is_silent:bool
|
|
||||||
async def get_command_args(message: types.Message) -> CommandArguments:
|
async def get_command_args(message: types.Message) -> CommandArguments:
|
||||||
"""Describe user data and arguments from message"""
|
"""Describe user data and arguments from message"""
|
||||||
|
|
||||||
|
|
|
@ -2,6 +2,7 @@ import config
|
||||||
from database import Member, Restriction
|
from database import Member, Restriction
|
||||||
from database import db
|
from database import db
|
||||||
|
|
||||||
|
|
||||||
def database_is_empty():
|
def database_is_empty():
|
||||||
if not db.get_columns("members"):
|
if not db.get_columns("members"):
|
||||||
db.create_tables([Member, Restriction])
|
db.create_tables([Member, Restriction])
|
||||||
|
@ -12,5 +13,6 @@ def database_is_empty():
|
||||||
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
async def notify_started_bot(bot):
|
async def notify_started_bot(bot):
|
||||||
await bot.send_message(config.second_group_id,"Bot successfully launched!")
|
await bot.send_message(config.second_group_id, "Bot successfully launched!")
|
||||||
|
|
|
@ -1,11 +1,11 @@
|
||||||
import re
|
import re
|
||||||
import typing
|
from typing import Optional
|
||||||
|
|
||||||
import datetime
|
import datetime
|
||||||
from load import types
|
from load import types
|
||||||
|
|
||||||
|
|
||||||
def parse_timedelta(value: str) -> typing.Optional[datetime.timedelta]:
|
def parse_timedelta(value: str) -> Optional[datetime.timedelta]:
|
||||||
regex = r'(?:(\d+)(?:d|д))?(?:(\d+)(?:h|ч))?(?:(\d+)(?:m|м))?(?:(\d+)(?:s|с))?'
|
regex = r'(?:(\d+)(?:d|д))?(?:(\d+)(?:h|ч))?(?:(\d+)(?:m|м))?(?:(\d+)(?:s|с))?'
|
||||||
specification = value.strip().replace(' ', '')
|
specification = value.strip().replace(' ', '')
|
||||||
match = re.fullmatch(regex, specification)
|
match = re.fullmatch(regex, specification)
|
||||||
|
@ -19,7 +19,7 @@ def parse_timedelta(value: str) -> typing.Optional[datetime.timedelta]:
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def parse_timedelta_from_message(message: types.Message) -> typing.Optional[datetime.timedelta]:
|
def parse_timedelta_from_message(message: types.Message) -> Optional[datetime.timedelta]:
|
||||||
_, *args = message.text.split()
|
_, *args = message.text.split()
|
||||||
|
|
||||||
if args:
|
if args:
|
||||||
|
|
|
@ -2,17 +2,17 @@ from pyrogram.client import Client
|
||||||
|
|
||||||
|
|
||||||
class TelegramClient:
|
class TelegramClient:
|
||||||
def __init__(self,api_id,api_hash,token):
|
def __init__(self, api_id, api_hash, token):
|
||||||
self.api_id = api_id
|
self.api_id = api_id
|
||||||
self.api_hash = api_hash
|
self.api_hash = api_hash
|
||||||
self.token = token
|
self.token = token
|
||||||
|
|
||||||
self.client = Client("session",
|
self.client = Client("session",
|
||||||
api_id=self.api_id,api_hash=self.api_hash,
|
api_id=self.api_id, api_hash=self.api_hash,
|
||||||
bot_token=self.token
|
bot_token=self.token
|
||||||
)
|
)
|
||||||
|
|
||||||
async def members_list(self,chat_id:int):
|
async def members_list(self, chat_id: int):
|
||||||
members = []
|
members = []
|
||||||
|
|
||||||
async for member in self.client.get_chat_members(chat_id):
|
async for member in self.client.get_chat_members(chat_id):
|
||||||
|
@ -27,10 +27,10 @@ class TelegramClient:
|
||||||
if (not member.user.is_bot):
|
if (not member.user.is_bot):
|
||||||
|
|
||||||
members.append({
|
members.append({
|
||||||
"id":member.user.id,
|
"id": member.user.id,
|
||||||
"status":str(member.status),
|
"status": str(member.status),
|
||||||
"first_name":member.user.first_name,
|
"first_name": member.user.first_name,
|
||||||
"username":username,
|
"username": username,
|
||||||
})
|
})
|
||||||
|
|
||||||
return members
|
return members
|
||||||
|
|
Reference in New Issue