项目框架、案例接口

.gitignore

@@ -0,0 +1,4 @@
+.vscode
+.idea
+
+__pycache__

app/case/case01.py

@@ -0,0 +1,98 @@
+from fastapi import APIRouter, Query, Form
+
+from app.common import util
+from app.common.schema import ApiResultSchema
+from app.common.response import ApiResult
+
+router = APIRouter()
+
+
+@router.post("/login", name="用户登录", description="""
+#### 测试数据
+- username: `zhangsan`
+- pwssword: `e10adc3949ba59abbe56e057f20f883e`
+- sign: `h9Ki9eMChBhjWU/aCeqlkLcCNBchnfIt3U2V6MSIF/Rl0WoD1iqM8TKH3U8ZmFoq`
+""", response_model=ApiResultSchema)
+async def login(
+        username: str = Form(..., title="用户名", description="用户名"),
+        password: str = Form(..., title="密码", description="密码，算法：`md5(password)`"),
+        sign: str = Form(..., title="签名", description="签名，算法：`aes(username+password)`"),
+):
+    if sign is None or util.aes_decrypt(_key, sign) != username + password:
+        return ApiResult.error_msg("签名不正确")
+    if username != _username:
+        return ApiResult.error_msg("用户名不存在")
+    if password != util.md5(_password):
+        return ApiResult.error_msg("密码不正确")
+    return ApiResult.success_data({"token": _token})
+
+
+@router.get("/list", name="小说列表", description="""
+## 测试数据
+- token: `e9d8edc1-400c-436f-880c-9cd8392745ca`
+""", response_model=ApiResultSchema)
+async def list(
+        token: str = Query(..., title="令牌", description="令牌"),
+):
+    if token.encode("utf-8") != _token:
+        return ApiResult.error_msg("无权限访问")
+    return ApiResult.success_data({"list": _list_data})
+
+
+# DATA
+_key = "4EFA6860A3539903".encode("utf-8")
+_username = "zhangsan"
+_password = "123456"
+_token = "e9d8edc1-400c-436f-880c-9cd8392745ca".encode("utf-8")
+_list_data = [
+    {
+        "image": "http://static.zongheng.com/upload/cover/d0/27/d027db9b2ee2695e8a3d4804dc20da39.jpeg",
+        "name": "死囚生存游戏",
+        "desc": "2050年，世界各国联合研发出一款用于惩戒重犯的虚拟现实系统：【溟河系统】，该系统可剥夺犯人意识并永久禁锢在虚拟世界中，以达到人性化惩戒重犯的目的。充斥杀机的战场，蛰伏的攻击型生物，13场搏杀游戏，12个死刑犯，3条溟河法则，1个幸存者名额，淘汰者将永远留在系统，成为行尸走肉，为自己【赎罪】；获胜者将重获自由，前罪全免，获得【救赎】……一念天堂，一念地狱，杀与被杀，屠夫与羔羊，你选择哪个？ 【注：溟河三法则：当第一声警报响起时，城市的消防、警卫、医疗、安保等系统会彻底瘫痪，监狱系统将对外开放，游戏进入犯罪无罪的状态。当第二声警报响起时，玩家可以开始屠杀其他玩家，12个玩家的游戏正式开始。当第三声警报响起时，游戏中将会有攻击型生物、猎人出没，追杀游戏玩家】 【注：多主角，剧情向，游戏线与现实线交叉。】"
+    },
+    {
+        "image": "http://static.zongheng.com/upload/cover/54/be/54be6939cd7eeb3d173c09555149a3c5.jpeg",
+        "name": "阴阳灵探",
+        "desc": "鬼朔和王大灵是被警校开除的学生，二人合开了见鬼灵侦探社，谁知竟接到了一桩桩奇异的案件。。。原来这世间除了有人行凶外，还有其它。。。"
+    },
+    {
+        "image": "http://static.zongheng.com/upload/cover/4c/c5/4cc5125da0ae3f9f17a35ad375cd1c56.jpeg",
+        "name": "乱古剑帝",
+        "desc": "人间的九月，下起了一场剑雨。我会循着远古战歌的召唤，亦是在无尽的杀伐中，轻声吟唱。"
+    },
+    {
+        "image": "http://static.zongheng.com/upload/cover/bd/ef/bdef0b3920193e62fcf217d2d63713c2.jpeg",
+        "name": "狂枭",
+        "desc": "生当作人杰，死亦为鬼雄。 这是老子的人生，全部所有内容。"
+    },
+    {
+        "image": "http://static.zongheng.com/upload/cover/f1/5a/f15ae195aba87bfb6eb0ecce403258e5.jpeg",
+        "name": "道断修罗",
+        "desc": "少年李夜，生下即渡劫，誓要走一条与众不同的修行之路。随先生上天山修行，披一肩风雨，斩一山飞雪......下山回城遇退婚风波，狠心女人欲要斩草除根......行一路烟雨，踏一江春水......少年国师笑傲南疆，灭敌于南云城前......遇外敌内匪，踏修罗刀山......身陷黑洞，降临修罗域，天途道断，且看我如何踏天而行，搅动两界风云。"
+    },
+    {
+        "image": "http://static.zongheng.com/upload/cover/f4/1f/f41f7836f0d7a1b9d81d4c5fa9e230aa.jpeg",
+        "name": "血色圣歌",
+        "desc": "开场满级主动带飞零级菜鸡：信仰我，带你傲立九界之巅！作为传说中的至尊纯血再世，这口气，墨岚能忍？？ 他能！师父，爸爸，带我飞吧，我不想努力啦！"
+    },
+    {
+        "image": "http://static.zongheng.com/upload/cover/c5/27/c527797d2e392d00831add17d759101f.jpeg",
+        "name": "我怎么就不能飞升",
+        "desc": "泰山之巅，赵东来双手枕着脑袋，仰躺在一块大石头上，嘴里叼着一根不知道从哪拔来的野草。“贼老天……”赵东来望着天幕，轻声呢喃着：“我做了五十年的好事，救治了无数病人，连碰瓷的我都好心给他们治一治” “我就想问问，为什么我如今修为都快撑破这片天地了，怎么就还没到飞升境？” “我……怎么就不能飞升！”"
+    },
+    {
+        "image": "http://static.zongheng.com/upload/cover/c1/d1/c1d151d7630e03942ad3cb46be0be562.jpeg",
+        "name": "神诵九歌",
+        "desc": "大夏北境上，几个士兵发现亡灵来袭！ 有易部落内，首领之子即将迎来冒险！ 究竟是怎样的大手，才能弹奏出这壮丽的九歌？ 长夜将至，神、人、亡者的命运又当何去何从？"
+    },
+    {
+        "image": "http://static.zongheng.com/upload/cover/3b/18/3b182c63025a5b518a6a781943d22d7f.jpeg",
+        "name": "风云龙婿",
+        "desc": "三年潜伏，一朝归来，他依然是王者，曾经失去的，统统都要拿回来，在世人眼中，他是战神，但在心爱的人眼里，他只是个普通的男人。"
+    },
+    {
+        "image": "http://static.zongheng.com/upload/cover/ad/51/ad5199ae6850e50ab8e8d220ec8e2988.jpeg",
+        "name": "妖孽哪里走",
+        "desc": "“建国以后不许成精！”苏铭霸气的说道。某妖弱弱的问道：“那建国前成精的怎么处理？”"
+    }
+]

app/case/case02.py

@@ -0,0 +1,43 @@
+import base64
+import binascii
+import random
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Form
+
+from app.common import util
+from app.common.schema import ApiResultSchema
+from app.common.response import ApiResult
+
+router = APIRouter()
+
+
+@router.post("/search", name="查询网盘提取码", description="""
+#### 测试数据
+- url: `123456M3lrcWRwRm9OSmw5S2RQK3RjUEFMT3RYSld3MDVldDVDUTYxaWFVZjg5WT0=`
+- device_id: `123456`
+- timestamp: `1610640156`
+""", response_model=ApiResultSchema)
+async def search(
+        url: str = Form(..., title="百度网盘分享链接", description="百度网盘分享链接，算法：`device_id + base64(aes(url))`"),
+        device_id: str = Form(..., title="设备标识", description="设备标识"),
+        timestamp: str = Form(..., title="时间戳", description="时间戳"),
+):
+    if url is None or device_id is None or timestamp is None:
+        return ApiResult.error_msg("参数错误")
+    urlBase64 = url.replace(device_id, "")
+    try:
+        urlAES = base64.b64decode(urlBase64)
+    except:
+        return ApiResult.error_msg("参数错误")
+    urlFinal = util.aes_decrypt(_key, urlAES)
+    if urlFinal is None or urlFinal.isspace() or len(urlFinal) == 0:
+        return ApiResult.error_msg("参数错误")
+    if 1 - urlFinal.startswith("https://pan.baidu.com/"):
+        return ApiResult.error_msg("非百度网盘分享链接")
+    code = str(random.randint(1000, 9999))
+    codeHex = binascii.hexlify(code.encode("utf-8")).decode('utf-8')
+    return ApiResult.success_data({"code": codeHex})
+
+
+# DATA
+_key = "4EFA6860A3539903".encode("utf-8")

app/case/case03.py

@@ -0,0 +1,28 @@
+from fastapi import APIRouter, Depends, HTTPException, Query, Form
+
+from app.common.response import ApiResult
+from app.common.schema import ApiResultSchema
+from app.case.case01 import _username, _password, _list_data
+
+router = APIRouter()
+
+
+@router.post("/login", name="用户登录", description="""
+#### 测试数据
+- username: `zhangsan`
+- pwssword: `123456`
+""", response_model=ApiResultSchema)
+async def login(
+        username: str = Form(..., title="用户名", description="用户名"),
+        password: str = Form(..., title="密码", description="密码"),
+):
+    if username != _username:
+        return ApiResult.error_msg("用户名不存在")
+    if password != _password:
+        return ApiResult.error_msg("密码不正确")
+    return ApiResult.success()
+
+
+@router.get("/list", name="小说列表", description="", response_model=ApiResultSchema)
+async def list():
+    return ApiResult.success_data({"list": _list_data})

app/common/const.py

@@ -0,0 +1,4 @@
+SUCCESS = 0
+ERROR = 1
+SUCCESS_MSG = "SUCCESS"
+ERROR_MSG = "ERROR"

app/common/response.py

@@ -0,0 +1,49 @@
+from typing import Any
+
+from fastapi.encoders import jsonable_encoder
+
+from app.common import const
+
+
+class ApiResult:
+
+    def __init__(self, code, message, data) -> None:
+        self.code = code
+        self.message = message
+        self.data = data
+
+    code: int
+    message: str
+    data: Any
+
+    @staticmethod
+    def success():
+        return jsonable_encoder(ApiResult(const.SUCCESS, const.SUCCESS_MSG, data=None))
+
+    @staticmethod
+    def success_msg(message: str):
+        return jsonable_encoder(ApiResult(const.SUCCESS, message, data=None))
+
+    @staticmethod
+    def success_data(data):
+        return jsonable_encoder(ApiResult(const.SUCCESS, const.SUCCESS_MSG, data))
+
+    @staticmethod
+    def success_str_data(message: str, data):
+        return jsonable_encoder(ApiResult(const.SUCCESS, message, data))
+
+    @staticmethod
+    def error():
+        return jsonable_encoder(ApiResult(const.ERROR, const.ERROR, data=None))
+
+    @staticmethod
+    def error_msg(message: str):
+        return jsonable_encoder(ApiResult(const.ERROR, message, data=None))
+
+    @staticmethod
+    def error_data(data):
+        return jsonable_encoder(ApiResult(const.ERROR, const.ERROR, data))
+
+    @staticmethod
+    def error_str_data(message: str, data):
+        return jsonable_encoder(ApiResult(const.ERROR, message, data))

app/common/schema.py

@@ -0,0 +1,9 @@
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class ApiResultSchema(BaseModel):
+    code: int = Field(..., title="状态码", description="0=成功；!0：失败")
+    message: str = Field(..., title="响应消息", description="响应消息")
+    data: Any = Field(None, title="响应数据", description="响应数据")

app/common/util.py

@@ -0,0 +1,32 @@
+import base64
+import hashlib
+
+from Crypto.Cipher import AES
+
+
+def md5(text):
+    text = text.encode(encoding='utf-8')
+    m = hashlib.md5()
+    m.update(text)
+    return m.hexdigest()
+
+
+def aes_encrypt(key, data):
+    try:
+        mode = AES.MODE_ECB
+        padding = lambda s: s + (16 - len(s) % 16) * chr(16 - len(s) % 16)
+        cryptos = AES.new(key, mode)
+        cipher_text = cryptos.encrypt(padding(data).encode("utf-8"))
+        return base64.b64encode(cipher_text).decode("utf-8")
+    except Exception as e:
+        return ""
+
+
+def aes_decrypt(key, data):
+    try:
+        cryptos = AES.new(key, AES.MODE_ECB)
+        decrpytBytes = base64.b64decode(data)
+        meg = cryptos.decrypt(decrpytBytes).decode('utf-8')
+        return meg[:-ord(meg[-1])]
+    except Exception as e:
+        return ""

app/main.py

@@ -0,0 +1,61 @@
+from fastapi import Depends, FastAPI, Request, status
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
+from fastapi.staticfiles import StaticFiles
+from fastapi.responses import HTMLResponse
+from fastapi.templating import Jinja2Templates
+from fastapi.responses import JSONResponse
+
+from .case import case01, case02, case03
+
+app = FastAPI(
+    debug=False,
+    title="API Library",
+    description="",
+    version="1.0.0",
+)
+
+# 设置静态资源目录
+app.mount("/static", StaticFiles(directory="static"), name="static")
+
+# 设置页面模板目录
+templates = Jinja2Templates(directory="templates")
+
+
+# 定义接管RequestValidationError的方法，捕捉422报错并进行自定义处理
+@app.exception_handler(RequestValidationError)
+async def request_validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
+    return JSONResponse(
+        status_code=status.HTTP_200_OK,
+        content={
+            "code": 1,
+            "message": "Parameter error",
+            "data": jsonable_encoder(exc.errors())
+        },
+    )
+
+
+# 案例API
+app.include_router(
+    case01.router, 
+    prefix="/api/case/01",
+    tags=["案例01（小小图书）"],
+)
+app.include_router(
+    case02.router, 
+    prefix="/api/case/02",
+    tags=["案例02（网盘助手）"],
+)
+app.include_router(
+    case03.router, 
+    prefix="/api/case/03",
+    tags=["案例03（小小图书2）"],
+)
+
+# 其他API ...
+
+
+@app.get("/", response_class=HTMLResponse, include_in_schema=False)
+async def root(request: Request):
+    attr_text = "API Library"
+    return templates.TemplateResponse("index.html", {"request": request, "attr_text": attr_text})

requirements.txt

@@ -0,0 +1,3 @@
+pydantic==1.7.3
+fastapi==0.63.0
+pycryptodome==3.9.9

run.py

@@ -0,0 +1,5 @@
+# uvicorn app.main:app --reload
+import uvicorn
+
+if __name__ == '__main__':
+    uvicorn.run(app='app.main:app', host="0.0.0.0", port=8000, reload=True, debug=False)

templates/index.html

@@ -0,0 +1,20 @@
+<!DOCTYPE html>
+<html>
+<head>
+    <meta charset="UTF-8">
+    <link rel="shortcut icon" href="/static/favicon.png">
+    <title>FastAPI</title>
+    <style>
+        body { text-align: center }
+        h2 { font-weight: normal }
+        a { text-decoration: none }
+        a:hover { color: rgb(24, 111, 175) }
+        a { color: rgb(50, 50, 159) }
+    </style>
+</head>
+<body>
+    <h1>{{ attr_text }}</h1>
+    <h2><a href="/docs">Swagger UI</a></h2>
+    <h2><a href="/redoc">ReDoc UI</a></h2>
+</body>
+</html>