影刀RPA 断点续跑流程中断后从断点继续而不是重头开始作者林焱长流程最怕中断。采集了500条数据第501条时报错了如果从头开始跑前面的500条白干了。断点续跑就是解决这个问题的——记录执行状态中断后从断点继续。原理进度持久化每次处理完一条记录把当前位置存到文件里。重新启动时先读进度文件跳过已处理的部分。基础实现文件记录进度拼多多店群自动化上架方案importjsonimportos PROGRESS_FILEprogress.jsondefload_progress():读取断点ifos.path.exists(PROGRESS_FILE):withopen(PROGRESS_FILE,r)asf:returnjson.load(f)return{last_page:1,last_item:0,processed:0}defsave_progress(page,item,processed):保存断点withopen(PROGRESS_FILE,w)asf:json.dump({last_page:page,last_item:item,processed:processed,updated_at:datetime.now().isoformat()},f)# 主流程progressload_progress()start_pageprogress[last_page]totalprogress[processed]forpageinrange(start_page,total_pages1):itemsfetch_page(page)# 如果是从断点恢复跳过已处理的itemifpagestart_page:itemsitems[progress[last_item]:]fori,iteminenumerate(items):try:process_item(item)total1save_progress(page,i1,total)exceptExceptionase:print(f处理失败页面{page}第{i1}条{e})save_progress(page,i,total)raise# 抛出异常人工介入更精细数据库记录每条处理状态对于数据量大的场景用SQLite记录每条记录的状态importsqlite3 connsqlite3.connect(task_tracker.db)conn.execute(CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT UNIQUE, status TEXT DEFAULT pending, -- pending/processing/done/failed retry_count INTEGER DEFAULT 0, error_msg TEXT, created_at TEXT, updated_at TEXT))defget_pending_tasks(limit100):获取待处理的任务cursorconn.execute(SELECT task_id FROM tasks WHERE status IN (pending,failed) AND retry_count 3 ORDER BY id LIMIT ?,(limit,))return[row[0]forrowincursor.fetchall()]defmark_task(task_id,status,errorNone):更新任务状态ifstatusfailed:conn.execute(UPDATE tasks SET statusfailed, retry_countretry_count1, error_msg?, updated_at? WHERE task_id?,(error,datetime.now().isoformat(),task_id))else:conn.execute(UPDATE tasks SET status?, updated_at? WHERE task_id?,(status,datetime.now().isoformat(),task_id))conn.commit()# 使用task_idsget_pending_tasks()fortidintask_ids:try:process(tid)mark_task(tid,done)exceptExceptionase:mark_task(tid,failed,str(e))实战URL采集的完整断点续跑importsqlite3importrequestsdefsetup_db():connsqlite3.connect(crawler_state.db)conn.execute(CREATE TABLE IF NOT EXISTS urls (url TEXT PRIMARY KEY, status TEXT DEFAULT pending, data TEXT, error TEXT, crawled_at TEXT))conn.commit()returnconndefadd_urls(url_list):connsetup_db()forurlinurl_list:conn.execute(INSERT OR IGNORE INTO urls (url) VALUES (?),(url,))conn.commit()conn.close()defcrawl_with_resume(urls,headers):add_urls(urls)# 先把所有URL注册进数据库connsqlite3.connect(crawler_state.db)whileTrue:cursorconn.execute(SELECT url FROM urls WHERE statuspending LIMIT 1)rowcursor.fetchone()ifnotrow:break# 没有待采集的URL了urlrow[0]try:resprequests.get(url,headersheaders,timeout10)dataresp.text[:1000]conn.execute(UPDATE urls SET statusdone, data?, crawled_at? WHERE url?,(data,datetime.now().isoformat(),url))exceptExceptionase:conn.execute(UPDATE urls SET statusfailed, error? WHERE url?,(str(e),url))conn.commit()time.sleep(random.uniform(0.5,1.5))conn.close()踩坑实录坑1进度文件写入中途崩溃导致状态不一致如果在写进度文件时断电或关闭程序文件可能损坏。解决方法先写临时文件然后原子替换importtempfileimportshutildefsafe_save_progress(data):fd,tmp_pathtempfile.mkstemp()withos.fdopen(fd,w)asf:json.dump(data,f)shutil.move(tmp_path,PROGRESS_FILE)# 原子操作坑2数据变更导致断点偏移TEMU店群如何管理运营如果在断点期间数据源有新增/删除比如按序号翻页、第3页的内容变了断点续跑可能导致数据遗漏或重复。如果有唯一ID如URL、数据库主键用ID定位比用序号定位靠谱得多。坑3重试逻辑不当导致死循环失败的任务一直重试retry_count只增不减可能一直卡住。加个最大重试次数比如3次超过后标记为需要人工处理。坑4SQLite并发写入如果你的影刀流程是多线程的SQLite的并发写入可能报database is locked。SQLite默认是串行写入多线程要用WAL模式conn.execute(PRAGMA journal_modeWAL)写在最后断点续跑是能用和好用的分界线。花10分钟加上进度记录未来省你无数次从头跑流程的痛苦。不管是采集、处理还是推送只要步骤超过50步就值得加断点续跑。