build.cogs.databasecog.DatabaseCog.db_add_update_spots() - Code Metrics - Inspection of "Merge pull request #6 from BonHowi/BonHowi" - BonHowi/discordPogmareBot - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 4d254e...b95c28 )

by Bartosz

created 2021-09-09 01:24 UTC

DatabaseCog.db_add_update_spots() A

↳ Parent: build.cogs.databasecog

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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

spots = Table('spots', metadata_obj,
              Column('member_id', BigInteger, ForeignKey("member.id"), primary_key=True),
              Column('legendary', Integer),
              Column('rare', Integer),
              Column('common', Integer)
              )

spots_temp = Table('spots_temp', metadata_obj,
                   Column('member_id', BigInteger, ForeignKey("member.id"), primary_key=True),
                   Column('legendary', Integer),
                   Column('rare', Integer),
                   Column('common', Integer)
                   )

warn = Table('warn', metadata_obj,
             Column('id', Integer, primary_key=True),
             Column('member_id', BigInteger, ForeignKey("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(50), nullable=False),
               Column('monster_type', String(50), nullable=False)
               )


class DatabaseCog(cogbase.BaseCog):
    password = get_settings("DB_P")
    engine = create_engine(f"mysql+mysqldb://BonHowi:{password}@localhost/server_database")
    metadata_obj.create_all(engine)
    conn = engine.connect()

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

        # Connect to database
        password = get_settings("DB_P")
        self.engine = create_engine(f"mysql+mysqldb://BonHowi:{password}@localhost/server_database")
        metadata_obj.create_all(self.engine)
        self.conn = self.engine.connect()
        self.db_update_loop.start()

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

    def db_add_update_member(self, _member):
        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)

    def db_add_update_spots(self, spots_table, guild_member):
        stmt = insert(spots_table).values(
            member_id=guild_member.id, legendary=0,
            rare=0, common=0)
        do_update_stmt = stmt.on_duplicate_key_update(member_id=stmt.inserted.member_id)
        self.conn.execute(do_update_stmt)

    # Add or refresh all guild members and spots to database
    async def db_update(self):
        guild = self.bot.get_guild(self.bot.guild[0])
        # Member tables
        print(f"[{self.__class__.__name__}]: Refreshing member count")
        for guild_member in guild.members:
            self.db_add_update_member(guild_member)
        print(f"[{self.__class__.__name__}]: Member count refreshed")

        # Spots tables
        guild = self.bot.get_guild(self.bot.guild[0])
        for guild_member in guild.members:
            self.db_add_update_spots(spots, guild_member)
            self.db_add_update_spots(spots_temp, guild_member)

    @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):
        print(f'[{self.__class__.__name__}]: Waiting until Bot is ready')
        await self.bot.wait_until_ready()

    # Placeholder for simpler function(now it updates whole tables instead of one row)
    @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)

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

    @classmethod
    async def db_count_spot(cls, _id: int, monster_type: str):
        # Get member nr of spots for certain monster type
        stmt = select(spots.c.member_id, spots.c.legendary, spots.c.rare, spots.c.common).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).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)

    @classmethod
    async def db_save_coords(cls, _coords: str, _monster_type):
        stmt = insert(coords).values(coords=_coords, monster_type=_monster_type)
        cls.conn.execute(stmt)

    @classmethod
    async def db_clear_spots_temp_table(cls):
        stmt = delete(spots_temp)
        cls.conn.execute(stmt)

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

    @classmethod
    async def db_get_spots_df(cls):
        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)
        df = pd.read_sql(stmt, cls.conn)
        return df

    async def db_update_spotting_roles(self):
        pass

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

    @classmethod
    async def db_add_warn(cls, _member: int, _reason: str):
        stmt = insert(warn).values(member_id=_member, reason=_reason, date=datetime.now())
        cls.conn.execute(stmt)

    @classmethod
    async def db_get_warns(cls, _member: int):
        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]
        return warns_list, counter

    @classmethod
    async def db_remove_warns(cls, _member: int):
        stmt = delete(warn).where(warn.c.member_id == _member)
        cls.conn.execute(stmt)

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

    @classmethod
    async def db_get_member_stats(cls, _member: int):
        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)
        return df


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


1			import inspect
2			from discord.ext import commands, tasks
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
18			spots = Table('spots', metadata_obj,
19			Column('member_id', BigInteger, ForeignKey("member.id"), primary_key=True),
20			Column('legendary', Integer),
21			Column('rare', Integer),
22			Column('common', Integer)
23			)
24
25			spots_temp = Table('spots_temp', metadata_obj,
26			Column('member_id', BigInteger, ForeignKey("member.id"), primary_key=True),
27			Column('legendary', Integer),
28			Column('rare', Integer),
29			Column('common', Integer)
30			)
31
32			warn = Table('warn', metadata_obj,
33			Column('id', Integer, primary_key=True),
34			Column('member_id', BigInteger, ForeignKey("member.id")),
35			Column('reason', String(120), nullable=False),
36			Column('date', DateTime, nullable=False),
37			)
38
39			coords = Table('coords', metadata_obj,
40			Column('id', Integer, primary_key=True),
41			Column('coords', String(50), nullable=False),
42			Column('monster_type', String(50), nullable=False)
43			)
44
45
46			class DatabaseCog(cogbase.BaseCog):
47			password = get_settings("DB_P")
48			engine = create_engine(f"mysql+mysqldb://BonHowi:{password}@localhost/server_database")
49			metadata_obj.create_all(engine)
50			conn = engine.connect()
51
52			def __init__(self, base):
53			super().__init__(base)
54
55			# Connect to database
56			password = get_settings("DB_P")
57			self.engine = create_engine(f"mysql+mysqldb://BonHowi:{password}@localhost/server_database")
58			metadata_obj.create_all(self.engine)
59			self.conn = self.engine.connect()
60			self.db_update_loop.start()
61
62			# ----- BASE DATABASE OPERATIONS -----
63
64			def db_add_update_member(self, _member):
65			stmt = insert(member).values(
66			id=_member.id, name=_member.name,
67			display_name=_member.display_name)
68			do_update_stmt = stmt.on_duplicate_key_update(
69			name=stmt.inserted.name, display_name=stmt.inserted.display_name
70			)
71			self.conn.execute(do_update_stmt)
72
73			def db_add_update_spots(self, spots_table, guild_member):
74			stmt = insert(spots_table).values(
75			member_id=guild_member.id, legendary=0,
76			rare=0, common=0)
77			do_update_stmt = stmt.on_duplicate_key_update(member_id=stmt.inserted.member_id)
78			self.conn.execute(do_update_stmt)
79
80			# Add or refresh all guild members and spots to database
81			async def db_update(self):
82			guild = self.bot.get_guild(self.bot.guild[0])
83			# Member tables
84			print(f"[{self.__class__.__name__}]: Refreshing member count")
85			for guild_member in guild.members:
86			self.db_add_update_member(guild_member)
87			print(f"[{self.__class__.__name__}]: Member count refreshed")
88
89			# Spots tables
90			guild = self.bot.get_guild(self.bot.guild[0])
91			for guild_member in guild.members:
92			self.db_add_update_spots(spots, guild_member)
93			self.db_add_update_spots(spots_temp, guild_member)
94
95			@tasks.loop(hours=12)
96			async def db_update_loop(self):
97			await self.db_update()
98
99			@db_update_loop.before_loop
100			async def before_db_update_loop(self):
101			print(f'[{self.__class__.__name__}]: Waiting until Bot is ready')
102			await self.bot.wait_until_ready()
103
104			# Placeholder for simpler function(now it updates whole tables instead of one row)
105			@commands.Cog.listener()
106			async def on_member_join(self, _member):
107			self.db_add_update_member(_member)
108			self.db_add_update_spots(spots, _member)
109			self.db_add_update_spots(spots_temp, _member)
110
111			# ----- SPOTTING OPERATIONS -----
112
113			@classmethod
114			async def db_count_spot(cls, _id: int, monster_type: str):
115			# Get member nr of spots for certain monster type
116			stmt = select(spots.c.member_id, spots.c.legendary, spots.c.rare, spots.c.common).where(
117			spots.c.member_id == _id)
118			result = cls.conn.execute(stmt)
119			counter = []
120			for nr_of_kills in result.columns(monster_type):
121			counter = nr_of_kills[0]
122			stmt = update(spots).where(spots.c.member_id == _id).values({f"{monster_type}": counter + 1})
123			cls.conn.execute(stmt)
124
125			stmt = select(spots_temp.c.member_id, spots_temp.c.legendary, spots_temp.c.rare, spots_temp.c.common).where(
126			spots_temp.c.member_id == _id)
127			result = cls.conn.execute(stmt)
128			for nr_of_kills in result.columns(monster_type):
129			counter = nr_of_kills[0]
130			stmt = update(spots_temp).where(spots_temp.c.member_id == _id).values({f"{monster_type}": counter + 1})
131			cls.conn.execute(stmt)
132
133			@classmethod
134			async def db_save_coords(cls, _coords: str, _monster_type):
135			stmt = insert(coords).values(coords=_coords, monster_type=_monster_type)
136			cls.conn.execute(stmt)
137
138			@classmethod
139			async def db_clear_spots_temp_table(cls):
140			stmt = delete(spots_temp)
141			cls.conn.execute(stmt)
142
143			# ----- LEADERBOARD OPERATIONS -----
144
145			@classmethod
146			async def db_get_spots_df(cls):
147			stmt = select(member.c.display_name, spots.c.legendary, spots.c.rare, spots.c.common).select_from(member).join(
148			spots,
149			member.c.id == spots.c.member_id)
150			df = pd.read_sql(stmt, cls.conn)
151			return df
152
153			async def db_update_spotting_roles(self):
154			pass
155
156			# ----- WARN OPERATIONS -----
157
158			@classmethod
159			async def db_add_warn(cls, _member: int, _reason: str):
160			stmt = insert(warn).values(member_id=_member, reason=_reason, date=datetime.now())
161			cls.conn.execute(stmt)
162
163			@classmethod
164			async def db_get_warns(cls, _member: int):
165			stmt = select(warn.c.reason, warn.c.date).select_from(member).join(warn, member.c.id == warn.c.member_id).where(
166			member.c.id == _member)
167			result = cls.conn.execute(stmt)
168			date_warn = []
169			counter = 0
170			for warns in result.columns("reason", "date"):
171			reason_with_date = [warns[1], warns[0]]
172			date_warn.append(reason_with_date)
173			counter += 1
174			warns_list = [': \t'.join([str(elem) for elem in sublist]) for sublist in date_warn]
175			return warns_list, counter
176
177			@classmethod
178			async def db_remove_warns(cls, _member: int):
179			stmt = delete(warn).where(warn.c.member_id == _member)
180			cls.conn.execute(stmt)
181
182			# ----- MEMBER OPERATIONS -----
183
184			@classmethod
185			async def db_get_member_stats(cls, _member: int):
186			stmt = select(member.c.display_name, spots.c.legendary, spots.c.rare, spots.c.common).select_from(member).join(
187			spots,
188			member.c.id == spots.c.member_id).where(spots.c.member_id == _member)
189			df = pd.read_sql(stmt, cls.conn)
190			return df
191
192
193			def setup(bot: commands.Bot):
194			bot.add_cog(DatabaseCog(bot))
195

BonHowi / discordPogmareBot

Push — main ( 4d254e...b95c28 )

DatabaseCog.db_add_update_spots() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like