影刀RPA技术深度：Python协同高级实战指南——第三方库安装自定义模块与异常处理完全解析

作者：林焱

Python协同是影刀RPA的高级功能，能让影刀调用Python的强大生态。本文深入讲解第三方库安装、自定义模块、异常处理和与影刀变量交互，全部附带真实代码和报错案例。

一、Python脚本模块基础

影刀的Python脚本模块允许在流程中执行Python代码。

1.1 基本使用

在影刀中添加"Python脚本"指令：

  1. 拖拽"Python脚本"指令到流程中
  2. 在代码编辑器中编写Python代码
  3. 运行流程，查看输出

真实代码示例：

# 简单的Python脚本
import datetime
now = datetime.datetime.now()
print(f"当前时间：{now.strftime('%Y-%m-%d %H:%M:%S')}")

1.2 与影刀变量交互

Python脚本可以读取和设置影刀变量。

# 读取影刀变量
input_text = yingdao.get_variable('input_text')

# 处理
result = input_text.upper()

# 设置影刀变量
yingdao.set_variable('output_text', result)

真实配置参数：

在影刀"Python脚本"指令中：
在这里插入图片描述

  • 代码类型：Python 3
  • 超时时间：60秒（默认）
  • 出错时继续：否（默认）

二、第三方库安装完全攻略

2.1 方法一：使用pip安装

影刀自带Python环境，可以用pip安装第三方库。

操作步骤：

  1. 打开影刀安装目录下的Python文件夹
  2. 按住Shift键，右键点击空白处，选择"在此处打开命令窗口"
  3. 执行pip安装命令

真实命令：

pip install requests
pip install openpyxl
pip install beautifulsoup4
pip install lxml

真实报错案例：

错误：pip is not recognized as an internal or external command
原因：pip不在系统PATH中
解决：使用完整路径，如 C:\Users\用户名\AppData\Local\影子科技\影刀\Python\Scripts\pip.exe install requests

2.2 方法二：在Python脚本中安装

可以在Python脚本中动态安装第三方库。

import subprocess
import sys

def install_package(package):
    """安装Python包"""
    subprocess.check_call([sys.executable, "-m", "pip", "install", package])

# 安装requests库
install_package("requests")

注意：这种方法需要网络连接，且每次运行都会尝试安装，不适合生产环境。

2.3 方法三：使用requirements.txt

对于需要多个第三方库的场景，可以使用requirements.txt。

操作步骤：
在这里插入图片描述

  1. 创建requirements.txt文件
  2. 列出所有需要安装的库
  3. 执行pip安装命令

requirements.txt示例：

requests==2.31.0
openpyxl==3.1.2
beautifulsoup4==4.12.2
lxml==4.9.3

安装命令：

pip install -r requirements.txt

三、常用第三方库实战

3.1 requests库：HTTP请求

requests库是Python中最流行的HTTP客户端库。

真实代码示例：

import requests

def get_taobao_product(product_id):
    """获取淘宝商品信息"""
    url = f"https://detail.tmall.com/item.htm?id={product_id}"
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    response = requests.get(url, headers=headers, timeout=10)
    
    if response.status_code == 200:
        return response.text
    else:
        raise Exception(f"请求失败，状态码：{response.status_code}")

# 使用
try:
    html = get_taobao_product("12345678")
    print("获取成功")
except Exception as e:
    print(f"错误：{e}")

3.2 openpyxl库：Excel高级操作

拼多多店群自动化上架方案

openpyxl库可以处理影刀Excel指令无法处理的复杂场景。

真实代码示例：

import openpyxl
from openpyxl.styles import Font, PatternFill

def create_styled_excel():
    """创建带样式的Excel"""
    wb = openpyxl.Workbook()
    ws = wb.active
    
    # 写入数据
    ws['A1'] = "姓名"
    ws['B1'] = "年龄"
    
    # 设置样式
    header_font = Font(bold=True, color="FFFFFF")
    header_fill = PatternFill(start_color="4F81BD", end_color="4F81BD", fill_type="solid")
    
    ws['A1'].font = header_font
    ws['A1'].fill = header_fill
    ws['B1'].font = header_font
    ws['B1'].fill = header_fill
    
    # 保存
    wb.save("styled.xlsx")

# 使用
create_styled_excel()

3.3 beautifulsoup4库：HTML解析

beautifulsoup4库可以方便地解析HTML。

在这里插入图片描述

真实代码示例：

from bs4 import BeautifulSoup
import requests

def parse_jingdong_product(html):
    """解析京东商品页面"""
    soup = BeautifulSoup(html, 'html.parser')
    
    # 提取商品标题
    title = soup.select_one('.sku-name').text.strip()
    
    # 提取商品价格
    price = soup.select_one('.p-price i').text.strip()
    
    # 提取商品评价数
    comment = soup.select_one('.comment-count a').text.strip()
    
    return {
        'title': title,
        'price': price,
        'comment': comment
    }

# 使用
html = requests.get("https://item.jd.com/123456.html").text
product = parse_jingdong_product(html)
print(product)

四、自定义模块开发

4.1 创建自定义模块

可以将常用的功能封装成自定义模块。

操作步骤：

  1. 创建Python文件，定义函数
  2. 将Python文件放到影刀Python路径中
  3. 在Python脚本中导入自定义模块

真实代码示例：

# 文件：yingdao_utils.py
def extract_numbers(text):
    """从文本中提取所有数字"""
    import re
    return re.findall(r"\d+", text)

def format_date(date_str):
    """格式化日期"""
    from datetime import datetime
    date = datetime.strptime(date_str, "%Y-%m-%d")
    return date.strftime("%Y年%m月%d日")

使用自定义模块：

# 在影刀Python脚本中
import yingdao_utils

text = "订单号123456，金额999.99"
numbers = yingdao_utils.extract_numbers(text)
print(numbers)  # 输出：['123456', '999', '99']

4.2 模块搜索路径

Python会从一个特定的路径列表中搜索模块。

查看模块搜索路径：

import sys
print(sys.path)

添加自定义模块路径：

在这里插入图片描述

import sys
sys.path.append("C:/my_modules")

4.3 包（Package）开发

对于复杂的项目，可以开发包。

包结构：

my_package/
    __init__.py
    utils.py
    api.py

真实代码示例：

# 文件：my_package/__init__.py
from .utils import extract_numbers
from .api import get_product_info

__all__ = ['extract_numbers', 'get_product_info']

五、异常处理完全攻略

5.1 基本异常处理

Python使用try-except进行异常处理。

真实代码示例：

try:
    # 可能出错的代码
    result = 10 / 0
except ZeroDivisionError as e:
    # 处理异常
    print(f"错误：{e}")
finally:
    # 无论是否出错都会执行
    print("执行完成")

5.2 常见异常类型

异常类型 说明
Exception 所有异常的基类
ValueError 值错误
TypeError 类型错误
IndexError 索引错误
KeyError 键错误
IOError 输入输出错误
requests.exceptions.RequestException requests库异常

5.3 在影刀中的异常处理

影刀的Python脚本模块支持完整的异常处理。

在这里插入图片描述

真实配置参数：

在影刀"Python脚本"指令中：

  • 出错时继续：是
  • 错误保存到变量：python_error

错误处理代码：

try:
    import requests
    response = requests.get("https://www.taobao.com", timeout=5)
    yingdao.set_variable('status_code', response.status_code)
except Exception as e:
    yingdao.set_variable('python_error', str(e))

六、与影刀变量交互高级

6.1 基本交互

已经介绍过基本的变量交互，这里深入讲解。

读取影刀变量：

# 读取字符串变量
text = yingdao.get_variable('text_var')

# 读取数字变量
number = yingdao.get_variable('number_var')

# 读取列表变量
list_var = yingdao.get_variable('list_var')

设置影刀变量：

# 设置字符串变量
yingdao.set_variable('text_var', 'hello')

# 设置数字变量
yingdao.set_variable('number_var', 123)

# 设置列表变量
yingdao.set_variable('list_var', [1, 2, 3])

6.2 复杂数据结构交互

影刀变量支持列表和字典。

真实代码示例：

# 在Python中创建复杂数据结构
product = {
    'id': '123456',
    'name': '苹果手机',
    'price': 5999.00,
    'tags': ['电子产品', '手机']
}

![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/31d7e39c5860440abb49056af637c05d.png#pic_center)

# 设置到影刀变量
yingdao.set_variable('product', product)

# 在另一个Python脚本中读取
product = yingdao.get_variable('product')
print(product['name'])  # 输出：苹果手机

6.3 变量作用域

影刀变量有作用域。

  • 流程变量：在当前流程中有效
  • 全局变量：在所有流程中有效

在Python中操作全局变量：

# 读取全局变量
global_var = yingdao.get_variable('global_var', scope='global')

# 设置全局变量
yingdao.set_variable('global_var', 'value', scope='global')

七、真实报错信息与解决方案

7.1 报错：ModuleNotFoundError: No module named ‘requests’

可能原因：

  1. 没有安装requests库
  2. 安装了但Python路径不对

解决方案：

# 安装requests库
pip install requests

# 验证安装
python -c "import requests; print(requests.__version__)"

7.2 报错：yingdao is not defined

可能原因：

  1. 代码不在影刀Python脚本中运行
  2. 影刀版本太旧

解决方案：

  1. 确保代码在影刀"Python脚本"指令中
  2. 更新影刀到最新版本

7.3 报错：can not convert dict to string

在这里插入图片描述

可能原因：

  1. 尝试将字典直接设置为影刀变量
  2. 影刀变量类型不匹配

解决方案：

# 错误的写法
yingdao.set_variable('var', {'key': 'value'})  # 可能报错

# 正确的写法：先转换为字符串，或确保影刀变量类型支持字典
import json
yingdao.set_variable('var', json.dumps({'key': 'value'}))

八、12大核心模块中的Python应用

8.1 网页自动化模块

在网页自动化模块中，可以用Python执行JavaScript。

# 在影刀Python脚本中
import requests

# 用requests获取网页源码
response = requests.get("https://www.baidu.com")
html = response.text

# 用BeautifulSoup解析
from bs4 import BeautifulSoup
soup = BeautifulSoup(html, 'html.parser')
title = soup.title.text
print(title)

8.2 Excel自动化模块

虽然影刀有Excel指令，但某些高级操作需要用Python。

import openpyxl

# 打开Excel
wb = openpyxl.load_workbook('data.xlsx')
ws = wb.active

# 遍历所有单元格，查找特定值
for row in ws.iter_rows():
    for cell in row:
        if cell.value and '关键字' in str(cell.value):
            print(f"找到：{cell.value}，位置：{cell.coordinate}")

8.3 数据库模块

影刀有数据库指令，但某些复杂查询需要用Python。

import pymysql

# 连接数据库
conn = pymysql.connect(host='localhost', user='root', password='password', database='test')
cursor = conn.cursor()

# 执行复杂查询
sql = """
SELECT o.order_id, o.amount, c.customer_name
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
WHERE o.amount > %s
"""
cursor.execute(sql, (100,))
results = cursor.fetchall()

for row in results:
    print(row)

8.4 Python脚本模块

TEMU店群如何管理运营？

在这里插入图片描述

这就是本文的主题。

8.5 图像识别模块

可以用Python调用OpenCV进行高级图像识别。

import cv2
import numpy as np

# 读取图像
img = cv2.imread('screenshot.png')

# 转换为灰度图
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

# 模板匹配
template = cv2.imread('button.png', 0)
result = cv2.matchTemplate(gray, template, cv2.TM_CCOEFF_NORMED)
min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result)

print(f"匹配度：{max_val}")

8.6 文件处理模块

可以用Python进行高级文件操作。

import os
import shutil

# 批量创建文件夹
for i in range(1, 11):
    folder_name = f"文件夹{i}"
    if not os.path.exists(folder_name):
        os.makedirs(folder_name)

# 批量复制文件
for file in os.listdir('.'):
    if file.endswith('.txt'):
        shutil.copy(file, '备份')

8.7 定时任务模块

可以用Python计算复杂的定时规则。

import schedule
import time

def job():
    print("执行任务")

# 每天10:30执行
schedule.every().day.at("10:30").do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

8.8 API对接模块

可以用Python调用复杂的API。

import requests
import hashlib
import time

def call_api(app_id, app_secret):
    """调用API，包含签名计算"""
    timestamp = int(time.time())
    nonce = 'random_string'
    
    # 计算签名
    sign_string = f"{app_id}{timestamp}{nonce}{app_secret}"
    signature = hashlib.md5(sign_string.encode()).hexdigest()
    
    # 发送请求
    url = "https://api.example.com/data"
    params = {
        'app_id': app_id,
        'timestamp': timestamp,
        'nonce': nonce,
        'signature': signature
    }
    
    response = requests.get(url, params=params)
    return response.json()

8.9 邮件模块

可以用Python发送复杂的邮件。

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.image import MIMEImage

def send_email_with_image():
    """发送带图片的邮件"""
    msg = MIMEMultipart('related')
    msg['From'] = 'sender@example.com'
    msg['To'] = 'receiver@example.com'
    msg['Subject'] = '带图片的邮件'
    
    # 邮件正文
    body = """
    <html>
      <body>
        <p>这是一张图片：</p>
        <img src="cid:image1">
        ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/cfa6b71403084b15bd92d448ccd5db6d.png#pic_center)

      </body>
    </html>
    """
    msg.attach(MIMEText(body, 'html'))
    
    # 添加图片
    with open('image.png', 'rb') as f:
        img = MIMEImage(f.read())
        img.add_header('Content-ID', '<image1>')
        msg.attach(img)
    
    # 发送
    with smtplib.SMTP('smtp.example.com', 587) as server:
        server.starttls()
        server.login('sender@example.com', 'password')
        server.send_message(msg)

8.10 钉钉模块

可以用Python调用钉钉API。

import requests

def send_dingtalk_message(webhook, message):
    """发送钉钉消息"""
    data = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }
    
    response = requests.post(webhook, json=data)
    return response.json()

8.11 飞书模块

可以用Python调用飞书API。

import requests

def send_feishu_message(app_id, app_secret, chat_id, message):
    """发送飞书消息"""
    # 获取access_token
    token_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    token_data = {
        "app_id": app_id,
        "app_secret": app_secret
    }
    token_response = requests.post(token_url, json=token_data)
    access_token = token_response.json()['tenant_access_token']
    
    # 发送消息
    msg_url = "https://open.feishu.cn/open-apis/im/v1/messages"
    headers = {
        "Authorization": f"Bearer {access_token}"
    }
    msg_data = {
        "receive_id": chat_id,
        "msg_type": "text",
        "content": f'{{"text":"{message}"}}'
    }
    response = requests.post(msg_url, headers=headers, json=msg_data)
    return response.json()

8.12 企业微信模块

可以用Python调用企业微信API。

import requests

def send_wechat_message(corpid, corpsecret, agentid, userid, message):
    """发送企业微信消息"""
    # 获取access_token
    token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&corpsecret={corpsecret}"
    token_response = requests.get(token_url)
    access_token = token_response.json()['access_token']
    
    # 发送消息
    msg_url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}"
    msg_data = {
        "touser": userid,
        "msgtype": "text",
        "agentid": agentid,
        "text": {
            "content": message
        }
    }
    response = requests.post(msg_url, json=msg_data)
    return response.json()

九、高级技巧与最佳实践

9.1 代码复用

将常用代码封装成函数或模块。

# 文件：yingdao_common.py
def retry_on_failure(func, max_retries=3, *args, **kwargs):
    """重试装饰器"""
    for i in range(max_retries):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            if i == max_retries - 1:
                raise
            print(f"第{i+1}次尝试失败，重试...")

9.2 日志记录

在Python脚本中添加日志。

import logging

# 配置日志
logging.basicConfig(
    filename='yingdao.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

# 使用日志
logging.info("流程开始")
try:
    # 业务逻辑
    logging.info("业务执行成功")
except Exception as e:
    logging.error(f"业务执行失败：{e}")

9.3 性能优化

在这里插入图片描述

对于大量数据处理，注意性能优化。

# 不好的写法：重复打开文件
for i in range(100):
    with open('data.txt', 'a') as f:
        f.write(f"行{i}\n")

# 好的写法：一次打开文件
with open('data.txt', 'a') as f:
    for i in range(100):
        f.write(f"行{i}\n")

十、真实案例：淘宝订单自动抓取

场景：每天自动抓取淘宝订单，保存到Excel。

解决方案：

import requests
import json
import openpyxl
import datetime

def fetch_taobao_orders():
    """抓取淘宝订单"""
    # 这里需要真实的API接口，以下是示例代码
    url = "https://api.taobao.com/orders"
    params = {
        'access_token': yingdao.get_variable('taobao_token'),
        'start_date': (datetime.datetime.now() - datetime.timedelta(days=1)).strftime('%Y-%m-%d'),
        'end_date': datetime.datetime.now().strftime('%Y-%m-%d')
    }
    
    response = requests.get(url, params=params)
    if response.status_code == 200:
        return response.json()['orders']
    else:
        raise Exception(f"抓取失败：{response.status_code}")

def save_to_excel(orders):
    """保存订单到Excel"""
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "订单"
    
    # 写入表头
    ws.append(['订单号', '商品名称', '金额', '状态'])
    
    # 写入数据
    for order in orders:
        ws.append([
            order['order_id'],
            order['product_name'],
            order['amount'],
            order['status']
        ])
    
    # 保存
    filename = f"订单_{datetime.datetime.now().strftime('%Y%m%d')}.xlsx"
    wb.save(filename)
    return filename

# 主流程
try:
    orders = fetch_taobao_orders()
    filename = save_to_excel(orders)
    yingdao.set_variable('result', f"成功保存{len(orders)}条订单到{filename}")
except Exception as e:
    yingdao.set_variable('result', f"失败：{e}")

十一、总结

Python协同是影刀RPA的高级功能，能让影刀调用Python的强大生态。

关键要点：

  1. 第三方库安装有多种方法
  2. 自定义模块能提高代码复用性
  3. 异常处理能保证流程稳定
  4. 与影刀变量交互要注意数据类型
  5. Python可以扩展影刀的所有核心模块

下一篇我们将讲解HTTP请求高级技巧。

内容标签：#影刀RPA #RPA教程 #Python #第三方库 #异常处理

作者：林焱

# AI助手
Logo
CSDN-OPC开发者社区

这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。

更多推荐

cover

从零构建 Java 智能体 RAG 系统：Milvus 向量数据库实战指南

avatar CSDN-OPC开发者社区
cover

千禧年难题被 AI 拿下后，OpenAI、Anthropic、xAI 为何集体讨论踩刹车

avatar CSDN-OPC开发者社区
cover

提升办公效率：OpenClaw 本地自动化 AI 工具搭建实战教程

avatar CSDN-OPC开发者社区