影刀RPA 50个最常用的Python代码片段合集

影刀RPA 50个最常用的Python代码片段合集作者林焱什么情况用写RPA流程时有大量常用的代码片段会反复用到——发送HTTP请求、读写Excel、正则提取、日期处理、去重排序。每次都要翻之前的代码或搜索引擎效率很低。这里整理了50个经过实际验证的代码片段覆盖RPA日常开发的绝大部分场景。建议收藏随用随查。一、HTTP请求8个1. 通用GET请求importrequests resprequests.get(https://api.example.com/data,headers{Authorization:Bearer TOKEN},timeout30)dataresp.json()2. 通用POST请求JSONresprequests.post(https://api.example.com/create,json{name:张三,age:28},timeout30)3. 带重试的请求defrequest_with_retry(url,max_retries3):foriinrange(max_retries):try:resprequests.get(url,timeout30)resp.raise_for_status()returnresp.json()exceptExceptionase:ifimax_retries-1:raisetime.sleep(2**i)# 指数退避4. 文件上传withopen(file.pdf,rb)asf:resprequests.post(https://api.example.com/upload,files{file:f})5. 下载文件流式resprequests.get(https://example.com/large.zip,streamTrue)withopen(large.zip,wb)asf:forchunkinresp.iter_content(chunk_size8192):f.write(chunk)6. Session保持登录sessionrequests.Session()session.post(https://example.com/login,json{user:admin,pwd:123})# 后续请求自动带Cookiedatasession.get(https://example.com/dashboard).json()7. 设置代理proxies{http:http://127.0.0.1:7890,https:http://127.0.0.1:7890}resprequests.get(https://api.example.com,proxiesproxies)8. URL参数编码fromurllib.parseimporturlencode,quote params{keyword:笔记本电脑,page:1}urlfhttps://search.example.com?{urlencode(params)}# 单独编码quote(笔记本电脑) → %E7%AC%94%E8%AE%B0%E6%9C%AC%E7%94%B5%E8%84%91二、Excel/CSV操作8个9. 读取Excelimportpandasaspd dfpd.read_excel(data.xlsx,sheet_nameSheet1)dfpd.read_excel(data.xlsx,sheet_nameNone)# 读取所有Sheet返回字典10. 写入Excel多Sheetwithpd.ExcelWriter(output.xlsx)aswriter:df1.to_excel(writer,sheet_name汇总,indexFalse)df2.to_excel(writer,sheet_name明细,indexFalse)11. 读取CSV处理编码encodings[utf-8,gbk,gb2312,utf-8-sig]forencinencodings:try:dfpd.read_csv(data.csv,encodingenc)breakexceptUnicodeDecodeError:continue12. DataFrame常用操作df.head(10)# 前10行df.describe()# 统计描述df[列名].value_counts()# 值计数df.drop_duplicates()# 去重df.dropna()# 删除空行[video(video-dwFFlvcd-1783409315797)(type-csdn)(url-https://live.csdn.net/v/embed/525010)(image-https://v-blog.csdnimg.cn/asset/f4faa587144cb7070f19e8b36813806b/cover/Cover0.jpg)(title-店群矩阵自动化突破运营极限)]df.fillna(0)# 空值填0df.sort_values(金额,ascendingFalse)# 排序13. 数据筛选df[df[状态]已完成]df[(df[金额]1000)(df[城市].isin([北京,上海]))]df[df[备注].str.contains(紧急,naFalse)]14. 分组汇总df.groupby(部门).agg({销售额:sum,订单数:count})df.groupby([城市,月份])[金额].sum().unstack()15. 合并多个DataFrame# 纵向拼接pd.concat([df1,df2,df3],ignore_indexTrue)# 横向关联类似SQL JOINpd.merge(df1,df2,on订单号,howleft)16. 日期筛选df[日期]pd.to_datetime(df[日期])last_7_daysdf[df[日期]datetime.now()-timedelta(days7)]this_monthdf[df[日期].dt.monthdatetime.now().month]三、正则与文本处理7个17. 提取手机号importre phonesre.findall(r1[3-9]\d{9},text)18. 提取身份证号idsre.findall(r\d{17}[\dXx],text)19. 提取邮箱emailsre.findall(r[a-zA-Z0-9._%-][a-zA-Z0-9.-]\.[a-zA-Z]{2,},text)20. 提取URLurlsre.findall(rhttps?://[^\s{}|\\^\[\]],text)21. 提取日期datesre.findall(r\d{4}[-/年]\d{1,2}[-/月]\d{1,2}[日]?,text)22. 字符串清洗textre.sub(r\s,,text)# 去掉所有空白textre.sub(r[^],,text)# 去掉HTML标签texttext.replace(\u200b,).strip()# 去掉零宽空格23. 中文提取 vs 非中文提取chinese.join(re.findall(r[\u4e00-\u9fff],text))# 只保留中文non_chinesere.sub(r[\u4e00-\u9fff],,text)# 去掉中文四、日期时间6个24. 获取当前时间各种格式fromdatetimeimportdatetime,date,timedelta nowdatetime.now()print(now.strftime(%Y-%m-%d %H:%M:%S))# 2024-06-26 14:30:00print(now.strftime(%Y%m%d_%H%M%S))# 20240626_143000文件名用print(now.strftime(%Y年%m月%d日 %H:%M))# 2024年06月26日 14:3025. 日期加减yesterdaydatetime.now()-timedelta(days1)last_weekdatetime.now()-timedelta(weeks1)last_monthdatetime.now()-timedelta(days30)# 约1个月26. 月份加减精确fromdateutil.relativedeltaimportrelativedelta next_monthdatetime.now()relativedelta(months1)last_month_same_daydatetime.now()-relativedelta(months1)27. 获取本月第一天和最后一天importcalendar todaydate.today()first_daytoday.replace(day1)last_daytoday.replace(daycalendar.monthrange(today.year,today.month)[1])28. 字符串解析为日期dtdatetime.strptime(2024-06-26,%Y-%m-%d)dtdatetime.strptime(2024年6月26日 14:30,%Y年%m月%d日 %H:%M)29. 计算两个日期相差天数diff(datetime(2024,12,31)-datetime(2024,1,1)).days# 365五、文件操作5个30. 遍历文件夹importosforroot,dirs,filesinos.walk(./data):forfinfiles:print(os.path.join(root,f))31. 确保目录存在os.makedirs(./output/reports,exist_okTrue)32. 读写文本文件withopen(input.txt,r,encodingutf-8)asf:contentf.read()withopen(output.txt,w,encodingutf-8)asf:f.write(Hello World)33. 文件大小和时间size_mbos.path.getsize(file.pdf)/(1024*1024)mtimedatetime.fromtimestamp(os.path.getmtime(file.pdf))34. 临时文件importtempfilewithtempfile.NamedTemporaryFile(deleteFalse,suffix.csv)asf:f.write(bdata)temp_pathf.name# 用完删除os.unlink(temp_path)六、数据转换5个35. JSON ↔ Pythonimportjson objjson.loads({name:张三,age:28})# JSON → dicttextjson.dumps(obj,ensure_asciiFalse,indent2)# dict → JSONwithopen(data.json,r)asf:datajson.load(f)# 读JSON文件36. 字典列表 → DataFramedata[{name:张三,age:28},{name:李四,age:32}]dfpd.DataFrame(data)37. DataFrame → 字典列表recordsdf.to_dict(records)38. Base64编解码importbase64 encodedbase64.b64encode(bhello).decode()# aGVsbG8decodedbase64.b64decode(aGVsbG8).decode()# hello39. URL编解码fromurllib.parseimportquote,unquote encodedquote(笔记本电脑)# %E7%AC%94%E8%AE%B0%E6%9C%AC%E7%94%B5%E8%84%91decodedunquote(encoded)# 笔记本电脑七、数据库操作4个40. SQLite读写importsqlite3 connsqlite3.connect(data.db)dfpd.read_sql(SELECT * FROM users WHERE statusactive,conn)conn.execute(INSERT INTO logs VALUES (?, ?, ?),(1,info,success))conn.commit()conn.close()41. MySQL连接importpymysql connpymysql.connect(host127.0.0.1,port3306,userroot,passwordpwd,databasetest,charsetutf8mb4)[video(video-wcqXfyhf-1783409322478)(type-csdn)(url-https://live.csdn.net/v/embed/524992)(image-https://v-blog.csdnimg.cn/asset/b59aed2f01d4fe8583467562aaf4dcfd/cover/Cover0.jpg)(title-temu店群自动化报活动案例)]42. 批量插入data[(1,张三),(2,李四),(3,王五)]conn.executemany(INSERT INTO users (id, name) VALUES (?, ?),data)conn.commit()43. 数据库连接上下文管理器fromcontextlibimportcontextmanagercontextmanagerdefget_db():connsqlite3.connect(data.db)try:yieldconnfinally:conn.close()withget_db()asconn:dfpd.read_sql(SELECT * FROM orders,conn)八、影刀专用3个44. 接收影刀变量# 影刀传入的变量直接在Python节点中可用data_listjson.loads(youdao_variable_name)45. 输出给影刀后续节点# print的内容会被影刀捕获作为节点输出result{status:success,count:100}print(json.dumps(result,ensure_asciiFalse))46. 多行输出print(第一行输出)# 会被影刀读取为变量值importsysprint(调试信息,filesys.stderr)# 不会影响影刀的变量读取九、错误处理与日志2个47. 通用异常处理模板importtracebacktry:# 业务代码resultprocess_data()exceptValueErrorase:print(f数据格式错误{e})# 记录详细错误withopen(error.log,a)asf:f.write(f[{datetime.now()}] ValueError:{traceback.format_exc()}\n)exceptExceptionase:print(f未知错误{e})traceback.print_exc()48. 简单日志记录importlogging logging.basicConfig(levellogging.INFO,format%(asctime)s [%(levelname)s] %(message)s,handlers[logging.FileHandler(rpa.log,encodingutf-8),logging.StreamHandler()])logging.info(流程开始)logging.warning(处理异常数据)logging.error(流程失败,exc_infoTrue)十、实用工具2个49. 进度条显示defprogress_bar(current,total,prefix):percentcurrent/total*100bar_len40filledint(bar_len*current/total)bar█*filled░*(bar_len-filled)print(f\r{prefix}|{bar}|{percent:.1f}%,end)ifcurrenttotal:print()# 使用foriinrange(1,101):progress_bar(i,100,处理中)time.sleep(0.05)50. 哈希计算importhashlib texthello worldmd5_hashhashlib.md5(text.encode()).hexdigest()# 32位MD5sha256_hashhashlib.sha256(text.encode()).hexdigest()# 64位SHA256file_md5hashlib.md5(open(file.txt,rb).read()).hexdigest()# 文件MD5有什么坑坑1盲目复制粘贴这些代码片段是起点不是终点。直接复制不检查参数比如忘记改编码格式、timeout设太大或太小都会在你实际使用时出问题。坑2内存中的密码# ❌ 别这样写 API_KEY sk-abc123xyz456 # 放在代码片段里就分享出去了 # ✅ 从环境变量或配置文件读取  import os API_KEY os.environ.get(RPA_API_KEY)坑3编码问题无处不在中文环境下90%的问题和编码有关。记住房口诀读文件用encodingutf-8写入JSON加ensure_asciiFalseCSV先尝试utf-8再降级gbk。总结这50个代码片段是我在实际RPA开发中最频繁使用到的。保存成文件、加入代码片段工具如SnippetsLab、VS Code Snippets随用随查能省大量时间。但记住——理解每段代码在做什么比复制粘贴更重要。