# uvicorn main:app --host=10.10.3.10 --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 # db.gameinfo.createIndex({pubDate: -1}, {background: true}) # db.gameinfo.createIndex({discountLeftTime: -1}, {background: true}) # db.gameinfo.createIndex({isLowest: -1}, {background: true}) #db.gameinfo.createIndex({name: -1}, {background: true}) #db.gameinfo.createIndex({subName: -1}, {background: true}) # or db.gameinfo.ensureIndex({"isLowest":-1,"discountLeftTime":-1,"pubDate":-1}) tags_metadata = [ { "name": "common", "description": "Manage items. So _fancy_ they have their own docs.", "externalDocs": { "description": "Items external docs", "url": "https://fastapi.tiangolo.com/", }, }, ] orderby_list = { 1: "pubDate", # 上线时间 2: "discountLeftTime", #折扣截止时间 3: "isLowest", # 史低 4: "cutOff", # 折扣力度 5: "cuttime", # 价格变动时间 6: "price", # 当前价格 7: "mcScore" # 评分 } def create_app(): # application = FastAPI(dependencies=[Depends(get_token_header)], # openapi_tags=tags_metadata) application = FastAPI() 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, category: str = '', name: str = '', gameids: str = "", cutoff: bool = False, isLowest: bool = False, platform: int = 1, skip: int = 0, limit: int = 10, orderbyid: int = 1): db = request.app.state.mongo try: find_args = {} # db.gameinfo.find({'$or':[{"name":/アサツグトリ/},{"subName":'abc'}]}) if category: find_args['category'] = category if name: find_args['$or'] = [{ "name": re.compile(name) }, { "subName": re.compile(name) }] if cutoff: find_args["cutOff"] = {"$ne": 0} if isLowest: find_args["isLowest"] = {"$ne": 0} if gameids: ids = [] try: for item in gameids.split(","): ids.append(int(item)) find_args['oldGameId'] = {"$in": ids} except Exception as e: logger.error(f"get gamelist faild with gameids={gameids}") return JSONResponse( status_code=starlette.status. HTTP_500_INTERNAL_SERVER_ERROR, content=str(e), ) find_args["platform"] = platform if orderbyid not in orderby_list.keys(): return JSONResponse( status_code=starlette.status.HTTP_500_INTERNAL_SERVER_ERROR, content="orderbyid not define!", ) gamelist = db["gameinfo"].find( find_args, { "_id": 0, "oldGameId": 1, "name": 1, "banner": 1, "category": 1, "chinese": 1, "cutOff": 1, "discountLeftTime": 1, "isLowest": 1, "originPrice": 1, "platform": 1, "price": 1, "icon": 1, "mcScore": 1, "priceCountry": 1, "subName": 1 }).sort(orderby_list[orderbyid], -1).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, nums: int = 0): db = request.app.state.mongo if nums == 0: gameprice = db["gameprice"].find_one({"oldGameId": oldGameId}, {"_id": 0}) else: gameprice = db["gameprice"].find_one({"oldGameId": oldGameId}, {"_id": 0}) price = gameprice.get('prices', []) if price: gameprice['prices'] = price[:nums] else: logger.error(f"get price with {oldGameId} failed\n {gameprice}") gameprice['prices'] = [] 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}) if not gameinfoext: return JSONResponse( status_code=starlette.status.HTTP_200_OK, content="", ) dlc_data = gameinfoext.get('dlcList') new_data = list() for one_line in dlc_data: temp = {} temp['name'] = one_line['name'] temp['oldGameId'] = one_line['oldGameId'] temp['price'] = one_line['price'] temp['icon'] = one_line['icon'] new_data.append(temp) gameinfoext['dlcList'] = new_data 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)