build.cogs.databasecog.DatabaseCog.__init__() - Code Metrics - Inspection of "Update databasecog.py" - BonHowi/discordPogmareBot - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — BonHowi (148d69)

by Bartosz

created 2021-09-14 15:47 UTC

build.cogs.databasecog.DatabaseCog.init() A

↳ Parent: build.cogs.databasecog

Complexity

Conditions

Size

Total Lines	8
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	8
rs	10
c	0
b	0
f	0
cc	1
nop	2

from discord.ext import commands, tasks
from discord_slash import cog_ext
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)
               )


# noinspection PyPropertyAccess
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

    # ----- 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()

    @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)

    # Command for loading data from previous bot
    @cog_ext.cog_slash(name="updateSpotsWithOld", guild_ids=cogbase.GUILD_IDS,
                       description="Change N-Word channel name",
                       permissions=cogbase.PERMISSION_ADMINS)
    async def db_update_spots_old(self, ctx):
        self.conn = self.engine.connect()
        import json
        with open('server_files/old_base.json', 'r', encoding='utf-8-sig') as fp:
            old_db = json.load(fp)
        guild = self.bot.get_guild(self.bot.guild[0])

        for mem_id in old_db:
            if guild.get_member(int(mem_id)) is None:
                continue

            stmt = select(spots.c.member_id, spots.c.legendary, spots.c.rare, spots.c.common).where(
                spots.c.member_id == mem_id)
            result = self.conn.execute(stmt).fetchall()
            result = result[0]
            counter_lege = result[1]
            counter_rare = result[2]
            counter_common = result[3]

            counter_lege_old = old_db[mem_id]["type_1"] if "type_1" in old_db[mem_id] else 0
            counter_rare_old = old_db[mem_id]["type_0"] if "type_0" in old_db[mem_id] else 0
            counter_common_old = old_db[mem_id]["type_4"] if "type_4" in old_db[mem_id] else 0

            stmt = insert(spots).values(
                member_id=mem_id, legendary=counter_lege_old,
                rare=counter_rare_old, common=counter_common_old)
            do_update_stmt = stmt.on_duplicate_key_update(legendary=stmt.inserted.legendary + counter_lege,
                                                          rare=stmt.inserted.rare + counter_rare,
                                                          common=stmt.inserted.common + counter_common)
            self.conn.execute(do_update_stmt)

            stmt = insert(spots_temp).values(
                member_id=mem_id, legendary=counter_lege_old,
                rare=counter_rare_old, common=counter_common_old)
            do_update_stmt = stmt.on_duplicate_key_update(legendary=stmt.inserted.legendary + counter_lege,
                                                          rare=stmt.inserted.rare + counter_rare,
                                                          common=stmt.inserted.common + counter_common)
            self.conn.execute(do_update_stmt)

        await ctx.send(f"Spot tables updated with old data", delete_after=5.0)
        dt_string = self.bot.get_current_time()
        print(f'({dt_string})\t[{self.__class__.__name__}]: Spot tables updated with old data')
        self.conn.close()

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

    # Update spots tables
    @classmethod
    async def db_count_spot(cls, _id: int, monster_type: str):
        cls.conn = cls.engine.connect()
        # Get member nr of spots for certain monster type
        stmt = select(spots.c.member_id, spots.c.legendary, spots.c.rare, spots.c.common, spots.c.event1,
                      spots.c.event2).where(
            spots.c.member_id == _id)
        result = cls.conn.execute(stmt)
        counter = []
        for nr_of_kills in result.columns(monster_type):
            counter = nr_of_kills[0]
        stmt = update(spots).where(spots.c.member_id == _id).values({f"{monster_type}": counter + 1})
        cls.conn.execute(stmt)

        stmt = select(spots_temp.c.member_id, spots_temp.c.legendary, spots_temp.c.rare, spots_temp.c.common,
                      spots_temp.c.event1,
                      spots_temp.c.event2).where(
            spots_temp.c.member_id == _id)
        result = cls.conn.execute(stmt)
        for nr_of_kills in result.columns(monster_type):
            counter = nr_of_kills[0]
        stmt = update(spots_temp).where(spots_temp.c.member_id == _id).values({f"{monster_type}": counter + 1})
        cls.conn.execute(stmt)
        cls.conn.close()

    # 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

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

BonHowi / discordPogmareBot

Branch — BonHowi (148d69)

build.cogs.databasecog.DatabaseCog.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

build.cogs.databasecog.DatabaseCog.init() A