提交表格填写纠错提示的代码

.gitignore

@@ -0,0 +1,2 @@
+venv
+__pycache__

data_verification.py

@@ -0,0 +1,399 @@
+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
+from datetime import datetime
+from fastapi.responses import FileResponse
+
+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 = "download_cache/" + 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[sheet_names[0]]
+    title_row_num = 0
+
+    # 读取前5行，正常应该有字段名了
+    row_range = sheet[1:5]
+    for i, r in enumerate(row_range):
+        for j, c in enumerate(r):
+            print(f"第{i + 1 }行，第{j}列，值：{c.value}")
+            if "户主编号" == c.value or "户编号" == 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 FileResponse(savename)
+
+
+# return {"code": 200, "msg": "分析完成，请点击下载查看分析结果", "fileName": file.filename}
+
+
+def calculate_age_from_id_number(id_number):
+    """根据传入的身份证号码，提取出出生年月日，结合代码运行时的当天时间，返回当前年龄（周岁）"""
+    age = 0
+    if len(id_number) != 18:
+        return age
+
+    birthday = id_number[6:14]
+    birthday = datetime.strptime(birthday, "%Y%m%d")
+    today = datetime.now()
+    age = today.year - birthday.year
+    if today.month < birthday.month or (today.month == birthday.month and today.day < birthday.day):
+        age -= 1
+
+    return age
+
+
+def verify_age_region_for_education(education, age):
+    """根据传入的学历和年龄判断该年龄是否符合该学历"""
+    # 义务教育年龄段
+    # if education == "" and (age >= 6 and age <= 15):
+    #     return False
+
+    education_for_age = {
+        "学龄前儿童": (0, 6),
+        "学前教育": (3, 6),
+        "小学": (6, 12),
+        "七年级": (12, 13),
+        "八年级": (13, 14),
+        "九年级": (14, 15),
+        "初中": (12, 15),
+        "高中": (15, 18),
+        "中专": (15, 18),
+        "中职": (15, 18),
+        "高职": (18, 22),
+        "高专": (18, 22),
+        "大专": (18, 20),
+        "本科": (18, 22),
+        "硕士": (22, 25),
+    }
+    for education_key in education_for_age.keys():
+        if education_key in education:
+            age_range = education_for_age[education_key]
+            if age < age_range[0] or age > age_range[1]:
+                return False
+            break
+
+    return True
+
+
+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)
+        check_deformity(ws, i, title_dict)
+        check_education_level(ws, i, title_dict)
+        check_risk_type(ws, i, title_dict)
+        check_assistance(ws, i, title_dict)
+
+
+# 筛查方法
+def check_poverty_causes(ws, row_num, title_dict):
+    """筛查主要致贫原因是否合规"""
+    main_reason = "主要致贫原因"
+    if main_reason not in title_dict:
+        return
+    poverty_causes = ws[f"{title_dict[main_reason]}{row_num}"].value
+    # 致贫原因列表
+    imageTypeList = ["因病", "因学", "因残", "因自然灾害", "因意外事故", "因产业项目失败", "因务工就业不稳", "缺劳动力", "因房", "因水", "其他（填写备注）"]
+    if poverty_causes not in imageTypeList:
+        target = ws[f"{title_dict[main_reason]}{row_num}"]
+        comment_and_fill_yellow_for(target, "21.监测对象致（返）贫风险非最新设计的风险类型")
+
+
+def check_identitycard_length(ws, row_num, title_dict):
+    """筛查身份证号码是否合规"""
+    info_number = "户主证件号码"
+    if info_number not in title_dict:
+        return
+    identitycard = ws[f"{title_dict[info_number]}{row_num}"].value
+    if len(identitycard) not in [15, 18, 20, 22]:
+        target = ws[f"{title_dict[info_number]}{row_num}"]
+        comment_and_fill_yellow_for(target, "31.监测对象家庭成员证件号码位数异常（证件号码非15、18、20、22位）")
+
+
+def check_deformity(ws, row_num, title_dict):
+    """筛查是否不符合残疾劳动标准"""
+    condition = "健康状况"
+    skill = "劳动技能"
+    if condition not in title_dict or skill not in title_dict:
+        return
+    condition_value = ws[f"{title_dict[condition]}{row_num}"].value
+    skill_value = ws[f"{title_dict[skill]}{row_num}"].value
+    if "残疾" in condition_value and skill_value == "弱劳动力或半劳动力":
+        target = ws[f"{title_dict[skill]}{row_num}"]
+        comment_and_fill_yellow_for(target, "35.一、二级（重度）肢体残疾脱贫人口有普通劳动能力")
+
+
+def check_education_level(ws, row_num, title_dict):
+    """筛查在校生状况填写是否不符合年龄"""
+    info_number = "证件号码"
+    if info_number not in title_dict:
+        return
+    identitycard = ws[f"{title_dict[info_number]}{row_num}"].value
+    if len(identitycard) != 18:
+        return
+
+    education_key = "文化程度"
+    is_student_key = "在校生状况"
+
+    if education_key not in title_dict or is_student_key not in title_dict:
+        return
+    education = ws[f"{title_dict[education_key]}{row_num}"].value
+    is_student = ws[f"{title_dict[is_student_key]}{row_num}"].value
+
+    if len(education) == 0 and len(is_student) == 0:
+        target = ws[f"{title_dict[education_key]}{row_num}"]
+        comment_and_fill_yellow_for(target, "文化程度，在校生状况均未填写")
+
+        target_is_student = ws[f"{title_dict[education_key]}{row_num}"]
+        comment_and_fill_yellow_for(target_is_student, "文化程度，在校生状况均未填写")
+        return
+
+    if len(is_student) == 0:
+        return
+
+    age = calculate_age_from_id_number(identitycard)
+    age_is_reliable = verify_age_region_for_education(is_student, age)
+    if not age_is_reliable:
+        target = ws[f"{title_dict[is_student_key]}{row_num}"]
+        comment_and_fill_yellow_for(target, "在校生状况与年龄不符")
+
+
+def check_risk_type(ws, row_num, title_dict):
+    """筛查风险类型，家庭信息是否合规"""
+
+    def get_item_values_for(items):
+        result = []
+        for item in items:
+            if item not in title_dict:
+                continue
+            value = ws[f"{title_dict[item]}{row_num}"].value
+            if value is not None and len(value) > 0:
+                result.append((item, value))
+        return result
+
+    def get_key_for(i):
+        key = f"致贫风险{i}"
+        return key
+
+    risks = []
+    for i in range(1, 6):
+        key = get_key_for(i)
+        if key not in title_dict:
+            return
+        risk = ws[f"{title_dict[key]}{row_num}"].value
+        if risk is not None and len(risk) > 0:
+            risks.append((risk, i))
+
+    print(f"risks is {risks}")
+
+    # 定义：健康帮扶，"综合保障，社会帮扶，义务教育保障, 教育帮扶, 住房安全保障, 搬迁, 饮水安全保障, 产业帮扶, 就业帮扶, 金融帮扶, 公益岗位帮扶等常量
+    HEALTH_SUPPORT = "健康帮扶"
+    COMPREHENSIVE_GUARANTEE = "综合保障"
+    SOCIAL_SUPPORT = "社会帮扶"
+    OBLIGATORY_EDUCATION_GUARANTEE = "义务教育保障"
+    EDUCATION_SUPPORT = "教育帮扶"
+    HOUSING_SECURITY_GUARANTEE = "住房安全保障"
+    RELOCATION = "搬迁"
+    DRINKING_WATER_SECURITY_GUARANTEE = "饮水安全保障"
+    INDUSTRY_SUPPORT = "产业帮扶"
+    EMPLOYMENT_SUPPORT = "就业帮扶"
+    FINANCIAL_SUPPORT = "金融帮扶"
+    PUBLIC_WELFARE_SUPPORT = "公益岗位帮扶"
+
+    must_selected_option = []
+    forbinddens_option = []
+    for risk, i in risks:
+        if risk == "因病":
+            must_selected_option = [HEALTH_SUPPORT, COMPREHENSIVE_GUARANTEE, SOCIAL_SUPPORT]
+            forbinddens_option = [
+                HOUSING_SECURITY_GUARANTEE,
+                DRINKING_WATER_SECURITY_GUARANTEE,
+                OBLIGATORY_EDUCATION_GUARANTEE,
+                EDUCATION_SUPPORT,
+            ]
+        elif risk == "因学":
+            must_selected_option = [OBLIGATORY_EDUCATION_GUARANTEE, EDUCATION_SUPPORT, SOCIAL_SUPPORT]
+            forbinddens_option = [
+                HOUSING_SECURITY_GUARANTEE,
+                DRINKING_WATER_SECURITY_GUARANTEE,
+                EDUCATION_SUPPORT,
+            ]
+        elif risk == "因残":
+            must_selected_option = [COMPREHENSIVE_GUARANTEE, SOCIAL_SUPPORT]
+            forbinddens_option = [
+                HOUSING_SECURITY_GUARANTEE,
+                DRINKING_WATER_SECURITY_GUARANTEE,
+                OBLIGATORY_EDUCATION_GUARANTEE,
+                EDUCATION_SUPPORT,
+            ]
+        elif risk == "因产业项目失败" or risk == "因务工就业不稳" or risk == "缺劳动力":
+            must_selected_option = []
+            forbinddens_option = [
+                HOUSING_SECURITY_GUARANTEE,
+                DRINKING_WATER_SECURITY_GUARANTEE,
+                OBLIGATORY_EDUCATION_GUARANTEE,
+                EDUCATION_SUPPORT,
+                HEALTH_SUPPORT,
+            ]
+        elif risk == "住房出现安全问题":
+            must_selected_option = [HOUSING_SECURITY_GUARANTEE, RELOCATION]
+            forbinddens_option = [
+                DRINKING_WATER_SECURITY_GUARANTEE,
+                HEALTH_SUPPORT,
+                OBLIGATORY_EDUCATION_GUARANTEE,
+                EDUCATION_SUPPORT,
+                COMPREHENSIVE_GUARANTEE,
+            ]
+        elif risk == "家庭成员中有义务教育阶段适龄儿童少年失学辍学":
+            must_selected_option = [OBLIGATORY_EDUCATION_GUARANTEE]
+            forbinddens_option = [
+                INDUSTRY_SUPPORT,
+                EMPLOYMENT_SUPPORT,
+                FINANCIAL_SUPPORT,
+                PUBLIC_WELFARE_SUPPORT,
+                HOUSING_SECURITY_GUARANTEE,
+                DRINKING_WATER_SECURITY_GUARANTEE,
+                HEALTH_SUPPORT,
+                EDUCATION_SUPPORT,
+                COMPREHENSIVE_GUARANTEE,
+                SOCIAL_SUPPORT,
+            ]
+        elif risk == "饮水安全问题":
+            must_selected_option = [DRINKING_WATER_SECURITY_GUARANTEE]
+            forbinddens_option = [
+                INDUSTRY_SUPPORT,
+                EMPLOYMENT_SUPPORT,
+                FINANCIAL_SUPPORT,
+                PUBLIC_WELFARE_SUPPORT,
+                HOUSING_SECURITY_GUARANTEE,
+                OBLIGATORY_EDUCATION_GUARANTEE,
+                HEALTH_SUPPORT,
+                EDUCATION_SUPPORT,
+                COMPREHENSIVE_GUARANTEE,
+                SOCIAL_SUPPORT,
+            ]
+        elif risk == "家庭成员中有未参加城乡居民（职工）基本医疗保险":
+            must_selected_option = [HEALTH_SUPPORT, SOCIAL_SUPPORT]
+            forbinddens_option = [
+                INDUSTRY_SUPPORT,
+                EMPLOYMENT_SUPPORT,
+                FINANCIAL_SUPPORT,
+                PUBLIC_WELFARE_SUPPORT,
+                HOUSING_SECURITY_GUARANTEE,
+                OBLIGATORY_EDUCATION_GUARANTEE,
+                EDUCATION_SUPPORT,
+                COMPREHENSIVE_GUARANTEE,
+                DRINKING_WATER_SECURITY_GUARANTEE,
+            ]
+        must_selected = get_item_values_for(must_selected_option)
+        forbiddens = get_item_values_for(forbinddens_option)
+
+        # 如果option为空，代表无要求
+        must_selected_ok = len(must_selected) > 0 if len(must_selected_option) > 0 else True
+        forbiddens_ok = len(forbiddens) == 0 if len(forbinddens_option) > 0 else True
+
+        print(f"risk 为{risk}的必选项当前值：{must_selected},禁选项当前值：{forbiddens}")
+        if not must_selected_ok or not forbiddens_ok:
+            target = ws[f"{title_dict[get_key_for(i)]}{row_num}"]
+            if not must_selected_ok:
+                comment_and_fill_yellow_for(target, f"{get_key_for(i)} 未选择必选项，必选项为{must_selected_option}")
+            else:
+                forbiddens_keys = [x for x, y in forbiddens]
+                forbiddens_values = [y for x, y in forbiddens]
+                comment_and_fill_yellow_for(
+                    target,
+                    f"{get_key_for(i)} 填写了禁选项{forbiddens_keys}，填写的内容为{forbiddens_values}",
+                )
+
+
+def check_assistance(ws, row_num, title_dict):
+    """检查实施开发式帮扶填写状态是否有问题，如果填写了开发式帮扶，以下四项：产业帮扶、就业帮扶、金融帮扶、公益岗位帮扶有一项填写的是其他或者技能培训就不合规"""
+    assistance = "实施开发式帮扶措施情况"
+    assistance_value = ws[f"{title_dict[assistance]}{row_num}"].value
+    if assistance_value and len(assistance_value) > 0:
+        for type in ["产业帮扶", "就业帮扶", "金融帮扶", "公益岗位帮扶"]:
+            target = ws[f"{title_dict[type]}{row_num}"].value
+            for key in ["其他", "技能培训"]:
+                if key in target:
+                    comment_and_fill_yellow_for(target, f"实施开发式帮扶填写状态下，{type} 不允许选择 {key}")
+                    return
+
+
+# 填写了的话，剩下四项有一个是其他或者技能培训就不行
+
+# info_number = "户主证件号码"
+# identitycard = ws[f"{title_dict[info_number]}{row_num}"].value
+# if len(identitycard) not in [15, 18, 20, 22]:
+#     target = ws[f"{title_dict[info_number]}{row_num}"]
+#     comment_and_fill_yellow_for(target, "31.监测对象家庭成员证件号码位数异常（证件号码非15、18、20、22位）")
+
+
+# def get_item_values_for(ws, row_num, title_dict, items):
+#     result = []
+#     for item in items:
+#         if item not in title_dict:
+#             continue
+#         value = ws[f"{title_dict[item]}{row_num}"].value
+#         if value is not None:
+#             result.append(value)
+#     return result
+
+
+def comment_and_fill_yellow_for(target, comment):
+    target.comment = Comment(text=comment, author="system")
+    yellow_fill = PatternFill(patternType="solid", fgColor="FFFF00")
+    target.fill = yellow_fill
+
+
+if __name__ == "__main__":
+    # result = calculate_age_from_id_number("532801200607144126")
+    # print(result)
+    uvicorn.run("data_verification:app", host="localhost", port=8000, reload=True)

pyproject.toml

@@ -0,0 +1,2 @@
+[tool.ruff]
+line-length = 120

requirements.txt

@@ -0,0 +1,4 @@
+fastapi
+python-multipart
+openpyxl
+uvicorn