V1.2.0 alpha 1

- 支持使用 api 提交物品信息
- 支持 JWT 鉴权

routes/backend/__init__.py

@@ -0,0 +1,2 @@
+from . import admin
+from . import session

routes/backend/admin.py

@@ -0,0 +1,74 @@
+from nicegui import app
+from typing import Annotated
+from fastapi import Depends
+from fastapi import HTTPException, status
+from jwt import InvalidTokenError
+import jwt, JWT
+from model import database
+from model import token as Token
+from model.response import DefaultResponse
+
+async def is_admin(token: Annotated[str, Depends(JWT.oauth2_scheme)]):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Login required",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, JWT.SECRET_KEY, algorithms=["HS256"])
+        username = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+    except InvalidTokenError:
+        raise credentials_exception
+    if not username == await database.Database().get_setting('account'):
+        raise credentials_exception
+    token_data = Token.TokenData(username=username)
+    return True
+
+@app.get('/api/items')
+async def get_items(
+    is_admin: Annotated[str, Depends(is_admin)],
+    id: int = None,
+    key: str = None):
+    items = await database.Database().get_object(id=id, key=key)
+    return DefaultResponse(data=items)
+
+@app.post('/api/items')
+async def add_items(
+    is_admin: Annotated[str, Depends(is_admin)],
+    key: str,
+    name: str,
+    icon: str,
+    phone: str):
+    try:
+        await database.Database().add_object(
+            key=key, name=name, icon=icon, phone=phone)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    else:
+        return DefaultResponse()
+
+@app.patch('/api/items')
+async def update_items(
+    is_admin: Annotated[str, Depends(is_admin)],
+    id: int,
+    **kwargs):
+    try:
+        await database.Database().update_object(
+            id=id, **kwargs)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    else:
+        return DefaultResponse()
+
+@app.delete('/api/items')
+async def delete_items(
+    is_admin: Annotated[str, Depends(is_admin)],
+    id: int):
+    try:
+        await database.Database().delete_object(id=id)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    else:
+        return DefaultResponse()

routes/backend/session.py

@@ -0,0 +1,51 @@
+from nicegui import app
+from typing import Annotated
+from datetime import datetime, timedelta, timezone
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+from fastapi import APIRouter
+import jwt, JWT
+
+from model.token import Token
+from model import database
+from tool import verify_password
+
+Router = APIRouter()
+
+def create_access_token(data: dict, expires_delta: timedelta | None = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.now(timezone.utc) + expires_delta
+    else:
+        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, JWT.SECRET_KEY, algorithm='HS256')
+    return encoded_jwt
+
+async def authenticate_user(username: str, password: str):
+    # 验证账号和密码
+    account = await database.Database().get_setting('account')
+    stored_password = await database.Database().get_setting('password')
+
+    if account != username or not verify_password(stored_password, password):
+        return False
+    
+    return {'is_authenticated': True}
+
+# FastAPI 登录路由 / FastAPI login route
+@app.post("/api/token")
+async def login_for_access_token(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+) -> Token:
+    user = await authenticate_user(form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(hours=1)
+    access_token = create_access_token(
+        data={"sub": form_data.username}, expires_delta=access_token_expires
+    )
+    return Token(access_token=access_token, token_type="bearer")