feat: add party role

70149414 · ccran · 447b8d67 · 70149414 · 70149414 · 70149414
Commit 70149414 authored Jul 03, 2026 by ccran
10 changed files
--- a/.gitignore
+++ b/.gitignore
@@ -7,12 +7,13 @@
 # Keep Python source files
 !**/*.py
-!**/*.doc
+# !**/*.doc
-!**/*.docx
+# !**/*.docx
-!**/*.xlsx
+# !**/*.xlsx
-!**/*.pdf
+# !**/*.pdf
-!**/*.xls
+# !**/*.xls
 !workflow/**
+!data/**.xlsx
 !README.md

--- a/core/config.py
+++ b/core/config.py
@@ -12,16 +12,27 @@ use_docker = False
 #     model: str = "deepseek-v4-pro"
+# @dataclass
+# class LLMConfig:
+#     base_url: str = "http://172.21.107.80:9002/v1"
+#     api_key: str = "none"
+#     model: str = "Qwen2-72B-Instruct"
 @dataclass
 class LLMConfig:
    base_url: str = "http://192.168.252.71:9002/v1"
    api_key: str = "none"
    model: str = "Qwen2-72B-Instruct"
+# 最大分片数量
+min_single_chunk_size = 2000
+max_single_chunk_size = 100000
+max_chunk_page = 10
-# MAX_SINGLE_CHUNK_SIZE=100000
+# MAX_SINGLE_CHUNK_SIZE = 100000
-MERGE_RULE_PROMPT = False
 MAX_SINGLE_CHUNK_SIZE = 5000
+# MAX_SINGLE_CHUNK_SIZE = 2000
+MERGE_RULE_PROMPT = False
 META_KEY = "META"
 DEFAULT_RULESET_ID = "通用"
 ## 规则集ID列表，需与rules.xlsx中的sheet名称保持一致!!!
@@ -33,7 +44,7 @@ ALL_RULESET_IDS = [
    "金盘",
    "金盘简化",
    "麓发测试",
-    "麓发标准",
+    "通用（原始）",
    "金盘B类",
    "合同信息提取（合同组）",
    "合同信息提取（技术部）",
@@ -45,11 +56,13 @@ MAX_WORKERS = 10
 FILE_SUFFIX = "-审核批注"
 ## 关键参数**
-use_non_fastgpt_llm = False
+use_non_fastgpt_llm = True
 use_lufa = False
 use_jp_machine = True
+debug_mode = False
 ## 关键参数**
+max_model_len = 131072
 ocr_url = 'http://192.168.252.71:8202/openapi/ocrUploadFile'
 if use_lufa:
    outer_backend_url = "http://znkf.lgfzgroup.com:48081"
@@ -67,9 +80,9 @@ else:
        base_fastgpt_url = "http://192.168.252.71:18088"
        base_backend_url = "http://192.168.252.71:48080"
    else:
-        outer_backend_url = "http://172.21.107.45:48080"
+        outer_backend_url = "https://172.21.107.45:48080"
        base_fastgpt_url = "http://172.21.107.45:3030"
-        base_backend_url = "http://172.21.107.45:48080"
+        base_backend_url = "http://172.21.107.45:1122"
        ocr_url = "http://172.21.107.45:8202/openapi/ocrUploadFile"
    segment_review_api_key = (
        "fastgpt-vLu2JHAfqwEq5FUQhvATFDK0yDS6fs804v7KwWBMyU4sRrHzh4UGl89Zpa"
@@ -108,7 +121,3 @@ LLM = {
 }
 doc_support_formats = [".docx", ".doc", ".wps"]
 pdf_support_formats = [".txt", ".md", ".pdf"]
-# 最大分片数量
-min_single_chunk_size = 2000
-max_single_chunk_size = 20000
-max_chunk_page = 10
--- a/core/tools/party_role.py
+++ b/core/tools/party_role.py
+from __future__ import annotations
+from typing import Dict
+from loguru import logger
+from core.tool import tool, tool_func
+from core.tools.segment_llm import LLMTool
+PARTY_ROLE_SYSTEM_PROMPT = """
+你是一名专业合同审查律师。
+"""
+PARTY_ROLE_USER_PROMPT = """
+请分析指定公司在合同中的实际商业角色。
+注意：
+- 不要仅依据“甲方/乙方”字样判断
+- 必须结合实际权利义务关系判断
+重点考虑：
+- 谁提出需求
+- 谁支付费用
+- 谁接受服务或交付成果
+- 谁拥有验收、管理、监督权
+- 谁负责提供产品或服务
+输出要求：
+只输出一个 JSON：
+{{
+  "party_role": "demand_side | supplier_side | unclear",
+  "reason": ""
+}}
+其中：
+- demand_side：需求方、委托人、采购方、客户、甲方等
+- supplier_side：供应商、服务方、承包方、实施方、乙方等
+- unclear：无法明确判断
+reason 要用一句话简要说明核心依据。
+禁止输出任何Markdown、代码块或额外文本。
+待分析公司：
+{company_name}
+合同内容：
+{contract_text}
+"""
+VALID_PARTY_ROLES = {"demand_side", "supplier_side", "mixed", "unclear"}
+PARTY_ROLES_MAP={
+    "demand_side": "需求方、委托人、采购方、客户、甲方等",
+    "supplier_side": "供应商、服务方、承包方、实施方、乙方等",
+    "unclear": "",
+}
+@tool("party_role", "识别指定公司在合同中的实际商业角色")
+class PartyRoleTool(LLMTool):
+    def __init__(self) -> None:
+        super().__init__(PARTY_ROLE_SYSTEM_PROMPT)
+    @tool_func(
+        {
+            "type": "object",
+            "properties": {
+                "company_name": {"type": "string"},
+                "contract_text": {"type": "string"},
+            },
+            "required": ["company_name", "contract_text"],
+        }
+    )
+    def run(self, company_name: str, contract_text: str) -> Dict[str, str]:
+        company_name = (company_name or "").strip()
+        contract_text = (contract_text or "").strip()
+        if not company_name or not contract_text:
+            return {"party_role": "", "reason": ""}
+        user_content = PARTY_ROLE_USER_PROMPT.format(
+            company_name=company_name,
+            contract_text=contract_text,
+        )
+        try:
+            resp = self.run_with_loop(self.chat_async(self.build_messages(user_content)))
+            data = self.parse_first_json(resp)
+            party_role = str(data.get("party_role", "")).strip()
+            reason = str(data.get("reason", "")).strip()
+            if party_role in VALID_PARTY_ROLES:
+                return {"party_role": PARTY_ROLES_MAP.get(party_role, ""), "reason": reason}
+        except Exception as exc:
+            logger.error("Party role detection LLM failed: %s", exc)
+        return {"party_role": "", "reason": ""}
+if __name__ == "__main__":
+    tool = PartyRoleTool()
+    print(
+        tool.run(
+            company_name="麓发公司",
+            contract_text="麓发委托乙公司开发系统，甲公司支付费用并负责验收。",
+        )
+    )
--- a/data/rules.xlsx
+++ b/data/rules.xlsx
--- a/data/rules_backup.xlsx
+++ b/data/rules_backup.xlsx
--- a/demo/demo.py
+++ b/demo/demo.py
-for _ in range(input()):
-    try:
-        eval(raw_input())
-        print("YES")
-    except TypeError:
-        print("NO")
-    except:
-        print("NO")
--- a/demo/spire_doc_revision_compare.py
+++ b/demo/spire_doc_revision_compare.py
+import argparse
+import asyncio
+import difflib
+import json
+from itertools import zip_longest
+from pathlib import Path
+from typing import Any
+from openpyxl import Workbook, load_workbook
+from openpyxl.styles import Alignment, Font
+from core.config import LLMConfig
+from utils.common_util import extract_json
+from utils.openai_util import OpenAITool
+WORD_SUFFIXES = {".doc", ".docx", ".wps"}
+EXCEL_SUFFIXES = {".xlsx", ".xlsm", ".xls"}
+EXCEL_HEADER_NAMES = {"修改前", "修订前", "变更前", "before", "before_revision"}
+RULE_EXTRACT_SYSTEM_PROMPT = """你是资深合同审查规则分析助手。
+你的任务是根据单条合同条款的“修订前/修订后”对比文本，反推该条修订体现出的审查项和审查规则。
+输出要求：
+1. 只输出 JSON，不要输出 Markdown、解释或多余文本。
+2. JSON 格式必须是对象，包含：
+   - rule_title：审查项名称，简短准确，例如“纠纷解决审查”
+   - rule_detail：审查规则明细，用中文分点表述，例如“1）我司只接受……\\n2）我司只接受……”
+3. 只提取能从该条修订前后对比中合理归纳出的规则，不要编造无法支撑的规则。
+4. 如果该条差异无法归纳出审查规则，输出 {"rule_title": "", "rule_detail": ""}。
+"""
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Compare Word revisions or two-column Excel diff rows and write results to Excel.")
+    parser.add_argument("--input_file", 
+                        default=r'/home/ccran/lufa-contract/demo/test/5、[修订后复审版]麓谷发展集团2026-2027年质量_修订对比_2col.xlsx',
+                        help="Path to the Word document with revisions or a two-column Excel file.")
+    parser.add_argument(
+        "-o",
+        "--output",
+        help="Path to the output Excel file. Defaults to '<input_stem>_修订对比.xlsx'.",
+    )
+    parser.add_argument(
+        "--show-text",
+        action="store_true",
+        help="Print full text before and after accepting revisions.",
+    )
+    parser.add_argument(
+        "--no-diff-print",
+        action="store_true",
+        help="Do not print unified diff to the console.",
+    )
+    parser.add_argument(
+        "--no-rule-extract",
+        action="store_true",
+        help="Do not call LLM to extract review rules from revision diff rows.",
+    )
+    parser.add_argument(
+        "--batch-size",
+        type=int,
+        default=10,
+        help="Number of single-row prompts to run concurrently in one batch. Defaults to 10.",
+    )
+    return parser.parse_args()
+def get_default_output_file(input_file: Path) -> Path:
+    return input_file.with_name(f"{input_file.stem}_修订对比.xlsx")
+def read_revision_texts(input_file: Path) -> tuple[str, str]:
+    from spire.doc import Document
+    doc = Document()
+    try:
+        doc.LoadFromFile(str(input_file))
+        before_text = doc.GetText()
+        if doc.HasChanges:
+            doc.AcceptChanges()
+        after_text = doc.GetText()
+        return before_text, after_text
+    finally:
+        doc.Close()
+def cell_to_text(value: Any) -> str:
+    if value != value:
+        return ""
+    return "" if value is None else str(value).strip()
+def is_excel_header_row(before_text: str, after_text: str) -> bool:
+    before_normalized = before_text.strip().lower()
+    after_normalized = after_text.strip().lower()
+    return before_normalized in EXCEL_HEADER_NAMES and after_normalized in {
+        "修改后",
+        "修订后",
+        "变更后",
+        "after",
+        "after_revision",
+    }
+def read_excel_diff_rows(input_file: Path) -> list[tuple[str, str]]:
+    suffix = input_file.suffix.lower()
+    if suffix in {".xlsx", ".xlsm"}:
+        wb = load_workbook(input_file, read_only=True, data_only=True)
+        try:
+            ws = wb.active
+            rows: list[tuple[str, str]] = []
+            for row_index, row in enumerate(ws.iter_rows(min_row=1, values_only=True), start=1):
+                before_line = cell_to_text(row[0] if len(row) > 0 else None)
+                after_line = cell_to_text(row[1] if len(row) > 1 else None)
+                if row_index == 1 and is_excel_header_row(before_line, after_line):
+                    continue
+                if not before_line and not after_line:
+                    continue
+                rows.append((before_line, after_line))
+            return rows
+        finally:
+            wb.close()
+    if suffix == ".xls":
+        try:
+            import pandas as pd
+        except ImportError as exc:
+            raise RuntimeError("读取 .xls 需要安装 pandas/xlrd，建议另存为 .xlsx 后再运行。") from exc
+        try:
+            df = pd.read_excel(input_file, header=None, usecols=[0, 1], dtype=str)
+        except Exception as exc:
+            raise RuntimeError("读取 .xls 失败，建议将文件另存为 .xlsx 后再运行。") from exc
+        rows = []
+        for row_index, row in df.iterrows():
+            before_line = cell_to_text(row.iloc[0])
+            after_line = cell_to_text(row.iloc[1])
+            if row_index == 0 and is_excel_header_row(before_line, after_line):
+                continue
+            if not before_line and not after_line:
+                continue
+            rows.append((before_line, after_line))
+        return rows
+    raise ValueError(f"unsupported Excel file type: {suffix}")
+def build_unified_diff(before_lines: list[str], after_lines: list[str]) -> list[str]:
+    return list(
+        difflib.unified_diff(
+            before_lines,
+            after_lines,
+            fromfile="before_revision",
+            tofile="after_revision",
+            lineterm="",
+        )
+    )
+def build_diff_rows(before_lines: list[str], after_lines: list[str]) -> list[tuple[str, str]]:
+    rows: list[tuple[str, str]] = []
+    matcher = difflib.SequenceMatcher(None, before_lines, after_lines)
+    for tag, before_start, before_end, after_start, after_end in matcher.get_opcodes():
+        if tag == "equal":
+            continue
+        before_part = before_lines[before_start:before_end]
+        after_part = after_lines[after_start:after_end]
+        if tag == "replace":
+            rows.extend(
+                (before_line or "", after_line or "")
+                for before_line, after_line in zip_longest(before_part, after_part)
+            )
+        elif tag == "delete":
+            rows.extend((before_line, "") for before_line in before_part)
+        elif tag == "insert":
+            rows.extend(("", after_line) for after_line in after_part)
+    return rows
+def write_diff_rows_to_excel(
+    diff_rows: list[tuple[str, str]],
+    output_file: Path,
+    rule_results: list[dict[str, str]] | None = None,
+) -> None:
+    wb = Workbook()
+    ws = wb.active
+    ws.title = "revision_compare"
+    ws.append(["修订前", "修订后", "审查项", "审查规则"])
+    for cell in ws[1]:
+        cell.font = Font(bold=True)
+        cell.alignment = Alignment(horizontal="center", vertical="center")
+    for index, (before_line, after_line) in enumerate(diff_rows):
+        rule = rule_results[index] if rule_results and index < len(rule_results) else {}
+        ws.append(
+            [
+                before_line,
+                after_line,
+                rule.get("rule_title", ""),
+                rule.get("rule_detail", ""),
+            ]
+        )
+    for row in ws.iter_rows(min_row=2):
+        for cell in row:
+            cell.alignment = Alignment(wrap_text=True, vertical="top")
+        if row[1].value:
+            row[1].font = Font(color="FF0000")
+    ws.column_dimensions["A"].width = 80
+    ws.column_dimensions["B"].width = 80
+    ws.column_dimensions["C"].width = 24
+    ws.column_dimensions["D"].width = 80
+    ws.freeze_panes = "A2"
+    wb.save(output_file)
+def chunk_list(items: list[tuple[str, str]], batch_size: int) -> list[list[tuple[int, str, str]]]:
+    if batch_size <= 0:
+        raise ValueError("--batch-size must be greater than 0")
+    indexed_items = [
+        (index, before_line, after_line)
+        for index, (before_line, after_line) in enumerate(items, start=1)
+    ]
+    return [
+        indexed_items[start : start + batch_size]
+        for start in range(0, len(indexed_items), batch_size)
+    ]
+def build_rule_extract_user_prompt(before_line: str, after_line: str) -> str:
+    return (
+        "请根据以下单条合同修订前后对比文本，提取该条修订体现出的审查项和审查规则。\n"
+        "注意：修订后的文本通常代表更符合我司要求的表达，请结合修订前被替换或删除的内容归纳规则。\n"
+        f"{json.dumps({'修订前': before_line, '修订后': after_line}, ensure_ascii=False, indent=2)}"
+    )
+def normalize_single_rule_response(value: Any) -> dict[str, str]:
+    if isinstance(value, list):
+        rules = [item for item in value if isinstance(item, dict)]
+    elif isinstance(value, dict) and isinstance(value.get("rules"), list):
+        rules = [item for item in value["rules"] if isinstance(item, dict)]
+    elif isinstance(value, dict):
+        rules = [value]
+    else:
+        rules = []
+    normalized_rules = [
+        {
+            "rule_title": str(item.get("rule_title", "")).strip(),
+            "rule_detail": str(item.get("rule_detail", "")).strip(),
+        }
+        for item in rules
+        if str(item.get("rule_title", "")).strip() or str(item.get("rule_detail", "")).strip()
+    ]
+    if not normalized_rules:
+        return {"rule_title": "", "rule_detail": ""}
+    return {
+        "rule_title": "\n".join(rule["rule_title"] for rule in normalized_rules if rule["rule_title"]),
+        "rule_detail": "\n\n".join(rule["rule_detail"] for rule in normalized_rules if rule["rule_detail"]),
+    }
+def build_rule_extract_messages(before_line: str, after_line: str) -> list[dict[str, str]]:
+    return [
+        {"role": "system", "content": RULE_EXTRACT_SYSTEM_PROMPT},
+        {"role": "user", "content": build_rule_extract_user_prompt(before_line, after_line)},
+    ]
+def parse_rule_response(response: str) -> dict[str, str]:
+    parsed_items = extract_json(response)
+    parsed = parsed_items[0] if parsed_items else {}
+    return normalize_single_rule_response(parsed)
+async def extract_review_rules_async(
+    diff_rows: list[tuple[str, str]],
+    batch_size: int,
+) -> list[dict[str, str]]:
+    llm = OpenAITool(LLMConfig())
+    results_by_index: dict[int, dict[str, str]] = {}
+    batches = chunk_list(diff_rows, batch_size)
+    for batch_index, batch_rows in enumerate(batches, start=1):
+        start_index = batch_rows[0][0]
+        end_index = batch_rows[-1][0]
+        print(f"正在并发提取第 {batch_index}/{len(batches)} 批审查规则，行 {start_index}-{end_index}...")
+        messages_list = [
+            build_rule_extract_messages(before_line, after_line)
+            for _, before_line, after_line in batch_rows
+        ]
+        responses = await llm.mul_chat(messages_list)
+        for (row_index, _, _), response in zip(batch_rows, responses):
+            results_by_index[row_index] = parse_rule_response(response)
+    results: list[dict[str, str]] = []
+    for index in range(1, len(diff_rows) + 1):
+        results.append(results_by_index.get(index, {"rule_title": "", "rule_detail": ""}))
+    return results
+def extract_review_rules(diff_rows: list[tuple[str, str]], batch_size: int) -> list[dict[str, str]]:
+    return asyncio.run(extract_review_rules_async(diff_rows, batch_size))
+def build_diff_rows_from_word(
+    input_file: Path,
+    show_text: bool,
+    print_diff: bool,
+) -> list[tuple[str, str]]:
+    before_text, after_text = read_revision_texts(input_file)
+    if show_text:
+        print("===== 修订前文本 =====")
+        print(before_text)
+        print("===== 修订后文本 =====")
+        print(after_text)
+    if before_text == after_text:
+        return []
+    before_lines = before_text.splitlines()
+    after_lines = after_text.splitlines()
+    diff_lines = build_unified_diff(before_lines, after_lines)
+    diff_rows = build_diff_rows(before_lines, after_lines)
+    if print_diff and diff_rows:
+        print("===== Diff 对比结果 =====")
+        print("\n".join(diff_lines))
+    return diff_rows
+def load_diff_rows(input_file: Path, show_text: bool, print_diff: bool) -> list[tuple[str, str]]:
+    suffix = input_file.suffix.lower()
+    if suffix in EXCEL_SUFFIXES:
+        diff_rows = read_excel_diff_rows(input_file)
+        print(f"从 Excel 读取到 {len(diff_rows)} 条前后对比。")
+        return diff_rows
+    if suffix in WORD_SUFFIXES:
+        return build_diff_rows_from_word(input_file, show_text, print_diff)
+    raise ValueError(f"不支持的输入文件类型: {suffix}")
+def compare_revisions(
+    input_file: Path,
+    output_file: Path,
+    show_text: bool,
+    print_diff: bool,
+    extract_rules: bool,
+    batch_size: int,
+) -> bool:
+    diff_rows = load_diff_rows(input_file, show_text, print_diff)
+    if not diff_rows:
+        print("没有检测到前后对比内容，不写入 Excel。")
+        return False
+    rule_results = extract_review_rules(diff_rows, batch_size) if extract_rules else None
+    write_diff_rows_to_excel(diff_rows, output_file, rule_results)
+    print(f"Excel Diff 对比结果已写入: {output_file}")
+    return True
+def main() -> None:
+    args = parse_args()
+    input_file = Path(args.input_file)
+    output_file = Path(args.output) if args.output else get_default_output_file(input_file)
+    compare_revisions(
+        input_file=input_file,
+        output_file=output_file,
+        show_text=args.show_text,
+        print_diff=not args.no_diff_print,
+        extract_rules=not args.no_rule_extract,
+        batch_size=args.batch_size,
+    )
+if __name__ == "__main__":
+    main()
--- a/demo/spire_doc_revision_compare_demo.py
+++ b/demo/spire_doc_revision_compare_demo.py
-from spire.doc import *
+import argparse
+import asyncio
 import difflib
+import json
+from itertools import zip_longest
+from pathlib import Path
+from typing import Any
+from openpyxl import Workbook
+from openpyxl.styles import Alignment, Font
-# 将文本写入文件
+from core.config import LLMConfig
-def WriteAllText(fname: str, text: str):
+from utils.common_util import extract_json
-    with open(fname, "w", encoding="utf-8") as fp:
+from utils.openai_util import OpenAITool
-        fp.write(text)
-inputFile = "/home/ccran/lufa-contract/demo/修订测试.docx"
+RULE_EXTRACT_SYSTEM_PROMPT = """你是资深合同审查规则分析助手。
+你的任务是根据单条合同条款的“修订前/修订后”对比文本，反推该条修订体现出的审查项和审查规则。
-# 创建Document类的对象
+输出要求：
-doc = Document()
+1. 只输出 JSON，不要输出 Markdown、解释或多余文本。
+2. JSON 格式必须是对象，包含：
+   - rule_title：审查项名称，简短准确，例如“纠纷解决审查”
+   - rule_detail：审查规则明细，用中文分点表述，例如“1）我司只接受……\\n2）我司只接受……”
+3. 只提取能从该条修订前后对比中合理归纳出的规则，不要编造无法支撑的规则。
+4. 如果该条差异无法归纳出审查规则，输出 {"rule_title": "", "rule_detail": ""}。
+"""
-# 加载Word文档
-doc.LoadFromFile(inputFile)
-before_text = doc.GetText()
-print("===== 修订前文本 =====")
-print(before_text)
-# 检查文档是否有未接受的修订
+def parse_args() -> argparse.Namespace:
-if doc.HasChanges:
+    parser = argparse.ArgumentParser(description="Compare Word revision text and write diff rows to Excel.")
-    # 接受所有修订
+    parser.add_argument("--input_file", default=r'/home/ccran/lufa-contract/demo/test/2、[修改版]麓谷投发天际汽车盘活项目可行性.docx',
-    doc.AcceptChanges()
+                        help="Path to the Word document with revisions.")
+    parser.add_argument(
+        "-o",
+        "--output",
+        help="Path to the output Excel file. Defaults to '<input_stem>_修订对比.xlsx'.",
+    )
+    parser.add_argument(
+        "--show-text",
+        action="store_true",
+        help="Print full text before and after accepting revisions.",
+    )
+    parser.add_argument(
+        "--no-diff-print",
+        action="store_true",
+        help="Do not print unified diff to the console.",
+    )
+    parser.add_argument(
+        "--no-rule-extract",
+        action="store_true",
+        help="Do not call LLM to extract review rules from revision diff rows.",
+    )
+    return parser.parse_args()
+def get_default_output_file(input_file: Path) -> Path:
+    return input_file.with_name(f"{input_file.stem}_修订对比.xlsx")
+def read_revision_texts(input_file: Path) -> tuple[str, str]:
+    from spire.doc import Document
+    doc = Document()
+    try:
+        doc.LoadFromFile(str(input_file))
+        before_text = doc.GetText()
+        if doc.HasChanges:
+            doc.AcceptChanges()
+        after_text = doc.GetText()
+        return before_text, after_text
+    finally:
+        doc.Close()
+def build_unified_diff(before_lines: list[str], after_lines: list[str]) -> list[str]:
+    return list(
+        difflib.unified_diff(
+            before_lines,
+            after_lines,
+            fromfile="before_revision",
+            tofile="after_revision",
+            lineterm="",
+        )
+    )
+def build_diff_rows(before_lines: list[str], after_lines: list[str]) -> list[tuple[str, str]]:
+    rows: list[tuple[str, str]] = []
+    matcher = difflib.SequenceMatcher(None, before_lines, after_lines)
+    for tag, before_start, before_end, after_start, after_end in matcher.get_opcodes():
+        if tag == "equal":
+            continue
+        before_part = before_lines[before_start:before_end]
+        after_part = after_lines[after_start:after_end]
+        if tag == "replace":
+            rows.extend(
+                (before_line or "", after_line or "")
+                for before_line, after_line in zip_longest(before_part, after_part)
+            )
+        elif tag == "delete":
+            rows.extend((before_line, "") for before_line in before_part)
+        elif tag == "insert":
+            rows.extend(("", after_line) for after_line in after_part)
+    return rows
+def write_diff_rows_to_excel(
+    diff_rows: list[tuple[str, str]],
+    output_file: Path,
+    rule_results: list[dict[str, str]] | None = None,
+) -> None:
+    wb = Workbook()
+    ws = wb.active
+    ws.title = "revision_compare"
+    ws.append(["修订前", "修订后", "审查项", "审查规则"])
+    for cell in ws[1]:
+        cell.font = Font(bold=True)
+        cell.alignment = Alignment(horizontal="center", vertical="center")
+    for index, (before_line, after_line) in enumerate(diff_rows):
+        rule = rule_results[index] if rule_results and index < len(rule_results) else {}
+        ws.append(
+            [
+                before_line,
+                after_line,
+                rule.get("rule_title", ""),
+                rule.get("rule_detail", ""),
+            ]
+        )
+    for row in ws.iter_rows(min_row=2):
+        for cell in row:
+            cell.alignment = Alignment(wrap_text=True, vertical="top")
+        if row[1].value:
+            row[1].font = Font(color="FF0000")
+    ws.column_dimensions["A"].width = 80
+    ws.column_dimensions["B"].width = 80
+    ws.column_dimensions["C"].width = 24
+    ws.column_dimensions["D"].width = 80
+    ws.freeze_panes = "A2"
+    wb.save(output_file)
+def build_rule_extract_user_prompt(before_line: str, after_line: str) -> str:
+    return (
+        "请根据以下单条合同修订前后对比文本，提取该条修订体现出的审查项和审查规则。\n"
+        "注意：修订后的文本通常代表更符合我司要求的表达，请结合修订前被替换或删除的内容归纳规则。\n\n"
+        f"{json.dumps({'修订前': before_line, '修订后': after_line}, ensure_ascii=False, indent=2)}"
+    )
+def normalize_single_rule_response(value: Any) -> dict[str, str]:
+    if isinstance(value, list):
+        rules = [item for item in value if isinstance(item, dict)]
+    elif isinstance(value, dict) and isinstance(value.get("rules"), list):
+        rules = [item for item in value["rules"] if isinstance(item, dict)]
+    elif isinstance(value, dict):
+        rules = [value]
+    else:
+        rules = []
+    normalized_rules = [
+        {
+            "rule_title": str(item.get("rule_title", "")).strip(),
+            "rule_detail": str(item.get("rule_detail", "")).strip(),
+        }
+        for item in rules
+        if str(item.get("rule_title", "")).strip() or str(item.get("rule_detail", "")).strip()
+    ]
+    if not normalized_rules:
+        return {"rule_title": "", "rule_detail": ""}
+    return {
+        "rule_title": "\n".join(rule["rule_title"] for rule in normalized_rules if rule["rule_title"]),
+        "rule_detail": "\n\n".join(rule["rule_detail"] for rule in normalized_rules if rule["rule_detail"]),
+    }
+async def extract_one_review_rule_async(
+    llm: OpenAITool,
+    before_line: str,
+    after_line: str,
+) -> dict[str, str]:
+    messages = [
+        {"role": "system", "content": RULE_EXTRACT_SYSTEM_PROMPT},
+        {"role": "user", "content": build_rule_extract_user_prompt(before_line, after_line)},
+    ]
+    response = await llm.chat(messages)
+    parsed_items = extract_json(response)
+    parsed = parsed_items[0] if parsed_items else {}
+    return normalize_single_rule_response(parsed)
+async def extract_review_rules_async(diff_rows: list[tuple[str, str]]) -> list[dict[str, str]]:
+    llm = OpenAITool(LLMConfig())
+    results: list[dict[str, str]] = []
+    for index, (before_line, after_line) in enumerate(diff_rows, start=1):
+        print(f"正在提取第 {index}/{len(diff_rows)} 条审查规则...")
+        result = await extract_one_review_rule_async(llm, before_line, after_line)
+        results.append(result)
+    return results
+def extract_review_rules(diff_rows: list[tuple[str, str]]) -> list[dict[str, str]]:
+    return asyncio.run(extract_review_rules_async(diff_rows))
+def compare_revisions(
+    input_file: Path,
+    output_file: Path,
+    show_text: bool,
+    print_diff: bool,
+    extract_rules: bool,
+) -> bool:
+    before_text, after_text = read_revision_texts(input_file)
+    if show_text:
+        print("===== 修订前文本 =====")
+        print(before_text)
+        print("===== 修订后文本 =====")
+        print(after_text)
+    if before_text == after_text:
+        print("修订前后文本一致，无差异，不写入 Excel。")
+        return False
+    before_lines = before_text.splitlines()
+    after_lines = after_text.splitlines()
+    diff_lines = build_unified_diff(before_lines, after_lines)
+    diff_rows = build_diff_rows(before_lines, after_lines)
+    if not diff_rows:
+        print("没有检测到真实差异，不写入 Excel。")
+        return False
+    if print_diff:
+        print("===== Diff 对比结果 =====")
+        print("\n".join(diff_lines))
-after_text = doc.GetText()
+    rule_results = extract_review_rules(diff_rows) if extract_rules else None
-print("===== 修订后文本 =====")
+    write_diff_rows_to_excel(diff_rows, output_file, rule_results)
-print(after_text)
+    print(f"Excel Diff 对比结果已写入: {output_file}")
+    return True
-before_lines = before_text.splitlines()
-after_lines = after_text.splitlines()
-diff_lines = list(
+def main() -> None:
-    difflib.unified_diff(
+    args = parse_args()
-        before_lines,
+    input_file = Path(args.input_file)
-        after_lines,
+    output_file = Path(args.output) if args.output else get_default_output_file(input_file)
-        fromfile="before_revision",
+    compare_revisions(
-        tofile="after_revision",
+        input_file=input_file,
-        lineterm="",
+        output_file=output_file,
+        show_text=args.show_text,
+        print_diff=not args.no_diff_print,
+        extract_rules=not args.no_rule_extract,
    )
-)
-print("===== Diff 对比结果 =====")
-if diff_lines:
-    print("\n".join(diff_lines))
-else:
-    print("修订前后文本一致，无差异。")
-doc.Close()
+if __name__ == "__main__":
+    main()
--- a/main.py
+++ b/main.py
@@ -19,6 +19,7 @@ from core.config import (
    pdf_support_formats,
    MERGE_RULE_PROMPT,
    use_lufa,
+    max_model_len
 )
 from core.tools.segment_summary import SegmentSummaryTool
 from core.tools.segment_review import SegmentReviewTool
@@ -29,6 +30,7 @@ from core.tools.reflect_retry import ReflectRetryTool
 from core.tools.segment_merger import SegmentMergerTool
 from core.tools.fact_merger import FactMergerTool
 from core.tools.ruleset_router import RulesetRouterTool
+from core.tools.party_role import PartyRoleTool
 from core.memory import Finding
 from core.memory import FINDING_KEY_MERGE, FINDING_KEY_REFLECT, FINDING_KEY_REVIEW
@@ -44,6 +46,7 @@ reflect_tool = ReflectRetryTool()
 merger_tool = SegmentMergerTool()
 fact_merger_tool = FactMergerTool()
 ruleset_router_tool = RulesetRouterTool()
+party_role_tool = PartyRoleTool()
 @app.post("/sleep")
@@ -84,6 +87,17 @@ class RulesetRouteResponse(BaseModel):
    reason: str = ""
+class PartyRoleRequest(BaseModel):
+    company_name: str = Field(..., description="待分析公司名称")
+    contract_text: str = Field(..., description="合同全文或合同片段内容")
+class PartyRoleResponse(BaseModel):
+    company_name: str
+    party_role: str
+    reason: Optional[str] = ""
 @app.post("/rulesets/route", response_model=RulesetRouteResponse)
 def route_ruleset(payload: RulesetRouteRequest) -> RulesetRouteResponse:
    question = (payload.question or "").strip()
@@ -98,6 +112,26 @@ def route_ruleset(payload: RulesetRouteRequest) -> RulesetRouteResponse:
    )
+@app.post("/contracts/party-role", response_model=PartyRoleResponse)
+def detect_party_role(payload: PartyRoleRequest) -> PartyRoleResponse:
+    company_name = (payload.company_name or "").strip()
+    contract_text = (payload.contract_text or "").strip()
+    if not company_name:
+        raise HTTPException(status_code=400, detail="company_name cannot be empty")
+    if not contract_text:
+        raise HTTPException(status_code=400, detail="contract_text cannot be empty")
+    result = party_role_tool.run(
+        company_name=company_name,
+        contract_text=contract_text[:max_model_len],  # TODO 合同文本过长时，优先保留包含公司名称的片段内容，后续优化
+    )
+    return PartyRoleResponse(
+        company_name=company_name,
+        party_role=result.get("party_role", ""),
+        reason=result.get("reason", ""),
+    )
 @app.post("/documents/parse", response_model=DocumentParseResponse)
 async def parse_document(payload: DocumentParseRequest) -> DocumentParseResponse:
    if not payload.urls:

--- a/utils/tesseract_ocr_util.py
+++ b/utils/tesseract_ocr_util.py
+import asyncio
+import os
+import re
+from urllib import parse
+from urllib.parse import urlparse
+import fitz
+from loguru import logger
+from utils.common_util import random_str
+from utils.http_util import download_file, url_replace_fastgpt
+class TesseractOCRUtil:
+    def __init__(self, lang='chi_sim+eng'):
+        self.lang = lang
+    def ocr_image(self, file_path):
+        import pytesseract
+        from PIL import Image
+        with Image.open(file_path) as img:
+            return pytesseract.image_to_string(img, lang=self.lang)
+    async def ocr_image_async(self, path_list):
+        tasks = [asyncio.to_thread(self.ocr_image, file_path) for file_path in path_list]
+        responses = await asyncio.gather(*tasks)
+        res_dict = {}
+        for file_path, content in zip(path_list, responses):
+            page_num = int(self.get_pdf_2_img_page_num(file_path))
+            res_dict[page_num] = content
+        logger.info(f'tesseract ocr_image_async finish. all pages:{len(res_dict)}')
+        return [res_dict[key] for key in sorted(res_dict)]
+    def set_pdf_2_img_page(self, path, page_idx):
+        return f'{path}_{page_idx + 1}.png'
+    def get_pdf_2_img_page_num(self, path):
+        match = re.search(r'_(\d+)\.png$', path)
+        if not match:
+            raise ValueError(f'Invalid pdf page image path: {path}')
+        return match.group(1)
+    def pdf_2_img(self, path, zoom_x=2, zoom_y=2):
+        pdf = fitz.open(path)
+        pdf_list = []
+        for pg in range(0, pdf.page_count):
+            page = pdf[pg]
+            trans = fitz.Matrix(zoom_x, zoom_y)
+            pm = page.get_pixmap(matrix=trans, alpha=False)
+            dest_png = self.set_pdf_2_img_page(path, pg)
+            pm.save(dest_png)
+            pdf_list.append(dest_png)
+        pdf.close()
+        return pdf_list
+    def ocr_download_path(self, url):
+        logger.info(f'tesseract ocr url:{url}')
+        url = url_replace_fastgpt(url)
+        url_parsed = urlparse(url)
+        query_dict = parse.parse_qs(url_parsed.query)
+        if 'filename' in query_dict:
+            filename = query_dict.get('filename')[0]
+        else:
+            filename = f'{random_str()}.pdf'
+        dest_path = f'ocr/{filename}'
+        download_file(url, dest_path)
+        return dest_path
+    async def ocr_result_pdf(self, dest_path):
+        pdf_list = self.pdf_2_img(dest_path)
+        try:
+            return await self.ocr_image_async(pdf_list)
+        finally:
+            for pdf in pdf_list:
+                if os.path.exists(pdf):
+                    os.remove(pdf)
+if __name__ == '__main__':
+    ocr_util = TesseractOCRUtil()
+    result = asyncio.run(ocr_util.ocr_result_pdf('../demo/2020100593中建大成建筑（B类）.pdf'))
+    print(f'len(result):{len(result)}')
+    print(result)