jump_collect/main.py

# uvicorn main:app --host=127.0.0.1 --port=8030 --reload
from pydantic.fields import T
from config.config import settings
from fastapi import Depends, FastAPI, BackgroundTasks, Request
from fastapi.responses import JSONResponse
from dependencies import get_token_header
from scripts.common.mongodb import get_mongo
from typing import Optional
from scripts.logger import logger
from fastapi.middleware.cors import CORSMiddleware
import starlette
import re
from pydantic import BaseModel, Field, EmailStr
import json
# from apscheduler.events import EVENT_JOB_EXECUTED
# from jobs.jobs import Schedule, job_execute
tags_metadata = [
    {
        "name": "common",
        "description": "Manage items. So _fancy_ they have their own docs.",
        "externalDocs": {
            "description": "Items external docs",
            "url": "https://fastapi.tiangolo.com/",
        },
    },
]


def create_app():
    application = FastAPI(dependencies=[Depends(get_token_header)],
                          openapi_tags=tags_metadata)
    return application


app = create_app()
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class FindArgs(BaseModel):
    oldGameId: int
    name: str


@app.on_event("startup")
async def startup_event():
    app.state.mongo = await get_mongo()


@app.on_event("shutdown")
async def shutdown_event():
    app.state.mongo.close()
    await app.state.mongo.wait_close()


@app.get("/getPlatform")
async def getPlatform(request: Request, platformAlias: str):
    db = request.app.state.mongo
    existing_platform = db["platform"].find_one(
        {"platformAlias": platformAlias}, {"_id": 0})
    logger.info(f"starting get Platform with {platformAlias}!")
    return JSONResponse(
        status_code=starlette.status.HTTP_200_OK,
        content=existing_platform,
    )


@app.get("/getgamelist")
async def getgamelist(
    request: Request,
    cutoff: bool,
    isLowest: bool,
    platform: int = 1,
    skip: int = 0,
    limit: int = 10,
):
    db = request.app.state.mongo

    try:
        find_args = {}
        if cutoff:
            find_args["cutOff"] = {"$ne": 0}
        if isLowest:
            find_args["isLowest"] = {"$ne": 0}
        find_args["platform"] = platform
        gamelist = db["gameinfo"].find(find_args, {
            "_id": 0,
            "oldGameId": 1,
            "name": 1
        }).sort("_id").skip(skip).limit(limit)
        logger.info(f"get gamelist with {find_args}!")
        return JSONResponse(
            status_code=starlette.status.HTTP_200_OK,
            content=list(gamelist),
        )
    except Exception as e:
        return JSONResponse(
            status_code=starlette.status.HTTP_500_INTERNAL_SERVER_ERROR,
            content=e,
        )


@app.get("/getgameinfo")
async def getgameinfo(request: Request, oldGameId: int = 0, name: str = ""):
    db = request.app.state.mongo
    if oldGameId:
        gameinfo = db["gameinfo"].find({"oldGameId": oldGameId}, {"_id": 0})
    elif name:
        gameinfo = db["gameinfo"].find({"name": re.compile(name)}, {"_id": 0})
    else:
        logger.error(f"get gameinfo with {oldGameId} {name}!")
        return JSONResponse(
            status_code=starlette.status.HTTP_400_BAD_REQUEST,
            content={"message": "参数未提供"},
        )
    logger.info(f"get gameinfo with {oldGameId} {name}!")
    return JSONResponse(
        status_code=starlette.status.HTTP_200_OK,
        content=list(gameinfo),
    )


@app.get("/getgameprice")
async def getgameprice(request: Request, oldGameId: int):
    db = request.app.state.mongo
    gameprice = db["gameprice"].find_one({"oldGameId": oldGameId}, {"_id": 0})
    logger.info(f"get gameprice with {oldGameId} !")
    return JSONResponse(
        status_code=starlette.status.HTTP_200_OK,
        content=gameprice,
    )


@app.get("/getgameinfoext")
async def getgameinfoext(request: Request, oldGameId: int):
    db = request.app.state.mongo
    gameinfoext = db["gameinfoext"].find_one({"oldGameId": oldGameId},
                                             {"_id": 0})
    logger.info(f"get gameinfoext with {oldGameId} !")
    return JSONResponse(
        status_code=starlette.status.HTTP_200_OK,
        content=gameinfoext,
    )


@app.get("/gethistoryprice")
async def gethistoryprice(request: Request, oldGameId: int):
    db = request.app.state.mongo
    history_price = db["history_price"].find_one({"oldGameId": oldGameId},
                                                 {"_id": 0})
    logger.info(f"get historyprice with {oldGameId} !")
    return JSONResponse(
        status_code=starlette.status.HTTP_200_OK,
        content=history_price,
    )


if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main:app',
                host="0.0.0.0",
                port=8030,
                reload=True,
                debug=True)