2026 技术观察：智能合同履约进入自动跟踪阶段，节点预警、付款校验和风险识别成为新核心

import json
from datetime import datetime
from collections import defaultdict


CONTRACTS = [
    {
        "contract_id": "CT001",
        "name": "办公设备采购合同",
        "supplier": "华东办公科技",
        "amount": 120000,
        "start_date": "2026-06-01",
        "end_date": "2026-09-30",
        "status": "active"
    },
    {
        "contract_id": "CT002",
        "name": "系统运维服务合同",
        "supplier": "星河云服务",
        "amount": 300000,
        "start_date": "2026-01-01",
        "end_date": "2026-12-31",
        "status": "active"
    },
    {
        "contract_id": "CT003",
        "name": "市场活动执行合同",
        "supplier": "蓝海传媒",
        "amount": 80000,
        "start_date": "2026-05-01",
        "end_date": "2026-07-20",
        "status": "active"
    }
]


MILESTONES = [
    {
        "contract_id": "CT001",
        "milestone_id": "M001",
        "type": "delivery",
        "due_date": "2026-07-15",
        "status": "completed"
    },
    {
        "contract_id": "CT001",
        "milestone_id": "M002",
        "type": "payment",
        "due_date": "2026-07-30",
        "status": "pending"
    },
    {
        "contract_id": "CT003",
        "milestone_id": "M003",
        "type": "acceptance",
        "due_date": "2026-07-05",
        "status": "pending"
    }
]

def parse_date(date_text):
    return datetime.strptime(date_text, "%Y-%m-%d")


def check_contract_expiry(contract, current_date):
    end_date = parse_date(contract["end_date"])
    days_left = (end_date - current_date).days

    if days_left < 0:
        level = "expired"
        message = "合同已到期。"
    elif days_left <= 15:
        level = "high"
        message = "合同即将到期，建议确认续约或关闭。"
    elif days_left <= 30:
        level = "medium"
        message = "合同将在 30 天内到期，建议提前跟进。"
    else:
        level = "normal"
        message = "合同期限状态正常。"

    return {
        "contract_id": contract["contract_id"],
        "days_left": days_left,
        "expiry_level": level,
        "message": 30523.t.kuaisou.com 
    }

def check_milestone_risk(milestone, current_date):
    due_date = parse_date(milestone["due_date"])
    days_left = (due_date - current_date).days

    if milestone["status"] == "completed":
        return {
            "milestone_id": milestone["milestone_id"],
            "contract_id": milestone["contract_id"],
            "risk_level": "normal",
            "days_left": days_left,
            "message": "履约节点已完成。"
        }

    if days_left < 0:
        level = "high"
        message = "履约节点已逾期。"
    elif days_left <= 5:
        level = "medium"
        message = "履约节点即将到期。"
    else:
        level = "normal"
        message = "履约节点状态正常。"

    return {
        "milestone_id": milestone["milestone_id"],
        "contract_id": milestone["contract_id"],
        "type": milestone["type"],
        "risk_level": level,
        "days_left": days_left,
        "message": message
    }

INVOICES = [
    {
        "invoice_id": "INV001",
        "contract_id": "CT001",
        "amount": 60000
    },
    {
        "invoice_id": "INV002",
        "contract_id": "CT001",
        "amount": 70000
    },
    {
        "invoice_id": "INV003",
        "contract_id": "CT002",
        "amount": 100000
    }
]


def check_invoice_amount(contract, invoices):
    related = [
        item for item in invoices
        if item["contract_id"] == contract["contract_id"]
    ]

    total_invoice_amount = sum(
        item["amount"]
        for item in related
    )

    if total_invoice_amount > contract["amount"]:
        level = "high"
        message = "累计发票金额超过合同金额。"
    elif total_invoice_amount > contract["amount"] * 0.9:
        level = "medium"
        message = "累计发票金额接近合同金额上限。"
    else:
        level = "normal"
        message = "发票金额处于合同范围内。"

    return {
        "contract_id": contract["contract_id"],
        "contract_amount": contract["amount"],
        "invoice_amount": total_invoice_amount,
        "risk_level": 30503.t.kuaisou.com 
        "message": message
    }

SUPPLIER_HISTORY = {
    "华东办公科技": {
        "delay_count": 1,
        "quality_issue_count": 0
    },
    "星河云服务": {
        "delay_count": 0,
        "quality_issue_count": 1
    },
    "蓝海传媒": {
        "delay_count": 3,
        "quality_issue_count": 2
    }
}


def evaluate_supplier_risk(contract):
    history = SUPPLIER_HISTORY.get(
        contract["supplier"],
        {
            "delay_count": 0,
            "quality_issue_count": 0
        }
    )

    score = history["delay_count"] * 2
    score += history["quality_issue_count"] * 3

    if score >= 7:
        level = "high"
    elif score >= 3:
        level = "medium"
    elif score > 0:
        level = "low"
    else:
        level = "normal"

    return {
        "contract_id": contract["contract_id"],
        "supplier": contract["supplier"],
        "supplier_risk_score": score,
        "supplier_risk_level": level,
        "history": 31226.t.kuaisou.com
    }

def generate_contract_suggestions(expiry, milestone_risks, invoice_risk, supplier_risk):
    suggestions = []

    if expiry["expiry_level"] in ["expired", "high", "medium"]:
        suggestions.append("建议业务负责人确认合同续约、关闭或补充协议。")

    for item in milestone_risks:
        if item["risk_level"] == "high":
            suggestions.append(f"履约节点 {item['milestone_id']} 已逾期，建议立即跟进。")
        elif item["risk_level"] == "medium":
            suggestions.append(f"履约节点 {item['milestone_id']} 即将到期，建议提前提醒责任人。")

    if invoice_risk["risk_level"] in ["high", "medium"]:
        suggestions.append(invoice_risk["message"])

    if supplier_risk["supplier_risk_level"] in ["high", "medium"]:
        suggestions.append("供应商历史履约风险较高，建议加强验收和付款审核。")

    if not suggestions:
        suggestions.append("当前合同履约状态正常，建议持续监控。")

    return suggestions

def run_contract_performance_monitor():
    current_date = datetime(2026, 7, 9)

    results = []

    for contract in CONTRACTS:
        expiry = check_contract_expiry(
            contract,
            current_date
        )

        related_milestones = [
            item for item in MILESTONES
            if item["contract_id"] == contract["contract_id"]
        ]

        milestone_risks = [
            check_milestone_risk(item, current_date)
            for item in related_milestones
        ]

        invoice_risk = check_invoice_amount(
            contract,
            INVOICES
        )

        supplier_risk = evaluate_supplier_risk(
            contract
        )

        suggestions = generate_contract_suggestions(
            expiry,
            milestone_risks,
            invoice_risk,
            supplier_risk
        )

        results.append({
            "contract": contract,
            "expiry": expiry,
            "milestone_risks": milestone_risks,
            "invoice_risk": invoice_risk,
            "supplier_risk": supplier_risk,
            "suggestions": 31227.t.kuaisou.com 
        })

    report = {
        "report_name": "智能合同履约跟踪报告",
        "contract_count": len(CONTRACTS),
        "results": results,
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_contract_performance_monitor()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))