python_extract_service/data_verification.py

'''
coding:utf-8
@Software:PyCharm
@Time:2023/2/20 13:17
@Author:zhangnf
@Desc:
'''

from fastapi import FastAPI, UploadFile ,File
from openpyxl import load_workbook
import warnings
from openpyxl.utils.cell import coordinate_from_string
from openpyxl.comments import Comment
from openpyxl.styles import PatternFill
warnings.filterwarnings('ignore')
import uvicorn

app = FastAPI()
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

@app.post("/uploadfile")
async def create_upload_file(file: UploadFile = File(...)):
    # print(file.filename)
    contents = await file.read()

    savename = "/data/download/" + file.filename
    #savename = "uploadfile/" + file.filename
    with open(savename,"wb" ) as f:
        f.write(contents)
    # 读取excel表
    workbook = load_workbook(savename)
    # 获取指定的sheet
    sheet_names = workbook.sheetnames
    if "脱贫户信息查询" in sheet_names:
        sheet = workbook["脱贫户信息查询"]
        title_row_num = 0

        # 读取前5行，正常应该有字段名了
        row_range = sheet[1:5]
        for r in row_range:
            for c in r:
                if ("户主编号" == c.value):
                    title_row_num = c.row
        # 获取字段名对应的列
        title_dict = {}
        title_rows = sheet[title_row_num]
        # 遍历字段名所在行的所有单元格
        for title_cell in title_rows:
            x, y = coordinate_from_string(title_cell.coordinate)
            title_dict[title_cell.value] = x

        # print(title_dict)


        # 开始读取表格内容
        # print(sheet.max_row)
        read_data(sheet,title_row_num+1,sheet.max_row,title_dict)

        # 保存文档
        workbook.save(savename)

        return {"code" :200 ,"msg": "分析完成，请点击下载查看分析结果","fileName":file.filename}

    else:
        print("读取不到指定的sheet页")
        return {"code" :500,"msg": "读取不到指定的sheet页--脱贫户信息查询"}

def read_data(ws,start_row,end_row,title_dict):

    #监测对象致（返）贫风险非最新设计的风险类型
    for i in range(start_row, end_row):
        check_poverty_causes(ws,i,title_dict)
        check_identitycard_length(ws, i, title_dict)

def check_poverty_causes(ws,row_num,title_dict):
    poverty_causes = ws[f"{title_dict['主要致贫原因']}{row_num}"].value
    # 致贫原因列表
    imageTypeList = ['因病', '因学', '因残', '因自然灾害','因意外事故','因产业项目失败','因务工就业不稳','缺劳动力','因房','因水','其他（填写备注）']
    if poverty_causes not in imageTypeList:
        ws[f"{title_dict['主要致贫原因']}{row_num}"].comment = Comment(text="21.监测对象致（返）贫风险非最新设计的风险类型", author="system")
        yellow_fill = PatternFill(patternType='solid', fgColor='FFFF00')
        ws[f"{title_dict['主要致贫原因']}{row_num}"].fill = yellow_fill

def check_identitycard_length(ws,row_num,title_dict):
    identitycard = ws[f"{title_dict['户主证件号码']}{row_num}"].value
    if len(identitycard) not in [15,18,20,22]:
        ws[f"{title_dict['户主证件号码']}{row_num}"].comment = Comment(text="31.监测对象家庭成员证件号码位数异常（证件号码非15、18、20、22位）", author="system")
        yellow_fill = PatternFill(patternType='solid', fgColor='FFFF00')
        ws[f"{title_dict['户主证件号码']}{row_num}"].fill = yellow_fill

if __name__ == '__main__':
    uvicorn.run('data_verification:app',host='localhost',port=8000,reload=True)