reformat/naas

base.py

@@ -54,7 +54,10 @@ def base_head():
 @app.get("/{path}", include_in_schema=False)
 def disallow_get_any(request: Request, var: Any = None):
     path = request.path_params["path"]
-    if not (isinstance(path, str) and path.split("/", maxsplit=1) == "widget"):
+    if not (
+        isinstance(path, str)
+        and (path.split("/", maxsplit=1) == "widget" or path == "misc/no")
+    ):
         return util.get_blocked_response()
     else:
         logging.info("OK, %s", path)

endpoints/lyric_search.py

@@ -170,8 +170,8 @@ class LyricSearch(FastAPI):
         if not result:
             if not data.lrc:
                 await self.notifier.send(
-                "DEBUG", f"Could not locate lyrics, request was:\n`{data}`"
-            )
+                    "DEBUG", f"Could not locate lyrics, request was:\n`{data}`"
+                )
             return JSONResponse(
                 content={
                     "err": True,

endpoints/misc.py

@@ -1,6 +1,8 @@
 import logging
 import time
 import os
+import json
+import random
 from typing import Optional, Annotated
 from fastapi import FastAPI, Request, UploadFile, Response, HTTPException, Form
 from fastapi.responses import JSONResponse
@@ -22,12 +24,16 @@ class Misc(FastAPI):
         self.redis_client = redis.Redis(password=private.REDIS_PW)
         self.radio = radio
         self.activity_image: Optional[bytes] = None
+        self.nos_json_path: str = os.path.join("/usr/local/share/naas/reasons.json")
+        self.nos: list[str] = []
+        self.last_5_nos: list[str] = []
         self.endpoints: dict = {
             "widget/redis": self.homepage_redis_widget,
             "widget/sqlite": self.homepage_sqlite_widget,
             "widget/lyrics": self.homepage_lyrics_widget,
             "widget/radio": self.homepage_radio_widget,
             "misc/get_activity_image": self.get_activity_image,
+            "misc/no": self.no,
         }
 
         for endpoint, handler in self.endpoints.items():
@@ -39,6 +45,30 @@ class Misc(FastAPI):
             "/misc/upload_activity_image", self.upload_activity_image, methods=["POST"]
         )
 
+        logging.debug("Loading NaaS reasons")
+
+        with open(self.nos_json_path, "r") as f:
+            self.nos = json.loads(f.read())
+
+        logging.debug("Loaded %s reasons", len(self.nos))
+
+    def get_no(self) -> str:
+        try:
+            no = random.choice(self.nos)
+            if no in self.last_5_nos:
+                return self.get_no()  # recurse
+            self.last_5_nos.append(no)
+            if len(self.last_5_nos) >= 5:
+                self.last_5_nos.pop(0)
+            return no
+        except Exception as e:
+            logging.debug("Exception: %s", str(e))
+            return "No."
+
+    async def no(self) -> JSONResponse:
+        """NaaS"""
+        return JSONResponse(content={"no": self.get_no()})
+
     async def upload_activity_image(
         self, image: UploadFile, key: Annotated[str, Form()], request: Request
     ) -> Response: