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

作者：林焱

什么情况用

写RPA流程时，有大量常用的代码片段会反复用到——发送HTTP请求、读写Excel、正则提取、日期处理、去重排序。每次都要翻之前的代码或搜索引擎，效率很低。

这里整理了50个经过实际验证的代码片段，覆盖RPA日常开发的绝大部分场景。建议收藏，随用随查。

一、HTTP请求（8个）

1. 通用GET请求

importrequests resp=requests.get("https://api.example.com/data",headers={"Authorization":"Bearer TOKEN"},timeout=30)data=resp.json()

2. 通用POST请求（JSON）

resp=requests.post("https://api.example.com/create",json={"name":"张三","age":28},timeout=30)![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/41fce8b6a8244091a967b8ec162d2fcf.png#pic_center)

3. 带重试的请求

defrequest_with_retry(url,max_retries=3):foriinrange(max_retries):try:resp=requests.get(url,timeout=30)resp.raise_for_status()returnresp.json()exceptExceptionase:ifi==max_retries-1:raisetime.sleep(2**i)# 指数退避

4. 文件上传

withopen("file.pdf","rb")asf:resp=requests.post("https://api.example.com/upload",files={"file":f})

5. 下载文件（流式）

resp=requests.get("https://example.com/large.zip",stream=True)withopen("large.zip","wb")asf:forchunkinresp.iter_content(chunk_size=8192):f.write(chunk)

6. Session保持登录

session=requests.Session()session.post("https://example.com/login",json={"user":"admin","pwd":"123"})# 后续请求自动带Cookiedata=session.get("https://example.com/dashboard").json()

7. 设置代理

proxies={"http":"http://127.0.0.1:7890","https":"http://127.0.0.1:7890"}resp=requests.get("https://api.example.com",proxies=proxies)

8. URL参数编码

fromurllib.parseimporturlencode,quote params={"keyword":"笔记本电脑","page":1}url=f"https://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. 读取Excel

importpandasaspd df=pd.read_excel("data.xlsx",sheet_name="Sheet1")df=pd.read_excel("data.xlsx",sheet_name=None)# 读取所有Sheet，返回字典

10. 写入Excel（多Sheet）

withpd.ExcelWriter("output.xlsx")aswriter:df1.to_excel(writer,sheet_name="汇总",index=False)df2.to_excel(writer,sheet_name="明细",index=False)

11. 读取CSV（处理编码）

encodings=['utf-8','gbk','gb2312','utf-8-sig']forencinencodings:try:df=pd.read_csv("data.csv",encoding=enc)breakexceptUnicodeDecodeError:continue

12. 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("金额",ascending=False)# 排序

13. 数据筛选

df[df["状态"]=="已完成"]df[(df["金额"]>1000)&(df["城市"].isin(["北京","上海"]))]df[df["备注"].str.contains("紧急",na=False)]

14. 分组汇总

df.groupby("部门").agg({"销售额":"sum","订单数":"count"})df.groupby(["城市","月份"])["金额"].sum().unstack()

15. 合并多个DataFrame

# 纵向拼接pd.concat([df1,df2,df3],ignore_index=True)# 横向关联（类似SQL JOIN）pd.merge(df1,df2,on="订单号",how="left")

16. 日期筛选

df["日期"]=pd.to_datetime(df["日期"])last_7_days=df[df["日期"]>=datetime.now()-timedelta(days=7)]this_month=df[df["日期"].dt.month==datetime.now().month]

三、正则与文本处理（7个）

17. 提取手机号

importre phones=re.findall(r'1[3-9]\d{9}',text)

18. 提取身份证号

ids=re.findall(r'\d{17}[\dXx]',text)

19. 提取邮箱

emails=re.findall(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',text)

20. 提取URL

urls=re.findall(r'https?://[^\s<>"{}|\\^`\[\]]+',text)

21. 提取日期

dates=re.findall(r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}[日]?',text)

22. 字符串清洗

text=re.sub(r'\s+','',text)# 去掉所有空白text=re.sub(r'<[^>]+>','',text)# 去掉HTML标签text=text.replace('\u200b','').strip()# 去掉零宽空格

23. 中文提取 vs 非中文提取

chinese=''.join(re.findall(r'[\u4e00-\u9fff]',text))# 只保留中文non_chinese=re.sub(r'[\u4e00-\u9fff]','',text)# 去掉中文

四、日期时间（6个）

24. 获取当前时间各种格式

fromdatetimeimportdatetime,date,timedelta now=datetime.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:30

25. 日期加减

yesterday=datetime.now()-timedelta(days=1)last_week=datetime.now()-timedelta(weeks=1)last_month=datetime.now()-timedelta(days=30)# 约1个月

26. 月份加减（精确）

fromdateutil.relativedeltaimportrelativedelta next_month=datetime.now()+relativedelta(months=1)last_month_same_day=datetime.now()-relativedelta(months=1)

27. 获取本月第一天和最后一天

importcalendar today=date.today()first_day=today.replace(day=1)last_day=today.replace(day=calendar.monthrange(today.year,today.month)[1])

28. 字符串解析为日期

dt=datetime.strptime("2024-06-26","%Y-%m-%d")dt=datetime.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_ok=True)

32. 读写文本文件

withopen("input.txt","r",encoding="utf-8")asf:content=f.read()withopen("output.txt","w",encoding="utf-8")asf:f.write("Hello World")

33. 文件大小和时间

size_mb=os.path.getsize("file.pdf")/(1024*1024)mtime=datetime.fromtimestamp(os.path.getmtime("file.pdf"))

34. 临时文件

importtempfilewithtempfile.NamedTemporaryFile(delete=False,suffix=".csv")asf:f.write(b"data")temp_path=f.name# 用完删除os.unlink(temp_path)

六、数据转换（5个）

35. JSON ↔ Python

importjson obj=json.loads('{"name":"张三","age":28}')# JSON → dicttext=json.dumps(obj,ensure_ascii=False,indent=2)# dict → JSONwithopen("data.json","r")asf:data=json.load(f)# 读JSON文件

36. 字典列表 → DataFrame

data=[{"name":"张三","age":28},{"name":"李四","age":32}]df=pd.DataFrame(data)

37. DataFrame → 字典列表

records=df.to_dict("records")

38. Base64编解码

importbase64 encoded=base64.b64encode(b"hello").decode()# aGVsbG8=decoded=base64.b64decode("aGVsbG8=").decode()# hello

39. URL编解码

fromurllib.parseimportquote,unquote encoded=quote("笔记本电脑")# %E7%AC%94%E8%AE%B0%E6%9C%AC%E7%94%B5%E8%84%91decoded=unquote(encoded)# 笔记本电脑

七、数据库操作（4个）

40. SQLite读写

importsqlite3 conn=sqlite3.connect("data.db")df=pd.read_sql("SELECT * FROM users WHERE status='active'",conn)conn.execute("INSERT INTO logs VALUES (?, ?, ?)",(1,"info","success"))conn.commit()conn.close()

41. MySQL连接

importpymysql conn=pymysql.connect(host="127.0.0.1",port=3306,user="root",password="pwd",database="test",charset="utf8mb4")[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()![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/00042aaf17c046ee980af46513641154.png#pic_center)

43. 数据库连接上下文管理器

fromcontextlibimportcontextmanager@contextmanagerdefget_db():conn=sqlite3.connect("data.db")try:yieldconnfinally:conn.close()withget_db()asconn:df=pd.read_sql("SELECT * FROM orders",conn)

八、影刀专用（3个）

44. 接收影刀变量

# 影刀传入的变量直接在Python节点中可用data_list=json.loads(youdao_variable_name)

45. 输出给影刀后续节点

# print的内容会被影刀捕获作为节点输出result={"status":"success","count":100}print(json.dumps(result,ensure_ascii=False))

46. 多行输出

print("第一行输出")# 会被影刀读取为变量值importsysprint("调试信息",file=sys.stderr)# 不会影响影刀的变量读取

九、错误处理与日志（2个）

47. 通用异常处理模板

importtracebacktry:# 业务代码result=process_data()exceptValueErrorase:print(f"数据格式错误：{e}")# 记录详细错误![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/6b2f5b77575e4dbfba147cf08640056e.png#pic_center)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(level=logging.INFO,format='%(asctime)s [%(levelname)s] %(message)s',handlers=[logging.FileHandler("rpa.log",encoding="utf-8"),logging.StreamHandler()])logging.info("流程开始")logging.warning("处理异常数据")logging.error("流程失败",exc_info=True)

十、实用工具（2个）

49. 进度条显示

defprogress_bar(current,total,prefix=""):percent=current/total*100bar_len=40filled=int(bar_len*current/total)bar="█"*filled+"░"*(bar_len-filled)print(f"\r{prefix}|{bar}|{percent:.1f}%",end="")ifcurrent==total:print()# 使用foriinrange(1,101):progress_bar(i,100,"处理中")time.sleep(0.05)

50. 哈希计算

importhashlib text="hello world"md5_hash=hashlib.md5(text.encode()).hexdigest()# 32位MD5sha256_hash=hashlib.sha256(text.encode()).hexdigest()# 64位SHA256file_md5=hashlib.md5(open("file.txt","rb").read()).hexdigest()# 文件MD5

有什么坑

坑1：盲目复制粘贴

这些代码片段是起点，不是终点。直接复制不检查参数，比如忘记改编码格式、timeout设太大或太小，都会在你实际使用时出问题。

坑2：内存中的密码

# ❌ 别这样写 API_KEY = "sk-abc123xyz456" # 放在代码片段里就分享出去了 # ✅ 从环境变量或配置文件读取 ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/782c7a4584d445a7949b504f54edc5a9.png#pic_center) import os API_KEY = os.environ.get("RPA_API_KEY")

坑3：编码问题无处不在

中文环境下，90%的问题和编码有关。记住房口诀：读文件用encoding="utf-8"，写入JSON加ensure_ascii=False，CSV先尝试utf-8再降级gbk。

总结：这50个代码片段是我在实际RPA开发中最频繁使用到的。保存成文件、加入代码片段工具（如SnippetsLab、VS Code Snippets），随用随查能省大量时间。但记住——理解每段代码在做什么，比复制粘贴更重要。