110 lines
3.1 KiB
Python
110 lines
3.1 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
AI Agent信息流API路由模块，提供信息流的增删改查功能
|
||
"""
|
||
|
||
import logging
|
||
from fastapi import APIRouter, HTTPException, status, Query, Depends
|
||
from pydantic import BaseModel
|
||
from typing import Dict, Any, List, Optional
|
||
from datetime import datetime
|
||
|
||
from cryptoai.routes.user import get_current_user
|
||
from cryptoai.utils.db_manager import get_db_manager
|
||
|
||
# 配置日志
|
||
logger = logging.getLogger("feed_router")
|
||
|
||
# 创建路由
|
||
router = APIRouter()
|
||
|
||
# 请求模型
|
||
class AgentFeedCreate(BaseModel):
|
||
"""创建信息流请求模型"""
|
||
agent_name: str
|
||
content: str
|
||
avatar_url: Optional[str] = None
|
||
|
||
# 响应模型
|
||
class AgentFeedResponse(BaseModel):
|
||
"""信息流响应模型"""
|
||
id: int
|
||
agent_name: str
|
||
avatar_url: Optional[str] = None
|
||
content: str
|
||
create_time: datetime
|
||
|
||
@router.post("", response_model=Dict[str, Any], status_code=status.HTTP_201_CREATED)
|
||
async def create_feed(feed: AgentFeedCreate) -> Dict[str, Any]:
|
||
"""
|
||
创建新的AI Agent信息流
|
||
|
||
Args:
|
||
feed: 信息流创建请求
|
||
|
||
Returns:
|
||
创建成功的状态信息
|
||
"""
|
||
try:
|
||
# 获取数据库管理器
|
||
db_manager = get_db_manager()
|
||
|
||
# 保存信息流
|
||
success = db_manager.save_agent_feed(
|
||
agent_name=feed.agent_name,
|
||
content=feed.content,
|
||
avatar_url=feed.avatar_url
|
||
)
|
||
|
||
if not success:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="保存信息流失败"
|
||
)
|
||
|
||
return {
|
||
"status": "success",
|
||
"message": "信息流创建成功"
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"创建信息流失败: {str(e)}")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail=f"创建信息流失败: {str(e)}"
|
||
)
|
||
|
||
@router.get("", response_model=List[AgentFeedResponse])
|
||
async def get_feeds(
|
||
agent_name: Optional[str] = Query(None, description="AI Agent名称，可选"),
|
||
limit: int = Query(20, description="返回的最大记录数，默认20条"),
|
||
skip: int = Query(0, description="跳过的记录数，默认0条"),
|
||
current_user: Dict[str, Any] = Depends(get_current_user)
|
||
) -> List[AgentFeedResponse]:
|
||
"""
|
||
获取AI Agent信息流列表
|
||
|
||
Args:
|
||
agent_name: 可选，指定获取特定Agent的信息流
|
||
limit: 返回的最大记录数，默认20条
|
||
|
||
Returns:
|
||
信息流列表
|
||
"""
|
||
try:
|
||
# 获取数据库管理器
|
||
db_manager = get_db_manager()
|
||
|
||
# 获取信息流
|
||
feeds = db_manager.get_agent_feeds(agent_name=agent_name, limit=limit, skip=skip)
|
||
|
||
return feeds
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取信息流失败: {str(e)}")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail=f"获取信息流失败: {str(e)}"
|
||
) |