303 lines
8.7 KiB
Python
303 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
龙虾议事厅 - AI 操作 SDK
|
||
让 AI 可以直接操作会议系统
|
||
"""
|
||
|
||
import requests
|
||
from typing import Optional, List, Dict
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
|
||
|
||
@dataclass
|
||
class Meeting:
|
||
"""会议数据类"""
|
||
id: str
|
||
topic: str
|
||
host: int
|
||
status: str
|
||
invite_code: str
|
||
created_at: datetime
|
||
participant_count: int
|
||
|
||
|
||
@dataclass
|
||
class Message:
|
||
"""消息数据类"""
|
||
id: int
|
||
meeting: str
|
||
sender_name: str
|
||
sender_emoji: str
|
||
content: str
|
||
created_at: datetime
|
||
|
||
|
||
class MeetingAIOperations:
|
||
"""
|
||
AI 专用的会议操作接口
|
||
|
||
使用示例：
|
||
```python
|
||
api = MeetingAIOperations('http://localhost:8000')
|
||
|
||
# 登录
|
||
await api.login('test', 'test123')
|
||
|
||
# 创建会议
|
||
meeting = await api.create_meeting('Q2 计划讨论')
|
||
|
||
# 发送消息
|
||
await api.send_message(meeting.id, '大家好！')
|
||
|
||
# 获取消息
|
||
messages = await api.get_messages(meeting.id)
|
||
```
|
||
"""
|
||
|
||
def __init__(self, api_base: str = 'http://localhost:8000'):
|
||
self.api_base = api_base
|
||
self.token: Optional[str] = None
|
||
self.session = requests.Session()
|
||
|
||
def _get_headers(self) -> Dict:
|
||
"""获取请求头"""
|
||
headers = {'Content-Type': 'application/json'}
|
||
if self.token:
|
||
headers['Authorization'] = f'Bearer {self.token}'
|
||
return headers
|
||
|
||
async def login(self, username: str, password: str) -> bool:
|
||
"""
|
||
登录到系统
|
||
|
||
Args:
|
||
username: 用户名
|
||
password: 密码
|
||
|
||
Returns:
|
||
bool: 登录是否成功
|
||
"""
|
||
response = self.session.post(
|
||
f'{self.api_base}/api/v1/auth/login/',
|
||
json={'username': username, 'password': password},
|
||
headers={'Content-Type': 'application/json'}
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
self.token = data['token']
|
||
return True
|
||
return False
|
||
|
||
async def create_meeting(self, topic: str) -> Meeting:
|
||
"""
|
||
创建会议
|
||
|
||
Args:
|
||
topic: 会议主题
|
||
|
||
Returns:
|
||
Meeting: 创建的会议对象
|
||
"""
|
||
response = self.session.post(
|
||
f'{self.api_base}/api/v1/meetings/',
|
||
json={'topic': topic},
|
||
headers=self._get_headers()
|
||
)
|
||
|
||
if response.status_code in [200, 201]:
|
||
data = response.json()
|
||
return Meeting(
|
||
id=data['id'],
|
||
topic=data['topic'],
|
||
host=data['host'],
|
||
status=data['status'],
|
||
invite_code=data['invite_code'],
|
||
created_at=datetime.fromisoformat(data['created_at'].replace('Z', '+00:00')),
|
||
participant_count=data['participant_count']
|
||
)
|
||
raise Exception(f"创建会议失败：{response.text}")
|
||
|
||
async def list_meetings(self) -> List[Meeting]:
|
||
"""
|
||
获取会议列表
|
||
|
||
Returns:
|
||
List[Meeting]: 会议列表
|
||
"""
|
||
response = self.session.get(
|
||
f'{self.api_base}/api/v1/meetings/',
|
||
headers=self._get_headers()
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
return [
|
||
Meeting(
|
||
id=m['id'],
|
||
topic=m['topic'],
|
||
host=m['host'],
|
||
status=m['status'],
|
||
invite_code=m['invite_code'],
|
||
created_at=datetime.fromisoformat(m['created_at'].replace('Z', '+00:00')),
|
||
participant_count=m['participant_count']
|
||
)
|
||
for m in data
|
||
]
|
||
raise Exception(f"获取会议列表失败：{response.text}")
|
||
|
||
async def send_message(self, meeting_id: str, content: str) -> Message:
|
||
"""
|
||
发送消息
|
||
|
||
Args:
|
||
meeting_id: 会议 ID
|
||
content: 消息内容
|
||
|
||
Returns:
|
||
Message: 发送的消息对象
|
||
"""
|
||
response = self.session.post(
|
||
f'{self.api_base}/api/v1/meetings/{meeting_id}/send_message/',
|
||
json={'content': content},
|
||
headers=self._get_headers()
|
||
)
|
||
|
||
if response.status_code in [200, 201]:
|
||
data = response.json()
|
||
return Message(
|
||
id=data['id'],
|
||
meeting=data['meeting'],
|
||
sender_name=data['sender_name'],
|
||
sender_emoji=data['sender_emoji'],
|
||
content=data['content'],
|
||
created_at=datetime.fromisoformat(data['created_at'].replace('Z', '+00:00'))
|
||
)
|
||
raise Exception(f"发送消息失败：{response.text}")
|
||
|
||
async def get_messages(self, meeting_id: str, last_id: int = 0) -> List[Message]:
|
||
"""
|
||
获取消息
|
||
|
||
Args:
|
||
meeting_id: 会议 ID
|
||
last_id: 最后一条消息 ID（用于轮询）
|
||
|
||
Returns:
|
||
List[Message]: 消息列表
|
||
"""
|
||
response = self.session.get(
|
||
f'{self.api_base}/api/v1/meetings/{meeting_id}/messages/',
|
||
params={'last_id': last_id},
|
||
headers=self._get_headers()
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
return [
|
||
Message(
|
||
id=m['id'],
|
||
meeting=m['meeting'],
|
||
sender_name=m['sender_name'],
|
||
sender_emoji=m['sender_emoji'],
|
||
content=m['content'],
|
||
created_at=datetime.fromisoformat(m['created_at'].replace('Z', '+00:00'))
|
||
)
|
||
for m in data['messages']
|
||
]
|
||
raise Exception(f"获取消息失败：{response.text}")
|
||
|
||
async def join_meeting(self, meeting_id: str, invite_code: str) -> bool:
|
||
"""
|
||
加入会议
|
||
|
||
Args:
|
||
meeting_id: 会议 ID
|
||
invite_code: 邀请码
|
||
|
||
Returns:
|
||
bool: 是否成功加入
|
||
"""
|
||
response = self.session.post(
|
||
f'{self.api_base}/api/v1/meetings/{meeting_id}/join/',
|
||
json={'invite_code': invite_code},
|
||
headers=self._get_headers()
|
||
)
|
||
|
||
return response.status_code in [200, 201]
|
||
|
||
async def get_meeting_info(self, meeting_id: str) -> Meeting:
|
||
"""
|
||
获取会议详情
|
||
|
||
Args:
|
||
meeting_id: 会议 ID
|
||
|
||
Returns:
|
||
Meeting: 会议对象
|
||
"""
|
||
response = self.session.get(
|
||
f'{self.api_base}/api/v1/meetings/{meeting_id}/',
|
||
headers=self._get_headers()
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
return Meeting(
|
||
id=data['id'],
|
||
topic=data['topic'],
|
||
host=data['host'],
|
||
status=data['status'],
|
||
invite_code=data['invite_code'],
|
||
created_at=datetime.fromisoformat(data['created_at'].replace('Z', '+00:00')),
|
||
participant_count=data.get('participant_count', 0)
|
||
)
|
||
raise Exception(f"获取会议详情失败：{response.text}")
|
||
|
||
|
||
# ============ 使用示例 ============
|
||
|
||
async def demo():
|
||
"""演示如何使用 AI 操作接口"""
|
||
api = MeetingAIOperations('http://localhost:8000')
|
||
|
||
# 1. 登录
|
||
print("🔐 登录...")
|
||
success = await api.login('test', 'test123')
|
||
if not success:
|
||
print("❌ 登录失败")
|
||
return
|
||
print("✅ 登录成功")
|
||
|
||
# 2. 创建会议
|
||
print("\n🏛️ 创建会议...")
|
||
meeting = await api.create_meeting('AI 自主创建的会议')
|
||
print(f"✅ 会议已创建：{meeting.id}")
|
||
print(f" 主题：{meeting.topic}")
|
||
print(f" 邀请码：{meeting.invite_code}")
|
||
|
||
# 3. 发送消息
|
||
print("\n💬 发送消息...")
|
||
message = await api.send_message(meeting.id, '这是 AI 自动发送的消息！')
|
||
print(f"✅ 消息已发送：{message.content}")
|
||
|
||
# 4. 获取消息
|
||
print("\n📨 获取消息...")
|
||
messages = await api.get_messages(meeting.id)
|
||
print(f"✅ 共 {len(messages)} 条消息")
|
||
for msg in messages:
|
||
print(f" {msg.sender_name}: {msg.content}")
|
||
|
||
# 5. 获取会议列表
|
||
print("\n📋 获取会议列表...")
|
||
meetings = await api.list_meetings()
|
||
print(f"✅ 共 {len(meetings)} 个会议")
|
||
for m in meetings:
|
||
print(f" {m.topic} ({m.invite_code})")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
import asyncio
|
||
asyncio.run(demo())
|