from utils import (upload_image_to_cos, get_image_memory_size, json_format, setup_logging, json_to_excel) from call_llm import call_tongyi_ocr_stream, call_tongyi_ocr from var import COS_URL, MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DB, OCR_MODEL from PIL import Image, ImageDraw import os import time import traceback import logging # 初始化日志 logger = setup_logging() class WorkflowResult: def __init__(self, local_file_path, cos_path_prefix, file_id, filename, json_output_path, excel_output_path): """初始化工作流结果对象 Args: local_file_path (str): 本地文件路径 cos_path_prefix (str): COS存储路径，不包含前面的域名部分 file_id (str): 前端识别用的id filename (str): filename json_output_path (str): JSON输出文件路径 excel_output_path (str): Excel输出文件路径 """ self.local_file_path = local_file_path self.cos_path_prefix = cos_path_prefix self.file_id = file_id self.filename = filename self.cos_path = f"{cos_path_prefix}/{file_id}/{filename}" self.json_output_path = json_output_path self.excel_output_path = excel_output_path self.cos_url_full = COS_URL + self.cos_path self.file_size = 0 self.ocr_json_result = None self.input_tokens = 0 self.output_tokens = 0 self.ocr_cost_time = 0 self.task_status = 'success' logger.info("WorkflowResult对象初始化完成: 文件=%s, file_id=%s", filename, file_id) logger.debug("本地文件路径: %s", local_file_path) logger.debug("COS路径前缀: %s", cos_path_prefix) logger.debug("JSON输出路径: %s", json_output_path) logger.debug("Excel输出路径: %s", excel_output_path) def llm_ocr_single_image(self): logger.info("开始OCR识别单张图片: %s", self.filename) system_prompt = f""" 你是一个专业的OCR工具，识别图片里表格的内容： 1. 检测出图片里完整的表格，如果表格不完整，则直接忽略 2. 表格数据前31行表示一年中每一天的数据，竖向记录，一列表示一个月，识别时也要竖向识别，按月份返回数据。切记这是真实数据，只有一月、三月、五月、七月、八月、十月、十二月，有31天。二月只有28或者29天，其余月份只有30天 3. 第32行记录的是平均数，若均值不存在，则对应值记录为None 数据返回为JSON格式:{{"1月": {{"data": [0.1, 0.2], "avg": 0.3}}, "2月": {{...}}}}，每个月份数据的长度，要与这个月的天数相符。如果某个月份存在数据缺失，则用null进行填充 """ logger.debug("准备处理图片: %s", self.cos_path) ocr_start_time = time.time() try: self.ocr_json_result, self.input_tokens, self.output_tokens = call_tongyi_ocr( system_prompt, "识别图中的表格", self.cos_url_full ) self.ocr_cost_time = int(time.time() - ocr_start_time) if self.ocr_json_result is None: self.task_status = 'ocr_failed' logger.error("OCR识别失败: %s", self.filename) else: with open(self.json_output_path, 'w', encoding='utf-8') as f: f.write(self.ocr_json_result) logger.info("OCR识别成功: 文件=%s, 耗时=%d秒, 输入token=%d, 输出token=%d", self.filename, self.ocr_cost_time, self.input_tokens, self.output_tokens) logger.debug("OCR结果已保存到: %s", self.json_output_path) except Exception as e: self.task_status = 'ocr_failed' logger.error("OCR识别过程中发生异常: %s", str(e), exc_info=True) raise def run_single_workflow(self): """处理单个文件的工作流 需要导入MySQL的数据: `id` bigint NOT NULL AUTO_INCREMENT COMMENT '唯一主键', `file_id` varchar(255) NOT NULL COMMENT 'web生成的id，也是cos的一个路径', `file_name` varchar(255) NOT NULL COMMENT '图片原始文件名', `file_storage_url` varchar(500) NOT NULL COMMENT '图片在云存储的地址(URL)', `ocr_json_result` longtext COMMENT 'OCR识别出的完整JSON结果', `file_size` bigint DEFAULT '0' COMMENT '图片文件大小（字节）', `task_status` varchar(20) NOT NULL DEFAULT 'uploading' COMMENT '任务状态: uploading(上传中), processing(识别中), success, failed', `ocr_engine` varchar(50) DEFAULT NULL COMMENT '使用的OCR引擎，如 tesseract, aliyun_ocr', `create_time` TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '任务开始时间', `ocr_cost_time` float DEFAULT 0 COMMENT 'OCR处理完成时间', `input_tokens` int unsigned DEFAULT NULL COMMENT '本次OCR调用消耗的输入Token数', `output_tokens` int unsigned DEFAULT NULL COMMENT '本次OCR调用消耗的输出Token数', """ logger.info("开始执行单文件工作流: %s", self.filename) # 检查文件类型 if not self.local_file_path.lower().endswith(('.png', '.jpg', '.jpeg')): logger.warning("跳过非图片文件: %s", self.local_file_path) return None, None # 上传到COS logger.info("开始上传文件到COS: %s", self.local_file_path) is_upload_success = upload_image_to_cos(self.local_file_path, self.cos_path) if not is_upload_success: self.task_status = 'upload_failed' logger.error("文件上传失败: %s", self.local_file_path) return None, None logger.info("文件上传成功: %s", self.cos_url_full) # 获取图片大小 try: image = Image.open(self.local_file_path) self.file_size = get_image_memory_size(image) logger.debug("图片大小: %d 字节", self.file_size) except Exception as e: logger.error("获取图片大小失败: %s", str(e)) self.file_size = 0 # OCR识别 self.llm_ocr_single_image() if self.task_status != 'success': logger.error("工作流处理失败，当前状态: %s", self.task_status) return None, None # 处理OCR结果 try: logger.info("开始处理OCR结果") json_res = json_format(self.ocr_json_result) data_detail = json_to_excel(json_res, self.excel_output_path) logger.info("OCR结果处理完成，Excel文件已生成") except Exception as e: logger.error("处理OCR结果时发生异常: %s", str(e), exc_info=True) return None, None logger.info("单文件工作流执行完成: %s", self.filename) return data_detail def db_insert_params(self): """获取用于数据库插入的参数元组 Returns: dict: 包含所有需要插入数据库的字段值 """ db_params = { "file_id": self.file_id, "filename": self.filename, "cos_url_full": self.cos_url_full, "ocr_json_result": self.ocr_json_result, "file_size": self.file_size, "task_status": self.task_status, "ocr_cost_time": self.ocr_cost_time, "input_tokens": self.input_tokens, "output_tokens": self.output_tokens } return db_params class WorkflowResultStream: def __init__(self, local_file_path, cos_path_prefix, file_id, filename, json_output_path, excel_output_path): """初始化工作流结果对象 Args: local_file_path (str): 本地文件路径 cos_path_prefix (str): COS存储路径，不包含前面的域名部分 file_id (str): 前端识别用的id filename (str): filename json_output_path (str): JSON输出文件路径 excel_output_path (str): Excel输出文件路径 """ logger = logging.getLogger(__name__) self.local_file_path = local_file_path self.cos_path_prefix = cos_path_prefix self.file_id = file_id self.filename = filename self.cos_path = f"{cos_path_prefix}/{file_id}/{filename}" self.json_output_path = json_output_path self.excel_output_path = excel_output_path self.cos_url_full = COS_URL + self.cos_path self.file_size = 0 # self.ocr_json_result = None # self.input_tokens = 0 # self.output_tokens = 0 # self.ocr_cost_time = 0 self.task_status = 'success' logger.info("WorkflowResult对象初始化完成: 文件=%s, file_id=%s", filename, file_id) logger.debug("本地文件路径: %s", local_file_path) logger.debug("COS路径前缀: %s", cos_path_prefix) logger.debug("JSON输出路径: %s", json_output_path) logger.debug("Excel输出路径: %s", excel_output_path) def llm_ocr_single_image(self): logger.info("开始OCR识别单张图片: %s", self.filename) system_prompt = f""" 你是一个专业的OCR工具，识别图片里表格的内容： 1. 检测出图片里完整的表格，如果表格不完整，则直接忽略 2. 表格数据前31行表示一年中每一天的数据，竖向记录，一列表示一个月，识别时也要竖向识别，按月份返回数据。切记这是真实数据，只有一月、三月、五月、七月、八月、十月、十二月，有31天。二月只有28或者29天，其余月份只有30天 3. 第32行记录的是平均数，若均值不存在，则对应值记录为None 数据返回为JSON格式:{{"1月": {{"data": [0.1, 0.2], "avg": 0.3}}, "2月": {{...}}}}，每个月份数据的长度，要与这个月的天数相符。如果某个月份存在数据缺失，则用null进行填充 """ logger.debug("准备处理图片: %s", self.cos_path) try: stream_output = call_tongyi_ocr_stream( system_prompt, "识别图中的表格", self.cos_url_full ) return stream_output except Exception as e: self.task_status = 'ocr_failed' logger.error("OCR识别过程中发生异常: %s", str(e), exc_info=True) raise def run_single_workflow(self): """处理单个文件的工作流 需要导入MySQL的数据: `id` bigint NOT NULL AUTO_INCREMENT COMMENT '唯一主键', `file_id` varchar(255) NOT NULL COMMENT 'web生成的id，也是cos的一个路径', `file_name` varchar(255) NOT NULL COMMENT '图片原始文件名', `file_storage_url` varchar(500) NOT NULL COMMENT '图片在云存储的地址(URL)', `ocr_json_result` longtext COMMENT 'OCR识别出的完整JSON结果', `file_size` bigint DEFAULT '0' COMMENT '图片文件大小（字节）', `task_status` varchar(20) NOT NULL DEFAULT 'uploading' COMMENT '任务状态: uploading(上传中), processing(识别中), success, failed', `ocr_engine` varchar(50) DEFAULT NULL COMMENT '使用的OCR引擎，如 tesseract, aliyun_ocr', `create_time` TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '任务开始时间', `ocr_cost_time` float DEFAULT 0 COMMENT 'OCR处理完成时间', `input_tokens` int unsigned DEFAULT NULL COMMENT '本次OCR调用消耗的输入Token数', `output_tokens` int unsigned DEFAULT NULL COMMENT '本次OCR调用消耗的输出Token数', """ logger.info("开始执行单文件工作流: %s", self.filename) # 检查文件类型 if not self.local_file_path.lower().endswith(('.png', '.jpg', '.jpeg')): logger.warning("跳过非图片文件: %s", self.local_file_path) return None, None # 上传到COS logger.info("开始上传文件到COS: %s", self.local_file_path) is_upload_success = upload_image_to_cos(self.local_file_path, self.cos_path) if not is_upload_success: self.task_status = 'upload_failed' logger.error("文件上传失败: %s", self.local_file_path) return None, None logger.info("文件上传成功: %s", self.cos_url_full) # 获取图片大小 try: image = Image.open(self.local_file_path) self.file_size = get_image_memory_size(image) logger.debug("图片大小: %d 字节", self.file_size) except Exception as e: logger.error("获取图片大小失败: %s", str(e)) self.file_size = 0 # OCR识别 stream_output = self.llm_ocr_single_image() if self.task_status != 'success': logger.error("工作流处理失败，当前状态: %s", self.task_status) return None logger.info("单文件工作流执行完成: %s", self.filename) return stream_output def db_insert_params(self): """获取用于数据库插入的参数元组 Returns: dict: 包含所有需要插入数据库的字段值 """ db_params = { "file_id": self.file_id, "filename": self.filename, "cos_url_full": self.cos_url_full, # "ocr_json_result": self.ocr_json_result, "file_size": self.file_size, "task_status": self.task_status, # "ocr_cost_time": self.ocr_cost_time, # "input_tokens": self.input_tokens, # "output_tokens": self.output_tokens } return db_params if __name__ == "__main__": logger.info("=== 水文数据分析程序开始运行 ===") try: local_file_path = r"D:\project\hydrological_data_analyse\input\102.jpg" cos_path_prefix = "images/ruiwen" json_output_path = r"D:\project\hydrological_data_analyse\output\102_processed.json" excel_output_path = r"D:\project\hydrological_data_analyse\output\102_output.xlsx" file_id = "testfileid123" filename = "102.jpg" logger.info("程序参数: 输入文件=%s, 输出JSON=%s, 输出Excel=%s", local_file_path, json_output_path, excel_output_path) # 检查输入文件是否存在 if not os.path.exists(local_file_path): logger.error("输入文件不存在: %s", local_file_path) exit(1) # 检查输出目录是否存在 output_dir = os.path.dirname(json_output_path) if not os.path.exists(output_dir): os.makedirs(output_dir) logger.info("创建输出目录: %s", output_dir) workflow = WorkflowResult(local_file_path, cos_path_prefix, file_id, filename, json_output_path, excel_output_path) data_detail, _image = workflow.run_single_workflow() if data_detail is not None: logger.info("程序执行成功完成") else: logger.error("程序执行过程中出现错误") except Exception as e: logger.critical("程序执行出现未预期错误: %s", str(e), exc_info=True) finally: logger.info("=== 水文数据分析程序运行结束 ===") logging.shutdown()