Add log

.env.example

@@ -4,3 +4,5 @@ XMPP_ROOMS=
 
 NANOBOT_WS_ENDPOINT=
 NANOBOT_WS_TOKEN=
+
+DEBUG=

src/log.py

@@ -0,0 +1,25 @@
+import logging
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+LEVELS = {
+    "debug": logging.DEBUG,
+    "info": logging.INFO,
+    "warn": logging.WARNING,
+    "warning": logging.WARNING,
+    "error": logging.ERROR,
+    "critical": logging.CRITICAL,
+}
+
+level_name = os.getenv("DEBUG", "info").lower()
+level = LEVELS.get(level_name, logging.INFO)
+
+logging.basicConfig(
+    level=level,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+)
+
+main_logger = logging.getLogger("picoclaw")

src/nanobot.py

@@ -1,15 +1,10 @@
 import asyncio
 import json
-import logging
 import os
 
 import websockets
 
-from dotenv import load_dotenv
-
-load_dotenv()
-
-logger = logging.getLogger(__name__)
+from log import main_logger
 
 NANOBOT_WS_ENDPOINT = os.getenv("NANOBOT_WS_ENDPOINT")
 NANOBOT_WS_TOKEN = os.getenv("NANOBOT_WS_TOKEN")
@@ -32,7 +27,7 @@ class Nanobot:
         self._ws = await websockets.connect(uri)
         ready = json.loads(await self._ws.recv())
         self._chat_id = ready["chat_id"]
-        logger.info("connected chat_id=%s", self._chat_id)
+        main_logger.info("connected chat_id=%s", self._chat_id)
 
     async def chat(self, message: str, timeout: float = 10):
         await self._ws.send(json.dumps({"content": message}))
@@ -41,15 +36,15 @@ class Nanobot:
             try:
                 raw = await asyncio.wait_for(self._ws.recv(), timeout=timeout)
             except asyncio.TimeoutError:
-                logger.info("ws: timeout (no data for %.1fs), generator done", timeout)
+                main_logger.info("ws: timeout (no data for %.1fs), generator done", timeout)
                 return
             frame = json.loads(raw)
             event = frame.get("event")
             text = frame.get("text", "")
-            logger.info("ws raw: %s", raw)
+            main_logger.debug("ws raw: %s", raw)
             if event == "message":
                 if frame.get("kind") == "progress":
-                    logger.info("ws: skipping progress message")
+                    main_logger.debug("ws: skipping progress message")
                     continue
                 yield text
                 return

src/server.py

@@ -1,35 +1,29 @@
 import asyncio
-import logging
 import os
 
-from dotenv import load_dotenv
-
 from nanobot import Nanobot
 from xmpp import XmppBot
-
-load_dotenv()
-
-logger = logging.getLogger(__name__)
+from log import main_logger
 
 
 async def main():
     nanobot = Nanobot()
     await nanobot.connect()
-    logger.info("Nanobot connected")
+    main_logger.info("Nanobot connected")
 
     jid = os.getenv("XMPP_USERNAME")
     password = os.getenv("XMPP_PASSWORD")
     rooms = os.getenv("XMPP_ROOMS", "")
     if not jid or not password:
-        logger.error("XMPP_USERNAME and XMPP_PASSWORD must be set in .env")
+        main_logger.error("XMPP_USERNAME and XMPP_PASSWORD must be set in .env")
         return
 
     async def on_message(body, msg_type, sender, xmpp_msg):
-        logger.info(">> nanobot: %s", body[:200])
+        main_logger.info(">> nanobot: %s", body[:200])
         async for reply in nanobot.chat(body):
             if not reply:
                 continue
-            logger.info("<< nanobot: %s", reply[:200])
+            main_logger.info("<< nanobot: %s", reply[:200])
             xmpp_msg.reply(reply).send()
 
     xmpp = XmppBot(jid, password, rooms=rooms, on_message=on_message)
@@ -42,9 +36,4 @@ async def main():
 
 
 if __name__ == "__main__":
-    level = logging.DEBUG if os.getenv("DEBUG") else logging.INFO
-    logging.basicConfig(
-        level=level,
-        format="%(asctime)s [%(levelname)s] %(message)s",
-    )
     asyncio.run(main())

src/xmpp.py

@@ -1,12 +1,7 @@
-import logging
 import os
 
 import slixmpp
-from dotenv import load_dotenv
-
-load_dotenv()
-
-logger = logging.getLogger(__name__)
+from log import main_logger
 
 
 class XmppBot(slixmpp.ClientXMPP):
@@ -32,7 +27,7 @@ class XmppBot(slixmpp.ClientXMPP):
     async def handle_message(self, msg):
         sender = msg["from"]
         body_preview = msg["body"][:100]
-        logger.info("<< [%s] %s: %s", msg["type"], sender, body_preview)
+        main_logger.info("<< [%s] %s: %s", msg["type"], sender, body_preview)
 
         if msg["type"] == "groupchat":
             if msg["mucnick"] == self.nickname:
@@ -70,15 +65,10 @@ class XmppBot(slixmpp.ClientXMPP):
 
 
 if __name__ == "__main__":
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s [%(levelname)s] %(message)s",
-    )
-
     jid = os.getenv("XMPP_USERNAME")
     password = os.getenv("XMPP_PASSWORD")
     if not jid or not password:
-        logger.error("XMPP_USERNAME and XMPP_PASSWORD must be set in .env")
+        main_logger.error("XMPP_USERNAME and XMPP_PASSWORD must be set in .env")
         exit(1)
 
     bot = XmppBot(jid, password, rooms=os.getenv("XMPP_ROOMS", ""))