Ttykuu

I'm posting here to ask for some help improving myself using python since I'm fairly new to it and come from C#.

The main functionality of this bot is parsing the Wargaming API for player related information.

Project structure:

project
| main.py
└───ENV
│ 
└───data
 | classes.py
 | dbcontext.py
 | getStats.py

My Main:

import asyncio
import logging
import os
import sys
import re
import json

import discord
from discord.ext import commands

from data import dbcontext, log, secret, getStats as GetStats
from data.translations import en, de, pl ,tr
from data.classes import Config, Ship, Player, Stats, ReturnVal, ErrorType

sys.path.append(os.path.join(os.path.dirname(__file__), "data"))


"""
WoWs-Stats Bot.
Creator Fuyune

Uses Discord.py framework by Rapptz
"""

####### Bot Basic Configuration #######

bot = commands.Bot(command_prefix="!")
logger = logging.getLogger('discord')
logger.setLevel(logging.ERROR)
handler = logging.FileHandler(filename='err.log', encoding='utf-8', mode='a')
handler.setFormatter(logging.Formatter(
 '%(asctime)s:%(levelname)s:%(name)s: %(message)s'))
logger.addHandler(handler)
bot.remove_command('help')
configs = []
regions = ["eu","ru","na","asia"]
langs = ["de","en","pl","tr"] #remember adding to imports too
amounts = 

####### Bot Events #######

@bot.event
async def on_ready():
 log.writeLog("init", "Ready")
 print("Bot Started")
 configs = []
 for x in bot.guilds:
 configs.append(dbcontext.getConfig(x.id))
 print("Connected to: " + x.name)
 log.writeLog("Connected", x.name)


@bot.command()
async def stats(ctx, *args):
 if ctx.message.guild is None:
 await ctx.send("**I'm not allowed to answer in Private messages.**")
 else:
 config = dbcontext.getConfig(ctx.guild.id)
 if (len(args) == 0) or (len(args) == 1 and args[0].lower() == "help") :
 await writeHelp(ctx,config)
 elif len(args) == 1 and args[0] != "help":
 playerObject = GetStats.getPlayer(config,args[0])
 if playerObject == None or playerObject.id == 0 :
 await writeError(ctx,config,ErrorType.UNKNOWN_PLAYER)
 else:
 playerStats = GetStats.getPlayerStats(config,playerObject)
 if playerStats.hidden:
 await writeError(ctx,config,ErrorType.HIDDEN_STATS)
 elif playerStats.damage == 0:
 await writeError(ctx,config,ErrorType.UNKNOWN_STATS)
 else:
 await writeAnswer(ctx,config,playerObject,playerStats)
 elif len(args) >= 2:
 playerObject = GetStats.getPlayer(config,args[0])
 if playerObject == None or playerObject.id == 0:
 await writeError(ctx,config,ErrorType.UNKNOWN_PLAYER)
 else:
 ship = dbcontext.getShip(" ".join(args[1:]))
 if ship.id == 0:
 await writeError(ctx,config,ErrorType.UNKNOWN_SHIP)
 else:
 shipStats = GetStats.getShipStats(config,playerObject,ship)
 if shipStats.hidden:
 await writeError(ctx,config,ErrorType.HIDDEN_STATS)
 elif shipStats.damage == 0:
 await writeError(ctx,config,ErrorType.UNKNOWN_STATS)
 else:
 await writeAnswer(ctx,config,playerObject,ship,shipStats)


@bot.command()
async def statsr(ctx, *args):
 if ctx.message.guild is None:
 await ctx.send("**I'm not allowed to answer in Private messages.**")
 else:
 config = dbcontext.getConfig(ctx.guild.id)
 regex = re.compile("^(![1-9][1]|![1-9])|(!s[1-4])$", re.IGNORECASE) # regex to check for valid season
 if (len(args) == 0) or (len(args) == 1 and args[0].lower() == "help") :
 await writeHelp(ctx,config)
 elif regex.match(args[0]):
 await writeError(ctx,config,ErrorType.UNKNOWN_SEASON)
 elif len(args) == 2 and args[0] != "help":
 season = convertSeason(args[0])
 playerObject = GetStats.getPlayer(config,args[1])
 if playerObject == None or playerObject.id == 0 :
 await writeError(ctx,config,ErrorType.UNKNOWN_PLAYER)
 else:
 playerStats = GetStats.getRankedStats(config,playerObject,season)
 if playerStats.hidden:
 await writeError(ctx,config,ErrorType.HIDDEN_STATS)
 elif playerStats.damage == 0:
 await writeError(ctx,config,ErrorType.UNKNOWN_STATS)
 else:
 await writeAnswer(ctx,config,playerObject,playerStats)
 elif len(args) > 2:
 season = convertSeason(args[0])
 playerObject = GetStats.getPlayer(config,args[1])
 if playerObject == None or playerObject.id == 0:
 await writeError(ctx,config,ErrorType.UNKNOWN_PLAYER)
 else:
 ship = dbcontext.getShip(" ".join(args[2:]))
 if ship.id == 0:
 await writeError(ctx,config,ErrorType.UNKNOWN_SHIP)
 else:
 shipStats = GetStats.getRankedStats(config,playerObject,season,ship)
 if shipStats.hidden:
 await writeError(ctx,config,ErrorType.HIDDEN_STATS)
 elif shipStats.damage == 0:
 await writeError(ctx,config,ErrorType.UNKNOWN_STATS)
 else:
 await writeAnswer(ctx,config,playerObject,ship,shipStats)
 else:
 await writeHelp(ctx,config)

####### Helper Functions #######

async def writeAnswer(ctx,config,*args):
 player = None
 ship = None
 stats = None
 embed = None
 translation = globals()[config.language]
 for arg in args: 
 if isinstance(arg,Player):
 player=arg
 elif isinstance(arg,Ship):
 ship=arg
 elif isinstance(arg,Stats):
 stats=arg
 if player != None and stats != None and ship != None:
 color = getColor(stats.avgWins)
 embed=discord.Embed(title=translation.title.format(username=player.name),url=GetStats.getPlayerLink(config,player), description=translation.description.format(shipname=ship.name), color=color)
 embed.set_author(name="WoWs-Stats-Bot",url="https://github.com/De-Wohli")
 embed.set_thumbnail(url=ship.url)
 embed.add_field(name=translation.battles, value=stats.battles, inline=True)
 embed.add_field(name=translation.avgDamage, value=stats.avgDamage, inline=True)
 embed.add_field(name=translation.winrate, value=":.2f%".format(stats.avgWins), inline=True)
 embed.set_footer(text=translation.footer)
 elif player != None and stats != None and ship == None:
 color = getColor(stats.avgWins)
 embed=discord.Embed(title=translation.title.format(username=player.name),url=GetStats.getPlayerLink(config,player), description=translation.general, color=color)
 embed.set_author(name="WoWs-Stats-Bot",url="https://github.com/De-Wohli")
 embed.add_field(name=translation.battles, value=stats.battles, inline=True)
 embed.add_field(name=translation.avgDamage, value=stats.avgDamage, inline=True)
 embed.add_field(name=translation.winrate, value=":.2f%".format(stats.avgWins), inline=True)
 embed.set_footer(text=translation.footer)
 if embed != None:
 await ctx.send(embed=embed)


async def writeHelp(ctx,config):
 color = discord.Color.teal()
 translation = globals()[config.language]
 embed=discord.Embed(title=translation.helpHeader, description=translation.helpDescription, color=color)
 embed.set_author(name="WoWs-Stats-Bot")
 embed.add_field(name="!stats [player]",value=translation.helpPlayer,inline=False)
 embed.add_field(name="!stats [player] [shipname]",value=translation.helpShip,inline=False)
 embed.add_field(name="!statsr [season] [player]",value=translation.helpRanked,inline=False)
 embed.add_field(name="!statsr [season] [player] [shipname]",value=translation.helpSRanked,inline=False)
 embed.set_footer(text="This bot was made by Fuyu_Kitsune")
 await ctx.send(embed=embed)


async def writeError(ctx,config,errorType):
 color = discord.Color.dark_teal()
 translation = globals()[config.language]
 errorText = translation.error[errorType.value]
 embed=discord.Embed(title="Error", description=errorText, color=color)
 embed.set_author(name="WoWs-Stats-Bot")
 embed.set_footer(text=translation.footer)
 await ctx.send(embed=embed)


def getColor(value):
 if value <= 40:
 return discord.Colour.red()
 elif value > 40 and value <= 45:
 return discord.Colour.orange()
 elif value > 45 and value <= 50:
 return discord.Colour.gold()
 elif value > 50 and value <= 53:
 return discord.Colour.green()
 elif value > 53 and value <= 56:
 return discord.Color.dark_green()
 elif value > 56 and value <= 60:
 return discord.Color.teal()
 elif value > 60 and value <= 66:
 return discord.Color.purple()
 elif value > 66:
 return discord.Colour.dark_purple()


def convertSeason(value):
 season = value
 if season == "s1":
 season = "101"
 elif season == "s2":
 season = "102"
 elif season == "s3":
 season = "103"
 elif season == "s4":
 season = "104"
 return season

This being the main method, here is where the parsing of the command takes place. The usual command is !stats [playername] [shipname] My primary concerns here are the async def stats(ctx, *args): and async def statsr(ctx, *args): functions.

data/classes.py:

from enum import Enum


class Ship:
 def __init__(self, id=0, name="", url=""):
 self.id = id
 self.name = name
 self.url = url

 def __eq__(self, other):
 return self.name == other


class Player:
 def __init__(self, id=0, name="", code="404"):
 self.id = id
 self.name = name
 self.code = code


class Stats:
 def __init__(self, battles=0, frags=0, damage_dealt=0, wins=0, hidden=False, code=404):
 self.hidden = hidden
 self.battles = battles
 self.frags = float(frags)
 self.damage = float(damage_dealt)
 self.wins = wins
 self.code = code

 @property
 def avgFrags(self):
 if self.battles == 0:
 return 0
 return round(self.frags / self.battles, 2)

 @property
 def avgDamage(self):
 if self.battles == 0:
 return 0
 return round(self.damage / self.battles, 2)

 @property
 def avgWins(self):
 if self.battles == 0:
 return 0
 return round(float(self.wins / self.battles), 4)*100


class Config:
 def __init__(self, serverId=0, region="eu", language="en"):
 self.serverId = serverId
 self.region = region
 self.language = language


class ReturnVal(Enum):
 SUCCESS = 0
 FAILED = 1
 DOUBLE = 2

class ErrorType(Enum):
 def __str__(self):
 return str(self.value)

 UNKNOWN_PLAYER = 0
 UNKNOWN_SHIP = 1
 UNKNOWN_STATS = 2
 HIDDEN_STATS = 3
 UNKNOWN_SEASON = 4

these are my model classes, since I'm coming from C# I'm not too sure if this would be an acceptable way of dealing with it in python.

data/dbcontext.py:

import sys
import os
import mysql.connector

import data.log as log
from data.classes import Config, ReturnVal, Ship
from data.secret import Secret


def connect():
 mydb = mysql.connector.connect(host=Secret.dbAddr,user=Secret.dbUser,passwd=Secret.dbPwd,database=Secret.dbName)
 return mydb


def getShip(name):
 try:
 con=connect()
 cursor = con.cursor()
 sql = 'SELECT id,Name,url FROM Ships WHERE name LIKE %s'
 val = (name,)
 cursor.execute(sql,val)
 rows = cursor.fetchone()
 if rows is None:
 sql = 'SELECT id,Name,url FROM Ships WHERE id = (SELECT id FROM Asn WHERE name LIKE %s)'
 val = (name,)
 cursor.execute(sql,val)
 rows = cursor.fetchone()
 if rows is None:
 return Ship()
 else:
 return Ship(id=rows[0], name=rows[1], url=rows[2])
 else:
 return Ship(id=rows[0], name=rows[1], url=rows[2])
 except Exception as e:
 log.writeLog("getShip", str(e))
 con.rollback()
 return Ship()
 finally:
 con.commit()
 con.close()



def addAsn(name, asn):
 try:
 con=connect()
 cursor = con.cursor()
 sql = 'SELECT id,Name,url FROM Ships WHERE name LIKE %s'
 val = (name,)
 cursor.execute(sql,val)
 rows = cursor.fetchone()
 if rows is None:
 return ReturnVal.FAILED
 else:
 sql = 'INSERT INTO Asn (name,id,url) VALUES(%s,%s,%s)'
 val = (asn,rows[0],rows[2])
 cursor.execute(sql,val)
 return ReturnVal.SUCCESS
 except mysql.connector.IntegrityError as e:
 return ReturnVal.DOUBLE
 except Exception as e:
 log.writeLog("getShip", str(e))
 con.rollback()
 return ReturnVal.FAILED
 finally:
 con.commit()
 con.close()

def getConfig(id):
 try:
 con = connect()
 cursor = con.cursor()
 sql = 'SELECT region,language FROM Config WHERE ServerId = %s'
 val = (id,)
 cursor.execute(sql,val)
 rows = cursor.fetchone()
 if rows is None:
 config = Config(serverId=id)
 sql = 'INSERT INTO Config(ServerID,region,language) VALUES(%s,%s,%s)'
 val = (config.serverId, config.region, config.language)
 cursor.execute(sql,val)
 return config
 else:
 return Config(serverId=id, region=rows[0], language=rows[1])
 except Exception as e:
 log.writeLog("getConfig", str(e))
 con.rollback()
 finally:
 con.commit()
 con.close()

def addConfig(id):
 try:
 con=connect()
 cursor = con.cursor()
 sql = 'INSERT INTO Config (ServerId, region, language) VALUES(%s,%s,%s)'
 val = (id,"eu","en")
 cursor.execute(sql,val)
 return ReturnVal.SUCCESS
 except Exception as e:
 log.writeLog("addConfig", str(e))
 con.rollback()
 return ReturnVal.FAILED
 finally:
 con.commit()
 con.close()

def updateConfig(config):
 try:
 con=connect()
 cursor = con.cursor()
 sql = 'UPDATE Config SET region = %s, language = %s WHERE ServerID = %s'
 val = (config.region, config.language, config.serverId)
 cursor.execute(sql,val)
 return ReturnVal.SUCCESS
 except Exception as e:
 log.writeLog("updateConfig", str(e))
 con.rollback()
 return ReturnVal.FAILED
 finally:
 con.commit()
 con.close()

In this file I'm handling all database access. The database stores mainly Ship names & IDs. Is this way of handling the database connection secure or is this vulnerable?

data/getStats.py:

import os
import sys
import requests
import json

sys.path.append(os.path.join(os.path.dirname(__file__), "lib"))
sys.path.append(os.path.join(os.path.dirname(__file__),"../data"))
from data.classes import Ship, Player, Stats
from data.secret import Secret
from data.api import api
from data import log


def getPlayer(config,playerName):
 try:
 url = api.psearch.format(reg=config.region,wgapi=Secret.api,playerName=playerName)
 response = requests.get(url)
 statuscode = response.status_code
 response = response.json()
 if response["status"] == "ok":
 if response["meta"]["count"] == 0:
 return Player(code=200)
 else:
 nick = response["data"][0]["nickname"]
 pid = response["data"][0]["account_id"]
 newPlayer = Player(name = nick, id = pid,code = statuscode)
 return newPlayer
 else:
 return Player(code=statuscode)
 except Exception as e:
 print(str(e))
 log.writeLog("getPlayer",str(e))
 return Player(code=200)


def getPlayerLink(config,player):
 link = str.format("-",str(api.plink).format(reg=config.region),player.id,player.name)
 return link


def getPlayerStats(config,player):
 try:
 url = api.pstats.format(reg=config.region,wgapi=Secret.api,accountID=player.id)
 response = requests.get(url)
 statuscode = response.status_code
 response = response.json()
 if(response["status"] == "ok"):
 if bool(response["data"]):
 if bool(response["meta"]["hidden"]):
 st = Stats(1,1,1,1,True,200)
 else:
 battles = response["data"][str(player.id)]["statistics"]['pvp']['battles']
 wins = response["data"][str(player.id)]["statistics"]['pvp']['wins']
 frags = response["data"][str(player.id)]["statistics"]['pvp']['frags']
 damage_dealt= response["data"][str(player.id)]["statistics"]['pvp']['damage_dealt']
 st = Stats(battles,frags,damage_dealt,wins,False,statuscode)
 else:
 return Stats(code=200)
 return st
 else:
 return Stats(code=statuscode)
 except Exception as e:
 print(str(e))
 log.writeLog("getPlayerStats",str(e))
 return Stats(code=200)


def getShipStats(config,player,ship):
 try:
 url = api.sstats.format(reg=config.region,wgapi=Secret.api,accountID=player.id,shipID=ship.id)
 response = requests.get(url)
 statuscode = response.status_code
 response = response.json()
 if(response["status"] == "ok"):
 if bool(response["meta"]["hidden"]):
 st = Stats(1,1,1,1,True,200)
 elif bool(response["data"]) and not response["data"][str(player.id)] == None :
 battles = response["data"][str(player.id)][0]['pvp']['battles']
 wins = response["data"][str(player.id)][0]['pvp']['wins']
 frags = response["data"][str(player.id)][0]['pvp']['frags']
 damage_dealt= response["data"][str(player.id)][0]['pvp']['damage_dealt']
 st = Stats(battles,frags,damage_dealt,wins,False,statuscode)
 else:
 return Stats(code=200)
 return st
 else:
 return Stats(code=statuscode)
 except Exception as e:
 print(str(e))
 log.writeLog("getShipStats",str(e))
 return Stats(code=200)


def getRankedStats(config,player,season,ship = None):
 try:
 if ship is not None:
 url = api.rsstats.format(reg=config.region,wgapi=Secret.api,accountID=player.id,shipID = ship.id)
 else:
 url = api.rpstats.format(reg=config.region,wgapi=Secret.api,accountID=player.id)
 response = requests.get(url)
 statuscode = response.status_code
 response = response.json()
 stats = Stats()
 if(response["status"] == "ok"):
 if bool(response["meta"]["hidden"]):
 stats = Stats(1,1,1,1,True,200)
 elif bool(response["data"]) and not response["data"][str(player.id)] == None :
 if ship is not None:
 seasons = response["data"][str(player.id)][0]["seasons"]
 if season in seasons:
 currentSeason = response["data"][str(player.id)][0]["seasons"][season]
 else:
 return Stats(code=200)
 else:
 seasons = response["data"][str(player.id)]["seasons"]
 if season in seasons:
 currentSeason = response["data"][str(player.id)]["seasons"][season]
 else:
 return Stats(code=200)
 r = []
 r.append(currentSeason["rank_solo"])
 r.append(currentSeason["rank_div2"])
 r.append(currentSeason["rank_div3"])
 for x in r:
 if x is not None:
 stats.wins += x["wins"]
 stats.damage += x["damage_dealt"]
 stats.battles += x["battles"]
 stats.frags += x["frags"]
 stats.code = 200
 else:
 return Stats(code=200)
 return stats
 else:
 return Stats(code=statuscode)
 pass
 except Exception as e:
 log.writeLog("getRankedStats",str(e))
 pass

this is the communication class for querying the API endpoints. The endpoints are stored in a different file data/api.py containing a class with the variables stored. i.e.

class api:
 psearch = "https://api.worldofwarships.reg/wows/account/list/?application_id=wgapi&search=playerName"

The code as shown here is currently working and being hosted on a linux server without too many troubles yet but I'm curious to know about the, probably many things, i could improve.
If needed I can provide the answers I'm getting from the API if that would be of any concern to the code I've posted.

In general I'm seeking assistance to improve my python programming aswell as improve performance / stability of my code. Any information aswell as critique is welcome.

Thank you.