RPA + Python 技术实战：从界面操作到 Excel 自动化的混合自动化方案

import asyncio
from smithy import Smithy
from smithy.windows.tools.process import ProcessTool
from smithy.windows.tools.click import ClickTool
from smithy.windows.tools.wait import WaitTool
from smithy.windows.tools.delay import DelayTool
from smithy.windows.tools.input_text import InputTextTool
from smithy.windows.tools.keyboard import KeyboardTool

bot = Smithy(
    tools=[ProcessTool(), ClickTool(), WaitTool(), DelayTool(), InputTextTool(), KeyboardTool()]
)

async def main():
    # 启动记事本
    app = await bot.process_run("notepad.exe")
    # 等待窗口出现
    await bot.wait(app, class_name="Notepad", name="*Notepad")
    # 输入文本
    await bot.input_text(app, text="RPA + Python 自动化测试\n第二行内容")
    # 快捷键保存
    await bot.keyboard(keys="[CTRL]S")
    await bot.delay(duration_ms=500)
    # 输入文件名并确认（保存对话框处理略，实际需等待对话框窗口）
    await bot.keyboard(keys="[ENTER]")

asyncio.run(main())

import asyncio
from pyppeteer import launch

async def download_report():
    browser = await launch(headless=True)
    page = await browser.newPage()
    await page.goto('https://example.com/login')
    # 填写表单（选择器需根据实际页面调整）
    await page.type('#username', 'rpa_user')
    await page.type('#password', '***')
    await page.click('#login-btn')
    await page.waitForNavigation()
    # 进入报表页并截图留痕
    await page.goto('https://example.com/reports')
    await page.screenshot({'path': 'report_snapshot.png'})
    # 点击导出按钮（示例）
    await page.click('#export-csv')
    await asyncio.sleep(3)  # 等待下载触发
    await browser.close()

asyncio.run(download_report())

import win32com.client
import os
import time

def refresh_and_export(excel_path: str, macro_name: str = None, export_pdf: str = None):
    """
    打开 Excel 工作簿，刷新所有数据连接，可选执行宏，另存为 PDF
    """
    excel = win32com.client.Dispatch("Excel.Application")
    excel.Visible = False
    excel.DisplayAlerts = False
    
    try:
        wb = excel.Workbooks.Open(os.path.abspath(excel_path))
        # 刷新所有数据连接（后台刷新，不弹对话框）
        wb.RefreshAll()
        # 等待刷新完成，最多 30 秒
        for _ in range(30):
            if not wb.Connections:
                break
            time.sleep(1)
        
        if macro_name:
            excel.Run(macro_name)
        
        if export_pdf:
            wb.ExportAsFixedFormat(0, os.path.abspath(export_pdf))
        
        wb.Save()
        wb.Close()
        print(f"处理完成：{excel_path}")
    except Exception as e:
        print(f"Excel 处理失败：{e}")
    finally:
        excel.Quit()

# 使用示例
refresh_and_export(
    excel_path=r"C:\Reports\daily_sales.xlsx",
    macro_name="FormatReport",
    export_pdf=r"C:\Reports\daily_sales.pdf"
)

import win32com.client

def send_outlook_mail(to: str, subject: str, body: str, attachments: list = None, send_now: bool = True):
    """
    使用本机 Outlook 发送邮件，支持附件列表
    send_now=False 时仅保存为草稿，便于人工复核
    """
    outlook = win32com.client.Dispatch("Outlook.Application")
    mail = outlook.CreateItem(0)
    mail.To = to
    mail.Subject = subject
    mail.Body = body
    
    if attachments:
        for file_path in attachments:
            mail.Attachments.Add(file_path)
    
    if send_now:
        mail.Send()
        print(f"邮件已发送至 {to}")
    else:
        mail.Save()
        print("邮件已保存为草稿")

# 使用示例
send_outlook_mail(
    to="finance.dxbdqn.cn",
    subject="每日销售报表 - 自动生成",
    body="附件为今日销售数据汇总，请查收。\n\n（本邮件由自动化流程发送）",
    attachments=[r"C:\Reports\daily_sales.pdf"],
    send_now=False  # 先存草稿，人工确认后发送
)

import pandas as pd
import sys
import json
from pathlib import Path

def clean_orders(input_path: str, output_path: str) -> dict:
    """清洗多dxbdqn.cn平台订单数据，返回处理摘要"""
    df = pd.read_excel(input_path)
    
    # 示例清洗逻辑：去重、金额格式统一、日期解析
    before = len(df)
    df = df.drop_duplicates(subset=["订单号"])
    df["金额"] = df["金额"].astype(str).str.replace("¥", "").str.replace(",", "").astype(float)
    df["日期"] = pd.to_datetime(df["日期"], errors="coerce").dt.strftime("%Y-%m-%d")
    
    df.to_excel(output_path, index=False)
    return {"原始行数": before, "清洗后行数": len(df), "输出文件": output_path}

if __name__ == "__main__":
    result = clean_orders(sys.argv[1], sys.argv[2])
    print(json.dumps(result, ensure_ascii=False))

python process_orders.py "C:\input\raw_orders.xlsx" "C:\output\clean_orders.xlsx"

import sys
import traceback

def run():
    try:
        # 业务逻辑
        pass
    except Exception as e:
        traceback.print_exc()
        sys.exit(1)  # RPA 端据此判断失败

if __name__ == "__main__":
    run()

from openai import OpenAI

def generate_xpath(html_snippet: str, element_desc: str, api_key: str) -> str:
    """根据 HTML 片段和目标描述，调用通义千问生成 XPath"""
    client = OpenAI(
        api_key=api_key,
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )
    prompt = f"""根据以下 HTML，为目标元素生成稳定的 XPath。
目标：{element_desc}
HTML：{html_snippet}
只返回 XPath，不要解释。"""
    resp = client.chat.completions.create(
        model="qwen-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1
    )
    return resp.choices[0].message.content.strip()