OpenClaw生态整合：连接Notion、飞书、钉钉的智能工作流
为什么需要生态整合？
OpenClaw作为AI助手，如果孤立运行，价值有限。通过生态整合，你可以：
– **打通信息孤岛**：连接不同平台的数据和工作流 – **自动化跨平台协作**：减少手动复制粘贴 – **智能决策支持**：基于多平台数据做出更好决策 – **提升工作效率**：一个指令触发多个平台操作
核心整合方案
1. Notion ↔ OpenClaw 双向同步
Notion作为知识管理工具，OpenClaw作为执行引擎，两者结合实现：
– **自动内容同步**：Notion中的文章自动发布到WordPress – **任务状态更新**：OpenClaw执行结果自动更新到Notion – **数据收集分析**：从Notion提取数据进行分析和报告
notion_integration.py
import requests import json from datetime import datetime
class NotionOpenClawIntegration: def __init__(self, notion_token, database_id): self.notion_token = notion_token self.database_id = database_id self.headers = { 'Authorization': f'Bearer {notion_token}', 'Notion-Version': '2022-06-28', 'Content-Type': 'application/json' }
def fetch_content_for_publishing(self): """从Notion获取待发布内容""" try: # 查询Notion数据库 query_url = f'https://api.notion.com/v1/databases/{self.database_id}/query' response = requests.post(query_url, headers=self.headers)
if response.status_code == 200: data = response.json() articles = []
for page in data.get('results', []): # 解析页面属性 properties = page.get('properties', {})
article = { 'id': page['id'], 'title': self._get_property_value(properties.get('Title', {})), 'content': self._get_page_content(page['id']), 'status': self._get_property_value(properties.get('Status', {})), 'category': self._get_property_value(properties.get('Category', {})), 'tags': self._get_property_value(properties.get('Tags', {}), is_list=True), 'created_time': page.get('created_time', '') }
# 只获取状态为"待发布"的文章 if article['status'] == '待发布': articles.append(article)
return articles else: print(f"❌ Notion查询失败: {response.status_code}") return []
except Exception as e: print(f"❌ 获取Notion内容失败: {e}") return []
def _get_property_value(self, property_data, is_list=False): """从Notion属性中提取值""" if not property_data: return [] if is_list else ''
prop_type = property_data.get('type')
if prop_type == 'title': titles = property_data.get('title', []) return titles[0].get('plain_text', '') if titles else '' elif prop_type == 'rich_text': rich_texts = property_data.get('rich_text', []) return rich_texts[0].get('plain_text', '') if rich_texts else '' elif prop_type == 'select': select_data = property_data.get('select', {}) return select_data.get('name', '') elif prop_type == 'multi_select': multi_selects = property_data.get('multi_select', []) return [item.get('name', '') for item in multi_selects] if is_list else ', '.join([item.get('name', '') for item in multi_selects]) elif prop_type == 'date': date_data = property_data.get('date', {}) return date_data.get('start', '') else: return ''
def _get_page_content(self, page_id): """获取Notion页面完整内容""" try: blocks_url = f'https://api.notion.com/v1/blocks/{page_id}/children' response = requests.get(blocks_url, headers=self.headers)
if response.status_code == 200: blocks = response.json().get('results', []) content = []
for block in blocks: block_type = block.get('type') block_data = block.get(block_type, {})
if block_type == 'paragraph': rich_text = block_data.get('rich_text', []) if rich_text: text = ''.join([rt.get('plain_text', '') for rt in rich_text]) content.append(text) elif block_type == 'heading_1': rich_text = block_data.get('rich_text', []) if rich_text: text = ''.join([rt.get('plain_text', '') for rt in rich_text]) content.append(f'# {text}') elif block_type == 'heading_2': rich_text = block_data.get('rich_text', []) if rich_text: text = ''.join([rt.get('plain_text', '') for rt in rich_text]) content.append(f'## {text}') elif block_type == 'heading_3': rich_text = block_data.get('rich_text', []) if rich_text: text = ''.join([rt.get('plain_text', '') for rt in rich_text]) content.append(f'### {text}') elif block_type == 'code': rich_text = block_data.get('rich_text', []) if rich_text: code_text = ''.join([rt.get('plain_text', '') for rt in rich_text]) language = block_data.get('language', '') content.append(f'``{language}\n{code_text}\n`')
return '\n\n'.join(content) else: print(f"❌ 获取页面内容失败: {response.status_code}") return ''
except Exception as e: print(f"❌ 获取Notion页面内容失败: {e}") return ''
def update_publishing_status(self, page_id, status='已发布', wordpress_url=''): """更新Notion中的发布状态""" try: update_url = f'https://api.notion.com/v1/pages/{page_id}'
update_data = { 'properties': { 'Status': { 'select': { 'name': status } }, 'Published_URL': { 'url': wordpress_url }, 'Published_Date': { 'date': { 'start': datetime.now().isoformat() } } } }
response = requests.patch(update_url, headers=self.headers, json=update_data)
if response.status_code == 200: print(f"✅ Notion状态更新成功: {page_id} -> {status}") return True else: print(f"❌ Notion状态更新失败: {response.status_code}") return False
except Exception as e: print(f"❌ 更新Notion状态失败: {e}") return False
使用示例：Notion到WordPress自动发布
def notion_to_wordpress_automation(): """Notion到WordPress的自动化发布流程"""
# 初始化集成 notion_integration = NotionOpenClawIntegration( notion_token='your_notion_token_here', database_id='your_database_id_here' )
# 初始化WordPress发布器 from content_automation import ContentAutomation wp_publisher = ContentAutomation( wp_url='https://www.aixianbao.cn/wp-json/wp/v2', username='laoyou', password='dQZHPam8F2DB1xAE7cHNqRJb' )
# 1. 从Notion获取待发布内容 print("📥 从Notion获取待发布内容...") articles = notion_integration.fetch_content_for_publishing()
if not articles: print("📭 Notion中没有待发布的内容") return
print(f"📄 找到 {len(articles)} 篇待发布文章")
# 2. 发布到WordPress for article in articles: print(f"🚀 发布文章: {article['title']}")
# 保存为临时Markdown文件 import tempfile import os
with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False, encoding='utf-8') as f: f.write(f"# {article['title']}\n\n") f.write(article['content']) temp_file = f.name
try: # 发布到WordPress post_id = wp_publisher.publish_article(temp_file, category_id=61)
if post_id: # 3. 更新Notion状态 wordpress_url = f"https://www.aixianbao.cn/?p={post_id}" notion_integration.update_publishing_status( page_id=article['id'], status='已发布', wordpress_url=wordpress_url )
print(f"✅ 文章发布完成: {article['title']} -> ID:{post_id}") else: print(f"❌ 文章发布失败: {article['title']}")
finally: # 清理临时文件 os.unlink(temp_file)
print("🎉 Notion到WordPress自动化发布完成")
2. 飞书（Feishu）机器人集成
飞书作为团队协作工具，OpenClaw可以作为智能助手集成：
- **自动消息通知**：任务完成、系统状态、数据报告 - **智能问答**：通过飞书机器人回答技术问题 - **工作流触发**：飞书消息触发OpenClaw任务执行
feishu_integration.py
import requests import json import hmac import hashlib import base64 import time
class FeishuOpenClawIntegration: def __init__(self, app_id, app_secret): self.app_id = app_id self.app_secret = app_secret self.access_token = None self.token_expire_time = 0
def _get_access_token(self): """获取飞书访问令牌""" current_time = time.time()
# 检查令牌是否过期（提前5分钟刷新） if self.access_token and current_time < self.token_expire_time - 300: return self.access_token
url = 'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal' data = { 'app_id': self.app_id, 'app_secret': self.app_secret }
response = requests.post(url, json=data)
if response.status_code == 200: result = response.json() if result.get('code') == 0: self.access_token = result['tenant_access_token'] self.token_expire_time = current_time + result['expire'] print("✅ 飞书访问令牌获取成功") return self.access_token else: print(f"❌ 飞书令牌获取失败: {result.get('msg')}") return None else: print(f"❌ 飞书API请求失败: {response.status_code}") return None
def send_message(self, receive_id_type='open_id', receive_id=None, msg_type='text', content=None): """发送飞书消息""" token = self._get_access_token() if not token: return False
url = 'https://open.feishu.cn/open-apis/im/v1/messages' headers = { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json' }
params = { 'receive_id_type': receive_id_type }
# 构建消息内容 if msg_type == 'text': message_content = { 'text': content } elif msg_type == 'post': message_content = { 'post': content } elif msg_type == 'image': message_content = { 'image_key': content } else: message_content = { 'text': str(content) }
data = { 'receive_id': receive_id, 'msg_type': msg_type, 'content': json.dumps(message_content, ensure_ascii=False) }
response = requests.post(url, headers=headers, params=params, json=data)
if response.status_code == 200: result = response.json() if result.get('code') == 0: print(f"✅ 飞书消息发送成功: {receive_id}") return True else: print(f"❌ 飞书消息发送失败: {result.get('msg')}") return False else: print(f"❌ 飞书API请求失败: {response.status_code}") return False
def send_article_published_notification(self, article_title, article_url, author='OpenClaw AI'): """发送文章发布通知""" # 构建富文本消息 post_content = { 'zh_cn': { 'title': '🎉 新文章发布通知', 'content': [ [{ 'tag': 'text', 'text': f'{author} 刚刚发布了新文章：\n' }], [{ 'tag': 'a', 'text': article_title, 'href': article_url }], [{ 'tag': 'text', 'text': '\n\n📊 文章已自动优化SEO并发布到网站。' }], [{ 'tag': 'text', 'text': '\n⏰ 发布时间：' }, { 'tag': 'text', 'text': time.strftime('%Y-%m-%d %H:%M:%S'), 'style': { 'bold': True } }] ] } }
# 发送到团队群组或个人 return self.send_message( receive_id_type='chat_id', receive_id='your_chat_id_here', # 团队群组ID msg_type='post', content=post_content )
def send_system_status_report(self, system_name, status, details=None): """发送系统状态报告""" status_emoji = { 'healthy': '✅', 'warning': '⚠️', 'error': '❌', 'maintenance': '🔧' }
emoji = status_emoji.get(status, '📊')
message = f"{emoji} **{system_name} 系统状态报告**\n\n" message += f"**状态**: {status}\n" message += f"**时间**: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
if details: message += "\n**详细信息**:\n" for key, value in details.items(): message += f"- {key}: {value}\n"
return self.send_message( receive_id_type='chat_id', receive_id='your_chat_id_here', msg_type='text', content=message )
使用示例：集成到自动化工作流
def integrate_feishu_notifications(): """集成飞书通知到自动化工作流"""
feishu = FeishuOpenClawIntegration( app_id='your_app_id', app_secret='your_app_secret' )
# 定义通知函数 def notify_article_published(article_title, article_url): """文章发布通知""" return feishu.send_article_published_notification(article_title, article_url)
def notify_daily_report(stats): """每日报告通知""" details = { '文章发布数': stats.get('articles_published', 0), '总字数': stats.get('words_written', 0), 'SEO评分': stats.get('seo_score', 0), '访问量': stats.get('traffic_visits', 0) }
return feishu.send_system_status_report( system_name='OpenClaw内容系统', status='healthy', details=details )
def notify_error(error_type, error_message, context=None): """错误通知""" details = { '错误类型': error_type, '错误信息': error_message }
if context: details.update(context)
return feishu.send_system_status_report( system_name='OpenClaw系统', status='error', details=details )
return { 'article_published': notify_article_published, 'daily_report': notify_daily_report, 'error': notify_error }
3. 钉钉（DingTalk）机器人集成
钉钉作为企业通讯工具，OpenClaw可以集成实现：
- **智能考勤提醒**：自动发送考勤统计 - **项目进度同步**：自动更新项目状态 - **数据报表推送**：定时发送业务数据报告
``python
dingtalk_integration.py
import requests import json import time import hashlib import base64 import hmac
class DingTalkOpenClawIntegration: def __init__(self, webhook_url, secret=None): self.webhook_url = webhook_url self.secret = secret
def _generate_signature(self, timestamp): """生成钉钉签名""" if not self.secret: return None
string_to_sign = f'{timestamp}\n{self.secret}' hmac_code = hmac.new( self.secret.encode('utf-8'), string_to_sign.encode('utf-8'), digestmod=hashlib.sha256 ).digest()
sign = base64.b64encode(hmac_code).decode('utf-8') return
暂无评论内容