Sheldan/Sissi
1
0
Fork 0
mirror of https://github.com/Sheldan/Sissi.git synced 2026-01-26 19:21:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

[SIS-xxx] adding rendering of current Debra donation information

Browse Source 
updating for Debra 2023 campaign
adding internal rest api for debra information
adding a debra button to receive information
This commit is contained in:
Sheldan
2023-11-09 01:01:49 +01:00
parent 7449c05462
commit c6f20d617d
73 changed files with 1078 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
python/tools/migrations/warn-migration/dto.py Normal file
View File

@@ -0,0 +1,11 @@
class LegacyWarning:
level = 0
reason = ''
author_id = 0
user_id = 0
date = 0
duration = 0
until = 0
mod_log_message_id = 0
mod_log_channel_id = 0

32
python/tools/migrations/warn-migration/legacy_loader.py Normal file
View File

@@ -0,0 +1,32 @@
import json
from datetime import datetime
from dto import LegacyWarning
def load_all_warnings():
with open('settings.json') as warnings:
lines = warnings.read()
warnings_obj = json.loads(lines)
warned_users = warnings_obj['260']['MODLOGS']['297910194841583616']
all_warning_dtos = []
for user in warned_users:
warnings = warned_users[user]['x']
for warning in warnings:
warning_dto = LegacyWarning()
warning_dto.level = warning['level']
warning_dto.author_id = warning['author']
warning_dto.reason = warning['reason']
if 'duration' in warning and warning['duration'] is not None:
warning_dto.duration = int(float(warning['duration']))
if 'until' in warning:
warning_dto.until = warning['until']
warning_dto.date = datetime.fromtimestamp(int(warning['time']))
warning_dto.user_id = user
if 'modlog_message' in warning:
warning_dto.mod_log_channel_id = warning['modlog_message']['channel_id']
warning_dto.mod_log_message_id = warning['modlog_message']['message_id']
all_warning_dtos.append(warning_dto)
print(f'loaded {len(all_warning_dtos)} warnings.')
return all_warning_dtos

20
python/tools/migrations/warn-migration/main.py Normal file
View File

@@ -0,0 +1,20 @@
import os
import sqlalchemy as db
from legacy_loader import load_all_warnings
from warning_importer import import_warnings
server_id = os.getenv('SERVER_ID')
all_warnings = load_all_warnings()
db_host = os.getenv('DB_HOST')
db_port = os.getenv('DB_PORT')
db_database = os.getenv('DB_NAME')
db_user = os.getenv('DB_USER')
db_password = os.getenv('DB_PASS')
engine = db.create_engine('postgresql://%s:%s@%s:%s/%s' % (db_user, db_password, db_host, db_port, db_database))
with engine.connect() as con:
with con.begin():
import_warnings(server_id, all_warnings, con)

147
python/tools/migrations/warn-migration/warning_importer.py Normal file
View File

@@ -0,0 +1,147 @@
from datetime import timedelta
from sqlalchemy.sql import text
from dto import LegacyWarning
def import_warnings(server_id, warnings, connection):
user_ids = [warn.user_id for warn in warnings]
user_ids.extend([warn.author_id for warn in warnings])
create_users(server_id, user_ids, connection)
channel_ids = [warn.mod_log_channel_id for warn in warnings if warn.mod_log_channel_id is not None]
create_channels(server_id, channel_ids, connection)
warning_id = 0
mute_id = 0
for warning in warnings:
if warning.level == 5 or warning.level == 4:
import_ban(server_id, warning, connection)
if warning.level == 1:
warning_id += 1
import_warning(server_id, warning, connection, warning_id)
if warning.level == 2:
mute_id += 1
import_mute(server_id, warning, connection, mute_id)
if warning.level == 3:
import_kick(server_id, warning, connection)
connection.execute(f"INSERT INTO counter(counter, server_id, counter_key) VALUES ({warning_id}, {server_id}, 'WARNINGS')")
connection.execute(f"INSERT INTO counter(counter, server_id, counter_key) VALUES ({mute_id}, {server_id}, 'MUTES')")
def import_mute(server_id, warning: LegacyWarning, connection, mute_id):
statement = text("""INSERT INTO infraction(server_id, infraction_user_in_server_id, infraction_creator_user_in_server_id,
description, type, created, points, decayed)
VALUES(:server_id,
(select user_in_server_id from user_in_server where user_id = :muted_user_id and server_id = :server_id),
(select user_in_server_id from user_in_server where user_id = :author_id and server_id = :server_id),
:text, 'mute', :created, :points, false) returning id""")
infraction_id = connection.execute(statement, {'server_id': server_id, 'author_id': warning.author_id, 'muted_user_id': warning.user_id,
'text': warning.reason if warning.reason is not None else 'No reason provided', 'created': warning.date, 'points': 20}).fetchone()[0]
statement = text("""INSERT INTO infraction_parameter(key, value, infraction_id, created)
VALUES(:key, :value, :infraction_id, :created)""")
if warning.duration is not None:
duration = str(int(warning.duration / 60)) + 'm' if warning.duration < 60 * 60 else str(int(warning.duration / 60 / 60)) + 'h'
duration = duration if warning.duration < 60 * 60 * 24 else str(int(warning.duration / 60 / 60 / 24)) + 'd'
else:
duration = '0s'
connection.execute(statement, {'key': 'DURATION', 'value': duration, 'infraction_id': infraction_id, 'created': warning.date})
print(f'Creating infraction for mute for {warning.user_id}')
statement = text("""INSERT INTO mute(id, server_id, muted_user_in_server_id, muting_user_in_server_id,
reason, mute_date, target_date, mute_ended, created, infraction_id)
VALUES(:mute_id, :server_id,
(select user_in_server_id from user_in_server where user_id = :warned_user_id and server_id = :server_id),
(select user_in_server_id from user_in_server where user_id = :author_id and server_id = :server_id),
:text, :created, :target_date, true, :created, :infraction_id)""")
connection.execute(statement, {'mute_id': mute_id, 'server_id': server_id, 'author_id': warning.author_id, 'warned_user_id': warning.user_id,
'text': warning.reason if warning.reason is not None else 'No reason provided', 'created': warning.date, 'infraction_id': infraction_id,
'target_date': warning.date + timedelta(seconds=warning.duration)})
print(f'Creating warning for {warning.user_id}')
def import_warning(server_id, warning: LegacyWarning, connection, warning_id):
statement = text("""INSERT INTO infraction(server_id, infraction_user_in_server_id, infraction_creator_user_in_server_id,
description, type, created, points, decayed)
VALUES(:server_id,
(select user_in_server_id from user_in_server where user_id = :warned_user_id and server_id = :server_id),
(select user_in_server_id from user_in_server where user_id = :author_id and server_id = :server_id),
:text, 'warn', :created, :points, false) returning id""")
infraction_id = connection.execute(statement, {'server_id': server_id, 'author_id': warning.author_id, 'warned_user_id': warning.user_id,
'text': warning.reason if warning.reason is not None else 'No reason provided', 'created': warning.date, 'points': 10}).fetchone()[0]
print(f'Creating infraction for warning for {warning.user_id}')
statement = text("""INSERT INTO warning(id, server_id, warned_user_in_server_id, warning_user_in_server_id,
reason, warn_date, created, decayed, infraction_id)
VALUES(:warn_id, :server_id,
(select user_in_server_id from user_in_server where user_id = :warned_user_id and server_id = :server_id),
(select user_in_server_id from user_in_server where user_id = :author_id and server_id = :server_id),
:text, :created, :created, false, :infraction_id)""")
connection.execute(statement, {'warn_id': warning_id, 'server_id': server_id, 'author_id': warning.author_id, 'warned_user_id': warning.user_id,
'text': warning.reason if warning.reason is not None else 'No reason provided', 'created': warning.date, 'infraction_id': infraction_id})
print(f'Creating warning for {warning.user_id}')
def import_ban(server_id, warning: LegacyWarning, connection):
statement = text("""INSERT INTO infraction(server_id, infraction_user_in_server_id, infraction_creator_user_in_server_id,
description, type, created, points, decayed)
VALUES(:server_id,
(select user_in_server_id from user_in_server where user_id = :banned_user_id and server_id = :server_id),
(select user_in_server_id from user_in_server where user_id = :author_id and server_id = :server_id),
:text, 'ban', :created, :points, false)""")
connection.execute(statement, {'server_id': server_id, 'author_id': warning.author_id, 'banned_user_id': warning.user_id,
'text': warning.reason if warning.reason is not None else 'No reason provided', 'created': warning.date, 'points': 150})
print(f'Creating ban for {warning.user_id}')
def import_kick(server_id, warning: LegacyWarning, connection):
statement = text("""INSERT INTO infraction(server_id, infraction_user_in_server_id, infraction_creator_user_in_server_id,
description, type, created, points, decayed)
VALUES(:server_id,
(select user_in_server_id from user_in_server where user_id = :kicked_user_id and server_id = :server_id),
(select user_in_server_id from user_in_server where user_id = :author_id and server_id = :server_id),
:text, 'kick', :created, :points, false)""")
connection.execute(statement, {'server_id': server_id, 'author_id': warning.author_id, 'kicked_user_id': warning.user_id,
'text': warning.reason if warning.reason is not None else 'No reason provided', 'created': warning.date, 'points': 50})
print(f'Creating kick for {warning.user_id}')
def create_users(server_id: int, user_ids, con):
created_users = {}
for user_id in user_ids:
if not does_user_exist(user_id, con) and user_id not in created_users:
create_user(user_id, con)
create_user_in_server(user_id, server_id, con)
created_users[user_id] = 1
def does_user_exist(user_id, con):
statement = text("""SELECT count(1) FROM auser where id = :id""")
return con.execute(statement, {'id': user_id}).fetchone()[0] == 1
def create_user(user_id, con):
statement = text("""INSERT INTO auser(id) VALUES(:id)""")
print(f'Creating user {user_id}')
con.execute(statement, {'id': user_id})
def create_channels(server_id: int, channel_ids, con):
channels_to_create = {}
for channel_id in channel_ids:
if not does_channel_exist(channel_id, con):
channels_to_create[channel_id] = 1
for channel_id in channels_to_create:
create_channel(channel_id, server_id, con)
def does_channel_exist(channel_id, con):
statement = text("""SELECT count(1) FROM channel where id = :id""")
return con.execute(statement, {'id': channel_id}).fetchone()[0] == 1
def create_channel(channel_id, server_id, con):
statement = text("""INSERT INTO channel(id, server_id, type, deleted) VALUES(:id, :server_id, 'TEXT', false)""")
print(f'Creating channel {channel_id}')
con.execute(statement, {'id': channel_id, 'server_id': server_id})
def create_user_in_server(user_id, server_id, con):
statement = text("""INSERT INTO user_in_server(server_id, user_id) VALUES(:server_id, :user_id) returning user_in_server_id""")
return con.execute(statement, {'user_id': user_id, 'server_id': server_id}).fetchone()[0]