data_stardard.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386
  1. import os
  2. import time
  3. import asyncio
  4. import io
  5. import csv
  6. import datetime
  7. import httpx
  8. import json
  9. import uuid
  10. # --- LangChain Imports ---
  11. from langchain_openai import ChatOpenAI
  12. from langchain_core.prompts import ChatPromptTemplate
  13. from langchain_core.output_parsers import JsonOutputParser
  14. # --- 核心 Parser ---
  15. class TransactionParserAgent:
  16. def __init__(self, api_key: str, multimodal_api_url: str, base_url: str = "https://api.deepseek.com"):
  17. # 1. 初始化 LangChain ChatOpenAI 客户端
  18. # DeepSeek 完全兼容 OpenAI 接口,使用 ChatOpenAI 是标准做法
  19. self.llm = ChatOpenAI(
  20. model="deepseek-chat",
  21. api_key=api_key,
  22. base_url=base_url,
  23. temperature=0.1,
  24. max_retries=3, # LangChain 内置重试机制
  25. # 配置 httpx 客户端以优化超时和连接 (LangChain 允许透传 http_client)
  26. http_client=httpx.Client(
  27. timeout=httpx.Timeout(300.0, read=300.0, connect=60.0),
  28. limits=httpx.Limits(max_keepalive_connections=5, max_connections=10)
  29. )
  30. )
  31. self.multimodal_api_url = multimodal_api_url
  32. # 定义 JSON 解析器
  33. self.parser = JsonOutputParser()
  34. # 初始化API调用跟踪
  35. self.api_calls = []
  36. async def _invoke_miner_u(self, file_path: str) -> str:
  37. """调用 MinerU 并提取纯行数据 (保持 httpx 调用不变,因为这不是 LLM)"""
  38. miner_start_time = time.perf_counter()
  39. print("\n" + "=" * 40)
  40. print("📌 【步骤1 - 数据提取】 开始执行")
  41. dealRows = 0
  42. try:
  43. # MinerU 是独立服务,继续使用原生 httpx
  44. async with httpx.AsyncClient() as client:
  45. with open(file_path, 'rb') as f:
  46. files = {'file': (os.path.basename(file_path), f)}
  47. data = {'folderId': 'text'}
  48. print("🔄数据提取中...")
  49. response = await client.post(self.multimodal_api_url, files=files, data=data, timeout=120.0)
  50. if response.status_code == 200:
  51. res_json = response.json()
  52. full_md_list = []
  53. for element in res_json.get('convert_json', []):
  54. if 'md' in element:
  55. full_md_list.append(element['md'])
  56. if 'rows' in element:
  57. dealRows+=len(element['rows'])
  58. print(f"📊 提取结果:共提取 {dealRows-1} 条数据")
  59. return "\n\n".join(full_md_list)
  60. return ""
  61. except Exception as e:
  62. print(f"❌ MinerU 调用异常: {e}")
  63. return ""
  64. finally:
  65. print(f"✅ 【步骤1 - 数据提取】 执行完成")
  66. print(f"⏱️ 执行耗时:{ time.perf_counter() - miner_start_time:.2f} 秒")
  67. def _get_csv_prompt_template(self) -> ChatPromptTemplate:
  68. """
  69. 构造 LangChain 的 Prompt 模板
  70. """
  71. system_template = """
  72. # Role
  73. 你是一个高精度的银行账单转换工具。
  74. # Task
  75. 将输入的 Markdown 表格行转换为 JSON 数组。
  76. # Field Rules
  77. 1. txId: 如果输入数据中有交易流水号则直接使用,如果没有,从 T{start_id:04d} 开始递增生成。
  78. 2. txDate: 交易日期,格式为YYYY-MM-DD
  79. 3. txTime: 交易时间,格式为HH:mm:ss (未知填 00:00:00)
  80. 4. txAmount: 交易金额,绝对值数字
  81. 5. txBalance: 交易后余额。浮点数,移除千分位逗号。
  82. 6. txDirection: 交易方向。必须根据以下逻辑判断只输出“收入”或“支出”:
  83. - 若有“借/贷”列:“借”通常为支出,“贷”通常为收入(除非是信用卡,需结合表头)。
  84. - 若有“收入/支出”分列:按列归类。
  85. - 若金额带正负号:"+"为收入,"-"为支出。
  86. - 如果无符号,请结合表头判断。
  87. 7. txSummary: 摘要、用途、业务类型等备注。
  88. 8. txCounterparty: 交易对手方(名称及账号,如有)。
  89. # Constraints
  90. - **强制输出格式**:
  91. 1. 严格返回一个包含对象的 JSON 数组。
  92. 2. 每个对象必须包含上述 8 个字段名作为 Key。
  93. 3. 不要输出任何解释文字或 Markdown 代码块标签。
  94. # Anti-Hallucination Rules
  95. - 不得根据上下文推断任何未在原始数据中明确出现的字段
  96. - 不得计算或猜测余额
  97. - 不得根据常识补全对手方名称
  98. - 若字段缺失,必须返回空字符串 ""
  99. """
  100. user_template = """# Input Data
  101. {chunk_data}
  102. # Output
  103. JSON Array:
  104. """
  105. return ChatPromptTemplate.from_messages([
  106. ("system", system_template),
  107. ("user", user_template)
  108. ])
  109. async def parse_to_csv(self, file_path: str) -> str:
  110. # 1. 获取完整 Markdown 文本并按行切分
  111. md_text = await self._invoke_miner_u(file_path)
  112. if not md_text:
  113. return ""
  114. # 记录开始时间(使用time.perf_counter获取高精度时间)
  115. switch_start_time = time.perf_counter()
  116. print("\n" + "=" * 40)
  117. print("📌 【步骤2 - 标准化转换】 开始执行")
  118. # 初步切分
  119. raw_lines = md_text.splitlines()
  120. # 提取真正的第一行作为基准表头
  121. clean_lines = [l.strip() for l in raw_lines if l.strip()]
  122. if len(clean_lines) < 2: return ""
  123. # --- 【核心改进:动态寻找表头】 ---
  124. table_header = ""
  125. header_index = 0
  126. header_keywords = ["余额", "金额", "账号", "日期", "借/贷", "摘要"]
  127. for idx, line in enumerate(clean_lines):
  128. # 如果某一行包含 2 个以上关键词,且含有 Markdown 表格分隔符 '|'
  129. hit_count = sum(1 for kw in header_keywords if kw in line)
  130. if hit_count >= 2 and "|" in line:
  131. table_header = line
  132. header_index = idx
  133. break
  134. if not table_header:
  135. table_header = clean_lines[0]
  136. header_index = 0
  137. data_rows = []
  138. for line in clean_lines[header_index + 1:]:
  139. if all(c in '|- ' for c in line): continue
  140. if line == table_header: continue
  141. # 过滤掉一些 MinerU 可能在表格末尾产生的页码或无关文字
  142. if "|" not in line: continue
  143. data_rows.append(line)
  144. csv_header = "txId,txDate,txTime,txAmount,txDirection,txBalance,txSummary,txCounterparty,createdAt\n"
  145. csv_content = csv_header
  146. batch_size = 15
  147. global_tx_counter = 1
  148. # 构建 LCEL Chain: Prompt -> LLM -> Parser
  149. chain = self._get_csv_prompt_template() | self.llm | self.parser
  150. # 2. 分块处理
  151. for i in range(0, len(data_rows), batch_size):
  152. chunk = data_rows[i: i + batch_size]
  153. context_chunk = [table_header] + chunk
  154. chunk_str = "\n".join(context_chunk)
  155. # 1. 记录开始时间(使用time.perf_counter获取高精度时间)
  156. start_time = time.perf_counter()
  157. print(f"🔄 正在通过LLM转换批次 {i // batch_size + 1},包含 {len(chunk)} 条数据...")
  158. # print(f"待转换的数据块:\n{chunk_str}")
  159. try:
  160. # --- LangChain 调用 ---
  161. # 使用 ainvoke 异步调用链
  162. # 记录API调用开始时间
  163. call_start_time = datetime.datetime.now()
  164. data_data = await chain.ainvoke({
  165. "start_id": global_tx_counter,
  166. "chunk_data": chunk_str
  167. })
  168. # 记录API调用结束时间
  169. call_end_time = datetime.datetime.now()
  170. # 记录API调用结果 - 简化版:只保存提示词和结果数据
  171. call_id = f"api_llm_数据转换_{'{:.2f}'.format((call_end_time - call_start_time).total_seconds())}"
  172. # 从chain中提取提示词(如果可能)
  173. prompt_content = ""
  174. try:
  175. # 尝试从chain获取最后的消息内容
  176. if hasattr(chain, 'get_prompts'):
  177. prompts = chain.get_prompts()
  178. if prompts:
  179. prompt_content = str(prompts[-1])
  180. else:
  181. # 如果无法获取,构造基本的提示词信息
  182. prompt_content = f"转换批次数据,start_id: {global_tx_counter}, chunk_data: {chunk_str[:200]}..."
  183. except:
  184. prompt_content = f"转换批次数据,start_id: {global_tx_counter}, chunk_data: {chunk_str[:200]}..."
  185. api_call_info = {
  186. "call_id": call_id,
  187. "start_time": call_start_time.isoformat(),
  188. "end_time": call_end_time.isoformat(),
  189. "duration": (call_end_time - call_start_time).total_seconds(),
  190. "prompt": prompt_content,
  191. "input_params": {
  192. "start_id": global_tx_counter,
  193. "chunk_data": chunk_str
  194. },
  195. "llm_result": data_data
  196. }
  197. self.api_calls.append(api_call_info)
  198. # 保存API结果到文件 (Markdown格式,更易阅读)
  199. # 使用运行ID创建独立的文件夹
  200. run_id = os.environ.get('FLOW_RUN_ID', 'default')
  201. api_results_dir = f"api_results_{run_id}"
  202. os.makedirs(api_results_dir, exist_ok=True)
  203. timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
  204. filename = f"{timestamp}_{call_id}.md"
  205. filepath = os.path.join(api_results_dir, filename)
  206. try:
  207. with open(filepath, 'w', encoding='utf-8') as f:
  208. f.write("# 数据转换结果\n\n")
  209. f.write("## 调用信息\n\n")
  210. f.write(f"- 调用ID: {call_id}\n")
  211. f.write(f"- 开始时间: {call_start_time.isoformat()}\n")
  212. f.write(f"- 结束时间: {call_end_time.isoformat()}\n")
  213. f.write(f"- 执行时长: {(call_end_time - call_start_time).total_seconds():.2f} 秒\n")
  214. f.write("\n## 提示词入参\n\n")
  215. f.write("```\n")
  216. f.write(api_call_info["prompt"])
  217. f.write("\n```\n\n")
  218. f.write("## 输入参数\n\n")
  219. f.write("```json\n")
  220. f.write(json.dumps(api_call_info["input_params"], ensure_ascii=False, indent=2))
  221. f.write("\n```\n\n")
  222. f.write("## LLM返回结果\n\n")
  223. f.write("```json\n")
  224. f.write(json.dumps(api_call_info["llm_result"], ensure_ascii=False, indent=2))
  225. f.write("\n```\n")
  226. print(f"[API_RESULT] 保存API结果文件: {filepath}")
  227. except Exception as e:
  228. print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
  229. # print(f"💡 LLM 返回数据: {data_data}")
  230. # 兼容处理:LangChain Parser 通常会直接返回 List 或 Dict
  231. if isinstance(data_data, dict):
  232. # 尝试寻找 transactions 键,如果没有则假设整个 dict 就是我们要的对象(虽然罕见)
  233. batch_data = data_data.get("transactions", [data_data])
  234. # 如果取出来还是 dict (例如单条记录),包一层 list
  235. if isinstance(batch_data, dict):
  236. batch_data = [batch_data]
  237. elif isinstance(data_data, list):
  238. batch_data = data_data
  239. else:
  240. batch_data = []
  241. if batch_data:
  242. output = io.StringIO()
  243. createdAtStr = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
  244. writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL, lineterminator='\n')
  245. print(f"✅ 批次转换成功,包含 {len(batch_data)} 条记录。")
  246. for item in batch_data:
  247. writer.writerow([
  248. item.get("txId", ""),
  249. item.get("txDate", ""),
  250. item.get("txTime", ""),
  251. item.get("txAmount", ""),
  252. item.get("txDirection", ""),
  253. item.get("txBalance", ""),
  254. item.get("txSummary", ""),
  255. item.get("txCounterparty", ""),
  256. createdAtStr
  257. ])
  258. batch_csv_string = output.getvalue()
  259. csv_content += batch_csv_string
  260. global_tx_counter += len(batch_data)
  261. except Exception as e:
  262. print(f"⚠️ 批次执行失败: {e}")
  263. finally:
  264. end_time = time.perf_counter()
  265. elapsed_time = end_time - start_time
  266. print(f"⏱️ 执行耗时: {elapsed_time:.2f} 秒")
  267. print(f"📊 转换结果:共转换 {global_tx_counter - 1} 条数据")
  268. print(f"✅ 【步骤2 - 标准化转换】 执行完成")
  269. return csv_content
  270. async def parse_and_save_to_file(self, file_path: str, output_dir: str = "output") -> str:
  271. """
  272. 供 Workflow 调用:解析并保存文件,返回全路径名
  273. """
  274. current_script_path = os.path.abspath(__file__)
  275. current_dir = os.path.dirname(current_script_path)
  276. file_full_name = os.path.basename(file_path)
  277. file_name = os.path.splitext(file_full_name)[0] # 不带后缀 11111
  278. output_dir = os.path.normpath(os.path.join(current_dir, "..", "..", output_dir))
  279. os.makedirs(output_dir, exist_ok=True)
  280. timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
  281. file_name = f"{file_name}_data_standard_{timestamp}.csv"
  282. full_path = os.path.join(output_dir, file_name)
  283. csv_result = await self.parse_to_csv(file_path)
  284. if csv_result:
  285. with open(full_path, "w", encoding="utf-8") as f:
  286. f.write(csv_result)
  287. return full_path
  288. else:
  289. raise Exception("数据解析失败,未生成有效内容")
  290. async def run_workflow_task(self, input_file_path: str) -> dict:
  291. """
  292. 标准 Workflow 入口方法
  293. """
  294. # 1. 记录开始时间(使用time.perf_counter获取高精度时间)
  295. start_time = time.perf_counter()
  296. print(f"BEGIN---数据标准化任务开始---")
  297. try:
  298. print(f"待执行标准化的文件:{input_file_path}")
  299. api_results_dir = "data_files"
  300. saved_path = await self.parse_and_save_to_file(input_file_path, api_results_dir)
  301. return {
  302. "status": "success",
  303. "file_path": saved_path,
  304. "file_name": os.path.basename(saved_path),
  305. "timestamp": datetime.datetime.now().isoformat()
  306. }
  307. except Exception as e:
  308. return {
  309. "status": "error",
  310. "message": str(e)
  311. }
  312. finally:
  313. end_time = time.perf_counter()
  314. elapsed_time = end_time - start_time
  315. print(f"⏱️ 执行总耗时: {elapsed_time:.2f} 秒")
  316. print(f"END---数据标准化任务结束")
  317. # --- 运行 ---
  318. async def main():
  319. agent = TransactionParserAgent(
  320. api_key="sk-8634dbc2866540c4b6003bb5733f23d8",
  321. multimodal_api_url="http://103.154.31.78:20012/api/file/read"
  322. )
  323. current_script_path = os.path.abspath(__file__)
  324. current_dir = os.path.dirname(current_script_path)
  325. # 模拟 Workflow 传入一个待处理文件
  326. input_pdf = "data_files/11111.png"
  327. filepath = os.path.normpath(os.path.join(current_dir, "..", "..", input_pdf))
  328. if not os.path.exists(filepath):
  329. print(f"{filepath}文件不存在")
  330. return
  331. result = await agent.run_workflow_task(filepath)
  332. if result["status"] == "success":
  333. print(f"🎯 【数据标准化】任务完成!")
  334. else:
  335. print(f"❌ 任务失败: {result['message']}")
  336. if __name__ == "__main__":
  337. asyncio.run(main())