影刀RPA进阶教程：Python节点深度使用与自定义指令开发完全手册-平芜编程栈

影刀RPA进阶教程：Python节点深度使用与自定义指令开发完全手册

作者：林焱

影刀RPA的拖拽指令很方便，但遇到复杂逻辑就力不从心了——比如要用正则表达式提取字符串、要调用第三方API、要做复杂的数据转换。这时候就需要用Python节点。我之前做的一个项目里，90%的逻辑都是用Python节点写的。这篇文章把Python节点的高级用法和自定义指令开发方法完整写出来。

一、Python节点的基本使用

1.1 在影刀中使用Python节点

操作步骤： 1. 在影刀流程编辑页面，点击「指令」→「Python」 2. 拖拽「Python节点」到流程里 3. 在Python节点里写代码（注意：代码是在影刀的Python环境里运行的） 4. 运行流程，查看输出

重要：影刀的Python环境和你本地装的Python不是同一个。影刀自带Python 3.8（版本可能随更新变化）。

1.2 影刀Python节点的输入输出

# -*- coding: utf-8 -*-# 影刀Python节点的标准写法defmain():""" 影刀Python节点的入口函数 注意：影刀会调用这个函数，并捕获它的输出 """# 输出到影刀日志（相当于 print，但会被影刀捕获）print('Python节点开始执行...')# 返回值：影刀会把这个返回值存到变量里# 在影刀Python节点的「输出」配置里，可以指定存到哪个变量result={'status':'success','data':[1,2,3]}returnresult# 调用入口函数if__name__=='__main__':result=main()print(f'执行结果：{result}')

在影刀里的配置：

「输入」：可以传入影刀变量（如input_list）
「输出」：指定返回值存到哪个影刀变量（如python_result）

二、可复用的Python函数库

2.1 为什么需要函数库？

如果每段Python代码都写在流程里，会有两个问题：

代码重复（同样的逻辑写多次）
维护困难（改一个逻辑要改多处）

解决方法：把常用逻辑封装成函数库，存成本地.py文件，在影刀Python节点里调用。

2.2 创建函数库

# 文件名：C:/rpa/utils/data_utils.py# -*- coding: utf-8 -*-# 功能：数据处理的工具函数库importreimportjsonfromdatetimeimportdatetimedefclean_text(text):""" 清洗文本：去除表情、特殊字符、多余空格 :param text: 原始文本 :return: 清洗后的文本 """ifnottext:return''# 去除表情符号emoji_pattern=re.compile('['u'\U0001F600-\U0001F64F'u'\U0001F300-\U0001F5FF'u'\U0001F680-\U0001F6FF'u'\U0001F1E0-\U0001F1FF'']+',flags=re.UNICODE)text=emoji_pattern.sub('',text)# 去除HTML标签text=re.sub(r'<[^>]+>','',text)# 去除多余空格和换行text=re.sub(r'\s+',' ',text).strip()returntextdefparse_price(price_str):""" 解析价格字符串，转成浮点数 :param price_str: 价格字符串（如 "¥99.9", "$19.99", "99.9元"） :return: 价格（浮点数） """ifnotprice_str:return0.0# 去除货币符号和逗号price_str=re.sub(r'[¥$￥,，元]','',str(price_str))# 提取数字（支持小数点）match=re.search(r'[\d.]+',price_str)ifmatch:try:returnfloat(match.group())except:return0.0else:return0.0defparse_sales(sales_str):""" 解析销量字符串（如 "1万+", "5000+", "978"） :param sales_str: 销量字符串 :return: 销量（整数） """ifnotsales_str:return0sales_str=str(sales_str)# 处理"万"单位if'万'insales_str:num=re.findall(r'[\d.]+',sales_str)[0]returnint(float(num)*10000)# 处理"+"后缀if'+'insales_str:sales_str=sales_str.replace('+','')# 提取数字match=re.search(r'\d+',sales_str)ifmatch:returnint(match.group())else:return0defformat_timestamp(ts,fmt='%Y-%m-%d %H:%M:%S'):""" 将时间戳转成字符串 :param ts: 时间戳（秒） :param fmt: 时间格式 :return: 格式化的时间字符串 """returndatetime.fromtimestamp(ts).strftime(fmt)defsafe_json_loads(json_str,default=None):""" 安全的JSON解析（解析失败返回默认值） :param json_str: JSON字符串 :param default: 解析失败时返回的默认值 :return: 解析结果或默认值 """try:returnjson.loads(json_str)except:returndefaultifdefaultisnotNoneelse{}# 测试代码（直接运行这个文件时会执行）if__name__=='__main__':print(clean_text('Hello 世界！🎉'))print(parse_price('¥99.9'))print(parse_sales('1万+'))

2.3 在影刀中调用函数库

# 在影刀Python节点里调用自定义函数库# 方法1：直接 import（需要确保文件在Python路径里）importsys sys.path.append('C:/rpa/utils')# 添加函数库所在目录fromdata_utilsimportclean_text,parse_price,parse_salesdefmain():# 使用函数库title=clean_text('这是 一个标题🎉')price=parse_price('¥99.9')sales=parse_sales('1万+')return{'title':title,'price':price,'sales':sales}

注意：影刀的Python环境可能不支持sys.path.append()，这时候可以用方法2。

# 方法2：用 exec 动态加载（兼容性更好）defmain():# 读取函数库文件内容withopen('C:/rpa/utils/data_utils.py','r',encoding='utf-8')asf:code=f.read()# 执行代码，把函数加载到当前命名空间exec(code,globals())# 现在可以调用函数库里的函数了title=clean_text('这是 一个标题🎉')price=parse_price('¥99.9')return{'title':title,'price':price}

三、复杂数据清洗

3.1 用正则表达式提取结构化数据

# -*- coding: utf-8 -*-# 功能：用正则表达式从非结构化文本中提取结构化数据importredefextract_structured_data(text):""" 从商品描述文本中提取结构化数据 :param text: 商品描述文本 :return: 结构化数据字典 """data={'brand':'',# 品牌'model':'',# 型号'color':'',# 颜色'size':'',# 尺寸'material':'',# 材质'weight':'',# 重量}# 提取品牌（模式：品牌：XXX 或 brand:XXX）brand_match=re.search(r'品牌[：:]\s*(\S+)',text,re.IGNORECASE)ifbrand_match:data['brand']=brand_match.group(1)# 提取型号（模式：型号：XXX 或 model:XXX）model_match=re.search(r'型号[：:]\s*(\S+)',text,re.IGNORECASE)ifmodel_match:data['model']=model_match.group(1)# 提取颜色（模式：颜色：XXX 或 可选颜色：XXX）color_match=re.search(r'颜色[：:]\s*([\S\s]+?)(?:\n|$)',text,re.IGNORECASE)ifcolor_match:data['color']=color_match.group(1).strip()# 提取尺寸（模式：尺寸：XXX 或 长宽高：XXX）size_match=re.search(r'尺寸[：:]\s*([\S\s]+?)(?:\n|$)',text,re.IGNORECASE)ifsize_match:data['size']=size_match.group(1).strip()# 提取材质（模式：材质：XXX 或 面料：XXX）material_match=re.search(r'(?:材质|面料)[：:]\s*([\S\s]+?)(?:\n|$)',text,re.IGNORECASE)ifmaterial_match:data['material']=material_match.group(1).strip()# 提取重量（模式：重量：XXX 或 毛重：XXX）weight_match=re.search(r'重量[：:]\s*([\d.]+)\s*(?:g|kg|克|千克)',text,re.IGNORECASE)ifweight_match:weight_val=float(weight_match.group(1))unit=re.search(r'(g|kg|克|千克)',weight_match.group(0)).group(1)# 统一转成 gifunitin['kg','千克']:weight_val=weight_val*1000data['weight']=f'{weight_val}g'returndata# 示例：# text = "品牌：Nike\n型号：A123\n颜色：红色\n尺寸：30x20x10cm\n材质：聚酯纤维\n重量：500g"# print(extract_structured_data(text))

3.2 用NLP做情感分析（可选）

如果需要分析评论的情感倾向（正面/负面），可以调用百度NLP API。

# -*- coding: utf-8 -*-# 功能：用百度NLP API做情感分析importrequestsimportjsondefsentiment_analysis(text,access_token):""" 情感分析 :param text: 要分析的文本 :param access_token: 百度API的 Access Token :return: 情感倾向（0=负面，1=中性，2=正面）和置信度 """url=f'https://aip.baidu.com/rpc/2.0/nlp/v1/sentiment_classify?access_token={access_token}'payload={'text':text,'scene':'default'# 场景：default（通用） 或 custom（自定义）}resp=requests.post(url,json=payload,timeout=10)result=resp.json()if'items'inresult:item=result['items'][0]return{'sentiment':item['positive_prob'],# 正面概率'confidence':item['confidence'],# 置信度'label':item['label']# 0=负面，1=中性，2=正面}else:print(f'情感分析失败：{result}')returnNone# 使用方式：# 1. 去百度AI开放平台申请自然语言处理 API# 2. 获取 Access Token# 3. 调用 sentiment_analysis('这个产品很好用！', access_token)

四、用Python绕过影刀指令的限制

4.1 影刀指令的限制

影刀的指令是封装好的，有些场景不够灵活：

限制	描述	Python解决方案
「输入文本」指令不能模拟逐字输入	一次性填入，容易被检测	用Selenium的`send_keys()`逐字输入
「点击元素」指令不能模拟鼠标移动	直接点击，容易被检测	用`ActionChains`模拟鼠标移动

4.2 模拟逐字输入（绕过输入检测）

# -*- coding: utf-8 -*-# 功能：模拟人类的逐字输入fromselenium.webdriver.common.keysimportKeysimporttimeimportrandomdefhuman_like_typing(element,text):""" 模拟人类的逐字输入 :param element: 要输入的元素（Selenium元素对象） :param text: 要输入的文本 """forcharintext:# 输入一个字符element.send_keys(char)# 随机延迟（模拟打字速度）# 中文输入法切换会有额外延迟if'\u4e00'<=char<='\u9fff':# 中文字符time.sleep(random.uniform(0.2,0.5))else:# 英文/数字time.sleep(random.uniform(0.05,0.15))print(f'逐字输入完成：{text}')

4.3 用JavaScript获取隐藏元素的内容

# -*- coding: utf-8 -*-# 功能：用JavaScript获取隐藏元素的文本内容defget_hidden_element_text(driver,selector):""" 用JavaScript获取隐藏元素的文本 :param driver: 浏览器对象 :param selector: CSS选择器 :return: 元素的文本内容 """# 用JavaScript获取 textContent（即使是隐藏元素也能获取）script=f''' var elem = document.querySelector("{selector}"); if (elem) {{ return elem.textContent || elem.innerText || ''; }} return ''; '''result=driver.execute_script(script)returnresult.strip()

五、调试Python代码

5.1 影刀Python节点的调试困难

影刀Python节点的调试不太方便：

没有断点调试
错误信息显示不完整
无法单步执行

解决方法：先在本地Python环境里调试好，再复制到影刀里。

5.2 本地调试技巧

# -*- coding: utf-8 -*-# 功能：在本地调试影刀Python节点代码# 步骤1：模拟影刀的输入变量input_list=[1,2,3,4,5]# 模拟影刀传入的变量config={'param1':'value1'}# 模拟影刀传入的配置# 步骤2：把影刀Python节点的代码复制到本地函数里defprocess_data(input_list,config):""" 这是影刀Python节点里的主逻辑 """result=[]foritemininput_list:# 处理逻辑...processed=item*2result.append(processed)returnresult# 步骤3：在本地测试if__name__=='__main__':try:output=process_data(input_list,config)print(f'测试成功：{output}')exceptExceptionase:# 本地可以看到完整的错误堆栈importtraceback traceback.print_exc()

5.3 在影刀里增加日志输出

# -*- coding: utf-8 -*-# 功能：在影刀Python节点里增加详细的日志defmain():print('=== Python节点开始执行 ===')# 打印输入变量（用于调试）print(f'输入变量 input_list:{input_list}')print(f'输入变量 config:{config}')try:# 处理逻辑...result=process_data(input_list,config)# 打印输出结果print(f'处理结果：{result}')print('=== Python节点执行成功 ===')returnresultexceptExceptionase:# 打印详细错误信息importtraceback error_msg=traceback.format_exc()print(f'❌ Python节点执行失败：{error_msg}')# 返回错误标记（方便影刀流程做异常处理）return{'status':'error','error':str(e),'traceback':error_msg}

六、自定义指令开发

6.1 什么是自定义指令？

影刀支持用Python写自定义指令（类似封装一个可复用的指令）。开发完成后，可以在指令面板里直接拖拽使用。

6.2 开发自定义指令的步骤

步骤1：在影刀里创建自定义指令 1. 点击指令面板右上角的「...」 2. 选择「创建自定义指令」 3. 填写指令名称、描述、图标 4. 选择「Python」作为实现方式 步骤2：写指令的实现代码 （参考下面的代码示例） 步骤3：配置指令的输入输出参数 - 输入参数：定义指令需要哪些输入（如 url、headers） - 输出参数：定义指令返回什么（如 response、status_code） 步骤4：测试指令 - 在测试区输入测试数据 - 运行指令，查看输出 步骤5：发布指令 - 点击「发布」，指令会出现在指令面板里

6.3 自定义指令代码示例

# -*- coding: utf-8 -*-# 自定义指令：HTTP请求（增强版）# 指令名称：发送HTTP请求（增强）# 指令描述：发送HTTP请求，支持自动重试、超时控制、响应解析defmain(url,method='GET',headers=None,data=None,max_retry=3,timeout=10):""" 自定义指令的主函数 :param url: 请求URL（输入参数） :param method: 请求方法（输入参数） :param headers: 请求头（输入参数） :param data: 请求数据（输入参数） :param max_retry: 最大重试次数（输入参数） :param timeout: 超时时间（输入参数） :return: 响应数据（输出参数） """importrequestsimporttime# 参数默认值ifheadersisNone:headers={}# 重试逻辑forretryinrange(max_retry):try:print(f'发送请求（第{retry+1}次）：{method}{url}')ifmethod.upper()=='GET':resp=requests.get(url,headers=headers,timeout=timeout)elifmethod.upper()=='POST':resp=requests.post(url,headers=headers,json=data,timeout=timeout)else:return{'status':'error','message':f'不支持的HTTP方法：{method}'}# 请求成功print(f'请求成功：状态码{resp.status_code}')return{'status':'success','status_code':resp.status_code,'response_text':resp.text,'response_json':resp.json()if'application/json'inresp.headers.get('Content-Type','')elseNone,'headers':dict(resp.headers)}exceptrequests.exceptions.Timeout:print(f'请求超时（第{retry+1}次），{timeout}秒后重试...')time.sleep(timeout)exceptExceptionase:print(f'请求失败（第{retry+1}次）：{str(e)}')time.sleep(2)# 所有重试都失败return{'status':'error','message':f'请求失败（已重试{max_retry}次）'}

常见问题速查

#	现象	原因	解决方式
1	Python节点执行失败，报错`ModuleNotFoundError`	影刀Python环境缺少第三方库	用`pip`在影刀安装目录里安装，或用`exec`动态加载自定义代码
2	Python节点的返回值获取不到	没有在「输出」配置里指定变量	在Python节点的「输出」配置里，指定返回值存到哪个影刀变量
3	用`exec`加载函数库时报编码错误	函数库文件包含中文，编码不对	用`open(file, 'r', encoding='utf-8')`读取文件
4	Selenium操作报`ElementNotInteractableException`	元素被遮挡或不可见	用JavaScript直接操作元素：`driver.execute_script('arguments[0].click()', element)`
5	本地调试正常，放到影刀里就失败	影刀Python环境和本地环境不一致	在影刀Python节点里打印`sys.version`，检查Python版本
6	自定义指令发布后在指令面板找不到	没有发布成功，或发布到了错误的分类	检查自定义指令的「分类」设置，确保选对了分类
7	Python代码里用到了第三方库，换电脑后运行失败	第三方库没有安装	把第三方库的源码打包到函数库里，用`exec`加载（避免依赖安装）

内容标签

影刀RPAPython节点自定义指令数据清洗正则表达式Selenium调试技巧函数库HTTP请求NLP

转载请注明出处：飞行社社区，作者：林焱

影刀RPA进阶教程：Python节点深度使用与自定义指令开发完全手册