import os
import time
import asyncio
import io
import csv
import datetime
import httpx
import json
import uuid
# --- LangChain Imports ---
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.outputs import Generation
import re
class SafeJsonOutputParser(JsonOutputParser):
def parse_result(self, result, *, partial: bool = False):
if isinstance(result, list) and len(result) > 0:
generation = result[0]
elif isinstance(result, Generation):
generation = result
else:
raise ValueError(f"Unexpected result type: {type(result)}")
text = generation.text
# 1️⃣ 去 ...
text = re.sub(r".*?", "", text, flags=re.S).strip()
# 2️⃣ 去 ```json ``` 包裹
text = re.sub(r"^```(?:json)?|```$", "", text, flags=re.I | re.M).strip()
# 3️⃣ ⭐ 只截取 JSON 本体
match = re.search(r"(\[\s*{.*}\s*\]|\{\s*\".*\"\s*\})", text, flags=re.S)
if not match:
raise ValueError(f"Invalid json output after clean: {text[:200]}")
json_text = match.group(1)
return json.loads(json_text)
# --- 核心 Parser ---
class TransactionParserAgent:
def __init__(self, api_key: str, multimodal_api_url: str, base_url: str = "https://api.deepseek.com"):
# 1. 初始化 LangChain ChatOpenAI 客户端
# DeepSeek 完全兼容 OpenAI 接口,使用 ChatOpenAI 是标准做法
self.llm = ChatOpenAI(
model="deepseek-chat",
api_key=api_key,
base_url=base_url,
temperature=0.1,
max_retries=3, # LangChain 内置重试机制
# 配置 httpx 客户端以优化超时和连接 (LangChain 允许透传 http_client)
http_client=httpx.Client(
timeout=httpx.Timeout(300.0, read=300.0, connect=300.0),
limits=httpx.Limits(max_keepalive_connections=5, max_connections=10)
)
)
self.multimodal_api_url = multimodal_api_url
# 定义 JSON 解析器
self.parser = SafeJsonOutputParser()
# 初始化API调用跟踪
self.api_calls = []
def _validate_and_reconcile(self, parsed_data: list) -> list:
"""
金额校验与余额矫正逻辑:
1. 正序排列
2. 寻找锚点 (符合前后余额勾稽关系)
3. 双向推演
4. 熔断机制 (40%)
"""
if len(parsed_data) < 3:
return parsed_data
sorted_data = [];
# 按照日期和时间正序排列(最早的在第一条)
# 假设 txDate 为 YYYY-MM-DD, txTime 为 HH:mm:ss
def get_sort_key(x):
# 处理时间:空值用最大时间兜底
date = x.get('txDate', '9999-12-31')
time = x.get('txTime', '23:59:59')
# 处理txId:提取数字,异常则返回极大值(倒序时排最后)
try:
txid_num = int(''.join([c for c in x.get('txId', '') if c.isdigit()]))
except:
txid_num = float('inf')
return (date, time, -txid_num)
if parsed_data[0]["txDate"] > parsed_data[len(parsed_data)-1]["txDate"]:
sorted_data = sorted(parsed_data, key=get_sort_key)
def to_float(s):
try:
return round(float(s), 2)
except:
return 0.0
anchor_idx = -1
# 1. 寻找肯定正确的数据行(锚点)
# 逻辑:
# 1. 当前行余额 = 上一行余额 +- 当前行金额
# 当前行余额 = 下一行余额 -+ 下一行金额
for i in range(1, len(sorted_data) - 1):
p = sorted_data[i - 1]
c = sorted_data[i]
n = sorted_data[i + 1]
# 当前行余额
c_bal = to_float(c['txBalance'])
# 计算上一笔推导当前笔
p_bal = to_float(p['txBalance'])
c_amt = to_float(c['txAmount'])
calc_c_bal = round(p_bal + c_amt, 2) if c['txDirection'] == "收入" else round(p_bal - c_amt, 2)
# 计算当前笔推导下一笔
n_amt = to_float(n['txAmount'])
n_bal = to_float(n['txBalance'])
calc_n_bal = round(n_bal - n_amt, 2) if n['txDirection'] == "收入" else round(n_bal + n_amt, 2)
if c_bal == calc_c_bal and c_bal == calc_n_bal:
anchor_idx = i
break
if anchor_idx == -1:
print("⚠️ 无法找到勾稽关系吻合的锚点,判定为不连续流水,跳过矫正。")
return parsed_data
print(f"数据锚点index={anchor_idx}")
# 2. 从锚点开始向上下推演修正
new_data = [item.copy() for item in sorted_data]
fix_count = 0
print(sorted_data)
# 往下推 (未来) 当前行余额 = 下一行余额 -+ 下一行金额
for i in range(anchor_idx + 1, len(new_data)-1):
curr_bal = to_float(new_data[i]['txBalance'])
next_bal = to_float(new_data[i + 1]['txBalance'])
next_amt = to_float(new_data[i + 1]['txAmount'])
expected_bal = round(next_bal - next_amt, 2) if new_data[i+1]['txDirection'] == "收入" else round(next_bal + next_amt, 2)
if abs(to_float(curr_bal) - expected_bal) > 0.01:
print(f"【往下推】行{new_data[i]['txId']}余额为:{curr_bal},计算余额为:{expected_bal}")
new_data[i]['txBalance'] = str(expected_bal)
fix_count += 1
# 往上推 (过去) 当前行余额 = 上一行余额 +- 当前行金额
for i in range(anchor_idx - 1, 0, -1): # 关键:终止条件改为0,避免i=0
curr_bal = to_float(new_data[i]['txBalance'])
pre_bal = to_float(new_data[i - 1]['txBalance'])
curr_amt = to_float(new_data[i]['txAmount'])
expected_bal = round(pre_bal + curr_amt, 2) if new_data[i]['txDirection'] == "收入" else round(pre_bal - curr_amt, 2)
if abs(to_float(curr_bal) - expected_bal) > 0.01:
print(f"【往上推】行{new_data[i]['txId']}余额为:{curr_bal},计算余额为:{expected_bal}")
new_data[i]['txBalance'] = str(expected_bal)
fix_count += 1
# 3. 熔断判定
fix_ratio = fix_count / len(new_data)
if fix_ratio > 0.4:
print(f"⚠️ 修正比例 {fix_ratio:.2%} 超过 40%,怀疑数据非连续,放弃修正。")
return parsed_data
print(f"✅ 余额勾稽校验通过,自动矫正了 {fix_count} 条数据偏差。")
return new_data
async def _invoke_miner_u(self, file_path: str) -> str:
"""调用 MinerU 并提取纯行数据 (保持 httpx 调用不变,因为这不是 LLM)"""
miner_start_time = time.perf_counter()
print("\n" + "=" * 40)
print("📌 【步骤1 - 数据提取】 开始执行")
dealRows = 0
try:
# MinerU 是独立服务,继续使用原生 httpx
async with httpx.AsyncClient() as client:
with open(file_path, 'rb') as f:
files = {'file': (os.path.basename(file_path), f)}
data = {'folderId': 'text'}
print("🔄数据提取中...")
response = await client.post(self.multimodal_api_url, files=files, data=data, timeout=120.0)
if response.status_code == 200:
res_json = response.json()
full_md_list = []
for element in res_json.get('convert_json', []):
if 'md' in element:
full_md_list.append(element['md'])
if 'rows' in element:
dealRows+=len(element['rows'])
print(f"📊 提取结果:共提取 {dealRows-1} 条数据")
return "\n\n".join(full_md_list)
return ""
except Exception as e:
print(f"❌ MinerU 调用异常: {e}")
return ""
finally:
print(f"✅ 【步骤1 - 数据提取】 执行完成")
print(f"⏱️ 执行耗时:{ time.perf_counter() - miner_start_time:.2f} 秒")
def _get_csv_prompt_template(self) -> ChatPromptTemplate:
"""
构造 LangChain 的 Prompt 模板
"""
system_template = """
# Role
你是一个高精度的银行账单转换工具。
# Task
将输入的 Markdown 表格行转换为 JSON 数组。
# Field Rules
1. txId: 如果输入数据中有交易流水号则直接使用,如果没有,从 T{start_id:04d} 开始递增生成。
2. txDate: 交易日期,格式为YYYY-MM-DD
3. txTime: 交易时间,格式为HH:mm:ss (未知填 00:00:00)
4. txAmount: 交易金额,绝对值数字
5. txBalance: 交易后余额。浮点数,移除千分位逗号。
6. txDirection: 交易方向。必须根据以下逻辑判断只输出“收入”或“支出”:
- 若有“借/贷”列:“借”通常为支出,“贷”通常为收入(除非是信用卡,需结合表头)。
- 若有“收入/支出”分列:按列归类。
- 若金额带正负号:"+"为收入,"-"为支出。
- 如果无符号,请结合表头判断。
7. txSummary: 摘要、用途、业务类型等备注。
8. txCounterparty: 交易对手方(名称及账号,如有)。
# Constraints
- **强制输出格式**:
1. 严格返回一个包含对象的 JSON 数组。
2. 每个对象必须包含上述 8 个字段名作为 Key。
3. 不要输出任何解释文字或 Markdown 代码块标签。
# Anti-Hallucination Rules
- 不得根据上下文推断任何未在原始数据中明确出现的字段
- 不得计算或猜测余额
- 不得根据常识补全对手方名称
- 若字段缺失,必须返回空字符串 ""
"""
user_template = """# Input Data
{chunk_data}
# Output
JSON Array:
"""
return ChatPromptTemplate.from_messages([
("system", system_template),
("user", user_template)
])
async def parse_to_csv(self, file_path: str) -> str:
# 1. 获取完整 Markdown 文本并按行切分
md_text = await self._invoke_miner_u(file_path)
if not md_text:
return ""
# 记录开始时间(使用time.perf_counter获取高精度时间)
switch_start_time = time.perf_counter()
print("\n" + "=" * 40)
print("📌 【步骤2 - 标准化转换】 开始执行")
# 初步切分
raw_lines = md_text.splitlines()
# 提取真正的第一行作为基准表头
clean_lines = [l.strip() for l in raw_lines if l.strip()]
if len(clean_lines) < 2: return ""
# --- 【核心改进:动态寻找表头】 ---
table_header = ""
header_index = 0
header_keywords = ["余额", "金额", "账号", "日期", "借/贷", "摘要"]
for idx, line in enumerate(clean_lines):
# 如果某一行包含 2 个以上关键词,且含有 Markdown 表格分隔符 '|'
hit_count = sum(1 for kw in header_keywords if kw in line)
if hit_count >= 2 and "|" in line:
table_header = line
header_index = idx
break
if not table_header:
table_header = clean_lines[0]
header_index = 0
data_rows = []
for line in clean_lines[header_index + 1:]:
if all(c in '|- ' for c in line): continue
if line == table_header: continue
# 过滤掉一些 MinerU 可能在表格末尾产生的页码或无关文字
if "|" not in line: continue
data_rows.append(line)
csv_header = "txId,txDate,txTime,txAmount,txDirection,txBalance,txSummary,txCounterparty,createdAt\n"
csv_content = csv_header
batch_size = 15
global_tx_counter = 1
# 构建 LCEL Chain: Prompt -> LLM -> Parser
chain = self._get_csv_prompt_template() | self.llm | self.parser
# 2. 分块处理
for i in range(0, len(data_rows), batch_size):
chunk = data_rows[i: i + batch_size]
context_chunk = [table_header] + chunk
chunk_str = "\n".join(context_chunk)
# 1. 记录开始时间(使用time.perf_counter获取高精度时间)
start_time = time.perf_counter()
print(f"🔄 正在通过LLM转换批次 {i // batch_size + 1},包含 {len(chunk)} 条数据...")
# print(f"待转换的数据块:\n{chunk_str}")
try:
# --- LangChain 调用 ---
# 使用 ainvoke 异步调用链
# 记录API调用开始时间
call_start_time = datetime.datetime.now()
data_data = await chain.ainvoke({
"start_id": global_tx_counter,
"chunk_data": chunk_str
})
# 记录API调用结束时间
call_end_time = datetime.datetime.now()
# 记录API调用结果 - 简化版:只保存提示词和结果数据
call_id = f"api_llm_数据转换_{'{:.2f}'.format((call_end_time - call_start_time).total_seconds())}"
# 从chain中提取提示词(如果可能)
prompt_content = ""
try:
# 尝试从chain获取最后的消息内容
if hasattr(chain, 'get_prompts'):
prompts = chain.get_prompts()
if prompts:
prompt_content = str(prompts[-1])
else:
# 如果无法获取,构造基本的提示词信息
prompt_content = f"转换批次数据,start_id: {global_tx_counter}, chunk_data: {chunk_str[:200]}..."
except:
prompt_content = f"转换批次数据,start_id: {global_tx_counter}, chunk_data: {chunk_str[:200]}..."
api_call_info = {
"call_id": call_id,
"start_time": call_start_time.isoformat(),
"end_time": call_end_time.isoformat(),
"duration": (call_end_time - call_start_time).total_seconds(),
"prompt": prompt_content,
"input_params": {
"start_id": global_tx_counter,
"chunk_data": chunk_str
},
"llm_result": data_data
}
self.api_calls.append(api_call_info)
# 保存API结果到文件 (Markdown格式,更易阅读)
# 使用运行ID创建独立的文件夹
run_id = os.environ.get('FLOW_RUN_ID', 'default')
api_results_dir = f"api_results_{run_id}"
os.makedirs(api_results_dir, exist_ok=True)
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{timestamp}_{call_id}.md"
filepath = os.path.join(api_results_dir, filename)
try:
with open(filepath, 'w', encoding='utf-8') as f:
f.write("# 数据转换结果\n\n")
f.write("## 调用信息\n\n")
f.write(f"- 调用ID: {call_id}\n")
f.write(f"- 开始时间: {call_start_time.isoformat()}\n")
f.write(f"- 结束时间: {call_end_time.isoformat()}\n")
f.write(f"- 执行时长: {(call_end_time - call_start_time).total_seconds():.2f} 秒\n")
f.write("\n## 提示词入参\n\n")
f.write("```\n")
f.write(api_call_info["prompt"])
f.write("\n```\n\n")
f.write("## 输入参数\n\n")
f.write("```json\n")
f.write(json.dumps(api_call_info["input_params"], ensure_ascii=False, indent=2))
f.write("\n```\n\n")
f.write("## LLM返回结果\n\n")
f.write("```json\n")
f.write(json.dumps(api_call_info["llm_result"], ensure_ascii=False, indent=2))
f.write("\n```\n")
print(f"[API_RESULT] 保存API结果文件: {filepath}")
except Exception as e:
print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
# print(f"💡 LLM 返回数据: {data_data}")
# 兼容处理:LangChain Parser 通常会直接返回 List 或 Dict
if isinstance(data_data, dict):
# 尝试寻找 transactions 键,如果没有则假设整个 dict 就是我们要的对象(虽然罕见)
batch_data = data_data.get("transactions", [data_data])
# 如果取出来还是 dict (例如单条记录),包一层 list
if isinstance(batch_data, dict):
batch_data = [batch_data]
elif isinstance(data_data, list):
batch_data = data_data
else:
batch_data = []
if batch_data:
final_list = self._validate_and_reconcile(batch_data)
final_list = sorted(final_list, key=lambda x: (x['txId']))
output = io.StringIO()
createdAtStr = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
writer = csv.writer(output, quoting=csv.QUOTE_MINIMAL, lineterminator='\n')
print(f"✅ 批次转换成功,包含 {len(final_list)} 条记录。")
for item in final_list:
writer.writerow([
item.get("txId", ""),
item.get("txDate", ""),
item.get("txTime", ""),
item.get("txAmount", ""),
item.get("txDirection", ""),
item.get("txBalance", ""),
item.get("txSummary", ""),
item.get("txCounterparty", ""),
createdAtStr
])
batch_csv_string = output.getvalue()
csv_content += batch_csv_string
global_tx_counter += len(final_list)
except Exception as e:
print(f"⚠️ 批次执行失败: {e}")
finally:
end_time = time.perf_counter()
elapsed_time = end_time - start_time
print(f"⏱️ 执行耗时: {elapsed_time:.2f} 秒")
print(f"📊 转换结果:共转换 {global_tx_counter - 1} 条数据")
print(f"✅ 【步骤2 - 标准化转换】 执行完成")
print(f"⏱️ 执行总耗时:{time.perf_counter() - switch_start_time:.2f} 秒")
return csv_content
async def parse_and_save_to_file(self, file_path: str, output_dir: str = "output") -> str:
"""
供 Workflow 调用:解析并保存文件,返回全路径名
"""
current_script_path = os.path.abspath(__file__)
current_dir = os.path.dirname(current_script_path)
file_full_name = os.path.basename(file_path)
file_name = os.path.splitext(file_full_name)[0] # 不带后缀 11111
output_dir = os.path.normpath(os.path.join(current_dir, "..", "..", output_dir))
os.makedirs(output_dir, exist_ok=True)
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
file_name = f"{file_name}_data_standard_{timestamp}.csv"
full_path = os.path.join(output_dir, file_name)
csv_result = await self.parse_to_csv(file_path)
if csv_result:
with open(full_path, "w", encoding="utf-8") as f:
f.write(csv_result)
return full_path
else:
raise Exception("数据解析失败,未生成有效内容")
async def run_workflow_task(self, input_file_path: str) -> dict:
"""
标准 Workflow 入口方法
"""
# 1. 记录开始时间(使用time.perf_counter获取高精度时间)
start_time = time.perf_counter()
print(f"BEGIN---数据标准化任务开始---")
try:
print(f"待执行标准化的文件:{input_file_path}")
api_results_dir = "data_files"
saved_path = await self.parse_and_save_to_file(input_file_path, api_results_dir)
return {
"status": "success",
"file_path": saved_path,
"file_name": os.path.basename(saved_path),
"timestamp": datetime.datetime.now().isoformat()
}
except Exception as e:
return {
"status": "error",
"message": str(e)
}
finally:
end_time = time.perf_counter()
elapsed_time = end_time - start_time
print(f"⏱️ 执行总耗时: {elapsed_time:.2f} 秒")
print(f"END---数据标准化任务结束")
async def data_standize(api_key: str, base_url: str, multimodal_api_url: str, input_file_path: str) -> dict:
"""
数据标准化入口方法
"""
# 创建Agent
agent = TransactionParserAgent(
api_key="sk-8634dbc2866540c4b6003bb5733f23d8",
base_url=base_url,
multimodal_api_url=multimodal_api_url
)
# 执行标准化处理
return await agent.run_workflow_task(input_file_path)
# --- 运行 ---
async def main():
agent = TransactionParserAgent(
api_key="sk-8634dbc2866540c4b6003bb5733f23d8",
multimodal_api_url="http://103.154.31.78:20012/api/file/read"
)
current_script_path = os.path.abspath(__file__)
current_dir = os.path.dirname(current_script_path)
# 模拟 Workflow 传入一个待处理文件
input_pdf = "data_files/4.pdf"
filepath = os.path.normpath(os.path.join(current_dir, "..", "..", input_pdf))
if not os.path.exists(filepath):
print(f"{filepath}文件不存在")
return
result = await agent.run_workflow_task(filepath)
if result["status"] == "success":
print(f"🎯 【数据标准化】任务完成!")
else:
print(f"❌ 任务失败: {result['message']}")
if __name__ == "__main__":
asyncio.run(main())