251202更新

utils/client.py

@@ -87,6 +87,169 @@ class MySQLClient:
             raise RuntimeError("执行SQL查询并返回DATAFRAME发生其它异常")
 
 
+class CacheClient:
+    """缓存客户端"""
+
+    def __init__(self, cache_ttl: int = 360, database: str = "Caches.db"):
+        """
+        :param cache_ttl: 缓存生存时间（单位：天），默认为360天
+        :param database: 缓存数据库名称
+        """
+        try:
+            self.cache_ttl = cache_ttl
+
+            # 创建缓存数据库连接
+            self.connection = sqlite3.connect(
+                database=database,
+                check_same_thread=False,
+                timeout=30,  # 缓存数据库锁超时时间（单位：秒），默认为30秒，避免并发锁死
+            )
+
+            # 创建缓存数据库连接（使用SQLite）
+            self.cache_connection = sqlite3.connect(
+                database="SQLite.db", check_same_thread=False
+            )
+
+            # 创建缓存表
+            self.connection.execute(
+                """CREATE TABLE IF NOT EXISTS caches (
+                    guid TEXT PRIMARY KEY,
+                    scene TEXT,
+                    cache TEXT NOT NULL,
+                    timestamp REAL NOT NULL
+                )"""
+            )
+
+            # 创建时间戳索引（优化过期缓存查询效率）
+            self.connection.execute(
+                """CREATE INDEX IF NOT EXISTS index_timestamp ON caches(timestamp)"""
+            )
+
+            # 删除过期缓存
+            self.connection.execute(
+                "DELETE FROM caches WHERE timestamp < ?",
+                (time.time() - self.cache_ttl * 86400,),
+            )
+
+            # 提交事务
+            self.connection.commit()
+
+        except Exception as exception:
+            if self.connection:
+                self.connection.close()
+            raise f"{str(exception)}" from exception
+
+    def _query_response(self, guid: str) -> Optional[Dict]:
+        """
+        私有方法：根据guid查询有效缓存记录（未过期）
+        :param guid: 记录唯一标识
+        :return: 未过期的响应数据（Dict），不存在/过期/异常时返回None
+        """
+        if not self.cache_connection:
+            logger.error("查询失败：缓存数据库未连接")
+            return None
+
+        with threading.Lock():  # 线程锁，保证并发安全
+            cursor = None
+            try:
+                cursor = self.cache_connection.cursor()
+                # 查询条件：guid匹配 + 未过期
+                expire_time = time.time() - self.cache_ttl * 86400
+                cursor.execute(
+                    "SELECT response FROM caches WHERE guid = ? AND timestamp >= ?",
+                    (guid, expire_time),
+                )
+                result = cursor.fetchone()  # 获取单条记录（guid唯一）
+                if result:
+                    logger.info(f"查询缓存成功：guid={guid}")
+                    return json.loads(result[0])  # JSON字符串转Dict
+                logger.info(f"未查询到有效缓存：guid={guid}（不存在或已过期）")
+                return None
+            except json.JSONDecodeError as e:
+                logger.error(
+                    f"缓存数据解析失败（JSON格式错误）：guid={guid}", exc_info=True
+                )
+                return None
+            except Exception as e:
+                logger.error(f"查询缓存异常：guid={guid}", exc_info=True)
+                self.cache_connection.rollback()  # 异常回滚事务
+                return None
+            finally:
+                if cursor:
+                    cursor.close()  # 确保游标关闭，释放资源
+
+    def _save_response(self, guid: str, response: Dict) -> bool:
+        """
+        私有方法：添加/更新缓存记录（存在则覆盖，不存在则新增）
+        :param guid: 记录唯一标识
+        :param response: 待保存的响应数据（Dict）
+        :return: 保存成功返回True，失败返回False
+        """
+        if not self.cache_connection:
+            logger.error("保存失败：缓存数据库未连接")
+            return False
+
+        with threading.Lock():  # 线程锁，保证并发安全
+            cursor = None
+            try:
+                cursor = self.cache_connection.cursor()
+                # 转换Dict为JSON字符串（ensure_ascii=False支持中文）
+                response_str = json.dumps(response, ensure_ascii=False, indent=2)
+                # INSERT OR REPLACE：存在则更新，不存在则插入
+                cursor.execute(
+                    "INSERT OR REPLACE INTO caches (guid, response, timestamp) VALUES (?, ?, ?)",
+                    (guid, response_str, time.time()),  # timestamp存储当前时间戳
+                )
+                self.cache_connection.commit()  # 提交事务
+                logger.info(f"保存缓存成功：guid={guid}")
+                return True
+            except json.JSONEncoderError as e:
+                logger.error(
+                    f"缓存数据序列化失败（Dict转JSON错误）：guid={guid}", exc_info=True
+                )
+                self.cache_connection.rollback()
+                return False
+            except Exception as e:
+                logger.error(f"保存缓存异常：guid={guid}", exc_info=True)
+                self.cache_connection.rollback()  # 异常回滚事务
+                return False
+            finally:
+                if cursor:
+                    cursor.close()  # 确保游标关闭
+
+    def query_or_save_response(
+        self, guid: str, response: Optional[Dict] = None
+    ) -> Optional[Dict]:
+        """
+        二合一公开方法：支持查询记录 / 添加/更新记录（灵活复用）
+        :param guid: 记录唯一标识（必填）
+        :param response: 待保存的响应数据（可选）：
+                        - 不传：仅查询有效记录，返回Dict/None
+                        - 传入：添加/更新记录，返回保存后的有效记录/Dict
+        :return: 查询到的记录 / 保存后的记录 / 失败时返回None
+        """
+        # 参数校验：guid不能为空
+        if not guid or not isinstance(guid, str):
+            logger.error("guid无效：必须是非空字符串")
+            return None
+
+        # 仅查询模式（未传入response）
+        if response is None:
+            return self._query_response(guid)
+
+        # 添加/更新模式（传入response）：先保存，再查询返回最新记录
+        if self._save_response(guid, response):
+            return self._query_response(guid)
+        logger.error(f"保存缓存失败，无法返回记录：guid={guid}")
+        return None
+
+    def close(self):
+        """关闭数据库连接（程序退出时调用）"""
+        if self.cache_connection:
+            self.cache_connection.close()
+            self.cache_connection = None
+
+
 """
 封装urllib.request的相关操作
 使用方法：

票据理赔自动化/main.py

@@ -1,11 +1,9 @@
 # -*- coding: utf-8 -*-
 
 """
-普康健康_自动化录入
-
---优先使用深圳快瞳，就增值税发票、医疗发票优先使用深圳快瞳票据查验、其次使用深圳快瞳票据识别，最后使用本地识别
---优先考虑增值税发票
-
+根据现普康票据理赔自动化最小化实现
+功能清单
+https://liubiren.feishu.cn/docx/WFjTdBpzroUjQvxxrNIcKvGnneh?from=from_copylink
 """
 import hashlib
 import json
@@ -25,14 +23,32 @@ from jionlp import parse_location
 from zen import ZenDecision, ZenEngine
 
 from utils.client import Authenticator, HTTPClient
-from utils.ocr import fuzzy_match
+
+
+# from utils.ocr import fuzzy_match
 
 
 # -------------------------
-# 工具函数
+# 封装方法
 # -------------------------
 
 
+def image_read():
+    """本地读取影像件"""
+    # 影像件读取（默认转为单通道灰度图，实际需下载影像件）
+    image_ndarray = cv2.imread(globals()["image_path"].as_posix(), cv2.IMREAD_GRAYSCALE)
+    if image_ndarray is None:
+        return None
+
+    # 编码为图像字节流
+    # noinspection PyShadowingNames
+    success, image_bytes = cv2.imencode(f".{globals()["image_format"]}", image_ndarray)
+    if not success or image_bytes is None:
+        return None
+
+    return image_bytes
+
+
 def images_compression(**kwargs) -> tuple[str | None, str | None]:
     """影像件压缩并BASE64编码"""
 
@@ -1116,7 +1132,7 @@ def disease_diagnosis(**kwargs) -> str | None:
 
 
 # -------------------------
-# 主逻辑
+# 主逻辑部分
 # -------------------------
 
 
@@ -1146,15 +1162,15 @@ if __name__ == "__main__":
     # 加载赔案档案模版
     template = environment.get_template("template.html")
 
-    # 遍历工作目录中赔案目录，根据赔案创建赔案档案
+    # 遍历工作目录中赔案目录，根据赔案创建赔案档案（模拟自动化域就待自动化任务创建理赔档案）
     for case_path in [
         case_path for case_path in directory_path.iterdir() if case_path.is_dir()
     ]:
-
-        # 初始化赔案档案（简化版本）
+        # 初始化赔案档案（实际报案层包括保险分公司名称、报案渠道、批次号、报案号和报案时间等）
+        # 报案渠道包括：保司定义，例如中银项目包括总行和各地分行驻场报案和普康宝自助报案等
         dossier = {
             "报案层": {
-                "保险分公司": "中银保险有限公司广东分公司",
+                "保险分公司": "中银保险有限公司广东分公司",  # 设定：保险分公司
                 "赔案号": (case_number := case_path.stem),  # 设定：赔案目录名称为赔案号
             },
             "影像件层": [],
@@ -1162,30 +1178,41 @@ if __name__ == "__main__":
 
         # 遍历赔案目录中影像件地址
         for image_index, image_path in enumerate(
-            sorted(case_path.glob(pattern="*"), key=lambda x: x.stat().st_ctime), 1
+            sorted(
+                [
+                    image_path
+                    for image_path in case_path.glob(pattern="*")
+                    if image_path.is_file()
+                    and image_path.suffix.lower() in [".jpg", ".jpeg", ".png"]
+                ],  # 实际作业亦仅支持JPG、JPEG或PNG
+                key=lambda x: x.name,
+            ),
+            1,
         ):
-            dossier["影像件层"].append(
-                {
-                    "影像件序号": (image_index := f"{image_index:02d}"),
-                    "影像件名称": (image_name := image_path.name),
-                }
+            # 初始化影像件数据
+            image = {
+                "原始影像件": {
+                    "影像件地址": image_path.as_posix(),
+                    "影像件名称": (image_name := image_path.stem),
+                    "影像件格式": (image_format := image_path.suffix.lower()),
+                },
+            }
+
+            # 本地读取影像件（实际从影像件服务器下载）
+            image_bytes = image_read()
+            if image_bytes is None:
+                # noinspection PyTypeChecker
+                image["影像件唯一标识"] = None
+                continue
+
+            # 生成影像件唯一标识
+            # noinspection PyTypeChecker
+            image["影像件唯一标识"] = (
+                image_guid := hashlib.md5(image_bytes.tobytes()).hexdigest().upper()
             )
 
-            # 若影像件格式非JPG/JPEG/PNG则跳过该影像件
-            if (image_format := image_path.suffix.lower().lstrip(".")) not in [
-                "jpg",
-                "jpeg",
-                "png",
-            ]:
-                dossier["影像件层"][-1]["已分类"] = "否，不支持的影像件"
-                continue
-
-            # 影像件读取
-            image = cv2.imread(image_path.as_posix(), cv2.IMREAD_GRAYSCALE)
-            # 若发生异常则跳过该影像件
-            if image is None:
-                dossier["影像件层"][-1]["已分类"] = "否，读取异常"
-                continue
+            print(image)
+            exit()
 
             # 影像件压缩（输出BASE64编码）
             image_guid, image_base64 = images_compression()
@@ -1200,7 +1227,13 @@ if __name__ == "__main__":
             if image_type is None or image_orientation is None:
                 dossier["影像件层"][-1]["已分类"] = "否，影像件分类异常"
                 continue
-
+            #
+            dossier["影像件层"].append(
+                {
+                    "影像件序号": (image_index := f"{image_index:02d}"),
+                    "影像件名称": (image_name := image_path.name),
+                }
+            )
             # 若影像件方向非0度，则影像件旋正并在此压缩
             if image_orientation != "0度":
                 # 影像件旋正