api parse token

README.md

@@ -10,3 +10,8 @@ short_description: log displayer for all end
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+
+Please set following environment variables:
+
+- `hf_token`: hugginface token
+- `SECRET_KEY`: key of parsing user token

main.py

@@ -1,9 +1,9 @@
-from fastapi import FastAPI, Body
+from fastapi import FastAPI, Body, Header, Request
 from fastapi.middleware.cors import CORSMiddleware
-from panel.io.fastapi import add_applications
-import page
-from utils import beijing
-import logging
+from utils import beijing, parse_token
+from logging_helper import LoggingHelper
+from fastapi.templating import Jinja2Templates
+
 
 app = FastAPI(
     title=f"Log Displayer",
@@ -19,25 +19,55 @@ app.add_middleware(
     allow_headers=["*"],
 )
 print("Done\n")
-add_applications(page.page, app=app, title="Log Displayer")
-
 
-@app.post("/app")
-async def log_app(message: str = Body(..., embed=True)):
-    print(f"[APP] {message}")
-    logger = logging.getLogger("APP")
-    logger.debug(message)
-    return True
+LOGS_REPO_ID = "pgsoft/logs"
+local_dir = "data/logs"
+logger = LoggingHelper(
+    repo_id=LOGS_REPO_ID,
+    local_dir=local_dir,
+)
 
 
-@app.post("/game")
-async def log_game(message: str = Body(..., embed=True)):
-    print(f"[GAME] {message}")
-    logger = logging.getLogger("Game")
-    logger.debug(message)
+@app.post("/{end}")
+async def add_log(
+    end: str,
+    message: str = Body(..., embed=True),
+    token: str | None = Header(None),
+    source: str | None = Header("web"),
+):
+    print("Type: ", end)
+    print("From: ", source)
+    print("Token: ", token)
+    uid, username = parse_token(token)
+    timestamp = beijing().isoformat()
+    print("Timestamp: ", timestamp)
+    print("Message: ", message)
+    log = {
+        "type": end,
+        "source": source,
+        "uid": uid,
+        "username": username,
+        "token": token,
+        "content": message,
+        "timestamp": timestamp,
+    }
+    logger.addlog(log)
     return True
 
 
 @app.get("/healthcheck")
 async def health_check():
-    return True
+    return True
+
+
+templates = Jinja2Templates(directory="static")
+
+
+@app.get("")
+@app.get("/")
+async def root(request: Request):
+    data = logger.refresh()
+    return templates.TemplateResponse(
+        "index.html",
+        {"request": request, "data": data},
+    )

requirements.txt

@@ -4,3 +4,7 @@ transformers
 numpy
 torch
 aiohttp
+python-jose[cryptography]
+apscheduler==3.11.0
+huggingface-hub==0.34.4
+datasets==4.0.0

scratch/test_dataset_to_dict.py

@@ -0,0 +1,15 @@
+import datasets as ds
+import pandas as pd
+import glob
+
+dataset = ds.Dataset.from_dict({})
+files = glob.glob("**/*.json", root_dir="data/logs", recursive=True)
+for file in files:
+    path = "data/logs/" + file
+    temp = ds.Dataset.from_json(path)
+    dataset = ds.concatenate_datasets([dataset, temp])  # type: ignore
+df = dataset.to_pandas()
+assert isinstance(df, pd.DataFrame)
+print(len(df))
+res = df.to_dict(orient="records")
+print(res)

scratch/test_glob.py

@@ -0,0 +1,6 @@
+import glob
+
+
+path = "data/logs"
+files = glob.glob("**/*.json", root_dir=path, recursive=True)
+print(files)

utils.py

@@ -1,7 +1,63 @@
 from datetime import datetime
 from zoneinfo import ZoneInfo
+from typing import Any
+from jose import jwt
+import hashlib
+import json
+import uuid
+import os
+
+
+SECRET_KEY = os.environ.get("SECRET_KEY")
 
 
 def beijing():
     """get beijing time"""
     return datetime.now(tz=ZoneInfo("Asia/Shanghai"))
+
+
+def decode_jwt(token: str) -> dict[str, Any]:
+    """get payload in the jwt token"""
+    assert SECRET_KEY, "Please set the environment variable SECRET_KEY"
+    return jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
+
+
+def parse_token(token: str | None) -> tuple[str, str]:
+    """parse user token to get uid and username
+
+    :param token (str): jwt token
+    :return (str, str): (uid, username)
+    """
+    if not token:
+        uid, username = "no_uid", "Anonymous"
+    else:
+        payload: dict = decode_jwt(token)
+        uid = payload.get("uid", "no_uid")
+        username = payload.get("username", "Anonymous")
+    print(f"UID: {uid}")
+    print(f"Username: {username}")
+    return uid, username
+
+
+def md5(text: list[str | bytes] | str | bytes | None = None) -> str:
+    """generate the md5 hash code of the given text, if text is None,
+    return a random md5"""
+    code = hashlib.md5()
+    if text:
+        if isinstance(text, str):
+            text = text.encode("utf-8")
+            code.update(text)
+        elif isinstance(text, list):
+            for t in text:
+                if isinstance(t, str):
+                    t = t.encode("utf-8")
+                code.update(t)
+        else:
+            code.update(text)
+    else:
+        code.update(uuid.uuid4().bytes)
+    return code.hexdigest()
+
+
+def json_to_str(obj: dict | list) -> str:
+    return json.dumps(obj, separators=(",", ":"))