dataplatform/APPData/Router/MacroDataRouter.py

import datetime
import math
import pandas as pd
from fastapi import APIRouter, Depends, UploadFile, File, HTTPException
from sqlalchemy.orm import Session

from APPData.Crud import MacroDataCrud
from APPData.Schemas import CompanySchemas
from Utils.DataBase.MongoHelperUtils import get_mongodb
from Utils.DataBase.SqlAlchemyUtils import get_db

router = APIRouter(
    tags=["宏观数据"],
    prefix="/api/macro_data"
)


@router.post("/upload", summary="上传经济实力表")
async def func01(region: str, excel: UploadFile = File(...), db: Session = Depends(get_db)):
    mongodb = get_mongodb()
    contents = await excel.read()
    data = pd.read_excel(contents, skiprows=1)
    data = data.set_index('序号')

    for num in data.index.values:
        df_line = data.loc[num, data.columns.values].to_dict()
        df_line['省/直辖市'] = region
        for key, value in df_line.items():
            try:
                if math.isnan(value):
                    df_line[key] = None
            except TypeError:
                continue
        MacroDataCrud.upsert_data(mongodb=mongodb, data=df_line)
    return {"info": "Success"}


@router.post("/view", summary="查看数据")
def func03(req_body: CompanySchemas.MacroSearch):
    mongodb = get_mongodb()
    data = MacroDataCrud.find_data(mongodb=mongodb, condition=req_body)
    return data