影刀RPA 日志文件管理写入、轮转、清理作者林焱什么情况用你的影刀流程每天运行运行日志全存在一个文件里几个月后文件大到打不开你想给流程加上详细的运行日志但不知道怎么组织日志格式日志文件太多了需要自动清理旧日志日志文件管理是流程运维的基础。好的日志系统让你能快速定位问题、追踪运行历史、监控流程健康状态。没有日志的流程出了问题就像瞎子摸象。本文讲清楚怎么写日志、怎么实现日志轮转按大小/日期分割、怎么自动清理旧日志。怎么做一、基本日志写入importloggingimportosfromdatetimeimportdatetime# 方式1Python logging模块推荐 defsetup_logger(log_dirlogs,log_namerpa):配置日志系统ifnotos.path.exists(log_dir):os.makedirs(log_dir)log_fileos.path.join(log_dir,f{log_name}_{datetime.now().strftime(%Y%m%d)}.log)loggerlogging.getLogger(log_name)logger.setLevel(logging.DEBUG)# 避免重复添加handleriflogger.handlers:returnlogger# 文件handlerfile_handlerlogging.FileHandler(log_file,encodingutf-8)file_handler.setLevel(logging.DEBUG)# 控制台handlerconsole_handlerlogging.StreamHandler()[video(video-8R7FDOdN-1784394740268)(type-csdn)(url-https://live.csdn.net/v/embed/526818)(image-https://v-blog.csdnimg.cn/asset/582d14c3bd0451c5399cd990b56e2a0d/cover/Cover0.jpg)(title-拼多多店群自动化报活动上架)]console_handler.setLevel(logging.INFO)# 日志格式formatterlogging.Formatter(%(asctime)s [%(levelname)s] %(message)s,datefmt%Y-%m-%d %H:%M:%S)file_handler.setFormatter(formatter)console_handler.setFormatter(formatter)logger.addHandler(file_handler)logger.addHandler(console_handler)returnlogger# 使用loggersetup_logger(D:/rpa/logs,scrape)logger.info(流程启动)logger.debug(f配置参数: max_pages{50}, timeout{30})logger.warning(页面加载较慢已等待15秒)logger.error(数据解析失败JSON格式错误)logger.info(流程结束共采集128条数据)二、日志级别# 五个日志级别 logger.debug(调试信息变量值、中间结果)# 最详细生产环境通常关闭logger.info(常规信息流程开始、完成、进度)# 正常运行信息logger.warning(警告信息非预期但可继续的情况)# 需要关注但不影响运行logger.error(错误信息操作失败但流程继续)# 影响部分功能logger.critical(严重错误流程无法继续)# 必须立即处理# 实际使用示例 defscrape_page(url,logger):采集单页数据的日志示例logger.info(f开始采集:{url})try:responserequests.get(url,timeout30)logger.debug(f响应状态码:{response.status_code})ifresponse.status_code!200:logger.warning(f非200响应:{response.status_code})returnNonedataresponse.json()item_countlen(data.get(items,[]))logger.info(f采集成功:{item_count}条数据)returndataexceptrequests.Timeout:logger.error(f采集超时:{url})returnNoneexceptExceptionase:logger.error(f采集异常:{url}-{e},exc_infoTrue)returnNone三、日志轮转importloggingimportlogging.handlersimportosdefsetup_logger_with_rotation(log_dirlogs,log_namerpa,max_size10*1024*1024,backup_count5): 带日志轮转的日志系统 max_size: 单个日志文件最大大小字节默认10MB backup_count: 保留的旧日志文件数量 ifnotos.path.exists(log_dir):os.makedirs(log_dir)log_fileos.path.join(log_dir,f{log_name}.log)loggerlogging.getLogger(log_name)logger.setLevel(logging.DEBUG)iflogger.handlers:returnlogger# 按大小轮转file_handlerlogging.handlers.RotatingFileHandler(log_file,maxBytesmax_size,backupCountbackup_count,encodingutf-8)file_handler.setLevel(logging.DEBUG)formatterlogging.Formatter(%(asctime)s [%(levelname)s] %(message)s,datefmt%Y-%m-%d %H:%M:%S)file_handler.setFormatter(formatter)logger.addHandler(file_handler)returnloggerdefsetup_logger_with_daily_rotation(log_dirlogs,log_namerpa,backup_count30): 按日期轮转的日志系统 每天生成一个新日志文件保留backup_count天的日志 ifnotos.path.exists(log_dir):os.makedirs(log_dir)log_fileos.path.join(log_dir,f{log_name}.log)loggerlogging.getLogger(log_name)logger.setLevel(logging.DEBUG)iflogger.handlers:returnlogger# 按日期轮转file_handlerlogging.handlers.TimedRotatingFileHandler(log_file,whenmidnight,# 每天午夜轮转interval1,# 每1天backupCountbackup_count,# 保留30天encodingutf-8)file_handler.setLevel(logging.DEBUG)formatterlogging.Formatter(%(asctime)s [%(levelname)s] %(message)s,datefmt%Y-%m-%d %H:%M:%S)file_handler.setFormatter(formatter)logger.addHandler(file_handler)returnlogger# 使用loggersetup_logger_with_daily_rotation(D:/rpa/logs,scrape,backup_count30)# 文件结构# scrape.log ← 今天的日志# scrape.log.2026-06-30 ← 昨天的日志# scrape.log.2026-06-29 ← 前天的日志# ...最多保留30天四、自定义日志格式importloggingimportjsonfromdatetimeimportdatetimeclassJSONFormatter(logging.Formatter):JSON格式的日志方便日志分析工具处理defformat(self,record):log_entry{timestamp:datetime.fromtimestamp(record.created).isoformat(),level:record.levelname,message:record.getMessage(),module:record.module,line:record.lineno,}ifrecord.exc_info:log_entry[exception]self.formatException(record.exc_info)returnjson.dumps(log_entry,ensure_asciiFalse)classRPAFormatter(logging.Formatter):RPA专用日志格式defformat(self,record):# 添加流程信息ifhasattr(record,flow_name):prefixf[{record.flow_name}]else:prefixifhasattr(record,page_num):prefixf[P{record.page_num}]returnf{self.formatTime(record)}[{record.levelname}]{prefix}{record.getMessage()}# 使用JSON格式loggerlogging.getLogger(rpa_json)logger.setLevel(logging.DEBUG)handlerlogging.FileHandler(rpa.json.log,encodingutf-8)handler.setFormatter(JSONFormatter())logger.addHandler(handler)logger.info(流程启动,extra{flow_name:数据采集,page_num:1})五、日志清理importosimporttimefromdatetimeimportdatetime,timedeltadefclean_old_logs(log_dir,days_to_keep30,pattern*.log): 清理旧日志文件 days_to_keep: 保留最近多少天的日志 pattern: 日志文件匹配模式 importglob cutoff_timetime.time()-days_to_keep*86400deleted_count0freed_bytes0log_filesglob.glob(os.path.join(log_dir,pattern))log_filesglob.glob(os.path.join(log_dir,pattern.*))# 轮转的旧文件forfilepathinlog_files:ifos.path.getmtime(filepath)cutoff_time:file_sizeos.path.getsize(filepath)os.remove(filepath)deleted_count1freed_bytesfile_sizeprint(f删除:{os.path.basename(filepath)}({file_size/1024:.1f}KB))print(f清理完成: 删除{deleted_count}个文件释放{freed_bytes/1024/1024:.1f}MB)defclean_logs_by_size(log_dir,max_total_size_mb500): 按总大小清理日志 当日志目录总大小超过限制时删除最旧的日志 importglob max_bytesmax_total_size_mb*1024*1024log_filesglob.glob(os.path.join(log_dir,*.log*))# 按修改时间排序log_files.sort(keylambdaf:os.path.getmtime(f))total_sizesum(os.path.getsize(f)forfinlog_files)iftotal_sizemax_bytes:return# 从最旧的开始删除forfilepathinlog_files:iftotal_sizemax_bytes:breakfile_sizeos.path.getsize(filepath)os.remove(filepath)total_size-file_sizeprint(f删除:{os.path.basename(filepath)})# 使用保留30天日志clean_old_logs(D:/rpa/logs,days_to_keep30)# 或限制总大小500MBclean_logs_by_size(D:/rpa/logs,max_total_size_mb500)六、流程运行日志模板importloggingimporttimefromdatetimeimportdatetimeclassFlowLogger:流程运行日志管理器def__init__(self,flow_name,log_dirlogs):self.flow_nameflow_name self.start_timeNoneself.loggersetup_logger_with_daily_rotation(log_dir,flow_name)defstart(self):记录流程开始self.start_timetime.time()self.logger.info(f{*50})self.logger.info(f流程启动:{self.flow_name})self.logger.info(f开始时间:{datetime.now().strftime(%Y-%m-%d %H:%M:%S)})defstep(self,step_name,pageNone):记录步骤ifpage:self.logger.info(f步骤:{step_name}(第{page}页))else:self.logger.info(f步骤:{step_name})defsuccess(self,message,countNone):记录成功ifcountisnotNone:self.logger.info(f✓{message}({count}条))else:self.logger.info(f✓{message})defwarning(self,message):记录警告self.logger.warning(f⚠{message})deferror(self,message,exceptionNone):记录错误ifexception:self.logger.error(f✗{message}:{exception},exc_infoTrue)else:self.logger.error(f✗{message})deffinish(self,successTrue,summary):记录流程结束elapsedtime.time()-self.start_timeifself.start_timeelse0status成功ifsuccesselse失败self.logger.info(f流程结束:{status})self.logger.info(f耗时:{elapsed:.1f}秒)ifsummary:self.logger.info(f汇总:{summary})self.logger.info(f{*50})# 使用示例flow_logFlowLogger(电商数据采集,D:/rpa/logs)flow_log.start()try:flow_log.step(登录系统)# ... 登录逻辑 ...flow_log.success(登录成功)flow_log.step(采集商品列表,page1)# ... 采集逻辑 ...flow_log.success(第1页采集完成,count50)flow_log.step(采集商品列表,page2)# ... 采集逻辑 ...flow_log.warning(第2页部分图片加载失败)flow_log.success(第2页采集完成,count48)flow_log.step(导出数据)# ... 导出逻辑 ...flow_log.success(导出完成)flow_log.finish(successTrue,summary共采集98条导出至result.xlsx)exceptExceptionase:flow_log.error(流程异常中断,e)flow_log.finish(successFalse,summaryf错误:{e})有什么坑坑1日志中文乱码现象日志文件中的中文显示为乱码。原因FileHandler默认使用系统编码Windows上是GBK但日志内容中包含UTF-8编码的中文。解决创建FileHandler时指定encodingutf-8handlerlogging.FileHandler(app.log,encodingutf-8)坑2重复添加handler导致日志重复输出现象同一条日志被输出多次。原因多次调用setup_logger函数每次都添加了新的handler导致一条日志被多个handler输出。TEMU店群矩阵自动化运营核价报活动解决添加handler前检查是否已有handlerif logger.handlers: return logger。或者用logger.handlers.clear()清除旧handler后再添加。推荐用模块级单例模式确保logger只初始化一次。坑3日志文件被占用无法删除现象清理旧日志时最新的日志文件删除失败提示文件被占用。原因当前流程正在使用这个日志文件文件句柄没有释放。解决清理日志时跳过正在使用的文件检查文件修改时间是否在最近1分钟内。或者在清理前先关闭logger的handlerfor handler in logger.handlers: handler.close()。只清理旧的轮转文件不清理当前正在写入的文件。坑4日志级别设置不当现象生产环境日志太大debug信息太多或者出问题时日志信息不够只记了info没有debug。解决生产环境用INFO级别开发环境用DEBUG级别。通过配置文件控制日志级别不需要改代码。关键操作如金额计算、数据删除用WARNING或ERROR级别确保记录。调试信息用DEBUG级别生产环境关闭。坑5影刀的【输出调试信息】和Python日志混用现象日志信息分散在影刀的调试输出和Python日志文件中排查问题时需要两边都看。解决统一用Python logging模块管理日志。影刀的【输出调试信息】只用于开发调试生产环境用Python日志。在Python代码块中把关键日志同时输出到控制台影刀调试输出会捕获和日志文件loggersetup_logger()logger.info(这条信息同时出现在调试输出和日志文件中)