"""
合同审核规则引擎
迁移自 AuditScope 项目（D:\\Trae\\自研分析工具-财报），包含完整的 5 类合同风险检测规则。
核心功能：
1. 合同文本自动分类（销售/采购/租赁/担保/关联交易）
2. 5 类风险规则扫描（收入确认/售后回购/关联方/担保/租赁）
3. 风险聚合与总体等级评估
4. 证据溯源（命中关键词位置、上下文片段）
规则结构：
每条规则包含 keywords/reason/audit_focus/suggestion/level 五个字段，
确保每条风险检测结果都有审计依据和修改建议。
CAS 准则引用：
- CAS 14 收入准则（收入确认/售后回购）
- CAS 21 租赁准则（租赁识别）
- CAS 13 或有事项准则（担保/或有负债）
- CAS 37 金融工具列报（股权回购）
"""
import re
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Tuple, Optional
@dataclass
class ContractRiskItem:
"""
合同风险项数据结构
用于存储单条风险检测结果，包含风险类型、等级、命中条款原文、
风险原因、审计关注要点、修改建议等完整信息。
Attributes:
risk_type: 风险类型（如"收入确认风险"、"担保/或有负债风险"）
risk_level: 风险等级（"高风险"/"中风险"/"低风险"）
clause_text: 命中的条款原文上下文（用于证据溯源）
reason: 风险原因说明
audit_focus: 审计关注要点
suggestion: 修改建议
keyword: 命中的关键词
position_chars: 关键词在文本中的字符位置
position_line: 关键词在文本中的行号
"""
risk_type: str
risk_level: str
clause_text: str
reason: str
audit_focus: str
suggestion: str
keyword: str = ""
position_chars: int = -1
position_line: int = -1
@dataclass
class ContractAnalysisResult:
"""
合同分析结果数据结构
用于存储完整的合同风险分析结果，包含合同类型、总体风险等级、
所有命中的风险项、风险统计等信息。
Attributes:
contract_type: 合同类型（如"sales"/"lease"/"guarantee"等）
overall_level: 总体风险等级（"高风险"/"中风险"/"低风险"）
risks: 命中的风险项列表
risk_counts: 各级风险数量统计
"""
contract_type: str = "mixed"
overall_level: str = "低风险"
risks: List[ContractRiskItem] = field(default_factory=list)
risk_counts: Dict[str, int] = field(default_factory=lambda: {"高风险": 0, "中风险": 0, "低风险": 0})
def to_dict(self) -> dict:
"""转换为字典格式，便于 JSON 序列化"""
return {
"contract_type": self.contract_type,
"overall_level": self.overall_level,
"risks": [asdict(r) for r in self.risks],
"risk_counts": self.risk_counts,
"total_risks": len(self.risks),
}
REVENUE_RECOGNITION_RULES = [
{
"keywords": ["提前开票", "先开票后付款", "预开票", "提前开具发票"],
"reason": "可能存在收入确认时点早于控制权转移的风险，需要关注商品是否已经交付、客户是否已验收。",
"audit_focus": "检查合同约定的交付、验收、开票与付款条款；核对发货单、签收记录、客户验收单。",
"suggestion": "明确约定'客户验收合格后 X 日内开具发票'或'控制权转移后开具发票'。",
"level": "高风险",
},
{
"keywords": ["未验收即付款", "未验收付款", "无条件付款", "先付款后验收"],
"reason": "付款不依赖验收结果，可能存在控制权未转移即确认收入或预付账款无法收回的风险。",
"audit_focus": "关注付款条件是否与验收、控制权转移挂钩；检查期后验收和退货情况。",
"suggestion": "改为'验收合格后 X 日内付款'，明确验收标准和异议期。",
"level": "高风险",
},
{
"keywords": ["无条件确认收入", "一次性确认", "全额确认收入"],
"reason": "存在收入确认政策激进的风险，需要判断是否满足控制权转移、收入金额可可靠计量等条件。",
"audit_focus": "对照 CAS 14 收入准则五步法，核查合同识别、履约义务识别、交易价格分摊、控制权转移判断。",
"suggestion": "按履约义务分步确认收入，在控制权转移时点分别确认。",
"level": "中风险",
},
{
"keywords": ["控制权未转移", "未发货先确认", "未交付确认"],
"reason": "条款可能暗示在控制权转移前即确认收入，不符合 CAS 14 的规定。",
"audit_focus": "对照 CAS 14 第十三条控制权转移迹象核对：现时收款权、实物占有、法定所有权、主要风险报酬、客户已接受。",
"suggestion": "修改为控制权转移后确认收入，明确转移时点的判断标准。",
"level": "高风险",
},
{
"keywords": ["完工百分比", "进度确认", "按进度开票"],
"reason": "若采用投入法确认收入，需要确认履约进度能否可靠计量，是否符合投入与产出匹配原则。",
"audit_focus": "检查投入成本与履约进度是否合理，是否存在成本归集不准确或进度虚估。",
"suggestion": "优先选用产出法（如已转移产品/服务的价值），投入法需披露估算方法。",
"level": "中风险",
},
]
BUYBACK_RETURN_RULES = [
{
"keywords": ["无理由退货", "无条件退货", "可随时退货", "随时可退"],
"reason": "存在重大退货权，可能导致收入金额无法可靠计量，需按预期退回金额确认退款负债。",
"audit_focus": "估计退货率、核对历史退货记录、检查退款负债的计提是否充分（CAS 14 第十六条）。",
"suggestion": "约定合理的退货期和退货条件，明确超过退货期不得退回；或约定按实际销量结算。",
"level": "高风险",
},
{
"keywords": ["售后回购", "供应商回购", "保底回购", "到期回购", "强制回购"],
"reason": "售后回购可能实质为融资安排（CAS 14 第十三条），商品控制权未转移，不应确认收入。",
"audit_focus": "判断回购价是否等于原价加合理利息；如是，按融资处理；如回购价低于售价，可能为附有退货权的销售。",
"suggestion": "避免约定强制回购，改为协商回购或不约定回购条款。",
"level": "高风险",
},
{
"keywords": ["客户可退货", "未售出可退", "滞销可退", "卖不完退回"],
"reason": "寄售/保底销售模式下，控制权可能未转移，收入确认时点和金额存在风险。",
"audit_focus": "核对发货单和客户实际领用/销售记录，关注是否存在滞销退回。",
"suggestion": "改为'买断式销售'或在合同中明确控制权转移的条件不包含未售出退回。",
"level": "中风险",
},
{
"keywords": ["可变对价", "浮动价格", "价格保护", "返利", "销售返点"],
"reason": "可变对价的估计涉及重大判断，可能影响收入确认金额的可靠性。",
"audit_focus": "检查可变对价的估计方法、是否满足'极可能不会发生重大转回'的限制条件（CAS 14 第十六条）。",
"suggestion": "尽量明确固定价格；若有返利，约定合理的计算方法并披露估计的不确定性。",
"level": "中风险",
},
]
RELATED_PARTY_RULES = [
{
"keywords": ["关联方", "关联企业", "关联公司", "受同一控制", "同受控制"],
"reason": "关联方交易需要额外关注商业实质、价格公允性和披露完整性。",
"audit_focus": "获取关联方清单、核查交易价格是否公允（与独立第三方比较）、检查是否履行了必要审批程序。",
"suggestion": "按非关联方同等条件定价，保留定价依据，在年报中完整披露关联方交易。",
"level": "中风险",
},
{
"keywords": ["实际控制人", "大股东指定", "实控人指定", "控股股东指定"],
"reason": "涉及实控人的条款需要特别关注是否构成关联方资金占用或利益输送。",
"audit_focus": "核查交易对手是否为实控人控制/关联企业、检查资金流向是否闭环、关注商业实质。",
"suggestion": "避免直接与实控人个人发生大额交易；实控人控制的企业按关联交易履行审批和披露程序。",
"level": "高风险",
},
{
"keywords": ["指定账户", "指定收款", "指定付款账户", "代收代付", "受托支付"],
"reason": "收付款账户与签约主体不一致可能隐含资金占用、体外循环或虚假交易。",
"audit_focus": "核对银行流水、确认三方关系真实性、判断是否存在资金最终流向实控人或其关联方。",
"suggestion": "严格约定签约主体双方直接收付款，确需指定第三方账户的需补充三方协议。",
"level": "高风险",
},
{
"keywords": ["代垫款", "代垫费用", "代垫工资", "代付费用"],
"reason": "代垫款长期挂账可能构成关联方非经营性资金占用（监管关注事项）。",
"audit_focus": "检查其他应收款/其他应付款中长期代垫款的余额、账龄和回收计划。",
"suggestion": "及时结算代垫款，避免大额长期挂账；确需保留的按规定收取合理资金占用费。",
"level": "高风险",
},
{
"keywords": ["资金拆借", "无息借款", "无偿借款", "往来款", "借款不付息"],
"reason": "关联方无息拆借属于非公允关联交易，可能构成资金占用或利益输送。",
"audit_focus": "检查借款协议、是否有归还计划、是否计提合理利息、关注借款是否用于对方经营。",
"suggestion": "按市场利率收取利息，明确还款期限和方式；或按实际需要改为增资/减资。",
"level": "高风险",
},
]
GUARANTEE_RULES = [
{
"keywords": ["连带责任保证", "连带保证", "承担连带责任", "无限连带"],
"reason": "连带责任保证使企业承担重大或有负债风险，需评估被担保方履约能力和自身最大风险敞口。",
"audit_focus": "检查董事会/股东会决议、评估被担保方财务状况、测算最大风险敞口、判断是否满足预计负债确认条件。",
"suggestion": "改为一般保证；或约定最高担保额度；对外担保严格履行内部审批和披露程序。",
"level": "高风险",
},
{
"keywords": ["差额补足", "承担差额", "不足部分补足", "托底", "兜底"],
"reason": "差额补足义务可能构成金融负债或预计负债，需根据协议条款判断会计处理。",
"audit_focus": "按《监管规则适用指引——会计类第1号》区分补偿协议是否构成'向投资方提供保本保收益安排'。",
"suggestion": "尽量避免明股实债式的差额补足；确需约定的按准则要求进行会计处理和披露。",
"level": "高风险",
},
{
"keywords": ["回购承诺", "承诺回购", "投资方回购", "股权回购", "份额回购"],
"reason": "股权回购承诺可能构成金融负债（CAS 37），尤其是按固定价格或可确定价格回购。",
"audit_focus": "检查回购触发条件、回购价格是否固定/可确定、判断是否应确认为金融负债。",
"suggestion": "避免固定价格回购承诺；改为按公允价值协商回购或不约定回购。",
"level": "高风险",
},
{
"keywords": ["补偿义务", "补偿责任", "保证金不足补足", "兜底条款"],
"reason": "补偿/兜底条款可能形成重大或有负债，需要评估发生可能性和金额估计。",
"audit_focus": "按 CAS 13 或有事项准则判断：义务现时性、经济利益流出可能性、金额能否可靠估计。",
"suggestion": "尽量避免无条件兜底；明确触发条件和补偿上限；按准则计提预计负债或予以披露。",
"level": "中风险",
},
]
LEASE_RULES = [
{
"keywords": ["指定资产", "指定场地", "指定房屋", "指定设备", "指定标的"],
"reason": "合同中存在指定资产可能表明包含租赁（CAS 21），需要进一步判断资产是否可替换。",
"audit_focus": "对照 CAS 21 第四条判断三要素：是否存在已识别资产、是否转移控制权的几乎所有经济利益、客户是否主导使用。",
"suggestion": "按租赁和非租赁部分拆分合同，分别进行会计处理；无法拆分的按租赁整体处理。",
"level": "中风险",
},
{
"keywords": ["独占使用", "独占使用权", "排他使用", "排他独占", "排他性"],
"reason": "独占使用表明客户有权获得使用资产的几乎所有经济利益，属于租赁的核心特征之一。",
"audit_focus": "结合是否存在已识别资产、客户是否主导使用三要素综合判断。",
"suggestion": "若构成租赁，按 CAS 21 确认使用权资产和租赁负债；短期租赁（<1年）可简化处理。",
"level": "中风险",
},
{
"keywords": ["不可替换", "不可替代", "无法替换", "唯一标的", "专门为我使用"],
"reason": "资产不可替换 = 存在已识别资产（CAS 21），合同很可能包含租赁。",
"audit_focus": "关注资产性质、合同是否约定供应商替换权（实质可行使则不算已识别资产）。",
"suggestion": "拆分租赁和服务部分；或通过增加供应商替换权安排，使整体不构成租赁。",
"level": "中风险",
},
{
"keywords": ["长期使用", "长期租赁", "租期", "续租权", "优先承租权", "固定期限使用"],
"reason": "长期固定期限使用资产需要判断是否构成融资租赁或经营租赁（CAS 21 新旧准则均要求列示使用权资产）。",
"audit_focus": "核对租期占剩余使用寿命的比例、付款额现值占公允价值的比例、有无购买选择权。",
"suggestion": "按 CAS 21 确认使用权资产和租赁负债；短期租赁（≤12 个月）和低价值租赁可豁免。",
"level": "中风险",
},
]
ALL_RULES: List[Tuple[str, List[Dict]]] = [
("收入确认风险", REVENUE_RECOGNITION_RULES),
("售后回购/退货风险", BUYBACK_RETURN_RULES),
("关联方/资金占用风险", RELATED_PARTY_RULES),
("担保/或有负债风险", GUARANTEE_RULES),
("租赁识别风险", LEASE_RULES),
]
CONTRACT_TYPES = {
"sales": "销售合同",
"purchase": "采购合同",
"lease": "租赁合同",
"guarantee": "担保/回购合同",
"related_party": "关联交易合同",
"mixed": "混合/通用合同",
}
RISK_LEVEL_COLORS = {
"高风险": "#FF4B4B",
"中风险": "#FFA500",
"低风险": "#4CAF50",
}
def detect_contract_type(text: str) -> str:
"""
根据文本内容粗判合同类型
通过关键词匹配的方式，对合同文本进行初步分类。
计算每种合同类型的匹配得分，返回得分最高的类型。
Args:
text: 合同文本内容
Returns:
合同类型标识（sales/purchase/lease/guarantee/related_party/mixed）
当无法识别时返回 "mixed"
"""
if not text or len(text.strip()) < 20:
return "mixed"
indicators = {
"sales": ["销售", "采购", "买方", "卖方", "采购方", "供应方", "购销", "订货", "发货"],
"lease": ["租赁", "出租", "承租", "租期", "租金", "房屋租赁", "场地租赁", "设备租赁"],
"guarantee": ["担保", "保证", "保函", "差额补足", "回购承诺", "连带责任"],
"related_party": ["关联方", "实际控制人", "代垫", "资金拆借", "指定账户", "往来款"],
"purchase": ["采购", "供应商", "向甲方采购", "采购价款", "交付货物"],
}
scores: Dict[str, int] = {}
for ctype, words in indicators.items():
scores[ctype] = sum(1 for w in words if w in text)
if not any(scores.values()):
return "mixed"
best = max(scores, key=lambda k: scores[k])
return best if scores[best] > 0 else "mixed"
def _find_clause_context(text: str, keyword: str, window: int = 60) -> str:
"""
在文本中找出包含关键词的短语上下文
用于证据溯源功能，命中关键词时截取前后 window 个字符作为上下文片段。
同时进行文本清洗：去除换行符、压缩多余空白。
Args:
text: 完整文本
keyword: 命中的关键词
window: 上下文窗口大小（字符数）
Returns:
清洗后的上下文片段
"""
idx = text.find(keyword)
if idx < 0:
return keyword
start = max(0, idx - window)
end = min(len(text), idx + len(keyword) + window)
snippet = text[start:end].replace("\n", " ").replace("\r", " ").strip()
snippet = re.sub(r"\s+", " ", snippet)
return snippet
def scan_contract_risks(text: str, max_per_type: int = 3) -> List[ContractRiskItem]:
"""
对合同文本执行全部规则扫描，返回命中的风险条款列表
扫描策略：
1. 遍历所有规则类型和规则条目
2. 每条规则匹配第一个未被其他规则使用的关键词
3. 同一规则只生成一条记录（取首次命中位置）
4. 同一风险类型最多保留 max_per_type 条（按风险等级排序）
去重策略：
- 同一关键词全局只命中一次
- 同一规则多个关键词命中时合并为一条
- 同一风险类型最多保留 max_per_type 条，高风险优先
Args:
text: 合同文本内容
max_per_type: 每个风险类型最多保留的条数
Returns:
命中的风险项列表，已按风险类型分组和严重度排序
"""
raw: List[ContractRiskItem] = []
seen_keywords = set()
for risk_type, rules in ALL_RULES:
for rule in rules:
matched_kw = None
for kw in rule["keywords"]:
if kw in text and kw not in seen_keywords:
if matched_kw is None:
matched_kw = kw
seen_keywords.add(kw)
if matched_kw:
context = _find_clause_context(text, matched_kw)
pos_chars = text.find(matched_kw)
pos_line = text[:pos_chars].count("\n") + 1 if pos_chars >= 0 else -1
raw.append(ContractRiskItem(
risk_type=risk_type,
risk_level=rule["level"],
clause_text=context,
reason=rule["reason"],
audit_focus=rule["audit_focus"],
suggestion=rule["suggestion"],
keyword=matched_kw,
position_chars=pos_chars,
position_line=pos_line,
))
grouped: Dict[str, List[ContractRiskItem]] = {}
for r in raw:
grouped.setdefault(r.risk_type, []).append(r)
results: List[ContractRiskItem] = []
severity_weight = {"高风险": 3, "中风险": 2, "低风险": 1}
for risk_type, items in grouped.items():
items.sort(key=lambda x: -severity_weight.get(x.risk_level, 0))
results.extend(items[:max_per_type])
return results
def aggregate_overall_level(risks: List[ContractRiskItem]) -> str:
"""
根据命中风险综合评估总体风险等级
评估规则：
- 无风险 → "低风险"
- 高风险 >= 2 条，或高风险 >= 1 条且总风险 >= 3 条 → "高风险"
- 高风险 >= 1 条，或中风险 >= 2 条 → "中风险"
- 其他情况 → "低风险"
Args:
risks: 命中的风险项列表
Returns:
总体风险等级（"高风险"/"中风险"/"低风险"）
"""
if not risks:
return "低风险"
high = sum(1 for r in risks if r.risk_level == "高风险")
medium = sum(1 for r in risks if r.risk_level == "中风险")
total = len(risks)
if high >= 2 or (high >= 1 and total >= 3):
return "高风险"
if high >= 1 or medium >= 2:
return "中风险"
return "低风险"
def analyze_contract(contract_text: str) -> ContractAnalysisResult:
"""
分析合同文本，返回完整的风险检测结果
这是合同规则引擎的主入口函数，整合了合同分类、
风险扫描、等级评估三个步骤。
Args:
contract_text: 合同文本内容
Returns:
ContractAnalysisResult 对象，包含：
- contract_type: 合同类型
- overall_level: 总体风险等级
- risks: 命中的风险项列表
- risk_counts: 各级风险数量统计
"""
contract_type = detect_contract_type(contract_text)
risks = scan_contract_risks(contract_text)
overall_level = aggregate_overall_level(risks)
risk_counts = {"高风险": 0, "中风险": 0, "低风险": 0}
for r in risks:
risk_counts[r.risk_level] = risk_counts.get(r.risk_level, 0) + 1
return ContractAnalysisResult(
contract_type=contract_type,
overall_level=overall_level,
risks=risks,
risk_counts=risk_counts,
)