日常更新

from NUC
This commit is contained in:
liubiren 2025-12-10 10:50:58 +08:00
parent 6dc41a1623
commit 0b5388be5b
5 changed files with 24 additions and 93 deletions

Binary file not shown.

View File

@ -1,73 +0,0 @@
# -*- coding: utf-8 -*-
"""
就excel工作表进行处理包括就指定字段解析JSON
"""
from json import loads
from pandas import DataFrame, read_excel
from utils.pandas_extension import save_as_workbook
# 打开并读取指定工作表(默认以字符串读取)
sheet = read_excel(io="dataset.xlsx", sheet_name="Sheet1", dtype=str)
dataset = []
for index, row in sheet.iterrows():
# 就指定字段解析为JSON
response = loads(row["response"])
data = {}
# 根据深圳快瞳票据查验要求解析查验结果
# 状态码
status = response.get("status", "")
# 错误码
code = response.get("code", "")
# 若状态码为200且错误码为10000则定义为响应成功
if status == 200 and code == 10000:
# 查验类型若查验类型为003081则为医疗收费票据=003082则为增值税发票
match response.get("data").get(
"productCode"
):
# 解析医疗收费票据
case "003081":
data["发票号"] = response.get("data").get(
"billNumber"
)
# 查验结果
match response.get("data").get("flushedRed"):
case "true":
data["查验结果"] = "正常"
case "false":
data["查验结果"] = "已红冲"
# 解析增值税发票
case "003082":
data["发票号"] = (
response.get("data").get("details").get("number")
)
# 查验结果
match response.get("data").get("details").get(
"invoiceTypeNo"
):
case "0":
data["查验结果"] = "正常"
case "1":
data["查验结果"] = "无法查验"
case "2" | "3" | "7" | "8":
data["查验结果"] = "已红冲"
# 若状态码为400且错误码为10001或10100则定义为假票
elif status == 400 and (code == 10001 or code == 10100):
data["查验结果"] = "假票"
else:
data["查验结果"] = "无法查验"
dataset.append(data)
# 本地保存
save_as_workbook(worksheets=[("Sheet1", DataFrame(data=dataset, dtype=str))], workbook_name="results.xlsx")

Binary file not shown.

View File

@ -12,7 +12,6 @@ import pandas
from utils.pandas_extension import save_as_workbook
print("1 打开并读取Excel文件...", end="")
try:

View File

@ -5,6 +5,7 @@
功能清单
https://liubiren.feishu.cn/docx/WFjTdBpzroUjQvxxrNIcKvGnneh?from=from_copylink
"""
import json
import re
import uuid
@ -27,7 +28,6 @@ from utils.client import Authenticator, HTTPClient, CacheClient
# from utils.ocr import fuzzy_match
def idcard_extraction(**kwargs) -> dict | None:
"""居民身份证数据提取"""
@ -908,7 +908,7 @@ def common_extraction(**kwargs) -> dict | None:
# 规则模型初始化
def decision(rules_path: Path) -> ZenDecision:
def loader(path):
with open(path, "r") as file:
with open(path, "r", encoding="utf-8") as file:
return file.read()
return ZenEngine({"loader": loader}).get_decision(rules_path.as_posix())
@ -997,7 +997,6 @@ if __name__ == "__main__":
# 加载赔案档案模版
template = environment.get_template("template.html")
# -------------------------
# 自定义方法
# -------------------------
@ -1022,7 +1021,6 @@ if __name__ == "__main__":
# 若本地打开并读取影像件发生异常则抛出异常(实际作业需从影像件服务器下载并读取影像件,因签收时会转存,故必可下载)
raise RuntimeError("影像件打开并读取发生异常")
# noinspection PyShadowingNames
def image_serialize(image_format: str, image_ndarray: numpy.ndarray) -> str:
"""
@ -1042,7 +1040,6 @@ if __name__ == "__main__":
image_guid = md5(image_bytes).hexdigest().upper()
return image_guid
# noinspection PyShadowingNames
def images_classify(
image_guid: str, image_format: str, image_ndarray: numpy.ndarray
@ -1090,7 +1087,9 @@ if __name__ == "__main__":
break
# 影像件BASE64编码
image_base64 = b64encode(image_ndarray_encoded.tobytes()).decode("utf-8")
image_base64 = b64encode(image_ndarray_encoded.tobytes()).decode(
"utf-8"
)
if len(image_base64) <= image_size_specified:
return image_base64
@ -1110,7 +1109,9 @@ if __name__ == "__main__":
return None
# 影像件压缩
image_base64 = images_compress(image_format, image_ndarray, image_size_specified=2) # 深圳快瞳要求为2兆字节
image_base64 = images_compress(
image_format, image_ndarray, image_size_specified=2
) # 深圳快瞳要求为2兆字节
# TODO: 若影像件压缩发生异常则流转至人工处理
if image_base64 is None:
raise RuntimeError("影像件压缩发生异常")
@ -1118,7 +1119,9 @@ if __name__ == "__main__":
# 请求深圳快瞳影像件分类接口
response = http_client.post(
url=(url := "https://ai.inspirvision.cn/s/api/ocr/genalClassify"),
headers={"X-RequestId-Header": image_guid}, # 以影像件唯一标识作为请求唯一标识,用于双方联查
headers={
"X-RequestId-Header": image_guid
}, # 以影像件唯一标识作为请求唯一标识,用于双方联查
data={
"token": authenticator.get_token(
servicer="szkt"
@ -1180,18 +1183,17 @@ if __name__ == "__main__":
}[image_orientation],
)
# 旋正后影像件再次压缩
image_base64 = images_compress(image_format, image_ndarray, image_size_specified=2)
image_base64 = images_compress(
image_format, image_ndarray, image_size_specified=2
)
# TODO: 若旋正后影像件再次压缩发生异常则流转至人工处理
if image_base64 is None:
raise RuntimeError("旋正后影像件再次压缩发生异常")
return image_base64, image_type, image_orientation
# 遍历工作目录中赔案目录并创建赔案档案(模拟自动化域就待自动化任务创建理赔档案)
for case_path in [
x for x in directory_path.iterdir() if x.is_dir()
]:
for case_path in [x for x in directory_path.iterdir() if x.is_dir()]:
# 初始化赔案档案(实际报案层包括保险分公司名称、报案渠道、批次号、报案号和报案时间等)
# 报案渠道包括:保司定义,例如中银项目包括总行和各地分行驻场报案和普康宝自助报案等
dossier = {
@ -1228,10 +1230,14 @@ if __name__ == "__main__":
# 影像件序列化
# noinspection PyTypeChecker
image["影像件唯一标识"] = (image_guid := image_serialize(image_format, image_ndarray))
image["影像件唯一标识"] = (
image_guid := image_serialize(image_format, image_ndarray)
)
# 影像件分类并旋正(较初审自动化,无使能检查)
image_base64, image_type, image_orientation = images_classify(image_guid, image_format, image_ndarray)
image_base64, image_type, image_orientation = images_classify(
image_guid, image_format, image_ndarray
)
image["影像件类型"] = image_type
image["影像件方向"] = image_orientation
# 将影像件数据添加至影像件层
@ -1239,7 +1245,6 @@ if __name__ == "__main__":
# 影像件识别
print(dossier)
exit()
"""