影刀RPA 流程审计日志：谁在什么时候做了什么
作者：林焱
什么情况用
流程跑着跑着出了问题，你要排查，发现——不知道哪一步出的错、不知道什么时间跑的、不知道输入是什么。只能从头到尾加断点重跑一遍，整个过程像在抓瞎。
稍微正规一点的RPA实践都会要求：每个流程必须有运行日志，出了问题能在5分钟内定位。更进一步的需求是审计——老板想知道「这个自动发薪流程，上个月到底改了谁的工资？谁改的？」
核心场景：需要追踪流程的完整运行轨迹，支持问题排查和合规审计。
怎么做
第一步：设计日志的四个层次
拼多多店群自动化上架方案
不是所有的日志都一样重要。分层设计，方便按需查看。
┌─────────────────────────────────────────┐ │ Level 1: 运行摘要 │ │ 每次执行一条记录：时间、流程名、结果、耗时 │ ├─────────────────────────────────────────┤ │ Level 2: 步骤追踪 │ │ 流程中每个关键步骤的输入输出 │ ├─────────────────────────────────────────┤ │ Level 3: 详细调试 │ │ 变量值、中间计算结果（仅调试时开启） │ ├─────────────────────────────────────────┤ │ Level 4: 变更审计（敏感流程专用） │ │ 谁、何时、改了哪个字段、改前值、改后值 │ └─────────────────────────────────────────┘第二步：构建日志记录器
importjsonimportsqlite3importosfromdatetimeimportdatetimefromthreadingimportLockclassAuditLogger:""" 影刀流程审计日志系统 基于SQLite，轻量无需额外部署 """def__init__(self,db_path="audit_log.db"):self.db_path=db_path self.lock=Lock()self._init_db()def_init_db(self):"""初始化数据库表"""withsqlite3.connect(self.db_path)asconn:conn.executescript(""" -- 运行摘要表 CREATE TABLE IF NOT EXISTS run_summary ( id INTEGER PRIMARY KEY AUTOINCREMENT, flow_name TEXT NOT NULL, start_time TEXT NOT NULL, end_time TEXT, status TEXT DEFAULT 'running', -- running/success/failed error_message TEXT, duration_seconds REAL, triggered_by TEXT DEFAULT 'auto', input_params TEXT, output_summary TEXT, created_at TEXT DEFAULT (datetime('now', 'localtime')) ); -- 步骤追踪表 CREATE TABLE IF NOT EXISTS step_trace ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL, step_name TEXT NOT NULL, step_type TEXT, -- action/decision/python_node/wait input_data TEXT, output_data TEXT, status TEXT DEFAULT 'success', error_detail TEXT, execution_time REAL, timestamp TEXT DEFAULT (datetime('now', 'localtime')), FOREIGN KEY (run_id) REFERENCES run_summary(id) ); -- 变更审计表（敏感操作专用） CREATE TABLE IF NOT EXISTS change_audit ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL, target_table TEXT, -- 被修改的表/文件 target_field TEXT, -- 被修改的字段 target_record_id TEXT, -- 被修改的记录标识 old_value TEXT, new_value TEXT, operator TEXT DEFAULT 'RPA', change_reason TEXT, timestamp TEXT DEFAULT (datetime('now', 'localtime')), FOREIGN KEY (run_id) REFERENCES run_summary(id) ); -- 创建索引提升查询速度 CREATE INDEX IF NOT EXISTS idx_run_flow ON run_summary(flow_name); CREATE INDEX IF NOT EXISTS idx_run_time ON run_summary(start_time); CREATE INDEX IF NOT EXISTS idx_step_run ON step_trace(run_id); CREATE INDEX IF NOT EXISTS idx_change_run ON change_audit(run_id); """)defstart_run(self,flow_name,input_params=None):"""开始一次流程运行，返回run_id"""withself.lock:withsqlite3.connect(self.db_path)asconn:cursor=conn.execute("""INSERT INTO run_summary (flow_name, start_time, input_params) VALUES (?, ?, ?)""",(flow_name,datetime.now().isoformat(),json.dumps(input_params,ensure_ascii=False)))returncursor.lastrowiddefend_run(self,run_id,status="success",error_message=None,output_summary=None):"""结束一次运行，记录结果"""withself.lock:withsqlite3.connect(self.db_path)asconn:# 计算运行时长row=conn.execute("SELECT start_time FROM run_summary WHERE id=?",(run_id,)).fetchone()ifrow:start=datetime.fromisoformat(row[0])duration=(datetime.now()-start).total_seconds()else:duration=0conn.execute("""UPDATE run_summary SET end_time=?, status=?, error_message=?, duration_seconds=?, output_summary=? WHERE id=?""",(datetime.now().isoformat(),status,error_message,duration,json.dumps(output_summary,ensure_ascii=False),run_id))deflog_step(self,run_id,step_name,step_type="action",input_data=None,output_data=None,status="success",error_detail=None,execution_time=None):"""记录单个步骤"""withself.lock:withsqlite3.connect(self.db_path)asconn:conn.execute("""INSERT INTO step_trace (run_id, step_name, step_type, input_data, output_data, status, error_detail, execution_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",(run_id,step_name,step_type,json.dumps(input_data,ensure_ascii=False)ifinput_dataelseNone,json.dumps(output_data,ensure_ascii=False)ifoutput_dataelseNone,status,error_detail,execution_time))deflog_change(self,run_id,target_table,target_field,target_record_id,old_value,new_value,change_reason=""):"""记录数据变更（审计专用）"""withself.lock:withsqlite3.connect(self.db_path)asconn:conn.execute("""INSERT INTO change_audit (run_id, target_table, target_field, target_record_id, old_value, new_value, change_reason) VALUES (?, ?, ?, ?, ?, ?, ?)""",(run_id,target_table,target_field,target_record_id,str(old_value),str(new_value),change_reason))defquery_runs(self,flow_name=None,status=None,days=7):"""查询最近的运行记录"""withsqlite3.connect(self.db_path)asconn:conn.row_factory=sqlite3.Row query="SELECT * FROM run_summary WHERE 1=1"params=[]ifflow_name:query+=" AND flow_name = ?"params.append(flow_name)ifstatus:query+=" AND status = ?"params.append(status)query+=" AND created_at >= datetime('now', 'localtime', ?)"params.append(f'-{days}days')query+=" ORDER BY created_at DESC LIMIT 100"cursor=conn.execute(query,params)return[dict(row)forrowincursor.fetchall()]defquery_change_audit(self,target_table=None,days=7):"""查询变更审计记录"""withsqlite3.connect(self.db_path)asconn:conn.row_factory=sqlite3.Row query="SELECT * FROM change_audit WHERE 1=1"params=[]iftarget_table:query+=" AND target_table = ?"params.append(target_table)query+=" AND timestamp >= datetime('now', 'localtime', ?)"params.append(f'-{days}days')query+=" ORDER BY timestamp DESC LIMIT 500"cursor=conn.execute(query,params)return[dict(row)forrowincursor.fetchall()]第三步：在影刀流程中使用
Python节点用法——在影刀流程的关键节点插入日志记录：
# ===== 流程入口 Python节点 =====importjson# 初始化日志器（全局）logger=AuditLogger("D:/RPA_Logs/audit_log.db")# 开始本次运行run_id=logger.start_run(flow_name="每日销售报表生成",input_params={"report_date":report_date,"data_source":data_source})# 把run_id存到影刀变量中，后续节点都需要它print(run_id)# ===== 每个关键步骤后 Python节点 =====logger.log_step(run_id=run_id,step_name="从ERP导出销售数据",step_type="action",output_data={"row_count":1520,"file":"sales_20240626.csv"},execution_time=3.2)# ===== 有数据修改时 =====logger.log_change(run_id=run_id,target_table="员工薪资表",target_field="基本工资",target_record_id="EMP-0042",old_value=15000,new_value=16500,change_reason="年度调薪，涨幅10%")# ===== 流程结束 Python节点 =====logger.end_run(run_id=run_id,status="success"ifnothas_errorelse"failed",error_message=error_msgifhas_errorelseNone,output_summary={"reports_generated":3,"emails_sent":5})第四步：日志查询工具
classLogReporter:"""日志报表生成"""def__init__(self,logger):self.logger=loggerdefdaily_report(self,date_str=None):"""生成每日运行报告"""ifdate_strisNone:date_str=datetime.now().strftime("%Y-%m-%d")runs=self.logger.query_runs(days=1)total=len(runs)success=sum(1forrinrunsifr["status"]=="success")failed=sum(1forrinrunsifr["status"]=="failed")total_duration=sum(r["duration_seconds"]or0forrinruns)report=f""" ===== RPA运行日报{date_str}===== 运行概览： 总执行次数：{total}成功：{success}（{success/total*100:.1f}%） 失败：{failed}（{failed/total*100:.1f}%）  总耗时：{total_duration/60:.1f}分钟 失败明细： """forrinruns:ifr["status"]=="failed":report+=f" - [{r['flow_name']}]{r['error_message']}\n"returnreportdefweekly_audit_report(self):"""每周审计报告——变更记录汇总"""changes=self.logger.query_change_audit(days=7)# 统计变更最多的表和字段fromcollectionsimportCounter table_stats=Counter(c["target_table"]forcinchanges)report="\n===== 审计周报 =====\n\n变更统计：\n"fortable,countintable_stats.most_common():report+=f"{table}:{count}次变更\n"returnreport有什么坑
坑1：SQLite的并发写入问题
如果你的流程有多线程（多个子流程同时运行），同时写SQLite可能触发database is locked错误。
踩坑实录：做了一个多线程爬虫流程，5个线程同时往同一个SQLite写日志，十个里面有七个报database is locked。
解决方法：用threading.Lock保护写操作，或者换成SQLite的WAL模式（Write-Ahead Logging）。
conn.execute("PRAGMA journal_mode=WAL;")坑2：日志文件无限制增长
SQLite文件不会自动清理，一个流程每天产生几千条日志，三个月后审计库就几百MB了。
解决方法：加一个定时清理任务——保留最近30天的日志，更早的归档到文件或直接删除。
TEMU店群如何管理运营？
defcleanup_old_logs(db_path,keep_days=30):"""清理旧日志"""withsqlite3.connect(db_path)asconn:conn.execute("DELETE FROM step_trace WHERE run_id IN ""(SELECT id FROM run_summary WHERE created_at < datetime('now', 'localtime', ?))",(f'-{keep_days}days',))conn.execute("DELETE FROM run_summary WHERE created_at < datetime('now', 'localtime', ?)",(f'-{keep_days}days',))conn.execute("VACUUM;")# 回收空间坑3：日志里忘了脱敏
审计日志记录了输入参数、输出数据、变更值——如果不脱敏，你的日志文件就变成了敏感数据的聚集地。
解决方法：在log_step和log_change里内置脱敏逻辑，对手机号、身份证等敏感字段自动遮盖。
坑4：影刀Python节点的变量作用域
同一个流程里，第一个Python节点定义的logger对象，第二个Python节点是拿不到的。每个Python节点是独立进程。
解决方法：把logger初始化逻辑放在每个需要日志的Python节点里，用文件路径来保持一致性。或者把核心状态（如run_id）存到影刀的全局变量中。
坑5：审计日志的可视化
老板说「把上个月的审计报告给我」，你总不能给老板一个SQLite文件。
解决方法：写个脚本定期导出change_audit表到Excel，生成变更汇总报表。或者用影刀的Excel节点自动生成。
总结：审计日志不是可有可无的装饰，而是流程能上线的基本要求。关键三件事：记录每次运行的结果、追踪关键步骤的输入输出、敏感操作记录变更前后值。SQLite够轻量够用，关键是形成习惯——每写一个新流程，先把日志埋好。
