企业级RPA落地实战：30分钟搭建AI驱动的离线自动化流程

AI编程工具（生成脚本）
    ↓
一键转为可视化流程（无需重写）
    ↓
配置内网离线环境（数据不出本地）
    ↓
智能元素定位 + Web元素失效自动修复
    ↓
对接指纹浏览器（紫鸟/比特/HubStudio/AdsPower）
    ↓
打包为加密EXE + 授权验证
    ↓
API触发 / 定时执行 / IM工具远程控制

import pandas as pd
import time
import json
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

def export_and_process_orders():
    # 配置指纹浏览器（支持紫鸟、比特、HubStudio、AdsPower等）
    options = webdriver.ChromeOptions()
    options.add_experimental_option("debuggerAddress", "127.0.0.1:9222")
    driver = webdriver.Chrome(options=options)
    
    try:
        driver.get("https://your-backend.com/login")
        
        # 智能等待元素加载
        wait = WebDriverWait(driver, 10)
        username = wait.until(EC.presence_of_element_located((By.ID, "username")))
        username.send_keys("your_account")
        driver.find_element(By.ID, "password").send_keys("your_pass")
        driver.find_element(By.ID, "login-btn").click()
        
        # 进入订单页面
        wait.until(EC.element_to_be_clickable((By.LINK_TEXT, "订单管理"))).click()
        time.sleep(2)
        
        # 筛选昨日订单并导出
        driver.find_element(By.ID, "date-filter").click()
        driver.find_element(By.XPATH, "//option[text()='昨日']").click()
        driver.find_element(By.ID, "export-btn").click()
        
        # 等待下载完成
        time.sleep(5)
        
        # 数据清洗
        df = pd.read_csv("/downloads/yesterday_orders.csv")
        df.drop_duplicates(subset=["order_id"], inplace=True)
        df["amount"] = df["amount"].astype(float)
        df["create_time"] = pd.to_datetime(df["create_time"])
        
        # 写入本地Excel（数据全程本地处理，不出本机）
        output_path = "D:/自动化报表/昨日订单汇总.xlsx"
        df.to_excel(output_path, index=False)
        
        # 推送到企业微信（支持视觉颜色操作，无需依赖元素节点）
        send_to_wechat(output_path)
        
        print(f"处理完成，共{len(df)}条订单，文件保存至：{output_path}")
        
    except Exception as e:
        print(f"执行异常：{e}")
        # 异常时回调通知（支持钉钉、飞书、企微、个人微信）
        send_alert(f"流程异常：{str(e)}")
    finally:
        driver.quit()

def send_to_wechat(file_path):
    # 基于视觉颜色操作企业微信，无需依赖元素节点
    webhook = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
    # 文件上传逻辑...
    pass

def send_alert(msg):
    # 异常回调通知
    # 支持在钉钉、飞书、企微、个人微信内接收执行结果
    print(f"[ALERT] {msg}")

if __name__ == "__main__":
    export_and_process_orders()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
企业级订单数据自动化处理脚本
支持: 网页自动化、数据清洗、本地存储、异常重试、回调通知
"""

import os
import sys
import time
import json
import logging
from datetime import datetime, timedelta
from pathlib import Path

import pandas as pd
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, NoSuchElementException

# 配置日志（本地存储，不出本机）
log_dir = Path("D:/自动化日志")
log_dir.mkdir(exist_ok=True)
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler(log_dir / f"auto_{datetime.now().strftime('%Y%m%d')}.log", encoding='utf-8'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

class OrderAutomation:
    def __init__(self, debugger_port=9222):
        self.debugger_port = debugger_port
        self.driver = None
        self.wait = None
        self.retry_count = 3
        self.retry_delay = 2
        
    def init_browser(self):
        # 初始化浏览器，支持指纹浏览器对接
        options = webdriver.ChromeOptions()
        options.add_experimental_option("debuggerAddress", f"127.0.0.1:{self.debugger_port}")
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-dev-shm-usage")
        
        try:
            self.driver = webdriver.Chrome(options=options)
            self.wait = WebDriverWait(self.driver, 15)
            logger.info("浏览器初始化成功")
        except Exception as e:
            logger.error(f"浏览器启动失败：{e}")
            raise
    
    def safe_find_element(self, by, value, timeout=10):
        # 安全查找元素，带重试和自愈逻辑
        for attempt in range(self.retry_count):
            try:
                element = WebDriverWait(self.driver, timeout).until(
                    EC.presence_of_element_located((by, value))
                )
                return element
            except (TimeoutException, NoSuchElementException) as e:
                logger.warning(f"第{attempt+1}次查找元素失败：{value}，尝试自愈...")
                if attempt < self.retry_count - 1:
                    time.sleep(self.retry_delay)
                    # 自愈机制：基于属性片段降级匹配
                    try:
                        if by == By.ID:
                            fallback = self.driver.find_element(
                                By.XPATH, f"//*[contains(@id,'{value}')]"
                            )
                        elif by == By.CLASS_NAME:
                            fallback = self.driver.find_element(
                                By.XPATH, f"//*[contains(@class,'{value}')]"
                            )
                        elif by == By.LINK_TEXT:
                            fallback = self.driver.find_element(
                                By.XPATH, f"//*[contains(text(),'{value}')]"
                            )
                        else:
                            raise e
                        logger.info(f"元素自愈成功，降级匹配到：{value}")
                        return fallback
                    except Exception:
                        pass  # 自愈失败，进入下一轮重试
                else:
                    raise Exception(f"元素定位失败且自愈超时：{value}")
    
    def login(self, username, password):
        # 登录后台系统
        try:
            self.driver.get("https://your-backend.com/login")
            
            # 智能等待页面加载
            self.safe_find_element(By.ID, "username").send_keys(username)
            self.safe_find_element(By.ID, "password").send_keys(password)
            self.safe_find_element(By.ID, "login-btn").click()
            
            # 验证登录成功
            self.wait.until(EC.presence_of_element_located((By.CLASS_NAME, "dashboard")))
            logger.info("登录成功")
            return True
        except Exception as e:
            logger.error(f"登录失败：{e}")
            return False
    
    def export_orders(self):
        # 导出昨日订单
        try:
            yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
            
            # 进入订单管理
            self.safe_find_element(By.LINK_TEXT, "订单管理").click()
            time.sleep(2)
            
            # 设置日期筛选
            date_input = self.safe_find_element(By.ID, "date-range")
            date_input.clear()
            date_input.send_keys(f"{yesterday} ~ {yesterday}")
            
            # 点击查询
            self.safe_find_element(By.ID, "search-btn").click()
            time.sleep(3)
            
            # 导出数据
            self.safe_find_element(By.ID, "export-btn").click()
            
            # 等待下载完成（轮询检测最新csv文件，不依赖固定文件名）
            download_dir = Path("D:/Downloads")
            for _ in range(30):
                csv_files = sorted(
                    download_dir.glob("*.csv"),
                    key=lambda x: x.stat().st_mtime,
                    reverse=True
                )
                if csv_files:
                    newest = csv_files[0]
                    # 确认是60秒内的新文件，避免误取旧文件
                    if (datetime.now().timestamp() - newest.stat().st_mtime) < 60:
                        logger.info(f"文件下载成功：{newest}")
                        return str(newest)
                time.sleep(1)
            
            raise TimeoutException("文件下载超时")
            
        except Exception as e:
            logger.error(f"导出订单失败：{e}")
            raise
    
    def process_data(self, file_path):
        # 数据清洗与处理
        try:
            df = pd.read_csv(file_path, encoding='utf-8')
            logger.info(f"原始数据{len(df)}条")
            
            # 去重
            df.drop_duplicates(subset=["order_id"], keep='first', inplace=True)
            
            # 数据类型转换
            df["amount"] = pd.to_numeric(df["amount"], errors='coerce').fillna(0)
            df["create_time"] = pd.to_datetime(df["create_time"], errors='coerce')
            
            # 增加计算字段
            df["profit"] = df["amount"] * 0.15  # 假设利润率15%
            
            # 筛选有效数据
            df = df[df["status"] == "已完成"]
            
            # 保存到本地（数据不出本地）
            output_dir = Path("D:/自动化报表")
            output_dir.mkdir(exist_ok=True)
            output_file = output_dir / f"订单日报_{datetime.now().strftime('%Y%m%d')}.xlsx"
            
            with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
                df.to_excel(writer, sheet_name='订单明细', index=False)
                
                # 汇总表
                summary = pd.DataFrame({
                    '统计项': ['总订单数', '总金额', '总利润', '平均客单价'],
                    '数值': [
                        len(df),
                        df["amount"].sum(),
                        df["profit"].sum(),
                        df["amount"].mean()
                    ]
                })
                summary.to_excel(writer, sheet_name='汇总', index=False)
            
            logger.info(f"数据处理完成，输出文件：{output_file}")
            return str(output_file)
            
        except Exception as e:
            logger.error(f"数据处理失败：{e}")
            raise
    
    def send_notification(self, file_path, webhook_url):
        # 发送企业微信通知（含文件上传，失败自动降级为文本）
        try:
            # Step 1: 上传文件获取 media_id
            upload_url = webhook_url.replace("/send", "/upload_media") + "&type=file"
            with open(file_path, "rb") as f:
                files = {
                    "media": (
                        os.path.basename(file_path),
                        f,
                        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
                    )
                }
                upload_resp = requests.post(upload_url, files=files, timeout=30)
                upload_data = upload_resp.json()
            
            if upload_data.get("errcode") != 0:
                raise Exception(f"上传失败：{upload_data}")
            
            media_id = upload_data["media_id"]
            
            # Step 2: 发送文件消息
            msg = {
                "msgtype": "file",
                "file": {"media_id": media_id}
            }
            resp = requests.post(webhook_url, json=msg, timeout=10)
            logger.info(f"文件推送成功：{resp.status_code}")
            
        except Exception as e:
            logger.error(f"文件推送失败：{e}，降级为文本通知")
            # 降级：发送文本通知，确保主流程不受影响
            try:
                msg = {
                    "msgtype": "text",
                    "text": {
                        "content": (
                            f"订单日报已生成\n"
                            f"文件：{file_path}\n"
                            f"时间：{datetime.now().strftime('%H:%M:%S')}"
                        )
                    }
                }
                requests.post(webhook_url, json=msg, timeout=10)
                logger.info("已降级发送文本通知")
            except Exception as e2:
                logger.error(f"文本通知也失败：{e2}")
                # 最终兜底：仅记录本地日志，不影响主流程
    
    def send_alert(self, msg):
        # 异常回调通知（支持钉钉、飞书、企微、个人微信Webhook接入）
        alert_msg = f"[ALERT] {msg}"
        logger.error(alert_msg)
        # 实际项目中可在此扩展各平台Webhook推送
        # 钉钉: https://oapi.dingtalk.com/robot/send?access_token=xxx
        # 飞书: https://open.feishu.cn/open-apis/bot/v2/hook/xxx
        # 企微: 复用 send_notification 逻辑
    
    def run(self, config):
        # 主执行流程
        try:
            self.init_browser()
            
            if not self.login(config["username"], config["password"]):
                raise Exception("登录环节失败")
            
            csv_file = self.export_orders()
            excel_file = self.process_data(csv_file)
            
            if config.get("webhook"):
                self.send_notification(excel_file, config["webhook"])
            
            logger.info("=== 流程执行成功 ===")
            return {"status": "success", "output": excel_file}
            
        except Exception as e:
            logger.error(f"流程执行失败：{e}")
            # 异常回调通知（支持钉钉、飞书、企微、个人微信）
            self.send_alert(f"流程异常：{str(e)}")
            return {"status": "failed", "error": str(e)}
        finally:
            if self.driver:
                self.driver.quit()

# 配置文件（实际使用时可改为JSON读取或环境变量注入）
CONFIG = {
    "username": "your_account",
    "password": "your_password",
    "webhook": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key",
    "debugger_port": 9222
}

if __name__ == "__main__":
    automation = OrderAutomation(debugger_port=CONFIG["debugger_port"])
    result = automation.run(CONFIG)
    print(json.dumps(result, ensure_ascii=False, indent=2))