python_extract_service/data_verification.py

import uvicorn
import warnings
from fastapi import FastAPI, UploadFile, File
from openpyxl import load_workbook
from openpyxl.utils.cell import coordinate_from_string
from openpyxl.comments import Comment
from openpyxl.styles import PatternFill
from fastapi.middleware.cors import CORSMiddleware

warnings.filterwarnings("ignore")

app = FastAPI()

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 "脱贫户信息查询" not in sheet_names:
        print("读取不到指定的sheet页")
        return {"code": 500, "msg": "读取不到指定的sheet页--脱贫户信息查询"}

    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}


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)