金融政务场景私有化AI+RPA实战：全离线内网部署架构、EXE授权交付与审计日志闭环设计

import requests
import urllib3
from datetime import datetime

urllib3.disable_warnings()

# 内网业务系统调用示例：双向证书认证
def call_core_api(endpoint, payload, cert_path, key_path, ca_path):
    try:
        response = requests.post(
            url=endpoint,
            json=payload,
            cert=(cert_path, key_path),  # 客户端证书
            verify=ca_path,               # CA根证书校验
            timeout=30
        )
        # 审计日志：记录每一次API调用
        audit_log({
            "action": "api_call",
            "endpoint": endpoint,
            "status_code": response.status_code,
            "timestamp": datetime.now().isoformat()
        })
        return response.json()
    except Exception as e:
        audit_log({
            "action": "api_call_failed",
            "endpoint": endpoint,
            "error": str(e),
            "timestamp": datetime.now().isoformat()
        })
        raise

import hashlib
import json
import os
import shutil
from datetime import datetime

AUDIT_LOG_PATH = r"D:\rpa_audit\audit.log"
READONLY_SERVER = r"\\audit-server\rpa-logs"  # 只读日志服务器

def audit_log(record: dict):
    """
    审计日志写入：先算哈希，再写本地，最后同步只读服务器
    """
    record["log_id"] = f"RPA-{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
    record_str = json.dumps(record, ensure_ascii=False, sort_keys=True)
    
    # 计算当前记录哈希
    record_hash = hashlib.sha256(record_str.encode()).hexdigest()
    record["hash"] = record_hash
    
    # WAL机制：先写本地
    with open(AUDIT_LOG_PATH, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")
        f.flush()
        os.fsync(f.fileno())  # 强制刷盘
    
    # 同步到只读日志服务器（防篡改）
    try:
        sync_to_readonly_server(record)
    except Exception as e:
        # 同步失败也要记录，但不能阻断主流程
        fallback_record = {
            "level": "system",
            "action": "sync_failed",
            "error": str(e),
            "record_id": record["log_id"],
            "timestamp": datetime.now().isoformat()
        }
        with open(AUDIT_LOG_PATH, "a", encoding="utf-8") as f:
            f.write(json.dumps(fallback_record, ensure_ascii=False) + "\n")

def sync_to_readonly_server(record):
    """同步到只读服务器，实际生产环境用SFTP或共享目录"""
    # UNC路径直接用字符串拼接，避免os.path.join处理异常
    target = f"{READONLY_SERVER}\\{record['log_id']}.json"
    os.makedirs(READONLY_SERVER, exist_ok=True)
    with open(target, "w", encoding="utf-8") as f:
        json.dump(record, f, ensure_ascii=False)

# 使用示例：记录一次流程启动
audit_log({
    "level": "operation",
    "flow_name": "核心系统对账流程",
    "trigger": "api",  # api / manual / schedule
    "operator": "system",
    "business_no": "20250820001",
    "action": "flow_start",
    "result": "success"
})

from datetime import datetime

class ElementHealer:
    """
    Web元素AI自愈：当原始定位失效时，自动尝试修复
    """
    def __init__(self, original_xpath, ai_model):
        self.original_xpath = original_xpath
        self.ai_model = ai_model
        self.heal_count = 0
    
    def find_element(self, driver):
        try:
            return driver.find_element("xpath", self.original_xpath)
        except Exception:
            # 原始路径失效，触发AI自愈
            self.heal_count += 1
            screenshot = driver.get_screenshot_as_base64()
            
            # AI根据当前页面截图和原始意图，重新生成XPath
            new_xpath = self.ai_model.heal_xpath(
                screenshot=screenshot,
                original_intent="提交按钮",
                original_xpath=self.original_xpath
            )
            
            # 记录自愈事件到审计日志
            audit_log({
                "level": "system",
                "action": "element_healed",
                "original_xpath": self.original_xpath,
                "new_xpath": new_xpath,
                "heal_count": self.heal_count,
                "timestamp": datetime.now().isoformat()
            })
            
            self.original_xpath = new_xpath
            return driver.find_element("xpath", new_xpath)