RPA 调用 Python 脚本实战：自定义指令开发、第三方库集成、EXE 打包与离线部署

import pandas as pd

input_path = args["excel_path"]
df = pd.read_excel(input_path)

result = df[
    (df["金额"] > 1000) & (df["状态"] == "已完成")
].groupby("部门").agg({"金额": ["sum", "mean", "count"]}).reset_index()

args["output_json"] = result.to_json(orient="records", force_ascii=False)

# data_processor.py
import sys, json, pandas as pd
from datetime import datetime, timedelta

def process_sales_data(input_file, output_file, date_range_days=30):
    df = pd.read_excel(input_file)
    df = df.dropna(subset=["订单号", "金额"])
    df = df.drop_duplicates(subset=["订单号"], keep="first")
    df["订单日期"] = pd.to_datetime(df["订单日期"])
    cutoff = datetime.now() - timedelta(days=date_range_days)
    df = df[df["订单日期"] >= cutoff]
    mean_val, std_val = df["金额"].mean(), df["金额"].std()
    df["是否异常"] = abs(df["金额"] - mean_val) > 3 * std_val
    df["金额_环比"] = df["金额"].pct_change() * 100
    df.to_excel(output_file, index=False)
    return {"status": "success", "total_rows": len(df)}

if __name__ == "__main__":
    result = process_sales_data(sys.argv[1], sys.argv[2], int(sys.argv[3]) if len(sys.argv)>3 else 30)
    print(json.dumps(result, ensure_ascii=False))

# custom_commands/data_cleaner.py
import pandas as pd, numpy as np

class DataCleaner:
    @staticmethod
    def execute(config):
        df = pd.read_excel(config["input_path"])
        original = len(df)
        for op in config.get("operations", []):
            if op["type"] == "remove_duplicates":
                df = df.drop_duplicates(subset=op["params"]["subset_cols"])
            elif op["type"] == "fill_missing":
                s = op["params"].get("strategy", "mean")
                for c in df.select_dtypes(include=[np.number]).columns:
                    df[c] = df[c].fillna(df[c].mean() if s=="mean" else df[c].median())
        df.to_excel(config["output_path"], index=False)
        return {"status": "success", "original": original, "final": len(df)}

def main(args): return DataCleaner.execute(args)

import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.chart import BarChart, Reference

def create_report(data_path, output_path):
    wb = openpyxl.Workbook()
    ws = wb.active
    df = pd.read_excel(data_path)
    for col, h in enumerate(df.columns, 1):
        cell = ws.cell(row=1, column=col, value=h)
        cell.font = Font(bold=True, color="FFFFFF")
        cell.fill = PatternFill(start_color="4472C4", fill_type="solid")
    for r, row in enumerate(df.values, 2):
        for c, v in enumerate(row, 1):
            ws.cell(row=r, column=c, value=v)
    chart = BarChart()
    chart.add_data(Reference(ws, min_col=2, min_row=1, max_row=len(df)+1))
    ws.add_chart(chart, "H2")
    wb.save(output_path)

import requests
from bs4 import BeautifulSoup

def scrape(url_list):
    results = []
    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
    for url in url_list:
        resp = requests.get(url, headers=headers, timeout=15)
        soup = BeautifulSoup(resp.text, "html.parser")
        results.append({
            "title": soup.select_one("h1").get_text(strip=True) if soup.select_one("h1") else "",
            "price": soup.select_one("span.price").get_text(strip=True) if soup.select_one("span.price") else ""
        })
    return results

import pdfplumber

def extract_pdf_tables(pdf_path, output_excel):
    tables = []
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            for t in page.extract_tables():
                tables.append(pd.DataFrame(t[1:], columns=t[0]))
    if tables:
        pd.concat(tables).to_excel(output_excel, index=False)

class OrderDataProcessor:
    FIELD_MAP = {
        "订单金额": ["订单金额", "实付金额", "payment_amount", "total_amount"],
        "订单号": ["订单号", "order_id", "订单编号"],
        "下单时间": ["下单时间", "order_time", "create_time"]
    }
    RATES = {"CNY": 1.0, "USD": 7.25, "EUR": 7.85}

    @classmethod
    def standardize(cls, df, platform):
        mapping = {c: s for s, ps in cls.FIELD_MAP.items() for c in df.columns if c in ps}
        df = df.rename(columns=mapping)
        df["来源"] = platform
        return df

    @classmethod
    def parse_amount(cls, val, cur="CNY"):
        try:
            return float(str(val).replace(",","").replace("¥","")) * cls.RATES.get(cur,1)
        except: return 0

    @classmethod
    def process(cls, files):
        dfs = []
        for f in files:
            df = pd.read_excel(f["path"])
            df = cls.standardize(df, f["platform"])
            if "订单金额" in df.columns:
                df["金额_人民币"] = df["订单金额"].apply(lambda x: cls.parse_amount(x, f.get("currency","CNY")))
            dfs.append(df)
        merged = pd.concat(dfs).drop_duplicates(subset=["订单号","金额_人民币"], keep="first")
        return merged

授权策略:
  硬件指纹绑定:
    绑定方式: CPU序列号 + 主板UUID
    防破解: 授权文件与硬件指纹强关联，复制到其他机器无法运行
  功能白名单:
    基础版: [数据清洗, 格式转换]
    专业版: [基础版 + 数据分析, 图表生成]
    企业版: [全部功能]
  加密分享:
    算法: AES-256
    方式: 生成加密授权文件，接收方导入即激活

import requests, json

class AIManager:
    CONFIGS = {
        "deepseek": {"url": "https://api.deepseek.com/v1/chat/completions", "model": "deepseek-chat"},
        "kimi": {"url": "https://api.moonshot.cn/v1/chat/completions", "model": "moonshot-v1-8k"},
        "doubao": {"url": "https://ark.cn-beijing.volces.com/api/v3/chat/completions", "model": "doubao-pro-128k"},
        "wenxin": {"url": "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions", "model": "ernie-bot-4"}
    }
    def __init__(self, model, api_key):
        self.cfg = self.CONFIGS[model]
        self.key = api_key
    def chat(self, prompt, system="你是一个专业的数据处理助手"):
        r = requests.post(self.cfg["url"], headers={"Authorization": f"Bearer {self.key}"}, json={
            "model": self.cfg["model"], "messages": [{"role":"system","content":system},{"role":"user","content":prompt}]
        }, timeout=30)
        return r.json()["choices"][0]["message"]["content"]

import base64

def image_to_text(image_path, api_key):
    with open(image_path, "rb") as f:
        img = base64.b64encode(f.read()).decode()
    r = requests.post("https://api.moonshot.cn/v1/chat/completions",
        headers={"Authorization": f"Bearer {api_key}"},
        json={"model": "moonshot-v1-8k-vision-preview",
              "messages": [{"role": "user", "content": [
                  {"type": "text", "text": "提取图片中所有文字，JSON返回"},
                  {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img}"}}
              ]}]})
    return r.json()["choices"][0]["message"]["content"]

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import subprocess

def launch_browser(browser_type="ziniao", profile_id="profile_001"):
    configs = {
        "ziniao": {"exe": "C:\\Program Files\\紫鸟浏览器\\ZiniaoBrowser.exe", "port": 9222},
        "bitbrowser": {"exe": "C:\\Program Files\\比特浏览器\\BitBrowser.exe", "port": 9233},
        "hubstudio": {"exe": "C:\\Program Files\\Hubstudio\\Hubstudio.exe", "port": 9244},
        "adspower": {"api": "http://local.adspower.net:50325/api/v1/browser/start"}
    }
    cfg = configs[browser_type]
    if browser_type != "adspower":
        subprocess.Popen([cfg["exe"], f"--remote-debugging-port={cfg['port']}", f"--profile-id={profile_id}"])
        opt = Options()
        opt.add_experimental_option("debuggerAddress", f"127.0.0.1:{cfg['port']}")
        return webdriver.Chrome(options=opt)
    else:
        r = requests.get(cfg["api"], params={"user_id": profile_id}).json()
        opt = Options()
        opt.add_experimental_option("debuggerAddress", r["data"]["ws"]["selenium"])
        return webdriver.Chrome(options=opt)

def handle_dingtalk(message, cfg):
    ai = AIManager("deepseek", cfg["api_key"])
    parsed = json.loads(ai.chat(f"解析意图和参数，JSON返回：{message}"))
    # 触发本地 RPA
    result = requests.post("http://127.0.0.1:16888/api/run",
                          json={"flow": parsed["flow_name"], "params": parsed["parameters"]},
                          timeout=300).json()
    # 回调通知 IM
    reply = f"任务{'完成' if result['status']=='success' else '失败'}\n结果：{result.get('output','')}" 
    requests.post(cfg["webhook"], json={"text": reply})
    # Webhook 回调业务系统
    if cfg.get("callback_url"):
        requests.post(cfg["callback_url"], json={"task_id": result.get("id"), "status": result["status"]})
    return reply