#!/usr/bin/env python3.12 import os import logging from typing import LiteralString, Optional, Union import aiosqlite as sqlite3 from constructors import LoveHateException class DB: """LoveHate DB Utility Class""" def __init__(self, bot) -> None: self.db_path: str|LiteralString = os.path.join("/usr/local/share", "sqlite_dbs", "lovehate.db") async def get_wholovehates(self, thing: str, loves: bool = False, hates: bool = False) -> list[tuple]|bool: """ Get a list of users who have professed their love OR hatred for Args: thing (str): The to check loves (bool): Are we looking for loves? hates (bool): ...or are we looking for hates? Returns: list[tuple] """ query: str = "SELECT display_name FROM lovehate WHERE thing LIKE ? AND flag = ?" params: tuple = tuple() flag: Optional[int] = None if hates and loves: raise LoveHateException("Both hates and loves may not be True") elif hates: flag = -1 elif loves: flag = 1 elif not hates and not loves: raise LoveHateException("Neither loves nor hates were requested") params = (thing, flag,) async with sqlite3.connect(self.db_path, timeout=2) as db_conn: async with await db_conn.execute(query, params) as db_cursor: result: list[tuple] = await db_cursor.fetchall() logging.debug("Result: %s", result) if not result: return False return result async def get_lovehates(self, loves: bool = False, hates: bool = False, user: Optional[str] = None, thing: Optional[str] = None) -> Union[list[tuple], bool]: """ Get a list of either 1) what {user} loves/hates, or who loves/hates {thing}, depending on bools loves, hates Args: loves (bool): Are we looking for loves? hates (bool): ...OR are we looking for hates? user (Optional[str]): the user to query against thing (Optional[str]): ... OR the thing to query against Returns: Union[list[tuple], bool] """ query: str = "" params: tuple = tuple() if not user and not thing: raise LoveHateException("Neither a or was specified to query against") flag: Optional[int] = None if hates and loves: raise LoveHateException("Both hates and loves may not be True") elif hates: flag = -1 elif loves: flag = 1 elif not hates and not loves: raise LoveHateException("Neither loves nor hates were requested") if user: query = "SELECT thing FROM lovehate WHERE display_name LIKE ? AND flag == ?" params = (user, flag,) elif thing: query = "SELECT display_name FROM lovehate WHERE thing LIKE ? AND flag == ?" params = (thing, flag,) async with sqlite3.connect(self.db_path, timeout=2) as db_conn: async with await db_conn.execute(query, params) as db_cursor: result = await db_cursor.fetchall() if not result: return False return result async def check_existence(self, user: str, thing: str) -> Optional[int]: """ Determine whether a user is opinionated on a Args: user (str): The user to check thing (str): The thing to check if the user has an opinion on Returns: Optional[int] """ params = (user, thing,) async with sqlite3.connect(self.db_path, timeout=2) as db_conn: async with await db_conn.execute("SELECT id, flag FROM lovehate WHERE display_name LIKE ? AND thing LIKE ?", params) as db_cursor: result = await db_cursor.fetchone() if not result: return None (_, flag) = result return flag async def update(self, user: str, thing: str, flag: int) -> str: """ Updates the lovehate database, and returns an appropriate response Args: user (str): The user to update thing (str): The thing the user loves/hates/doesn't care about anymore flag (int): int representation of love (1), hate (-1), and dontcare (0) Returns: str """ if not flag in range(-1, 2): raise LoveHateException(f"Invalid flag {flag} specified, is this love (1), hate (-1), or dontcare? (0)") db_query: str = "" params: tuple = (user, thing,) already_opinionated: Optional[int] = await self.check_existence(user, thing) if already_opinionated: if flag == 0: db_query = "DELETE FROM lovehate WHERE display_name LIKE ? AND thing LIKE ?" else: loves_or_hates: str = "loves" if already_opinionated == -1: loves_or_hates = "hates" raise LoveHateException(f"But {user} already {loves_or_hates} {thing}...") else: match flag: case -1: db_query = "INSERT INTO lovehate(display_name, flag, thing) VALUES(?, -1, ?)" case 1: db_query = "INSERT INTO lovehate(display_name, flag, thing) VALUES(?, 1, ?)" case _: raise LoveHateException("Unknown error, default case matched") async with sqlite3.connect(self.db_path, timeout=2) as db_conn: async with await db_conn.execute(db_query, params) as db_cursor: await db_conn.commit() if db_cursor.rowcount != 1: raise LoveHateException(f"DB Error - RowCount: {db_cursor.rowcount} for INSERT query") match flag: case -1: return f"We're done here, {user} hates {thing}." case 0: return f"We're done here, {user} no longer cares one way or the other about {thing}." case 1: return f"We're done here, {user} loves {thing}." case _: raise LoveHateException("Unknown error, default case matched [2]")