import json
from typing import Any
from backend.evaluation import format_match_label
from backend.extract_schema import EXTRACT_FIELDS
DIFFICULTY_LABELS = {
"easy": "简单",
"medium": "中等",
"hard": "困难",
}
TASK_LABELS = {
"classify": "金融文本分类",
"extract": "金融信息抽取",
"match": "金融文本匹配",
}
def preview_text(text: str, max_length: int) -> str:
value = str(text or "").strip().replace("\n", " ")
if len(value) <= max_length:
return value
return f"{value[:max_length]}..."
def format_input_text(task: str, input_data: dict[str, Any], max_length: int) -> str:
if task == "match":
text_a = preview_text(input_data.get("text_a", ""), max_length)
text_b = preview_text(input_data.get("text_b", ""), max_length)
return f"文本 A: {text_a}\n文本 B: {text_b}"
return preview_text(input_data.get("text", ""), max_length)
def _format_value(value: Any) -> str:
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
def format_expected_lines(task: str, expected: dict[str, Any] | None) -> list[str]:
if not expected:
return ["标准: （无标准标签）"]
if task == "classify":
return [f"标准: category = {expected.get('category', '（无）')}"]
if task == "match":
return [f"标准: is_match = {_format_value(expected.get('is_match'))}"]
if task == "extract":
lines = ["标准:"]
has_value = False
for key in EXTRACT_FIELDS:
value = expected.get(key, [])
if isinstance(value, list) and value:
has_value = True
preview = "、".join(str(item) for item in value[:3])
suffix = "..." if len(value) > 3 else ""
lines.append(f" {key}: {preview}{suffix}")
if not has_value:
lines.append(" （无必填字段）")
return lines
return [f"标准: {json.dumps(expected, ensure_ascii=False)}"]
def format_result_lines(task: str, result: dict[str, Any] | None) -> list[str]:
if not result:
return ["模型: （无）"]
if task == "classify":
return [f"模型: category = {result.get('category', '（无）')}"]
if task == "extract":
lines = ["模型:"]
has_value = False
for key in EXTRACT_FIELDS:
value = result.get(key, [])
if isinstance(value, list) and value:
has_value = True
preview = "、".join(str(item) for item in value[:3])
suffix = "..." if len(value) > 3 else ""
lines.append(f" {key}: {preview}{suffix}")
elif isinstance(value, list):
lines.append(f" {key}: （空）")
if not has_value:
lines.append(" （空）")
return lines
if task == "match":
return [
f"模型: is_match = {_format_value(result.get('is_match'))}",
f" score = {result.get('score')}",
]
return [f"模型: {json.dumps(result, ensure_ascii=False)}"]
def format_summary_value(task: str, item: dict[str, Any]) -> str:
expected = item.get("expected")
result = item.get("result")
if task == "classify":
standard = expected.get("category", "（无）") if expected else "（无）"
actual = result.get("category", "（无）") if result else "（无）"
return f"标准={standard} | 模型={actual}"
if task == "match":
standard = _format_value(expected.get("is_match")) if expected else "（无）"
actual = _format_value(result.get("is_match")) if result else "（无）"
return f"标准={standard} | 模型={actual}"
if task == "extract":
label = format_match_label(item.get("is_correct"))
return label
return "（已返回）"
def print_results(
task: str,
results: list[dict[str, Any]],
*,
output_mode: str,
text_preview_length: int,
show_raw_output: bool,
output_path: str | None = None,
) -> None:
if output_mode == "json":
print(json.dumps(results, ensure_ascii=False, indent=2))
if output_path:
print(f"\n完整 JSON 已保存到: {output_path}")
return
if output_mode == "quiet":
success_count = sum(1 for item in results if item.get("status") == "success")
failed_count = len(results) - success_count
correct_count = sum(1 for item in results if item.get("is_correct") is True)
compared_count = sum(
1 for item in results if item.get("is_correct") is not None
)
print(
f"任务 {task} 完成：成功 {success_count} 条，失败 {failed_count} 条，"
f"正确 {correct_count}/{compared_count} 条。"
)
if output_path:
print(f"完整 JSON 已保存到: {output_path}")
return
task_label = TASK_LABELS.get(task, task)
print("=" * 60)
print(f"金融文本打标结果 | 任务: {task_label} | 共 {len(results)} 条")
print("=" * 60)
print()
for index, item in enumerate(results, start=1):
sample_id = item.get("id", f"sample_{index}")
difficulty = DIFFICULTY_LABELS.get(
item.get("difficulty", ""),
item.get("difficulty", "未知"),
)
status = "成功" if item.get("status") == "success" else "失败"
print(f"[{index}/{len(results)}] {sample_id} | 难度: {difficulty}")
print(format_input_text(task, item.get("input", {}), text_preview_length))
print()
for line in format_expected_lines(task, item.get("expected")):
print(line)
for line in format_result_lines(task, item.get("result")):
print(line)
if item.get("is_correct") is not None:
print(f"对比: {format_match_label(item.get('is_correct'))}")
print(f"状态: {status}")
if item.get("status") != "success":
print(f"错误: {item.get('error', '未知错误')}")
if show_raw_output and item.get("raw_output"):
print(f"原始返回: {item.get('raw_output')}")
if show_raw_output and item.get("status") == "success" and item.get("raw_output"):
print(f"原始返回: {item.get('raw_output')}")
print()
print("-" * 60)
print("汇总")
print("-" * 60)
for item in results:
sample_id = item.get("id", "unknown")
difficulty = DIFFICULTY_LABELS.get(
item.get("difficulty", ""),
item.get("difficulty", "未知"),
)
status = "成功" if item.get("status") == "success" else "失败"
compare = format_match_label(item.get("is_correct"))
summary = format_summary_value(task, item)
print(f"{sample_id:<12} {difficulty:<4} {compare:<6} {status:<4} {summary}")
if output_path:
print()
print(f"完整 JSON 已保存到: {output_path}")