liubiren
/
Python
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Activity

日常更新 

from NUC
This commit is contained in:
liubiren 2025-12-15 21:46:09 +08:00
parent 963ce19609
commit 2eeb1e998f
5 changed files with 302192 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
test.py
View File

@ -1,18 +0,0 @@
# -*- coding: utf-8 -*-
"""
根据现普康票据理赔自动化最小化实现
功能清单
https://liubiren.feishu.cn/docx/WFjTdBpzroUjQvxxrNIcKvGnneh?from=from_copylink
"""
from decimal import Decimal, ROUND_HALF_UP
bill_amount = Decimal("1.2223").quantize(
Decimal("0.00"),
rounding=ROUND_HALF_UP,
)
result = {"票据金额": bill_amount}
print(result)

33
utils/client.py
View File

@ -400,14 +400,14 @@ class HTTPClient:
def __init__(self, cache_ttl: int):
"""
初始化缓存数据库
:param cache_ttl: 缓存生存时间，单位为天
:param cache_ttl: 缓存生存时间，单位为秒
"""
# 初始化SQLite客户端
super().__init__(database=Path(__file__).parent.resolve() / "caches.db")
# 初始化缓存生存时间，单位为秒
self.cache_ttl = cache_ttl
# 初始化缓存表、时间戳索引和清理过期缓存
# 初始化缓存表和时间戳索引（不清理过期缓存）
try:
with self:
self._execute(
@ -416,10 +416,6 @@ class HTTPClient:
self._execute(
sql="""CREATE INDEX IF NOT EXISTS idx_timestamp ON caches(timestamp)"""
)
self._execute(
sql="DELETE FROM caches WHERE timestamp < ?",
parameters=(time.time() - self.cache_ttl,),
)
except Exception as exception:
raise RuntimeError(
f"初始化缓存数据库发生异常：{str(exception)}"
@ -439,10 +435,9 @@ class HTTPClient:
sql="SELECT cache FROM caches WHERE guid = ? AND timestamp >= ?",
parameters=(guid, time.time() - self.cache_ttl),
)
# 就查询结果JSON反序列化
if result is not None and "cache" in result:
return json.loads(result["cache"])
return None
return (
None if result is None else json.loads(result["cache"])
) # 就缓存JSON反序列化
except Exception as exception:
raise RuntimeError("查询并获取单条缓存发生异常") from exception
@ -493,7 +488,7 @@ class HTTPClient:
self.timeout = timeout
# 初始化使用缓存
self.cache_enabled = cache_enabled
# 初始化缓存生存时间，单位为秒
# 初始化缓存生存时间，单位由天转为秒
self.cache_ttl = cache_ttl * 86400
self.cache_client: Optional[HTTPClient.CacheClient] = None
@ -569,16 +564,16 @@ class HTTPClient:
Generator[bytes, None, None],
None,
]:
response = self._request(
method="GET",
parameters=self.Parameters(**{"stream_enabled": stream_enabled, **kwargs}),
)
"""
下载文件
:param stream_enabled: 使用流式传输
:param chunk_size: 流式传输的分块大小
"""
# 若是用流式传输则处理流式传输响应
response = self._request(
method="GET",
parameters=self.Parameters(**{"stream_enabled": stream_enabled, **kwargs}),
)
# 若使用流式传输则处理流式传输响应
if stream_enabled:
return self._process_stream_response(
response=response, chunk_size=chunk_size
@ -643,7 +638,7 @@ class HTTPClient:
except Exception as exception:
# noinspection PyBroadException
try:
response = response or getattr(exception, "response", None)
response = getattr(exception, "response", None)
status = (
response.json().get("status", response.status_code)
if response is not None
@ -656,7 +651,7 @@ class HTTPClient:
)
except Exception:
status = None
message = f"{method} {parameters["url"]} 请求发生异常：{str(exception).split("\n")[0]}"
message = f"{method} {parameters["url"]} 请求发生异常：{str(exception).splitlines()[0]}"
raise self.RequestException(status=status, message=message) from exception
# 处理响应
@ -666,7 +661,7 @@ class HTTPClient:
) -> Union[str, Tuple[str, bytes], Dict[str, Any], ElementTree.Element, None]:
# 若响应内容为空则返回None
content = response.content
if not response:
if not content:
return None
# 响应类型

BIN
票据理赔自动化/SQLite.db
View File

Binary file not shown.

162
票据理赔自动化/main.py
View File

@ -14,7 +14,7 @@ from datetime import datetime
from decimal import Decimal, ROUND_HALF_UP
from hashlib import md5
from pathlib import Path
from typing import Optional, Tuple
from typing import Any, Dict, Optional, Tuple
import cv2
import numpy
@ -23,19 +23,12 @@ from jinja2 import Environment, FileSystemLoader
from jionlp import parse_location
from zen import ZenDecision, ZenEngine
from utils.client import Authenticator, HTTPClient
from utils.client import Authenticator, HTTPClient, SQLiteClient
# from utils.ocr import fuzzy_match
client = SQLiteClient()
a = client._query_one("SELECT * FROM institutions")
print(a)
exit()
def common_extraction(**kwargs) -> dict | None:
"""通用数据提取"""
@ -279,7 +272,54 @@ if __name__ == "__main__":
# 影像件识别使能
recognition_enable = rule_engine(Path("rules/影像件识别使能.json"))
# 初始化JINJA2环境
class MasterData(SQLiteClient):
"""主数据"""
def __init__(self):
"""
初始化主数据
"""
# 初始化SQLite客户端
super().__init__(database="SQLite.db")
# 初始化购药及就医机构表
try:
with self:
self._execute(
sql="""CREATE TABLE IF NOT EXISTS institutions (institution TEXT PRIMARY KEY, type TEXT NOT NULL, province TEXT NOT NULL, city TEXT NOT NULL)"""
)
except Exception as exception:
raise RuntimeError(
f"初始化数据库发生异常：{str(exception)}"
) from exception
# noinspection PyShadowingNames
def query_institution_type(self, institution: str) -> Optional[Dict[str, Any]]:
"""
查询并获取单条购药及就医机构类型
:param institution: 购药及就医机构
:return: 购药及就医机构类型
"""
# noinspection PyBroadException
try:
with self:
# noinspection SqlResolve
result = self._query_one(
sql="SELECT type FROM institutions WHERE institution = ?",
parameters=(institution,),
)
return (
None if result is None else result["type"]
) # 返回购药及就医机构类型
except Exception as exception:
raise RuntimeError(
"查询并获取单条购药及就医机构类型发生异常"
) from exception
# 实例化主数据
master_data = MasterData()
# 实例化JINJA2环境
environment = Environment(loader=FileSystemLoader("."))
# 添加DATE过滤器
environment.filters["date"] = lambda date: (
@ -334,13 +374,13 @@ if __name__ == "__main__":
# noinspection PyShadowingNames
def image_classify(
image_guid: str, image_format: str, image_ndarray: numpy.ndarray
) -> Optional[Tuple[str, str, str]]:
) -> Optional[Tuple[str, str]]:
"""
影像件分类并旋正
:param image_guid: 影像件唯一标识
:param image_format: 影像件格式
:param image_ndarray: 影像件数据
:return: 压缩后影像件BASE64编码，影像件类型和影像件方向
:return: 压缩后影像件BASE64编码和影像件类型
"""
# noinspection PyShadowingNames
@ -480,7 +520,7 @@ if __name__ == "__main__":
if image_base64 is None:
raise RuntimeError("旋正后影像件再次压缩发生异常")
return image_base64, image_type, image_orientation
return image_base64, image_type
# noinspection PyShadowingNames
def image_recognize(
@ -566,9 +606,9 @@ if __name__ == "__main__":
address := parse_location(response["data"]["address"])
).get(
"province"
), # 就住址解析为省、地、县和详细地址
"地": address.get("city"),
"县": address.get("county"),
), # 就住址解析为省、市、区和详细地址
"市": address.get("city"),
"区": address.get("county"),
"详细地址": address.get("detail"),
}
)
@ -675,12 +715,15 @@ if __name__ == "__main__":
rounding=ROUND_HALF_UP,
), # 深圳快瞳票据查验接口中开票金额由字符串转为Decimal，保留两位小数
"姓名": response["data"]["details"]["buyer"],
"购药及就医机构": (
institution := response["data"]["details"]["seller"]
),
"备注": (
response["data"]["details"]["remark"]
if response["data"]["details"]["remark"]
else None
), # 深圳快瞳票据查验接口中备注由空字符转为None
"明细层": [
"费项层": [
{
"名称": item["name"],
"规格": (
@ -711,12 +754,31 @@ if __name__ == "__main__":
"items", []
)
],
"购药及就医机构": (
institution := response["data"]["details"]["seller"]
),
# 根据购药及就医机构查询其类型若为药店则购药及就医类型为药店购药，其它则为门诊就医
}
)
# 查询并获取单条购药及就医机构类型
institution_type = master_data.query_institution_type(
institution
)
# TODO: 若购药及就医机构类型为None则流转至主数据人工处理
if institution_type is None:
raise RuntimeError(
"查询并获取单条购药及就医机构类型发生异常"
)
receipt["购药及就医机构类型"] = institution_type
# 根据购药及就医机构类型匹配处理方法
match institution_type:
# 若购药及就医机构类型为药店，则根据
case "药店":
pass
case "私立医院":
pass
case _:
raise RuntimeError("")
# 门诊/住院收费票据
case "003081":
# noinspection PyTypeChecker
@ -1350,12 +1412,10 @@ if __name__ == "__main__":
):
# 初始化影像件数据
image = {
"原始影像件": {
"影像件编号": image_index,
"影像件地址": image_path.as_posix(), # 将影像件路径对象转为字符串
"影像件名称": (image_name := image_path.stem),
"影像件格式": (image_format := image_path.suffix.lower()),
},
"影像件编号": image_index,
"影像件地址": image_path.as_posix(), # 将影像件路径对象转为字符串
"影像件名称": (image_name := image_path.stem),
"影像件格式": (image_format := image_path.suffix.lower()),
}
# 本地打开并读取影像件
@ -1368,24 +1428,48 @@ if __name__ == "__main__":
)
# 影像件分类并旋正（较初审自动化无使能检查）
image_base64, image_type, image_orientation = image_classify(
image_base64, image_type = image_classify(
image_guid, image_format, image_ndarray
)
# image["影像件BASE64编码"] = image_base64
image["影像件类型"] = image_type
image["影像件方向"] = image_orientation
# 将影像件数据添加至影像件层
# TODO: 若影像件类型为居民身份证(国徽面)和居民身份证(头像面)需合并
dossier["影像件层"].append(image)
# 影像件识别并整合至赔案档案
image_recognize(
image_index,
image_guid,
image_format,
image_base64,
insurance_branch,
image_type,
)
# 就影像件层按照影像件类型排序
dossier["影像件层"].sort(
key=lambda x: [
"居民户口簿",
"居民身份证(国徽面)",
"居民身份证(头像面)",
"居民身份证(国徽、头像面)",
"中国港澳台地区及境外护照",
"理赔申请书",
"增值税发票",
"医疗门诊收费票据",
"医疗住院收费票据",
"医疗费用清单",
"银行卡",
"其它",
].index(x["影像件类型"])
)
print(dossier)
exit()
for image in dossier["影像件层"]:
pass
# 影像件识别并整合至赔案档案
image_recognize(
image_index,
image_guid,
image_format,
image_base64,
insurance_branch,
image_type,
)
"""

302055
票据理赔自动化/普康健康_医疗机构_截止至251202_已清洗.csv Normal file
View File

File diff suppressed because it is too large Load Diff