WIP: Moderation type registry #26

Closed
cswimr wants to merge 146 commits from aurora-3rd-party into main
Showing only changes of commit e5cdd3893f - Show all commits

View file

@ -1,35 +1,36 @@
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional from typing import Dict, List, Optional
from discord import Guild
from pydantic import BaseModel from pydantic import BaseModel
from aurora.utilities.database import connect
class Moderation(BaseModel): class Moderation(BaseModel):
moderation_id: int moderation_id: int
timestamp: int guild_id: int
timestamp: datetime
moderation_type: str moderation_type: str
target_type: str target_type: str
target_id: int target_id: int
moderator_id: int moderator_id: int
role_id: Optional[int]
duration: Optional[timedelta]
end_timestamp: Optional[datetime]
reason: Optional[str]
resolved: bool resolved: bool
resolved_by: Optional[int]
resolve_reason: Optional[str]
expired: bool expired: bool
duration: str
end_timestamp: int
reason: str
changes: List[Dict] changes: List[Dict]
metadata: Dict metadata: Dict
resolved_by: Optional[int] = None
resolve_reason: Optional[str] = None
role_id: Optional[int] = None
def __str__(self): def __str__(self):
return f"{self.moderation_type} {self.target_type} {self.target_id} {self.reason}" return f"{self.moderation_type} {self.target_type} {self.target_id} {self.reason}"
@classmethod @classmethod
def from_sql(cls, moderation_id: int, guild: Guild): def from_sql(cls, moderation_id: int, guild_id: int):
query = f"SELECT * FROM moderation_{guild.id} WHERE moderation_id = ?;" from aurora.utilities.database import connect
query = f"SELECT * FROM moderation_{guild_id} WHERE moderation_id = ?;"
with connect() as database: with connect() as database:
cursor = database.cursor() cursor = database.cursor()
@ -37,9 +38,33 @@ class Moderation(BaseModel):
result = cursor.fetchone() result = cursor.fetchone()
if result: if result:
moderation_data = dict(zip(cls.model_fields, result)) if result[7] != "NULL":
moderation = cls(**moderation_data) hours, minutes, seconds = map(int, result[7].split(':'))
duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
else:
duration = None
case = {
"moderation_id": int(result[0]),
"guild_id": int(guild_id),
"timestamp": datetime.fromtimestamp(result[1]),
"moderation_type": str(result[2]),
"target_type": str(result[3]),
"target_id": int(result[4]),
"moderator_id": int(result[5]),
"role_id": int(result[6]) if result[6] != "0" else None,
"duration": duration,
"end_timestamp": datetime.fromtimestamp(result[8]) if result[8] != 0 else None,
"reason": result[9] if result[9] != "NULL" else None,
"resolved": bool(result[10]),
"resolved_by": result[11] if result[11] != "NULL" else None,
"resolve_reason": result[12] if result[12] != "NULL" else None,
"expired": bool(result[13]),
"changes": json.loads(result[14]),
"metadata": json.loads(result[15]),
}
cursor.close() cursor.close()
return moderation
return cls(**case)
return None return None