影刀RPA 流程审计日志：谁在什么时候做了什么

影刀RPA 流程审计日志谁在什么时候做了什么作者林焱什么情况用流程跑着跑着出了问题你要排查发现——不知道哪一步出的错、不知道什么时间跑的、不知道输入是什么。只能从头到尾加断点重跑一遍整个过程像在抓瞎。稍微正规一点的RPA实践都会要求每个流程必须有运行日志出了问题能在5分钟内定位。更进一步的需求是审计——老板想知道「这个自动发薪流程上个月到底改了谁的工资谁改的」核心场景需要追踪流程的完整运行轨迹支持问题排查和合规审计。怎么做第一步设计日志的四个层次拼多多店群自动化上架方案不是所有的日志都一样重要。分层设计方便按需查看。┌─────────────────────────────────────────┐ │ Level 1: 运行摘要 │ │ 每次执行一条记录时间、流程名、结果、耗时 │ ├─────────────────────────────────────────┤ │ Level 2: 步骤追踪 │ │ 流程中每个关键步骤的输入输出 │ ├─────────────────────────────────────────┤ │ Level 3: 详细调试 │ │ 变量值、中间计算结果仅调试时开启 │ ├─────────────────────────────────────────┤ │ Level 4: 变更审计敏感流程专用 │ │ 谁、何时、改了哪个字段、改前值、改后值 │ └─────────────────────────────────────────┘第二步构建日志记录器importjsonimportsqlite3importosfromdatetimeimportdatetimefromthreadingimportLockclassAuditLogger: 影刀流程审计日志系统 基于SQLite轻量无需额外部署 def__init__(self,db_pathaudit_log.db):self.db_pathdb_path self.lockLock()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_paramsNone):开始一次流程运行返回run_idwithself.lock:withsqlite3.connect(self.db_path)asconn:cursorconn.execute(INSERT INTO run_summary (flow_name, start_time, input_params) VALUES (?, ?, ?),(flow_name,datetime.now().isoformat(),json.dumps(input_params,ensure_asciiFalse)))returncursor.lastrowiddefend_run(self,run_id,statussuccess,error_messageNone,output_summaryNone):结束一次运行记录结果withself.lock:withsqlite3.connect(self.db_path)asconn:# 计算运行时长rowconn.execute(SELECT start_time FROM run_summary WHERE id?,(run_id,)).fetchone()ifrow:startdatetime.fromisoformat(row[0])duration(datetime.now()-start).total_seconds()else:duration0conn.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_asciiFalse),run_id))deflog_step(self,run_id,step_name,step_typeaction,input_dataNone,output_dataNone,statussuccess,error_detailNone,execution_timeNone):记录单个步骤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_asciiFalse)ifinput_dataelseNone,json.dumps(output_data,ensure_asciiFalse)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_nameNone,statusNone,days7):查询最近的运行记录withsqlite3.connect(self.db_path)asconn:conn.row_factorysqlite3.Row querySELECT * FROM run_summary WHERE 11params[]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 100cursorconn.execute(query,params)return[dict(row)forrowincursor.fetchall()]defquery_change_audit(self,target_tableNone,days7):查询变更审计记录withsqlite3.connect(self.db_path)asconn:conn.row_factorysqlite3.Row querySELECT * FROM change_audit WHERE 11params[]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 500cursorconn.execute(query,params)return[dict(row)forrowincursor.fetchall()]第三步在影刀流程中使用Python节点用法——在影刀流程的关键节点插入日志记录# 流程入口 Python节点 importjson# 初始化日志器全局loggerAuditLogger(D:/RPA_Logs/audit_log.db)# 开始本次运行run_idlogger.start_run(flow_name每日销售报表生成,input_params{report_date:report_date,data_source:data_source})# 把run_id存到影刀变量中后续节点都需要它print(run_id)# 每个关键步骤后 Python节点 logger.log_step(run_idrun_id,step_name从ERP导出销售数据,step_typeaction,output_data{row_count:1520,file:sales_20240626.csv},execution_time3.2)# 有数据修改时 logger.log_change(run_idrun_id,target_table员工薪资表,target_field基本工资,target_record_idEMP-0042,old_value15000,new_value16500,change_reason年度调薪涨幅10%)# 流程结束 Python节点 logger.end_run(run_idrun_id,statussuccessifnothas_errorelsefailed,error_messageerror_msgifhas_errorelseNone,output_summary{reports_generated:3,emails_sent:5})第四步日志查询工具classLogReporter:日志报表生成def__init__(self,logger):self.loggerloggerdefdaily_report(self,date_strNone):生成每日运行报告ifdate_strisNone:date_strdatetime.now().strftime(%Y-%m-%d)runsself.logger.query_runs(days1)totallen(runs)successsum(1forrinrunsifr[status]success)failedsum(1forrinrunsifr[status]failed)total_durationsum(r[duration_seconds]or0forrinruns)reportf RPA运行日报{date_str} 运行概览 总执行次数{total}成功{success}{success/total*100:.1f}% 失败{failed}{failed/total*100:.1f}%  总耗时{total_duration/60:.1f}分钟 失败明细 forrinruns:ifr[status]failed:reportf - [{r[flow_name]}]{r[error_message]}\nreturnreportdefweekly_audit_report(self):每周审计报告——变更记录汇总changesself.logger.query_change_audit(days7)# 统计变更最多的表和字段fromcollectionsimportCounter table_statsCounter(c[target_table]forcinchanges)report\n 审计周报 \n\n变更统计\nfortable,countintable_stats.most_common():reportf{table}:{count}次变更\nreturnreport有什么坑坑1SQLite的并发写入问题如果你的流程有多线程多个子流程同时运行同时写SQLite可能触发database is locked错误。踩坑实录做了一个多线程爬虫流程5个线程同时往同一个SQLite写日志十个里面有七个报database is locked。解决方法用threading.Lock保护写操作或者换成SQLite的WAL模式Write-Ahead Logging。conn.execute(PRAGMA journal_modeWAL;)坑2日志文件无限制增长SQLite文件不会自动清理一个流程每天产生几千条日志三个月后审计库就几百MB了。解决方法加一个定时清理任务——保留最近30天的日志更早的归档到文件或直接删除。TEMU店群如何管理运营defcleanup_old_logs(db_path,keep_days30):清理旧日志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够轻量够用关键是形成习惯——每写一个新流程先把日志埋好。