Python集成OpenClaw实现自动化脚本开发实战_Python

在自动化办公日益重要的今天，如何将openclaw与python深度集成，开发高效的自动化脚本？本文将从环境配置到实战案例，手把手教你入门。

一、环境准备

1.1 安装openclaw python sdk

# 使用pip安装
pip install openclaw-python
# 验证安装
python -c "import openclaw; print(openclaw.__version__)"

1.2 配置环境变量

# ~/.bashrc 或 ~/.zshrc
export openclaw_api_key="your-api-key-here"
export openclaw_base_url="https://api.openclaw.ai/v1"

1.3 初始化项目

# 创建项目目录
mkdir my_automation_project
cd my_automation_project
# 初始化openclaw配置
openclaw init

二、基础api调用

2.1 发送消息

from openclaw import client

# 创建客户端
client = client(api_key="your-api-key")

# 发送简单消息
response = client.chat.send(
    message="帮我生成一份周报模板",
    agent="writing-assistant"
)

print(response.content)

2.2 使用agent

from openclaw.agents import agent

# 初始化agent
agent = agent(
    name="data-processor",
    instructions="你是一个数据处理专家，擅长数据清洗和分析"
)

# 执行任务
result = agent.run("分析这个csv文件的销售趋势", 
                   files=["sales_data.csv"])
print(result)

三、自动化办公实战

3.1 场景一：邮件自动处理

需求：每天自动读取未读邮件，分类并生成摘要。

import imaplib
import email
from openclaw import client
from datetime import datetime

class emailprocessor:
    def __init__(self):
        self.client = client()
        self.agent = self.client.create_agent(
            name="email-classifier",
            instructions="对邮件进行分类：工作/营销/垃圾邮件"
        )
    
    def fetch_unread_emails(self):
        """获取未读邮件"""
        mail = imaplib.imap4_ssl("imap.gmail.com")
        mail.login("user@gmail.com", "password")
        mail.select("inbox")
        
        _, search_data = mail.search(none, "unseen")
        email_ids = search_data[0].split()
        
        emails = []
        for e_id in email_ids[:10]:  # 限制处理数量
            _, data = mail.fetch(e_id, "(rfc822)")
            raw_email = data[0][1]
            email_message = email.message_from_bytes(raw_email)
            
            emails.append({
                "subject": email_message["subject"],
                "from": email_message["from"],
                "body": self.get_body(email_message)
            })
        
        return emails
    
    def classify_and_summarize(self, emails):
        """分类并生成摘要"""
        summary = []
        
        for mail in emails:
            # 使用openclaw分类
            category = self.agent.run(
                f"分类这封邮件：\n主题：{mail['subject']}\n内容：{mail['body'][:500]}"
            )
            
            summary.append({
                "subject": mail["subject"],
                "category": category,
                "sender": mail["from"]
            })
        
        return summary
    
    def get_body(self, msg):
        """提取邮件正文"""
        if msg.is_multipart():
            for part in msg.walk():
                if part.get_content_type() == "text/plain":
                    return part.get_payload(decode=true).decode()
        else:
            return msg.get_payload(decode=true).decode()

# 使用示例
processor = emailprocessor()
emails = processor.fetch_unread_emails()
summary = processor.classify_and_summarize(emails)

print(f"今天收到 {len(emails)} 封未读邮件")
for item in summary:
    print(f"[{item['category']}] {item['subject']}")

3.2 场景二：excel报表自动生成

需求：每周一自动生成销售报表并发送给团队。

import pandas as pd
from openclaw import client
import schedule
import time

class reportgenerator:
    def __init__(self):
        self.client = client()
    
    def generate_weekly_report(self):
        """生成周报"""
        # 读取数据
        df = pd.read_excel("sales_data.xlsx")
        
        # 数据分析
        summary = {
            "total_sales": df["amount"].sum(),
            "total_orders": len(df),
            "avg_order_value": df["amount"].mean(),
            "top_products": df.groupby("product")["amount"].sum().nlargest(5)
        }
        
        # 使用openclaw生成分析报告
        agent = self.client.create_agent("data-analyst")
        analysis = agent.run(
            f"分析销售数据并生成周报：\n{summary}",
            output_format="markdown"
        )
        
        # 保存报告
        with open(f"周报_{datetime.now().strftime('%y%m%d')}.md", "w") as f:
            f.write(analysis)
        
        # 发送邮件
        self.send_report(analysis)
    
    def send_report(self, content):
        """发送报告"""
        self.client.email.send(
            to=["team@company.com"],
            subject=f"销售周报 - {datetime.now().strftime('%y年%m月%d日')}",
            body=content
        )

# 定时任务
reporter = reportgenerator()
schedule.every().monday.at("09:00").do(reporter.generate_weekly_report)

# 保持运行
while true:
    schedule.run_pending()
    time.sleep(60)

3.3 场景三：会议纪要自动整理

需求：将会议录音转文字后，自动提取关键信息和待办事项。

from openclaw import client
import speech_recognition as sr

class meetingassistant:
    def __init__(self):
        self.client = client()
        self.agent = self.client.create_agent(
            name="meeting-summarizer",
            instructions="整理会议纪要，提取关键决策和待办事项"
        )
    
    def transcribe_audio(self, audio_file):
        """语音转文字"""
        recognizer = sr.recognizer()
        
        with sr.audiofile(audio_file) as source:
            audio = recognizer.record(source)
        
        try:
            text = recognizer.recognize_google(audio, language="zh-cn")
            return text
        except exception as e:
            return f"转录失败: {e}"
    
    def summarize_meeting(self, transcript):
        """总结会议内容"""
        prompt = f"""
        请整理以下会议内容：
        
        1. 提取关键决策
        2. 列出待办事项（含负责人）
        3. 记录时间节点
        4. 生成简洁的会议纪要
        
        会议内容：
        {transcript}
        """
        
        summary = self.agent.run(prompt)
        return summary
    
    def process_meeting(self, audio_file):
        """处理完整流程"""
        print("正在转录音频...")
        transcript = self.transcribe_audio(audio_file)
        
        print("正在整理纪要...")
        summary = self.summarize_meeting(transcript)
        
        # 保存结果
        output_file = f"会议纪要_{datetime.now().strftime('%y%m%d')}.md"
        with open(output_file, "w") as f:
            f.write(summary)
        
        print(f"会议纪要已保存至: {output_file}")
        return summary

# 使用示例
assistant = meetingassistant()
assistant.process_meeting("meeting_recording.wav")

四、进阶技巧

4.1 错误处理与重试

from openclaw import client
from tenacity import retry, stop_after_attempt, wait_exponential

class robustautomation:
    def __init__(self):
        self.client = client()
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    def call_with_retry(self, prompt):
        """带重试机制的api调用"""
        try:
            return self.client.chat.send(prompt)
        except exception as e:
            print(f"调用失败，准备重试: {e}")
            raise

4.2 批量处理优化

from concurrent.futures import threadpoolexecutor
import openclaw

class batchprocessor:
    def __init__(self):
        self.client = openclaw.client()
    
    def process_batch(self, items, max_workers=5):
        """批量处理"""
        with threadpoolexecutor(max_workers=max_workers) as executor:
            futures = [
                executor.submit(self.process_item, item)
                for item in items
            ]
            
            results = []
            for future in futures:
                try:
                    results.append(future.result())
                except exception as e:
                    results.append({"error": str(e)})
            
            return results
    
    def process_item(self, item):
        """处理单个项目"""
        agent = self.client.create_agent("data-processor")
        return agent.run(f"处理数据: {item}")

五、最佳实践

5.1 安全建议

api密钥管理：使用环境变量，不要硬编码
输入验证：对用户输入进行过滤
日志记录：记录关键操作，便于审计

5.2 性能优化

连接池：复用http连接
缓存策略：缓存不常变化的结果
异步处理：使用asyncio提高并发

5.3 调试技巧

import logging

# 开启调试日志
logging.basicconfig(level=logging.debug)

# 使用上下文管理器追踪性能
from contextlib import contextmanager
import time

@contextmanager
def timed_execution(name):
    start = time.time()
    yield
    elapsed = time.time() - start
    print(f"{name} 耗时: {elapsed:.2f}秒")

# 使用示例
with timed_execution("数据分析"):
    result = agent.run("分析大量数据...")