build.cogs.databasecog.DatabaseCog.db_add_update_spots() - Code Metrics - Inspection of "New functionalities (#17)" - BonHowi/discordPogmareBot - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( d12a1e...942891 )

by Bartosz

created 2021-09-24 15:53 UTC

DatabaseCog.db_add_update_spots() A

↳ Parent: build.cogs.databasecog

Complexity

Conditions

Size

Total Lines	7
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	7
dl	0
loc	7
rs	10
c	0
b	0
f	0
cc	1
nop	3

import os

import discord
from discord.ext import commands, tasks
from discord_slash import cog_ext, SlashContext

from modules.get_settings import get_settings
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey, \
    BigInteger, update, select, DateTime, delete
from sqlalchemy.dialects.mysql import insert
import cogs.cogbase as cogbase
from datetime import datetime
import pandas as pd

metadata_obj = MetaData()
member = Table('member', metadata_obj,
               Column('id', BigInteger, primary_key=True),
               Column('name', String(50), nullable=False),
               Column('display_name', String(50), nullable=False)
               )
fk_member_id = "member.id"

spots = Table('spots', metadata_obj,
              Column('member_id', BigInteger, ForeignKey(fk_member_id), primary_key=True),
              Column('legendary', Integer, default=0),
              Column('rare', Integer, default=0),
              Column('common', Integer, default=0),
              Column('event1', Integer, default=0),
              Column('event2', Integer, default=0)
              )

spots_temp = Table('spots_temp', metadata_obj,
                   Column('member_id', BigInteger, ForeignKey(fk_member_id), primary_key=True),
                   Column('legendary', Integer, default=0),
                   Column('rare', Integer, default=0),
                   Column('common', Integer, default=0),
                   Column('event1', Integer, default=0),
                   Column('event2', Integer, default=0)
                   )

warn = Table('warn', metadata_obj,
             Column('id', Integer, primary_key=True),
             Column('member_id', BigInteger, ForeignKey(fk_member_id)),
             Column('reason', String(120), nullable=False),
             Column('date', DateTime, nullable=False),
             )

coords = Table('coords', metadata_obj,
               Column('id', Integer, primary_key=True),
               Column('coords', String(100), nullable=False),
               Column('monster_type', String(20), nullable=False)
               )


class DatabaseCog(cogbase.BaseCog):
    user = get_settings("DB_U")
    password = get_settings("DB_P")
    engine = create_engine(f"mysql+mysqldb://{user}:{password}@localhost/server_database?charset=utf8mb4")
    metadata_obj.create_all(engine)
    conn = None

    def __init__(self, base):
        super().__init__(base)

        # Connect to database
        self.engine = DatabaseCog.engine
        metadata_obj.create_all(self.engine)
        self.db_update_loop.start()
        self.conn = DatabaseCog.conn

    def cog_unload(self):
        self.db_update_loop.cancel()

    # ----- BASE DATABASE OPERATIONS -----

    # Add or update member in member table
    def db_add_update_member(self, _member):
        self.conn = self.engine.connect()
        stmt = insert(member).values(
            id=_member.id, name=_member.name,
            display_name=_member.display_name)
        do_update_stmt = stmt.on_duplicate_key_update(
            name=stmt.inserted.name, display_name=stmt.inserted.display_name
        )
        self.conn.execute(do_update_stmt)
        self.conn.close()

    # Add or update spots in spots table
    def db_add_update_spots(self, spots_table, guild_member):
        self.conn = self.engine.connect()
        stmt = insert(spots_table).values(
            member_id=guild_member.id)
        do_update_stmt = stmt.on_duplicate_key_update(member_id=stmt.inserted.member_id)
        self.conn.execute(do_update_stmt)
        self.conn.close()

    # Add or refresh all guild members and spots to database
    async def db_update(self):
        self.conn = self.engine.connect()
        guild = self.bot.get_guild(self.bot.guild[0])
        dt_string = self.bot.get_current_time()
        print(f"({dt_string})\t[{self.__class__.__name__}]: Refreshing member and spots tables")
        for guild_member in guild.members:
            # Member tables
            self.db_add_update_member(guild_member)
            # Spots tables
            self.db_add_update_spots(spots, guild_member)
            self.db_add_update_spots(spots_temp, guild_member)
        dt_string = self.bot.get_current_time()
        print(f"({dt_string})\t[{self.__class__.__name__}]: Member and spots tables refreshed")
        self.conn.close()

    @tasks.loop(hours=12)
    async def db_update_loop(self):
        await self.db_update()
        await self.db_backup_database()

    @db_update_loop.before_loop
    async def before_db_update_loop(self):
        dt_string = self.bot.get_current_time()
        print(f'({dt_string})\t[{self.__class__.__name__}]: Waiting until Bot is ready')
        await self.bot.wait_until_ready()

    # Add member to database on member join
    @commands.Cog.listener()
    async def on_member_join(self, _member):
        self.db_add_update_member(_member)
        self.db_add_update_spots(spots, _member)
        self.db_add_update_spots(spots_temp, _member)

    # Backup database
    async def db_backup_database(self):
        now = datetime.now()
        cmd = f"mysqldump -u {get_settings('DB_U')} " \
              f"--result-file=database_backup/backup-{now.strftime('%m-%d-%Y')}.sql " \
              f"-p{get_settings('DB_P')} server_database"
        os.system(cmd)
        dt_string = self.bot.get_current_time()
        print(f'({dt_string})\t[{self.__class__.__name__}]: Database backed up')

    # ----- SPOTTING OPERATIONS -----

    # Update spots tables
    @classmethod
    async def db_count_spot(cls, _id: int, monster_type: str):
        cls.conn = cls.engine.connect()
        cls.db_count_spot_table(spots, _id, monster_type)
        cls.db_count_spot_table(spots_temp, _id, monster_type)
        cls.conn.close()

    @classmethod
    def db_count_spot_table(cls, table, _id: int, monster_type: str):
        stmt = select(table.c.member_id, table.c.legendary, table.c.rare, table.c.common,
                      table.c.event1,
                      table.c.event2).where(
            table.c.member_id == _id)
        result = cls.conn.execute(stmt)
        counter = 0
        for nr_of_kills in result.columns(monster_type, 'legendary'):
            counter = nr_of_kills[0]
        if monster_type == "event1":
            values = cls.db_count_spot_table_event(table, _id, monster_type, counter)
        else:
            values = {f"{monster_type}": counter + 1}
        stmt = update(table).where(table.c.member_id == _id).values(values)
        cls.conn.execute(stmt)

    @classmethod
    def db_count_spot_table_event(cls, table, _id, monster_type: str, counter: int):
        stmt = select(table.c.member_id, table.c.legendary, table.c.rare, table.c.common,
                      table.c.event1,
                      table.c.event2).where(
            table.c.member_id == _id)
        result = cls.conn.execute(stmt)
        counter_leg = 0
        for nr_of_kills_leg in result.columns('legendary'):
            counter_leg = nr_of_kills_leg[0]
        values = {f"{monster_type}": counter + 1, "legendary": counter_leg + 1}
        return values

    # Save coords from spotting channels to database
    @classmethod
    async def db_save_coords(cls, _coords: str, _monster_type):
        cls.conn = cls.engine.connect()
        stmt = insert(coords).values(coords=_coords, monster_type=_monster_type)
        cls.conn.execute(stmt)
        cls.conn.close()

    # Clear data from spots_temp table(for events etc)
    @classmethod
    async def db_clear_spots_temp_table(cls):
        cls.conn = cls.engine.connect()
        stmt = delete(spots_temp)
        cls.conn.execute(stmt)
        cls.conn.close()

    # ----- LEADERBOARD OPERATIONS -----

    # Return all members' spots
    @classmethod
    async def db_get_spots_df(cls):
        cls.conn = cls.engine.connect()
        stmt = select(spots.c.member_id, member.c.display_name, spots.c.legendary, spots.c.rare,
                      spots.c.common, spots.c.event1, spots.c.event2
                      ).select_from(member
                                    ).join(spots, member.c.id == spots.c.member_id)
        cls.conn.execute(stmt)
        df = pd.read_sql(stmt, cls.conn)
        cls.conn.close()
        return df

    # ----- WARN OPERATIONS -----

    # Add member's warn to database
    @classmethod
    async def db_add_warn(cls, _member: int, _reason: str):
        cls.conn = cls.engine.connect()
        stmt = insert(warn).values(member_id=_member, reason=_reason, date=datetime.now())
        cls.conn.execute(stmt)
        cls.conn.close()

    # Get member's warns from database
    @classmethod
    async def db_get_warns(cls, _member: int):
        cls.conn = cls.engine.connect()
        stmt = select(warn.c.reason, warn.c.date).select_from(member).join(warn, member.c.id == warn.c.member_id).where(
            member.c.id == _member)
        result = cls.conn.execute(stmt)
        date_warn = []
        counter = 0
        for warns in result.columns("reason", "date"):
            reason_with_date = [warns[1], warns[0]]
            date_warn.append(reason_with_date)
            counter += 1
        warns_list = [': \t'.join([str(elem) for elem in sublist]) for sublist in date_warn]
        cls.conn.close()
        return warns_list, counter

    # Remove all member's warns
    @classmethod
    async def db_remove_warns(cls, _member: int):
        cls.conn = cls.engine.connect()
        stmt = delete(warn).where(warn.c.member_id == _member)
        cls.conn.execute(stmt)
        cls.conn.close()

    # ----- MEMBER OPERATIONS -----

    # Return member's spots
    @classmethod
    async def db_get_member_stats(cls, _member: int):
        cls.conn = cls.engine.connect()
        stmt = select(member.c.display_name, spots.c.legendary, spots.c.rare, spots.c.common).select_from(member).join(
            spots,
            member.c.id == spots.c.member_id).where(spots.c.member_id == _member)
        df = pd.read_sql(stmt, cls.conn)
        cls.conn.close()
        return df

    @cog_ext.cog_slash(name="changeMemberSpots", guild_ids=cogbase.GUILD_IDS,
                       description=" ",
                       default_permission=False,
                       permissions=cogbase.PERMISSION_MODS)
    async def change_member_spots(self, ctx: SlashContext, user: discord.Member, spot_type: str, number: int):
        self.conn = self.engine.connect()
        stmt = f"""UPDATE server_database.spots SET {spot_type} = {number} """ \
               f"""WHERE (member_id = {user.id});"""
        self.conn.execute(stmt)
        await ctx.send(f"{user.display_name} spots changed", hidden=True)
        self.conn.close()

    # ----- COORDS OPERATIONS -----

    # Return coords
    @classmethod
    async def db_get_coords(cls):
        cls.conn = cls.engine.connect()
        stmt = select(coords.c.id, coords.c.coords, coords.c.monster_type).select_from(coords)
        df = pd.read_sql(stmt, cls.conn)
        cls.conn.close()
        return df


def setup(bot: commands.Bot):
    bot.add_cog(DatabaseCog(bot))


1			import os
2
3			import discord
4			from discord.ext import commands, tasks
5			from discord_slash import cog_ext, SlashContext
6
7			from modules.get_settings import get_settings
8			from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey, \
9			BigInteger, update, select, DateTime, delete
10			from sqlalchemy.dialects.mysql import insert
11			import cogs.cogbase as cogbase
12			from datetime import datetime
13			import pandas as pd
14
15			metadata_obj = MetaData()
16			member = Table('member', metadata_obj,
17			Column('id', BigInteger, primary_key=True),
18			Column('name', String(50), nullable=False),
19			Column('display_name', String(50), nullable=False)
20			)
21			fk_member_id = "member.id"
22
23			spots = Table('spots', metadata_obj,
24			Column('member_id', BigInteger, ForeignKey(fk_member_id), primary_key=True),
25			Column('legendary', Integer, default=0),
26			Column('rare', Integer, default=0),
27			Column('common', Integer, default=0),
28			Column('event1', Integer, default=0),
29			Column('event2', Integer, default=0)
30			)
31
32			spots_temp = Table('spots_temp', metadata_obj,
33			Column('member_id', BigInteger, ForeignKey(fk_member_id), primary_key=True),
34			Column('legendary', Integer, default=0),
35			Column('rare', Integer, default=0),
36			Column('common', Integer, default=0),
37			Column('event1', Integer, default=0),
38			Column('event2', Integer, default=0)
39			)
40
41			warn = Table('warn', metadata_obj,
42			Column('id', Integer, primary_key=True),
43			Column('member_id', BigInteger, ForeignKey(fk_member_id)),
44			Column('reason', String(120), nullable=False),
45			Column('date', DateTime, nullable=False),
46			)
47
48			coords = Table('coords', metadata_obj,
49			Column('id', Integer, primary_key=True),
50			Column('coords', String(100), nullable=False),
51			Column('monster_type', String(20), nullable=False)
52			)
53
54
55			class DatabaseCog(cogbase.BaseCog):
56			user = get_settings("DB_U")
57			password = get_settings("DB_P")
58			engine = create_engine(f"mysql+mysqldb://{user}:{password}@localhost/server_database?charset=utf8mb4")
59			metadata_obj.create_all(engine)
60			conn = None
61
62			def __init__(self, base):
63			super().__init__(base)
64
65			# Connect to database
66			self.engine = DatabaseCog.engine
67			metadata_obj.create_all(self.engine)
68			self.db_update_loop.start()
69			self.conn = DatabaseCog.conn
70
71			def cog_unload(self):
72			self.db_update_loop.cancel()
73
74			# ----- BASE DATABASE OPERATIONS -----
75
76			# Add or update member in member table
77			def db_add_update_member(self, _member):
78			self.conn = self.engine.connect()
79			stmt = insert(member).values(
80			id=_member.id, name=_member.name,
81			display_name=_member.display_name)
82			do_update_stmt = stmt.on_duplicate_key_update(
83			name=stmt.inserted.name, display_name=stmt.inserted.display_name
84			)
85			self.conn.execute(do_update_stmt)
86			self.conn.close()
87
88			# Add or update spots in spots table
89			def db_add_update_spots(self, spots_table, guild_member):
90			self.conn = self.engine.connect()
91			stmt = insert(spots_table).values(
92			member_id=guild_member.id)
93			do_update_stmt = stmt.on_duplicate_key_update(member_id=stmt.inserted.member_id)
94			self.conn.execute(do_update_stmt)
95			self.conn.close()
96
97			# Add or refresh all guild members and spots to database
98			async def db_update(self):
99			self.conn = self.engine.connect()
100			guild = self.bot.get_guild(self.bot.guild[0])
101			dt_string = self.bot.get_current_time()
102			print(f"({dt_string})\t[{self.__class__.__name__}]: Refreshing member and spots tables")
103			for guild_member in guild.members:
104			# Member tables
105			self.db_add_update_member(guild_member)
106			# Spots tables
107			self.db_add_update_spots(spots, guild_member)
108			self.db_add_update_spots(spots_temp, guild_member)
109			dt_string = self.bot.get_current_time()
110			print(f"({dt_string})\t[{self.__class__.__name__}]: Member and spots tables refreshed")
111			self.conn.close()
112
113			@tasks.loop(hours=12)
114			async def db_update_loop(self):
115			await self.db_update()
116			await self.db_backup_database()
117
118			@db_update_loop.before_loop
119			async def before_db_update_loop(self):
120			dt_string = self.bot.get_current_time()
121			print(f'({dt_string})\t[{self.__class__.__name__}]: Waiting until Bot is ready')
122			await self.bot.wait_until_ready()
123
124			# Add member to database on member join
125			@commands.Cog.listener()
126			async def on_member_join(self, _member):
127			self.db_add_update_member(_member)
128			self.db_add_update_spots(spots, _member)
129			self.db_add_update_spots(spots_temp, _member)
130
131			# Backup database
132			async def db_backup_database(self):
133			now = datetime.now()
134			cmd = f"mysqldump -u {get_settings('DB_U')} " \
135			f"--result-file=database_backup/backup-{now.strftime('%m-%d-%Y')}.sql " \
136			f"-p{get_settings('DB_P')} server_database"
137			os.system(cmd)
138			dt_string = self.bot.get_current_time()
139			print(f'({dt_string})\t[{self.__class__.__name__}]: Database backed up')
140
141			# ----- SPOTTING OPERATIONS -----
142
143			# Update spots tables
144			@classmethod
145			async def db_count_spot(cls, _id: int, monster_type: str):
146			cls.conn = cls.engine.connect()
147			cls.db_count_spot_table(spots, _id, monster_type)
148			cls.db_count_spot_table(spots_temp, _id, monster_type)
149			cls.conn.close()
150
151			@classmethod
152			def db_count_spot_table(cls, table, _id: int, monster_type: str):
153			stmt = select(table.c.member_id, table.c.legendary, table.c.rare, table.c.common,
154			table.c.event1,
155			table.c.event2).where(
156			table.c.member_id == _id)
157			result = cls.conn.execute(stmt)
158			counter = 0
159			for nr_of_kills in result.columns(monster_type, 'legendary'):
160			counter = nr_of_kills[0]
161			if monster_type == "event1":
162			values = cls.db_count_spot_table_event(table, _id, monster_type, counter)
163			else:
164			values = {f"{monster_type}": counter + 1}
165			stmt = update(table).where(table.c.member_id == _id).values(values)
166			cls.conn.execute(stmt)
167
168			@classmethod
169			def db_count_spot_table_event(cls, table, _id, monster_type: str, counter: int):
170			stmt = select(table.c.member_id, table.c.legendary, table.c.rare, table.c.common,
171			table.c.event1,
172			table.c.event2).where(
173			table.c.member_id == _id)
174			result = cls.conn.execute(stmt)
175			counter_leg = 0
176			for nr_of_kills_leg in result.columns('legendary'):
177			counter_leg = nr_of_kills_leg[0]
178			values = {f"{monster_type}": counter + 1, "legendary": counter_leg + 1}
179			return values
180
181			# Save coords from spotting channels to database
182			@classmethod
183			async def db_save_coords(cls, _coords: str, _monster_type):
184			cls.conn = cls.engine.connect()
185			stmt = insert(coords).values(coords=_coords, monster_type=_monster_type)
186			cls.conn.execute(stmt)
187			cls.conn.close()
188
189			# Clear data from spots_temp table(for events etc)
190			@classmethod
191			async def db_clear_spots_temp_table(cls):
192			cls.conn = cls.engine.connect()
193			stmt = delete(spots_temp)
194			cls.conn.execute(stmt)
195			cls.conn.close()
196
197			# ----- LEADERBOARD OPERATIONS -----
198
199			# Return all members' spots
200			@classmethod
201			async def db_get_spots_df(cls):
202			cls.conn = cls.engine.connect()
203			stmt = select(spots.c.member_id, member.c.display_name, spots.c.legendary, spots.c.rare,
204			spots.c.common, spots.c.event1, spots.c.event2
205			).select_from(member
206			).join(spots, member.c.id == spots.c.member_id)
207			cls.conn.execute(stmt)
208			df = pd.read_sql(stmt, cls.conn)
209			cls.conn.close()
210			return df
211
212			# ----- WARN OPERATIONS -----
213
214			# Add member's warn to database
215			@classmethod
216			async def db_add_warn(cls, _member: int, _reason: str):
217			cls.conn = cls.engine.connect()
218			stmt = insert(warn).values(member_id=_member, reason=_reason, date=datetime.now())
219			cls.conn.execute(stmt)
220			cls.conn.close()
221
222			# Get member's warns from database
223			@classmethod
224			async def db_get_warns(cls, _member: int):
225			cls.conn = cls.engine.connect()
226			stmt = select(warn.c.reason, warn.c.date).select_from(member).join(warn, member.c.id == warn.c.member_id).where(
227			member.c.id == _member)
228			result = cls.conn.execute(stmt)
229			date_warn = []
230			counter = 0
231			for warns in result.columns("reason", "date"):
232			reason_with_date = [warns[1], warns[0]]
233			date_warn.append(reason_with_date)
234			counter += 1
235			warns_list = [': \t'.join([str(elem) for elem in sublist]) for sublist in date_warn]
236			cls.conn.close()
237			return warns_list, counter
238
239			# Remove all member's warns
240			@classmethod
241			async def db_remove_warns(cls, _member: int):
242			cls.conn = cls.engine.connect()
243			stmt = delete(warn).where(warn.c.member_id == _member)
244			cls.conn.execute(stmt)
245			cls.conn.close()
246
247			# ----- MEMBER OPERATIONS -----
248
249			# Return member's spots
250			@classmethod
251			async def db_get_member_stats(cls, _member: int):
252			cls.conn = cls.engine.connect()
253			stmt = select(member.c.display_name, spots.c.legendary, spots.c.rare, spots.c.common).select_from(member).join(
254			spots,
255			member.c.id == spots.c.member_id).where(spots.c.member_id == _member)
256			df = pd.read_sql(stmt, cls.conn)
257			cls.conn.close()
258			return df
259
260			@cog_ext.cog_slash(name="changeMemberSpots", guild_ids=cogbase.GUILD_IDS,
261			description=" ",
262			default_permission=False,
263			permissions=cogbase.PERMISSION_MODS)
264			async def change_member_spots(self, ctx: SlashContext, user: discord.Member, spot_type: str, number: int):
265			self.conn = self.engine.connect()
266			stmt = f"""UPDATE server_database.spots SET {spot_type} = {number} """ \
267			f"""WHERE (member_id = {user.id});"""
268			self.conn.execute(stmt)
269			await ctx.send(f"{user.display_name} spots changed", hidden=True)
270			self.conn.close()
271
272			# ----- COORDS OPERATIONS -----
273
274			# Return coords
275			@classmethod
276			async def db_get_coords(cls):
277			cls.conn = cls.engine.connect()
278			stmt = select(coords.c.id, coords.c.coords, coords.c.monster_type).select_from(coords)
279			df = pd.read_sql(stmt, cls.conn)
280			cls.conn.close()
281			return df
282
283
284			def setup(bot: commands.Bot):
285			bot.add_cog(DatabaseCog(bot))
286

BonHowi / discordPogmareBot

Push — main ( d12a1e...942891 )

DatabaseCog.db_add_update_spots() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like