api/endpoints/karma.py

#!/usr/bin/env python3.12

import os
import time
import datetime
import aiosqlite as sqlite3

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel


class ValidKarmaUpdateRequest(BaseModel):
    """
    Requires authentication
    - **granter**: who updated the karma
    - **keyword**: keyword to update karma for
    - **flag**: either 0 (decrement) for --, or 1 (increment) for ++
    """

    granter: str
    keyword: str
    flag: int


class ValidKarmaRetrievalRequest(BaseModel):
    """
    - **keyword**: keyword to retrieve karma value of
    """

    keyword: str

class ValidTopKarmaRequest(BaseModel):
    """
    - **n**: Number of top results to return (default: 10)
    """
    n: int | None = 10

class KarmaDB:
    def __init__(self):
        self.db_path = os.path.join("/", "var", "lib", "singerdbs", "karma.db")

    async def get_karma(self, keyword: str) -> str | int:
        async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
            async with db_conn.execute("SELECT score FROM karma WHERE keyword = ? LIMIT 1", (keyword,)) as db_cursor:
                try:
                    (score,) = await db_cursor.fetchone()
                    return score
                except TypeError as e:
                    return {
                        'err': True,
                        'errorText': f'No records for {keyword}',
                    }
                
    async def get_top(self, n: int = 10):
        try:
            async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
                async with db_conn.execute("SELECT keyword, score FROM karma ORDER BY score DESC LIMIT ?", (n,)) as db_cursor:
                    return await db_cursor.fetchall()
        except Exception as e:
            print(traceback.format_exc())
            return
        
    async def update_karma(self, granter: str, keyword: str, flag: int):
        if not flag in [0, 1]:
            return
        
        modifier = "score + 1" if not flag else "score - 1"
        query = f"UPDATE karma SET score = {modifier}, last_change = ? WHERE keyword = ?"
        new_keyword_query = "INSERT INTO karma(keyword, score, last_change) VALUES(?, ?, ?)"
        friendly_flag = "++" if not flag else "--"
        audit_message = f"{granter} adjusted karma for {keyword} @ {datetime.datetime.now().isoformat()}: {friendly_flag}"
        audit_query = "INSERT INTO karma_audit(impacted_keyword, comment) VALUES(?, ?)"
        now = int(time.time())

        print(f"Audit message: {audit_message}\nKeyword: {keyword}")

        async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
            async with db_conn.execute(audit_query, (keyword, audit_message,)) as db_cursor:
                await db_conn.commit()
                await db_cursor.close()             
            async with db_conn.execute(query, (now, keyword,)) as db_cursor:
                if db_cursor.rowcount:
                    await db_conn.commit()
                    return True
                if db_cursor.rowcount < 1:  # Keyword does not already exist
                    await db_cursor.close()
                    new_val = 1 if not flag else -1
                    async with db_conn.execute(new_keyword_query, (keyword, new_val, now,)) as db_cursor:
                        if db_cursor.rowcount >= 1:
                            await db_conn.commit()
                            return True
                        else:
                            return False





class Karma(FastAPI):
    """Karma Endpoints"""   
    def __init__(self, app: FastAPI, util, constants, glob_state):  # pylint: disable=super-init-not-called
        self.app = app
        self.util = util
        self.constants = constants
        self.glob_state = glob_state
        self.db = KarmaDB()

        self.endpoints = {
            "karma/get": self.get_karma_handler,
            "karma/modify": self.modify_karma_handler,
            "karma/top": self.top_karma_handler,
            }

        for endpoint, handler in self.endpoints.items():
            app.add_api_route(f"/{endpoint}/", handler, methods=["POST"])

    
    async def top_karma_handler(self, request: Request, data: ValidTopKarmaRequest | None = None):
        """
        /karma/top/
        Get top keywords for karma
        Requires key
        """

        if not self.util.check_key(request.url.path, request.headers.get('X-Authd-With')):
            raise HTTPException(status_code=403, detail="Unauthorized")
                
        n = 10
        if data:
            n = data.n
            

        try:
            top10 = await self.db.get_top(n=n)
            return top10
        except Exception as e:
            print(traceback.format_exc())
            return {
                'err': True,
                'errorText': 'Exception occurred.',
            }
    
    async def get_karma_handler(self, data: ValidKarmaRetrievalRequest):
        """
        /karma/get/
        Get current karma value
        Requires key
        """
        
        keyword = data.keyword
        try:
            count = await self.db.get_karma(keyword)
            return {
                'keyword': keyword,
                'count': count,
            }
        except:
            print(traceback.format_exc())
            return {
                'err': True,
                'errorText': "Exception occurred."
            }
    
    async def modify_karma_handler(self, data: ValidKarmaUpdateRequest, request: Request):
        """
        /karma/update/
        Update karma count 
        Requires key
        """

        if not self.util.check_key(request.url.path, request.headers.get('X-Authd-With'), 2):
            raise HTTPException(status_code=403, detail="Unauthorized")
        
        if not data.flag in [0, 1]:
            return {
                'err': True,
                'errorText': 'Invalid request'
            }
        
        return {
            'success': await self.db.update_karma(data.granter, data.keyword, data.flag)
        }