feat: Move python version to 3.12; Add sql_typecast; Rewrite typevar…

… class; fix: some bugs;

Configs/config.py

@@ -11,7 +11,9 @@ def get_config(self, *keys) -> object | None:
         result: object | None = self.config
         while len(keys) > 0:
             key = keys.pop(0)
-            if (isinstance(result, list) and key < len(result)) or (isinstance(result, dict) and key in result.keys()):
+            if (isinstance(result, list) and key < len(result)) or (
+                isinstance(result, dict) and key in result.keys()
+            ):
                 result = result[key]
             else:
                 return None

Models/comic.py

@@ -0,0 +1,41 @@
+from pydantic import BaseModel
+
+from typing import List, Optional, Dict, Any
+
+
+class BaseComicInfo(BaseModel):
+    """BaseComicInfo"""
+    """作者，漫画作者"""
+    author: List[str]
+    """封面，漫画封面图片URL"""
+    cover: str
+    """标识符，漫画在所属平台的索引ID"""
+    id: str
+    """名称，漫画名称"""
+    name: str
+
+
+class ComicInfo(BaseModel):
+    """ComicInfo"""
+    """章节数，漫画章节数"""
+    chapters: Optional[int] = None
+    """评论量，漫画评论量"""
+    comments: Optional[int] = None
+    """简介，漫画简介"""
+    description: Optional[str] = None
+    """额外信息，源平台携带的其它漫画信息"""
+    extras: Optional[Dict[str, Any]] = None
+    """收藏量，漫画收藏量"""
+    favorites: Optional[int] = None
+    """已收藏，漫画是否已收藏"""
+    is_favorite: Optional[bool] = None
+    """已完结，漫画是否已完结"""
+    is_finished: Optional[bool] = None
+    """已阅读，漫画是否已阅读"""
+    is_viewed: Optional[bool] = None
+    """标签，漫画标签"""
+    tags: Optional[List[str]] = None
+    """更新时间，漫画最近的更新时间戳"""
+    updated_at: Optional[int] = None
+    """阅读量，漫画阅读量"""
+    views: Optional[int] = None

Models/plugins.py

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
 
 from Models.response import StandardResponse
+from Models.comic import BaseComicInfo
 
 
 class BasePlugin(ABC):
@@ -11,7 +12,20 @@ def on_load(self) -> bool:
     def on_unload(self) -> None:
         pass
 
-    def login(self, username: str, password: str) -> StandardResponse:
+    @abstractmethod
+    def search(self, keyword: str) -> list[BaseComicInfo]:
+        pass
+
+
+class IAuth:
+    @abstractmethod
+    def login(self, body: dict[str, str]) -> StandardResponse:
+        pass
+
+
+class IShaper:
+    @abstractmethod
+    def imager_shaper(self):
         pass
 
 
@@ -23,11 +37,16 @@ class Plugin:
     service: dict
     instance: BasePlugin
 
-    def __init__(self, name: str, version: str, cnm_version: str, source: list[str], service: dict,
-                 instance: BasePlugin):
+    def __init__(
+        self,
+        name: str,
+        version: str,
+        cnm_version: str,
+        source: list[str],
+        instance: BasePlugin,
+    ):
         self.name = name
         self.version = version
         self.cnm_version = cnm_version
         self.source = source
-        self.service = service
         self.instance = instance

Models/requests.py

@@ -7,12 +7,6 @@ class SourceStorageReq(BaseModel):
     key_pwd: str
 
 
-class SourceLoginReq(BaseModel):
-    account: str
-    password: str
-    extra_info: dict[str, str] | None = None
-
-
 class ComicSearchReq(BaseModel):
     sources: list[str]
     keyword: str

Models/response.py

@@ -24,10 +24,12 @@ def limit_exceeded(request: Request, exc: RateLimitExceeded):
         )
 
 
-class StandardResponse(BaseModel):
+class StandardResponse[T](BaseModel):
     status_code: int
     message: str | None
-    data: object | None
+    data: T | None
 
-    def __init__(self, status_code: int = 200, message: str | None = None, data: object | None = None):
+    def __init__(
+        self, status_code: int = 200, message: str | None = None, data: T | None = None
+    ):
         super().__init__(status_code=status_code, message=message, data=data)

Routers/comic.py

@@ -1,10 +1,12 @@
 from fastapi import APIRouter, Response, Depends, HTTPException, Form, Request
 
 from Models.user import User
+from Models.requests import ComicSearchReq
+from Models.response import ExceptionResponse, StandardResponse
 from Services.Database.database import get_db
 from Services.Limiter.limiter import limiter
 from Services.Security.user import get_current_user
-from Models.requests import ComicSearchReq
+from Services.Modulator.manager import plugin_manager
 
 comic_router = APIRouter(prefix="/comic")
 
@@ -16,9 +18,12 @@ async def search_comic(body: ComicSearchReq, user: User = Depends(get_current_us
 
 @comic_router.get("/{src_id}/album/{album_id}")
 async def get_album(src_id: str, album_id: str, user: User = Depends(get_current_user)):
-    pass
+    if (source := plugin_manager.get_source(src_id)) is None:
+        raise ExceptionResponse.not_found
 
 
 @comic_router.get("/{src_id}/album/{album_id}/images/{chapter_id}")
-async def get_chapter_images(src_id: str, album_id: str, chapter_id: str, user: User = Depends(get_current_user)):
+async def get_chapter_images(
+    src_id: str, album_id: str, chapter_id: str, user: User = Depends(get_current_user)
+):
     pass

Routers/core.py

@@ -0,0 +1,16 @@
+from fastapi import APIRouter, Response, Depends, HTTPException, Form, Request
+
+from Models.user import User
+from Models.requests import ComicSearchReq
+from Models.response import ExceptionResponse, StandardResponse
+from Services.Database.database import get_db
+from Services.Limiter.limiter import limiter
+from Services.Security.user import get_current_user
+from Services.Modulator.manager import plugin_manager
+
+core_router = APIRouter(prefix="/core")
+
+
+@core_router.get("/ping")
+async def get_status() -> StandardResponse:
+    return StandardResponse(message="pong")

Routers/user.py

@@ -1,5 +1,10 @@
 import logging
+import typing
 
+import sqlalchemy
+
+from sqlalchemy import Column, Text
+from Models.plugins import IAuth
 from fastapi import APIRouter, Response, Depends, HTTPException, Form, Request
 from fastapi.security import OAuth2PasswordRequestForm
 from passlib.context import CryptContext
@@ -8,14 +13,20 @@
 from datetime import datetime, timedelta
 
 from Services.Database.database import get_db
+from Services.Modulator.manager import plugin_manager
 from Services.Limiter.limiter import limiter
 from Services.Security.user import (
-    ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token, get_current_user,
-    encrypt_src_password, decrypt_src_password)
+    ACCESS_TOKEN_EXPIRE_MINUTES,
+    create_access_token,
+    get_current_user,
+    encrypt_src_password,
+    decrypt_src_password,
+)
 from Models.database import UserDb, PwdDb
 from Models.user import Token, User
 from Models.response import ExceptionResponse, StandardResponse
-from Models.requests import SourceStorageReq, SourceLoginReq
+from Models.requests import SourceStorageReq
+from Utils.convert import sql_typecast
 
 user_router = APIRouter(prefix="/user")
 pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
@@ -24,27 +35,44 @@
 
 @user_router.post("/reg")
 @limiter.limit("5/minute")
-async def user_reg(request: Request, email: str = Form(), username: str = Form(), password: str = Form(),
-                   db: Session = Depends(get_db)):
-    if db.query(UserDb).filter(
-            UserDb.email == email or UserDb.username == username).first() is not None:  # type: ignore
+async def user_reg(
+    request: Request,
+    email: str = Form(),
+    username: str = Form(),
+    password: str = Form(),
+    db: Session = Depends(get_db),
+):
+    if (
+        db.query(UserDb)
+        .filter(UserDb.email == email or UserDb.username == username)
+        .first()
+        is not None
+    ):
         raise HTTPException(status_code=409, detail="User already exists")
 
-    db.add(UserDb(
-        uid=uuid4().hex,
-        email=email,
-        username=username,
-        hashed_password=pwd_ctx.hash(password),
-        created_at=datetime.now(),
-    ))
+    db.add(
+        UserDb(
+            uid=uuid4().hex,
+            email=email,
+            username=username,
+            hashed_password=pwd_ctx.hash(password),
+            created_at=datetime.now(),
+        )
+    )
     db.commit()
     return Response(status_code=201)
 
 
 @user_router.post("/login")
 @limiter.limit("5/minute")
-async def user_login(request: Request, body: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
-    user: UserDb = db.query(UserDb).filter(UserDb.username == body.username).first()  # type: ignore
+async def user_login(
+    body: OAuth2PasswordRequestForm = Depends(),
+    db: Session = Depends(get_db),
+):
+    user: UserDb | None = (
+        db.query(UserDb).filter(UserDb.username == body.username).first()
+    )
+
     if user is not None and pwd_ctx.verify(body.password, user.hashed_password):
         token = create_access_token(
             data={"sub": user.email, "id": user.uid},
@@ -60,41 +88,93 @@ async def user_profile(user: User = Depends(get_current_user)):
     return StandardResponse(data=user)
 
 
-@user_router.post("/{src_id}/login")
-async def src_login(body: SourceLoginReq):
-    pass
+@user_router.get("/{src}/login")
+async def src_login_info(src: str):
+    if (source := plugin_manager.get_source(src)) is None or not isinstance(
+        source.instance, IAuth
+    ):
+        raise ExceptionResponse.not_found
+
+    return StandardResponse(data={"required": True, "items": source.service["login"]})
+
+
+@user_router.post("/{src}/login")
+async def src_login(src: str, body: dict[str, str]):
+    if (source := plugin_manager.get_source(src)) is None or not isinstance(
+        source.instance, IAuth
+    ):
+        raise ExceptionResponse.not_found
+
+    if (
+        not isinstance(source.service.get("login"), list)
+        or source.service["login"] == []
+    ):
+        raise ExceptionResponse.not_found
+
+    result = source.instance.login(body)
 
 
 @user_router.post("/{src_id}/encrypt")
-async def encrypt_src_pwd(src_id: str, body: SourceStorageReq, db: Session = Depends(get_db),
-                          user: User = Depends(get_current_user)):
-    record: PwdDb = db.query(PwdDb).filter(
-        PwdDb.src_id == src_id and PwdDb.uid == user.uid and PwdDb.account == body.account).first()  # type: ignore
+async def encrypt_src_pwd(
+    src_id: str,
+    body: SourceStorageReq,
+    db: Session = Depends(get_db),
+    user: User = Depends(get_current_user),
+):
+    record: PwdDb | None = (
+        db.query(PwdDb)
+        .filter(
+            PwdDb.src_id == src_id
+            and PwdDb.uid == user.uid
+            and PwdDb.account == body.account
+        )
+        .first()
+    )
 
     if record is None:
-        db.add(PwdDb(src_id=src_id, uid=user.uid, account=body.account,
-                     pwd=encrypt_src_password(body.key_pwd, body.src_pwd)))
+        db.add(
+            PwdDb(
+                src_id=src_id,
+                uid=user.uid,
+                account=body.account,
+                pwd=encrypt_src_password(body.key_pwd, body.src_pwd),
+            )
+        )
         db.commit()
         return Response(status_code=201)
     else:
-        record.pwd = encrypt_src_password(body.key_pwd, body.src_pwd)
+        record.pwd = sql_typecast(
+            encrypt_src_password(body.key_pwd, body.src_pwd), Column[typing.Text]
+        )
         db.commit()
         return Response(status_code=200)
 
 
-@user_router.get("/{src_id}/accounts")
-async def get_src_accounts(src_id: str, db: Session = Depends(get_db), user: User = Depends(get_current_user)):
-    records: list[PwdDb] = db.query(PwdDb).filter(
-        PwdDb.src_id == src_id and PwdDb.uid == user.uid).all()  # type: ignore
+@user_router.get("/{src}/accounts")
+async def get_src_accounts(
+    src: str, db: Session = Depends(get_db), user: User = Depends(get_current_user)
+):
+    records: list[PwdDb] = (
+        db.query(PwdDb).filter(PwdDb.src_id == src and PwdDb.uid == user.uid).all()
+    )
 
     return StandardResponse(data=[record.account for record in records])
 
 
-@user_router.get("/{src_id}/decrypt")
-async def decrypt_src_pwd(src_id: str, account: str, db: Session = Depends(get_db),
-                          user: User = Depends(get_current_user)):
-    record: PwdDb = db.query(PwdDb).filter(
-        PwdDb.src_id == src_id and PwdDb.uid == user.uid and PwdDb.account == account).first()  # type: ignore
+@user_router.get("/{src}/decrypt")
+async def decrypt_src_pwd(
+    src: str,
+    account: str,
+    db: Session = Depends(get_db),
+    user: User = Depends(get_current_user),
+):
+    record: PwdDb | None = (
+        db.query(PwdDb)
+        .filter(
+            PwdDb.src_id == src and PwdDb.uid == user.uid and PwdDb.account == account
+        )
+        .first()
+    )
 
     if record is None:
         raise ExceptionResponse.not_found