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

作者：林焱

---

什么情况用

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

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

---

一、HTTP请求（8个）

1. 通用GET请求

import requests
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. 带重试的请求

def request_with_retry(url, max_retries=3):
    for i in range(max_retries):
        try:
            resp = requests.get(url, timeout=30)
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            if i == max_retries - 1:
                raise
            time.sleep(2 ** i)  # 指数退避

4. 文件上传

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

5. 下载文件（流式）

resp = requests.get("https://example.com/large.zip", stream=True)
with open("large.zip", "wb") as f:
    for chunk in resp.iter_content(chunk_size=8192):
        f.write(chunk)

6. Session保持登录

session = requests.Session()
session.post("https://example.com/login", json={"user": "admin", "pwd": "123"})
# 后续请求自动带Cookie
data = 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参数编码

from urllib.parse import urlencode, 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

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

10. 写入Excel（多Sheet）

with pd.ExcelWriter("output.xlsx") as writer:
    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']
for enc in encodings:
    try:
        df = pd.read_csv("data.csv", encoding=enc)
        break
    except UnicodeDecodeError:
        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)           # 空值填0
df.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. 提取手机号

import re
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. 获取当前时间各种格式

from datetime import datetime, date, timedelta
now = datetime.now()
print(now.strftime("%Y-%m-%d %H:%M:%S"))     # 2024-06-26 14:30:00
print(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. 月份加减（精确）

from dateutil.relativedelta import relativedelta
next_month = datetime.now() + relativedelta(months=1)
last_month_same_day = datetime.now() - relativedelta(months=1)

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

import calendar
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. 遍历文件夹

import os
for root, dirs, files in os.walk("./data"):
    for f in files:
        print(os.path.join(root, f))

31. 确保目录存在

os.makedirs("./output/reports", exist_ok=True)

32. 读写文本文件

with open("input.txt", "r", encoding="utf-8") as f:
    content = f.read()

with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello World")

33. 文件大小和时间

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

34. 临时文件

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

---

六、数据转换（5个）

35. JSON ↔ Python

import json
obj = json.loads('{"name":"张三","age":28}')           # JSON → dict
text = json.dumps(obj, ensure_ascii=False, indent=2)   # dict → JSON
with open("data.json", "r") as f: 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编解码

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

39. URL编解码

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

---

七、数据库操作（4个）

40. SQLite读写

import sqlite3
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连接

import pymysql
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. 数据库连接上下文管理器

from contextlib import contextmanager

@contextmanager
def get_db():
    conn = sqlite3.connect("data.db")
    try:
        yield conn
    finally:
        conn.close()

with get_db() as conn:
    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("第一行输出")  # 会被影刀读取为变量值
import sys
print("调试信息", file=sys.stderr)  # 不会影响影刀的变量读取

---

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

47. 通用异常处理模板

import traceback

try:
    # 业务代码
    result = process_data()
except ValueError as e:
    print(f"数据格式错误：{e}")
    # 记录详细错误
    ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/6b2f5b77575e4dbfba147cf08640056e.png#pic_center)

    with open("error.log", "a") as f:
        f.write(f"[{datetime.now()}] ValueError: {traceback.format_exc()}\n")
except Exception as e:
    print(f"未知错误：{e}")
    traceback.print_exc()

48. 简单日志记录

import logging
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. 进度条显示

def progress_bar(current, total, prefix=""):
    percent = current / total * 100
    bar_len = 40
    filled = int(bar_len * current / total)
    bar = "█" * filled + "░" * (bar_len - filled)
    print(f"\r{prefix} |{bar}| {percent:.1f}%", end="")
    if current == total:
        print()

# 使用
for i in range(1, 101):
    progress_bar(i, 100, "处理中")
    time.sleep(0.05)

50. 哈希计算

import hashlib
text = "hello world"
md5_hash = hashlib.md5(text.encode()).hexdigest()     # 32位MD5
sha256_hash = hashlib.sha256(text.encode()).hexdigest() # 64位SHA256
file_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），随用随查能省大量时间。但记住——理解每段代码在做什么，比复制粘贴更重要。