再也不靠人肉找合同"坑"：我的智能合同风险审查全流程落地实录

def generate_html_layout(json_result: Dict[str, Any], issues: List[Dict]) -> str:
    """基于 JSON 生成 HTML 版面恢复，并标注风险点"""
    # 1. 解析 OCR / PP-StructureV3 返回的版面结果
    layout_result = json_result.get("layoutParsingResults", [{}])[0].get(
        "prunedResult", {}
    )

    # 2. 提取文本元素：行文本 + 坐标 + 所属页码等
    text_elements = _extract_ocr_text_elements(layout_result)
    # text_elements 结构示例：
    # [
    #   {"page_id": 0, "text": "合同编号：...", "bbox": [x1, y1, x2, y2], "block_id": 1},
    #   ...
    # ]

    # 3. 将风险点根据文本内容或 block_id 映射到对应的文本元素
    issue_map = _build_issue_map_by_block_id(issues)

    # 4. 逐个元素生成绝对定位的 <div>，并根据风险等级上色
    html_blocks = []
    for elem in text_elements:
        block_id = elem.get("block_id")
        related_issues = issue_map.get(block_id, [])
        risk_level = _get_max_risk_level(related_issues)

        style = _build_style_from_bbox(elem["bbox"], risk_level)
        content = _escape_html(elem["text"])

        html_blocks.append(
            f'<div class="ocr-text ocr-risk-{risk_level}" style="{style}">'
            f"{content}"
            "</div>"
        )

    # 5. 包装成完整 HTML 页面（包含背景、样式和脚本）
    return _wrap_as_full_html_page("\n".join(html_blocks))

def find_text_positions_in_json(
    clause_text: str, json_result: Dict[str, Any]
) -> List[Dict[str, Any]]:
    """通过文本匹配在 JSON 中查找条款的位置信息"""
    layout_results = json_result.get("layoutParsingResults", [])
    matches: List[Dict[str, Any]] = []

    for layout_idx, layout_item in enumerate(layout_results):
        pruned = layout_item.get("prunedResult", {})
        # 1. 从结构化结果中查找（parsing_res_list）
        for block in pruned.get("parsing_res_list", []):
            text = _normalize_text(block.get("text", ""))
            if _is_clause_match(clause_text, text):
                matches.append(
                    {
                        "layout_index": layout_idx,
                        "block_id": block.get("block_id"),
                        "bbox": block.get("block_bbox"),
                        "source": "parsing_res_list",
                    }
                )

        # 2. 从整体 OCR 结果中查找（overall_ocr_res）
        for ocr_block in pruned.get("overall_ocr_res", []):
            for line in ocr_block.get("res", []):
                text = _normalize_text(line.get("text", ""))
                if _is_clause_match(clause_text, text):
                    matches.append(
                        {
                            "layout_index": layout_idx,
                            "block_id": ocr_block.get("block_id"),
                            "bbox": line.get("rec_boxes"),
                            "source": "overall_ocr_res",
                        }
                    )

    return matches

def _parse_model_json(self, content: str, context: str) -> Optional[Any]:
    """尝试解析模型返回的 JSON，失败时使用 json_repair 做容错"""
    if not content:
        return None

    # 1. 基础清洗：去掉 Markdown 代码块包装、无关前后缀
    cleaned = content.strip()
    cleaned = cleaned.replace("```json", "").replace("```", "").strip()

    # 2. 先尝试标准 JSON 解析（成功则直接返回）
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        logger.warning(f"{context} 返回的 JSON 非严格格式，尝试使用 json_repair 进行修复")

    # 3. 如果失败，再尝试使用 json_repair 自动修复
    if repair_json is None:
        logger.error("未检测到 json_repair 库，无法自动修复无效 JSON")
        return None

    try:
        repaired = repair_json(cleaned, ensure_ascii=False)
        return json.loads(repaired)
    except Exception as e:
        logger.error(f"使用 json_repair 解析 {context} 结果失败: {str(e)}; 原始内容片段: {cleaned[:200]}")
        return None

from pathlib import Path
from typing import Callable, Dict, Any

def calc_file_hash(file_path: str, chunk_size: int = 1024 * 1024) -> str:
    """对文件内容做哈希，保证同内容同哈希、不同内容必然不同"""
    import hashlib

    h = hashlib.sha256()
    with open(file_path, "rb") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            h.update(chunk)
    return h.hexdigest()


def get_cache_key(file_path: str, model_name: str) -> str:
    file_hash = calc_file_hash(file_path)
    return f"{file_hash}_{model_name}"


def load_or_run_workflow(cache_key: str, runner: Callable[[], Dict[str, Any]]) -> Dict[str, Any]:
    """如果缓存存在则直接返回结果，否则执行工作流并写入缓存"""
    cached = load_from_cache(cache_key)
    if cached is not None:
        return cached

    result = runner()
    save_to_cache(cache_key, result)
    return result