影刀RPA 50个最常用的Python代码片段合集
影刀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)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: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("金额",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:3025. 日期加减
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()# hello39. 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()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}")# 记录详细错误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" # 放在代码片段里就分享出去了 # ✅ 从环境变量或配置文件读取  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），随用随查能省大量时间。但记住——理解每段代码在做什么，比复制粘贴更重要。