# 使用pip安裝
pip install openclaw-python
# 驗(yàn)證安裝
python -c "import openclaw; print(openclaw.__version__)"

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

# 創(chuàng)建項(xiàng)目目錄
mkdir my_automation_project
cd my_automation_project
# 初始化OpenClaw配置
openclaw init

from openclaw import Client

# 創(chuàng)建客戶端
client = Client(api_key="your-api-key")

# 發(fā)送簡(jiǎn)單消息
response = client.chat.send(
    message="幫我生成一份周報(bào)模板",
    agent="writing-assistant"
)

print(response.content)

from openclaw.agents import Agent

# 初始化Agent
agent = Agent(
    name="data-processor",
    instructions="你是一個(gè)數(shù)據(jù)處理專家，擅長(zhǎng)數(shù)據(jù)清洗和分析"
)

# 執(zhí)行任務(wù)
result = agent.run("分析這個(gè)CSV文件的銷售趨勢(shì)", 
                   files=["sales_data.csv"])
print(result)

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="對(duì)郵件進(jìn)行分類：工作/營(yíng)銷/垃圾郵件"
        )
    
    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]:  # 限制處理數(shù)量
            _, 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內(nèi)容：{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']}")

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):
        """生成周報(bào)"""
        # 讀取數(shù)據(jù)
        df = pd.read_excel("sales_data.xlsx")
        
        # 數(shù)據(jù)分析
        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生成分析報(bào)告
        agent = self.client.create_agent("data-analyst")
        analysis = agent.run(
            f"分析銷售數(shù)據(jù)并生成周報(bào)：\n{summary}",
            output_format="markdown"
        )
        
        # 保存報(bào)告
        with open(f"周報(bào)_{datetime.now().strftime('%Y%m%d')}.md", "w") as f:
            f.write(analysis)
        
        # 發(fā)送郵件
        self.send_report(analysis)
    
    def send_report(self, content):
        """發(fā)送報(bào)告"""
        self.client.email.send(
            to=["team@company.com"],
            subject=f"銷售周報(bào) - {datetime.now().strftime('%Y年%m月%d日')}",
            body=content
        )

# 定時(shí)任務(wù)
reporter = ReportGenerator()
schedule.every().monday.at("09:00").do(reporter.generate_weekly_report)

# 保持運(yùn)行
while True:
    schedule.run_pending()
    time.sleep(60)

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="整理會(huì)議紀(jì)要，提取關(guān)鍵決策和待辦事項(xiàng)"
        )
    
    def transcribe_audio(self, audio_file):
        """語(yǔ)音轉(zhuǎn)文字"""
        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"轉(zhuǎn)錄失敗: {e}"
    
    def summarize_meeting(self, transcript):
        """總結(jié)會(huì)議內(nèi)容"""
        prompt = f"""
        請(qǐng)整理以下會(huì)議內(nèi)容：
        
        1. 提取關(guān)鍵決策
        2. 列出待辦事項(xiàng)（含負(fù)責(zé)人）
        3. 記錄時(shí)間節(jié)點(diǎn)
        4. 生成簡(jiǎn)潔的會(huì)議紀(jì)要
        
        會(huì)議內(nèi)容：
        {transcript}
        """
        
        summary = self.agent.run(prompt)
        return summary
    
    def process_meeting(self, audio_file):
        """處理完整流程"""
        print("正在轉(zhuǎn)錄音頻...")
        transcript = self.transcribe_audio(audio_file)
        
        print("正在整理紀(jì)要...")
        summary = self.summarize_meeting(transcript)
        
        # 保存結(jié)果
        output_file = f"會(huì)議紀(jì)要_{datetime.now().strftime('%Y%m%d')}.md"
        with open(output_file, "w") as f:
            f.write(summary)
        
        print(f"會(huì)議紀(jì)要已保存至: {output_file}")
        return summary

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

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):
        """帶重試機(jī)制的API調(diào)用"""
        try:
            return self.client.chat.send(prompt)
        except Exception as e:
            print(f"調(diào)用失敗，準(zhǔn)備重試: {e}")
            raise

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):
        """處理單個(gè)項(xiàng)目"""
        agent = self.client.create_agent("data-processor")
        return agent.run(f"處理數(shù)據(jù): {item}")

import logging

# 開啟調(diào)試日志
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} 耗時(shí): {elapsed:.2f}秒")

# 使用示例
with timed_execution("數(shù)據(jù)分析"):
    result = agent.run("分析大量數(shù)據(jù)...")