mirror of
https://github.com/iio612/DEFENDER.git
synced 2026-02-13 19:24:23 +00:00
V5.0.1
This commit is contained in:
@@ -3,7 +3,6 @@ from datetime import datetime
|
||||
from typing import Union
|
||||
from core.base import Base
|
||||
|
||||
|
||||
class User:
|
||||
|
||||
@dataclass
|
||||
@@ -26,7 +25,14 @@ class User:
|
||||
pass
|
||||
|
||||
def insert(self, newUser: UserModel) -> bool:
|
||||
"""Insert a new User object
|
||||
|
||||
Args:
|
||||
newUser (UserModel): New userModel object
|
||||
|
||||
Returns:
|
||||
bool: True if inserted
|
||||
"""
|
||||
result = False
|
||||
exist = False
|
||||
|
||||
@@ -46,7 +52,15 @@ class User:
|
||||
return result
|
||||
|
||||
def update(self, uid: str, newNickname: str) -> bool:
|
||||
"""Update the nickname starting from the UID
|
||||
|
||||
Args:
|
||||
uid (str): UID of the user
|
||||
newNickname (str): New nickname
|
||||
|
||||
Returns:
|
||||
bool: True if updated
|
||||
"""
|
||||
result = False
|
||||
|
||||
for record in self.UID_DB:
|
||||
@@ -61,7 +75,14 @@ class User:
|
||||
return result
|
||||
|
||||
def delete(self, uid: str) -> bool:
|
||||
"""Delete the User starting from the UID
|
||||
|
||||
Args:
|
||||
uid (str): UID of the user
|
||||
|
||||
Returns:
|
||||
bool: True if deleted
|
||||
"""
|
||||
result = False
|
||||
|
||||
for record in self.UID_DB:
|
||||
@@ -76,7 +97,14 @@ class User:
|
||||
return result
|
||||
|
||||
def get_User(self, uidornickname: str) -> Union[UserModel, None]:
|
||||
"""Get The User Object model
|
||||
|
||||
Args:
|
||||
uidornickname (str): UID or Nickname
|
||||
|
||||
Returns:
|
||||
UserModel|None: The UserModel Object | None
|
||||
"""
|
||||
User = None
|
||||
for record in self.UID_DB:
|
||||
if record.uid == uidornickname:
|
||||
@@ -89,7 +117,14 @@ class User:
|
||||
return User
|
||||
|
||||
def get_uid(self, uidornickname:str) -> Union[str, None]:
|
||||
"""Get the UID of the user starting from the UID or the Nickname
|
||||
|
||||
Args:
|
||||
uidornickname (str): UID or Nickname
|
||||
|
||||
Returns:
|
||||
str|None: Return the UID
|
||||
"""
|
||||
uid = None
|
||||
for record in self.UID_DB:
|
||||
if record.uid == uidornickname:
|
||||
@@ -101,14 +136,21 @@ class User:
|
||||
return uid
|
||||
|
||||
def get_nickname(self, uidornickname:str) -> Union[str, None]:
|
||||
"""Get the Nickname starting from UID or the nickname
|
||||
|
||||
Args:
|
||||
uidornickname (str): UID or Nickname of the user
|
||||
|
||||
Returns:
|
||||
str|None: the nickname
|
||||
"""
|
||||
nickname = None
|
||||
for record in self.UID_DB:
|
||||
if record.nickname == uidornickname:
|
||||
nickname = record.nickname
|
||||
if record.uid == uidornickname:
|
||||
nickname = record.nickname
|
||||
self.log.debug(f'The value {uidornickname} -- {nickname}')
|
||||
self.log.debug(f'The value to check {uidornickname} -> {nickname}')
|
||||
return nickname
|
||||
|
||||
class Admin:
|
||||
@@ -217,6 +259,7 @@ class Admin:
|
||||
return nickname
|
||||
|
||||
class Channel:
|
||||
|
||||
@dataclass
|
||||
class ChannelModel:
|
||||
name: str
|
||||
@@ -227,6 +270,7 @@ class Channel:
|
||||
|
||||
def __init__(self, Base: Base) -> None:
|
||||
self.log = Base.logs
|
||||
self.Base = Base
|
||||
pass
|
||||
|
||||
def insert(self, newChan: ChannelModel) -> bool:
|
||||
@@ -288,16 +332,26 @@ class Channel:
|
||||
|
||||
return result
|
||||
|
||||
def delete_user_from_channel(self,chan_name: str, uid:str) -> bool:
|
||||
def delete_user_from_channel(self, chan_name: str, uid:str) -> bool:
|
||||
try:
|
||||
result = False
|
||||
|
||||
for record in self.UID_CHANNEL_DB:
|
||||
if record.name == chan_name:
|
||||
record.uids.remove(uid)
|
||||
for user_id in record.uids:
|
||||
if self.Base.clean_uid(user_id) == uid:
|
||||
record.uids.remove(user_id)
|
||||
self.log.debug(f'uid {uid} has been removed, here is the new object: {record}')
|
||||
result = True
|
||||
|
||||
for record in self.UID_CHANNEL_DB:
|
||||
if not record.uids:
|
||||
self.UID_CHANNEL_DB.remove(record)
|
||||
self.log.debug(f'Channel {record.name} has been removed, here is the new object: {record}')
|
||||
|
||||
return result
|
||||
except ValueError as ve:
|
||||
self.log.error(f'{ve}')
|
||||
|
||||
def get_Channel(self, name: str) -> Union[ChannelModel, None]:
|
||||
|
||||
@@ -319,4 +373,3 @@ class Channel:
|
||||
|
||||
self.log.debug(f'The mode of the channel {name} has been found: {mode}')
|
||||
return mode
|
||||
|
||||
|
||||
274
core/dataClass.py
Normal file
274
core/dataClass.py
Normal file
@@ -0,0 +1,274 @@
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Union
|
||||
|
||||
class User:
|
||||
|
||||
@dataclass
|
||||
class UserDataClass:
|
||||
uid: str
|
||||
nickname: str
|
||||
username: str
|
||||
hostname: str
|
||||
umodes: str
|
||||
vhost: str
|
||||
isWebirc: bool
|
||||
connexion_datetime: datetime = field(default=datetime.now())
|
||||
|
||||
UID_DB:list[UserDataClass] = []
|
||||
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
def insert(self, user: UserDataClass) -> bool:
|
||||
"""Insert new user
|
||||
|
||||
Args:
|
||||
user (UserDataClass): The User dataclass
|
||||
|
||||
Returns:
|
||||
bool: True if the record has been created
|
||||
"""
|
||||
exists = False
|
||||
inserted = False
|
||||
|
||||
for record in self.UID_DB:
|
||||
if record.uid == user.uid:
|
||||
exists = True
|
||||
print(f'{user.uid} already exist')
|
||||
|
||||
if not exists:
|
||||
self.UID_DB.append(user)
|
||||
print(f'New record with uid: {user.uid}')
|
||||
inserted = True
|
||||
|
||||
return inserted
|
||||
|
||||
def update(self, uid: str, newnickname: str) -> bool:
|
||||
"""Updating a single record with a new nickname
|
||||
|
||||
Args:
|
||||
uid (str): the uid of the user
|
||||
newnickname (str): the new nickname
|
||||
|
||||
Returns:
|
||||
bool: True if the record has been updated
|
||||
"""
|
||||
status = False
|
||||
for user in self.UID_DB:
|
||||
if user.uid == uid:
|
||||
user.nickname = newnickname
|
||||
status = True
|
||||
print(f'Updating record with uid: {uid}')
|
||||
|
||||
return status
|
||||
|
||||
def delete(self, uid: str) -> bool:
|
||||
"""Delete a user based on his uid
|
||||
|
||||
Args:
|
||||
uid (str): The UID of the user
|
||||
|
||||
Returns:
|
||||
bool: True if the record has been deleted
|
||||
"""
|
||||
status = False
|
||||
for user in self.UID_DB:
|
||||
if user.uid == uid:
|
||||
self.UID_DB.remove(user)
|
||||
status = True
|
||||
print(f'Removing record with uid: {uid}')
|
||||
|
||||
return status
|
||||
|
||||
def isexist(self, uidornickname:str) -> bool:
|
||||
"""do the UID or Nickname exist ?
|
||||
|
||||
Args:
|
||||
uidornickname (str): The UID or the Nickname
|
||||
|
||||
Returns:
|
||||
bool: True if exist or False if don't exist
|
||||
"""
|
||||
result = False
|
||||
for record in self.UID_DB:
|
||||
if record.uid == uidornickname:
|
||||
result = True
|
||||
if record.nickname == uidornickname:
|
||||
result = True
|
||||
|
||||
return result
|
||||
|
||||
def get_User(self, uidornickname) -> Union[UserDataClass, None]:
|
||||
|
||||
UserObject = None
|
||||
for record in self.UID_DB:
|
||||
if record.uid == uidornickname:
|
||||
UserObject = record
|
||||
elif record.nickname == uidornickname:
|
||||
UserObject = record
|
||||
|
||||
return UserObject
|
||||
|
||||
def get_uid(self, uidornickname:str) -> Union[str, None]:
|
||||
|
||||
uid = None
|
||||
for record in self.UID_DB:
|
||||
if record.uid == uidornickname:
|
||||
uid = record.uid
|
||||
if record.nickname == uidornickname:
|
||||
uid = record.uid
|
||||
|
||||
return uid
|
||||
|
||||
def get_nickname(self, uidornickname:str) -> Union[str, None]:
|
||||
|
||||
nickname = None
|
||||
for record in self.UID_DB:
|
||||
if record.nickname == uidornickname:
|
||||
nickname = record.nickname
|
||||
if record.uid == uidornickname:
|
||||
nickname = record.nickname
|
||||
|
||||
return nickname
|
||||
|
||||
class Admin:
|
||||
@dataclass
|
||||
class AdminDataClass:
|
||||
uid: str
|
||||
nickname: str
|
||||
username: str
|
||||
hostname: str
|
||||
umodes: str
|
||||
vhost: str
|
||||
level: int
|
||||
connexion_datetime: datetime = field(default=datetime.now())
|
||||
|
||||
UID_ADMIN_DB:list[AdminDataClass] = []
|
||||
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
def insert(self, admin: AdminDataClass) -> bool:
|
||||
"""Insert new user
|
||||
|
||||
Args:
|
||||
user (UserDataClass): The User dataclass
|
||||
|
||||
Returns:
|
||||
bool: True if the record has been created
|
||||
"""
|
||||
exists = False
|
||||
inserted = False
|
||||
|
||||
for record in self.UID_ADMIN_DB:
|
||||
if record.uid == admin.uid:
|
||||
exists = True
|
||||
print(f'{admin.uid} already exist')
|
||||
|
||||
if not exists:
|
||||
self.UID_ADMIN_DB.append(admin)
|
||||
print(f'New record with uid: {admin.uid}')
|
||||
inserted = True
|
||||
|
||||
return inserted
|
||||
|
||||
def update(self, uid: str, newnickname: str) -> bool:
|
||||
"""Updating a single record with a new nickname
|
||||
|
||||
Args:
|
||||
uid (str): the uid of the user
|
||||
newnickname (str): the new nickname
|
||||
|
||||
Returns:
|
||||
bool: True if the record has been updated
|
||||
"""
|
||||
status = False
|
||||
for admin in self.UID_ADMIN_DB:
|
||||
if admin.uid == uid:
|
||||
admin.nickname = newnickname
|
||||
status = True
|
||||
print(f'Updating record with uid: {uid}')
|
||||
|
||||
return status
|
||||
|
||||
def delete(self, uid: str) -> bool:
|
||||
"""Delete a user based on his uid
|
||||
|
||||
Args:
|
||||
uid (str): The UID of the user
|
||||
|
||||
Returns:
|
||||
bool: True if the record has been deleted
|
||||
"""
|
||||
status = False
|
||||
for admin in self.UID_ADMIN_DB:
|
||||
if admin.uid == uid:
|
||||
self.UID_ADMIN_DB.remove(admin)
|
||||
status = True
|
||||
print(f'Removing record with uid: {uid}')
|
||||
|
||||
return status
|
||||
|
||||
def isexist(self, uidornickname:str) -> bool:
|
||||
"""do the UID or Nickname exist ?
|
||||
|
||||
Args:
|
||||
uidornickname (str): The UID or the Nickname
|
||||
|
||||
Returns:
|
||||
bool: True if exist or False if don't exist
|
||||
"""
|
||||
result = False
|
||||
for record in self.UID_ADMIN_DB:
|
||||
if record.uid == uidornickname:
|
||||
result = True
|
||||
if record.nickname == uidornickname:
|
||||
result = True
|
||||
|
||||
return result
|
||||
|
||||
def get_Admin(self, uidornickname) -> Union[AdminDataClass, None]:
|
||||
|
||||
AdminObject = None
|
||||
for record in self.UID_ADMIN_DB:
|
||||
if record.uid == uidornickname:
|
||||
AdminObject = record
|
||||
elif record.nickname == uidornickname:
|
||||
AdminObject = record
|
||||
|
||||
return AdminObject
|
||||
|
||||
def get_uid(self, uidornickname:str) -> Union[str, None]:
|
||||
|
||||
uid = None
|
||||
for record in self.UID_ADMIN_DB:
|
||||
if record.uid == uidornickname:
|
||||
uid = record.uid
|
||||
if record.nickname == uidornickname:
|
||||
uid = record.uid
|
||||
|
||||
return uid
|
||||
|
||||
def get_nickname(self, uidornickname:str) -> Union[str, None]:
|
||||
|
||||
nickname = None
|
||||
for record in self.UID_ADMIN_DB:
|
||||
if record.nickname == uidornickname:
|
||||
nickname = record.nickname
|
||||
if record.uid == uidornickname:
|
||||
nickname = record.nickname
|
||||
|
||||
return nickname
|
||||
|
||||
def get_level(self, uidornickname:str) -> int:
|
||||
|
||||
level = 0
|
||||
for record in self.UID_ADMIN_DB:
|
||||
if record.uid == uidornickname:
|
||||
level = record.level
|
||||
if record.nickname == uidornickname:
|
||||
level = record.level
|
||||
|
||||
return level
|
||||
|
||||
@@ -723,7 +723,7 @@ class Irc:
|
||||
# ['@msgid=5sTwGdj349D82L96p749SY;time=2024-08-15T09:50:23.528Z', ':001', 'SJOIN', '1721564574', '#welcome', ':001JD94QH']
|
||||
# ['@msgid=bvceb6HthbLJapgGLXn1b0;time=2024-08-15T09:50:11.464Z', ':001', 'SJOIN', '1721564574', '#welcome', '+lnrt', '13', ':001CIVLQF', '+11ZAAAAAB', '001QGR10C', '*@0014UE10B', '001NL1O07', '001SWZR05', '001HB8G04', '@00BAAAAAJ', '0019M7101']
|
||||
cmd.pop(0)
|
||||
channel = cmd[3]
|
||||
channel = str(cmd[3]).lower()
|
||||
mode = cmd[4]
|
||||
len_cmd = len(cmd)
|
||||
list_users:list = []
|
||||
|
||||
@@ -210,6 +210,25 @@ class Votekick():
|
||||
|
||||
return response
|
||||
|
||||
def timer_vote_verdict(self, channel: str) -> None:
|
||||
|
||||
dnickname = self.Config.SERVICE_NICKNAME
|
||||
|
||||
for chan in self.VOTE_CHANNEL_DB:
|
||||
if chan.channel_name == channel:
|
||||
target_user = self.User.get_nickname(chan.target_user)
|
||||
if chan.vote_for > chan.vote_against:
|
||||
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :The user {self.Config.CONFIG_COLOR["gras"]}{target_user}{self.Config.CONFIG_COLOR["nogc"]} will be kicked from this channel')
|
||||
self.Irc.send2socket(f":{dnickname} KICK {channel} {target_user} Following the vote, you are not welcome in {channel}")
|
||||
elif chan.vote_for <= chan.vote_against:
|
||||
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :This user will stay on this channel')
|
||||
|
||||
# Init the system
|
||||
if self.init_vote_system(channel):
|
||||
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :System vote re initiated')
|
||||
|
||||
return None
|
||||
|
||||
def cmd(self, data:list) -> None:
|
||||
cmd = list(data).copy()
|
||||
|
||||
@@ -227,6 +246,8 @@ class Votekick():
|
||||
|
||||
command = str(cmd[0]).lower()
|
||||
dnickname = self.Config.SERVICE_NICKNAME
|
||||
fromuser = user
|
||||
|
||||
if len(fullcmd) >= 3:
|
||||
fromchannel = str(fullcmd[2]).lower() if self.Base.Is_Channel(str(fullcmd[2]).lower()) else None
|
||||
else:
|
||||
@@ -244,17 +265,13 @@ class Votekick():
|
||||
else:
|
||||
channel = None
|
||||
|
||||
|
||||
fromuser = user
|
||||
|
||||
match command:
|
||||
|
||||
case 'vote_cancel':
|
||||
try:
|
||||
if fromchannel is None:
|
||||
channel = str(cmd[1]).lower()
|
||||
else:
|
||||
channel = fromchannel
|
||||
if channel is None:
|
||||
self.Logs.error(f"The channel is not known, defender can't cancel the vote")
|
||||
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :You need to specify the channel => /msg {dnickname} vote_cancel #channel')
|
||||
|
||||
for vote in self.VOTE_CHANNEL_DB:
|
||||
if vote.channel_name == channel:
|
||||
@@ -280,6 +297,9 @@ class Votekick():
|
||||
|
||||
except KeyError as ke:
|
||||
self.Logs.error(f'Key Error: {ke}')
|
||||
except IndexError as ie:
|
||||
self.Irc.send2socket(f':{dnickname} NOTICE {fromuser} :/msg {dnickname} vote_cancel #channel')
|
||||
self.Logs.error(f'Index Error: {ie}')
|
||||
|
||||
case 'vote_against':
|
||||
try:
|
||||
@@ -299,7 +319,7 @@ class Votekick():
|
||||
|
||||
case 'vote_stat':
|
||||
try:
|
||||
channel = str(fullcmd[2]).lower()
|
||||
# channel = str(fullcmd[2]).lower()
|
||||
for chan in self.VOTE_CHANNEL_DB:
|
||||
if chan.channel_name == channel:
|
||||
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :Channel: {chan.channel_name} | Target: {self.User.get_nickname(chan.target_user)} | For: {chan.vote_for} | Against: {chan.vote_against} | Number of voters: {str(len(chan.voter_users))}')
|
||||
@@ -309,7 +329,7 @@ class Votekick():
|
||||
|
||||
case 'vote_verdict':
|
||||
try:
|
||||
channel = str(fullcmd[2]).lower()
|
||||
# channel = str(fullcmd[2]).lower()
|
||||
for chan in self.VOTE_CHANNEL_DB:
|
||||
if chan.channel_name == channel:
|
||||
target_user = self.User.get_nickname(chan.target_user)
|
||||
@@ -330,7 +350,7 @@ class Votekick():
|
||||
# submit nickname
|
||||
try:
|
||||
nickname_submitted = cmd[1]
|
||||
channel = str(fullcmd[2]).lower()
|
||||
# channel = str(fullcmd[2]).lower()
|
||||
uid_submitted = self.User.get_uid(nickname_submitted)
|
||||
user_submitted = self.User.get_User(nickname_submitted)
|
||||
|
||||
@@ -372,6 +392,9 @@ class Votekick():
|
||||
|
||||
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :{nickname_submitted} has been targeted for a vote')
|
||||
|
||||
self.Base.create_timer(60, self.timer_vote_verdict, (channel, ))
|
||||
self.Irc.send2socket(f':{dnickname} PRIVMSG {channel} :This vote will end after 60 secondes')
|
||||
|
||||
except KeyError as ke:
|
||||
self.Logs.error(f'Key Error: {ke}')
|
||||
except TypeError as te:
|
||||
@@ -380,7 +403,7 @@ class Votekick():
|
||||
case 'activate':
|
||||
try:
|
||||
# activate #channel
|
||||
channel = str(cmd[1]).lower()
|
||||
# channel = str(cmd[1]).lower()
|
||||
|
||||
self.insert_vote_channel(
|
||||
self.VoteChannelModel(
|
||||
@@ -402,7 +425,7 @@ class Votekick():
|
||||
case 'deactivate':
|
||||
try:
|
||||
# deactivate #channel
|
||||
channel = str(cmd[1]).lower()
|
||||
# channel = str(cmd[1]).lower()
|
||||
|
||||
self.Irc.send2socket(f":{dnickname} SAMODE {channel} -o {dnickname}")
|
||||
self.Irc.send2socket(f":{dnickname} PART {channel}")
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
{
|
||||
"version": "5.0.0"
|
||||
"version": "5.0.1"
|
||||
}
|
||||
Reference in New Issue
Block a user