3931 lines
144 KiB
Python
3931 lines
144 KiB
Python
from fastapi import FastAPI, WebSocket, Request, WebSocketDisconnect, HTTPException, Depends, UploadFile, File, Form, Query
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.responses import JSONResponse, StreamingResponse
|
||
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
||
from typing import Dict, Set, List, Optional
|
||
from datetime import datetime, timedelta, timezone
|
||
import asyncio
|
||
import time
|
||
import jwt
|
||
from passlib.context import CryptContext
|
||
from motor.motor_asyncio import AsyncIOMotorClient
|
||
from bson import ObjectId
|
||
from pydantic import BaseModel, Field, EmailStr
|
||
from redis import asyncio as aioredis
|
||
from typing import Any
|
||
from contextlib import asynccontextmanager
|
||
from jwt.exceptions import ExpiredSignatureError, InvalidSignatureError # 修改导入
|
||
import json
|
||
import csv
|
||
from io import StringIO
|
||
from openai import OpenAI
|
||
import os
|
||
import PyPDF2
|
||
import numpy as np
|
||
import zipfile
|
||
from io import BytesIO, StringIO
|
||
import aiohttp
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from functools import partial
|
||
import threading
|
||
from fastapi import BackgroundTasks
|
||
|
||
# Database Configuration
|
||
MONGODB_URL = "mongodb://lab:[email protected]:27017/lab"
|
||
REDIS_URL = "redis://:Obscura@[email protected]:6379"
|
||
|
||
upload_path = "/obscura/task/references"
|
||
|
||
# JWT Configuration
|
||
SECRET_KEY = "Obscura@2024" # 保持这个值不变
|
||
ALGORITHM = "HS256"
|
||
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 30
|
||
|
||
# Password hashing
|
||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||
|
||
# MongoDB setup
|
||
class PyObjectId(ObjectId):
|
||
"""
|
||
自定义ObjectId类，用于在Pydantic模型中处理MongoDB的ObjectId
|
||
"""
|
||
@classmethod
|
||
def __get_validators__(cls):
|
||
yield cls.validate
|
||
|
||
@classmethod
|
||
def validate(cls, v, handler):
|
||
if not ObjectId.is_valid(v):
|
||
raise ValueError("Invalid ObjectId")
|
||
return ObjectId(v)
|
||
|
||
@classmethod
|
||
def __get_pydantic_json_schema__(cls, _schema_cache, **_kwargs):
|
||
return {
|
||
'type': 'string',
|
||
'description': 'ObjectId',
|
||
'pattern': r'^[0-9a-fA-F]{24}$'
|
||
}
|
||
|
||
@classmethod
|
||
def __modify_schema__(cls, field_schema):
|
||
field_schema.update(
|
||
type='string',
|
||
description='ObjectId',
|
||
pattern=r'^[0-9a-fA-F]{24}$'
|
||
)
|
||
|
||
# Pydantic models
|
||
class UserModel(BaseModel):
|
||
"""
|
||
用户模型
|
||
包含用户基本信息：用户名、密码、邮箱、姓名、所属机构
|
||
"""
|
||
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
||
username: str
|
||
password: str
|
||
email: EmailStr
|
||
name: str
|
||
institution: str
|
||
|
||
class Config:
|
||
populate_by_name = True
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {ObjectId: str}
|
||
|
||
class ProjectModel(BaseModel):
|
||
"""
|
||
项目模型
|
||
包含项目信息：用户ID、项目名称、创建时间、描述
|
||
"""
|
||
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
||
user_id: Optional[PyObjectId] = None
|
||
project_name: str
|
||
create_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||
description: str
|
||
|
||
class Config:
|
||
populate_by_name = True
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {ObjectId: str}
|
||
|
||
class UserDeviceModel(BaseModel):
|
||
"""用户-设备关联模型"""
|
||
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
||
user_id: PyObjectId
|
||
device_id: PyObjectId
|
||
|
||
class Config:
|
||
populate_by_name = True
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {ObjectId: str}
|
||
|
||
# Database connection
|
||
class Database:
|
||
"""数据库连接管理类"""
|
||
client: AsyncIOMotorClient = None
|
||
|
||
db = Database()
|
||
|
||
async def get_database() -> AsyncIOMotorClient:
|
||
return db.client["lab"]
|
||
|
||
async def connect_to_mongo():
|
||
try:
|
||
db.client = AsyncIOMotorClient(MONGODB_URL)
|
||
# 验证连接
|
||
await db.client.admin.command('ping')
|
||
print("Successfully connected to MongoDB")
|
||
except Exception as e:
|
||
print(f"Could not connect to MongoDB: {e}")
|
||
raise
|
||
|
||
async def close_mongo_connection():
|
||
db.client.close()
|
||
|
||
# Redis setup
|
||
async def get_redis():
|
||
redis = aioredis.from_url(
|
||
REDIS_URL,
|
||
encoding="utf-8",
|
||
decode_responses=True,
|
||
)
|
||
return redis
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
# Startup
|
||
await connect_to_mongo()
|
||
yield
|
||
# Shutdown
|
||
await close_mongo_connection()
|
||
|
||
# 更新 FastAPI 实例化
|
||
app = FastAPI(lifespan=lifespan)
|
||
|
||
# CORS configuration
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
expose_headers=["*"]
|
||
)
|
||
|
||
# Security functions
|
||
def create_access_token(data: dict):
|
||
to_encode = data.copy()
|
||
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||
to_encode.update({"exp": expire})
|
||
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||
return encoded_jwt
|
||
|
||
async def get_current_user(token: str = Depends(oauth2_scheme)):
|
||
credentials_exception = HTTPException(
|
||
status_code=401,
|
||
detail="Could not validate credentials",
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|
||
try:
|
||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||
username: str = payload.get("sub")
|
||
if username is None:
|
||
raise credentials_exception
|
||
except (ExpiredSignatureError, InvalidSignatureError) as e: # 修改异常处理
|
||
raise HTTPException(
|
||
status_code=401,
|
||
detail=str(e),
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|
||
|
||
db = await get_database()
|
||
user = await db.users.find_one({"username": username})
|
||
if user is None:
|
||
raise credentials_exception
|
||
return UserModel(**user)
|
||
|
||
# Auth routes
|
||
@app.post("/lab/register")
|
||
async def register(user_data: UserModel):
|
||
"""
|
||
用户注册接口
|
||
"""
|
||
db = await get_database()
|
||
|
||
# 检查用户名是否已存在
|
||
if await db.users.find_one({"username": user_data.username}):
|
||
raise HTTPException(status_code=400, detail="Username has already been registered")
|
||
|
||
# 创建用户文档
|
||
user_dict = user_data.model_dump(exclude={"id"})
|
||
|
||
try:
|
||
result = await db.users.insert_one(user_dict)
|
||
return {
|
||
"message": "Registration successful",
|
||
"id": str(result.inserted_id)
|
||
}
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"Database Error: {str(e)}")
|
||
|
||
@app.post("/lab/token")
|
||
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
|
||
"""
|
||
用户登录接口
|
||
"""
|
||
db = await get_database()
|
||
user = await db.users.find_one({"username": form_data.username})
|
||
|
||
# 直接比较明文密码
|
||
if not user or form_data.password != user["password"]:
|
||
raise HTTPException(status_code=400, detail="Incorrect username or password")
|
||
|
||
access_token = create_access_token(data={"sub": user["username"]})
|
||
return {"access_token": access_token, "token_type": "bearer"}
|
||
|
||
# Project routes
|
||
@app.post("/lab/projects")
|
||
async def create_project(
|
||
project: ProjectModel, # 确保请求体符合这个模型
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
db = await get_database()
|
||
|
||
project_dict = {
|
||
"user_id": current_user.id,
|
||
"project_name": project.project_name,
|
||
"description": project.description,
|
||
"create_time": datetime.now(timezone.utc)
|
||
}
|
||
|
||
result = await db.projects.insert_one(project_dict)
|
||
return {
|
||
"message": "Project created successfully",
|
||
"id": str(result.inserted_id)
|
||
}
|
||
|
||
@app.get("/lab/projects")
|
||
async def get_projects(current_user: UserModel = Depends(get_current_user)):
|
||
"""
|
||
获取当前用户所有项
|
||
|
||
参数:
|
||
current_user: 当前登录用户，通过JWT token验证获取
|
||
|
||
返回:
|
||
projects: 项目列表，每个项目包含完整的项目信息
|
||
"""
|
||
db = await get_database()
|
||
|
||
# 初化项目列表
|
||
projects = []
|
||
# 异步查询数据库，获取该用户的所有项目
|
||
async for project in db.projects.find({"user_id": current_user.id}):
|
||
projects.append(ProjectModel(**project))
|
||
|
||
return projects
|
||
|
||
# Device routes
|
||
@app.post("/lab/user/devices/{serial_number}")
|
||
async def add_device_to_user(
|
||
serial_number: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""
|
||
通过序列号添加设备到用户设备列表
|
||
|
||
参数:
|
||
serial_number: 设备序列号
|
||
current_user: 当前登录用户
|
||
返回:
|
||
设备信息和添加状态
|
||
"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 查找具有该序列号的设备
|
||
device = await db.devices.find_one({
|
||
"serial_numbers": {
|
||
"$elemMatch": {
|
||
"serial": serial_number
|
||
}
|
||
}
|
||
})
|
||
|
||
if not device:
|
||
raise HTTPException(status_code=404, detail="Device not found or invalid serial number")
|
||
# 检查该序列号是否已被其他用户激活
|
||
active_device = await db.user_devices.find_one({
|
||
"serial_number": serial_number,
|
||
"status": "active"
|
||
})
|
||
if active_device:
|
||
raise HTTPException(status_code=400, detail="Device is already in use by another user")
|
||
|
||
# Check if this serial number has already been added
|
||
existing = await db.user_devices.find_one({
|
||
"user_id": current_user.id,
|
||
"serial_number": serial_number
|
||
})
|
||
if existing:
|
||
if existing["status"] == "active":
|
||
raise HTTPException(status_code=400, detail="This device is already active")
|
||
else:
|
||
# 如果设备存在但状态是inactive，则重新激活
|
||
await db.user_devices.update_one(
|
||
{"_id": existing["_id"]},
|
||
{"$set": {"status": "active"}}
|
||
)
|
||
return {
|
||
"message": "Device reactivated successfully",
|
||
"device": {
|
||
"user_device_id": str(existing["_id"]),
|
||
"device_id": str(existing["device_id"]),
|
||
"serial_number": serial_number,
|
||
"device_name": existing["device_name"],
|
||
"device_type": existing["device_type"],
|
||
"device_number": existing["device_number"],
|
||
"sensors": existing.get("sensors", []),
|
||
"status": "active"
|
||
}
|
||
}
|
||
|
||
# 创建新的关联记录，包含完整的设备信息、序列号和传感器信息
|
||
user_device_dict = {
|
||
"user_id": current_user.id,
|
||
"device_id": device["_id"],
|
||
"serial_number": serial_number,
|
||
"device_name": device["device_name"],
|
||
"device_type": device["device_type"],
|
||
"device_number": device["device_number"],
|
||
"sensors": device.get("sensors", []), # 添加传感器信息
|
||
"status": "active", # 添加状态字段
|
||
"add_time": datetime.now(timezone.utc) # 添加时间记录
|
||
}
|
||
# 只有当序列号状态为available时才更新为in_use
|
||
await db.devices.update_one(
|
||
{
|
||
"_id": device["_id"],
|
||
"serial_numbers": {
|
||
"$elemMatch": {
|
||
"serial": serial_number,
|
||
"status": "available" # 只匹配状态为available的序列号
|
||
}
|
||
}
|
||
},
|
||
{
|
||
"$set": {
|
||
"serial_numbers.$.status": "in_use"
|
||
}
|
||
}
|
||
)
|
||
|
||
result = await db.user_devices.insert_one(user_device_dict)
|
||
|
||
# 返回设备信息
|
||
return {
|
||
"message": "Device added successfully",
|
||
"device": {
|
||
"user_device_id": str(result.inserted_id),
|
||
"device_id": str(device["_id"]),
|
||
"serial_number": serial_number,
|
||
"device_name": device["device_name"],
|
||
"device_type": device["device_type"],
|
||
"device_number": device["device_number"],
|
||
"sensors": device.get("sensors", []), # 在响应中包含传感器信息
|
||
"status": "active" # 添加状态字段"status": "active" # 添加状态字段
|
||
}
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error adding device to user: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to add device: {str(e)}")
|
||
|
||
@app.get("/lab/userdevices")
|
||
async def get_user_devices(current_user: UserModel = Depends(get_current_user)):
|
||
"""
|
||
获取当前用户的所有传感器列表
|
||
|
||
返回:
|
||
List[Dict]: 传感器列表，包含当前用户关联的所有传感器信息
|
||
"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
devices = []
|
||
async for device in db.user_devices.find({"user_id": current_user.id}):
|
||
devices.append({
|
||
"user_device_id": str(device["_id"]),
|
||
"device_id": str(device["device_id"]),
|
||
"serial_number": device["serial_number"], # 添加序列号
|
||
"device_name": device["device_name"],
|
||
"device_type": device["device_type"],
|
||
"device_number": device["device_number"],
|
||
"sensors": device.get("sensors", []), # 在响应中包含传感器信息
|
||
"status": device.get("status", "active")
|
||
})
|
||
|
||
return devices
|
||
|
||
except Exception as e:
|
||
print(f"Error getting user devices: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to get user devices: {str(e)}")
|
||
|
||
|
||
class ExperimentStatus:
|
||
ACTIVE = "active"
|
||
COMPLETED = "completed"
|
||
|
||
# 修改实验模型，添加状态字段
|
||
class ExperimentModel(BaseModel):
|
||
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
||
project_id: PyObjectId
|
||
experiment_name: str
|
||
create_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||
description: str
|
||
status: str = Field(default=ExperimentStatus.ACTIVE) # 添加状态字段
|
||
|
||
class Config:
|
||
populate_by_name = True
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {ObjectId: str}
|
||
|
||
# 添加实验模型
|
||
class ExperimentModel(BaseModel):
|
||
"""
|
||
实验模型
|
||
包含实验信息：项目ID、实验名称、创建时间、描述
|
||
"""
|
||
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
||
project_id: PyObjectId
|
||
experiment_name: str
|
||
create_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||
description: str
|
||
status: str = Field(default=ExperimentStatus.ACTIVE) # 添加状态字段
|
||
|
||
class Config:
|
||
populate_by_name = True
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {ObjectId: str}
|
||
|
||
# 首先定义一个请求模型
|
||
class ExperimentCreate(BaseModel):
|
||
"""实验创建请求模型"""
|
||
project_id: str
|
||
experiment_name: str
|
||
description: str | None = None # 修改为可选字段，默认为None
|
||
|
||
@app.post("/lab/experiments")
|
||
async def create_experiment(
|
||
experiment: ExperimentCreate,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""创建新实验"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证项目是否存在且于当前用户
|
||
project = await db.projects.find_one({
|
||
"_id": ObjectId(experiment.project_id),
|
||
"user_id": current_user.id
|
||
})
|
||
if not project:
|
||
raise HTTPException(status_code=404, detail="Project not found or unauthorized access")
|
||
|
||
# 构造实验文档
|
||
experiment_dict = {
|
||
"project_id": ObjectId(experiment.project_id),
|
||
"experiment_name": experiment.experiment_name,
|
||
"description": experiment.description or "", # 如果None则使用空字符串
|
||
"create_time": datetime.now(timezone.utc),
|
||
"status": ExperimentStatus.ACTIVE
|
||
}
|
||
|
||
|
||
result = await db.experiments.insert_one(experiment_dict)
|
||
|
||
return {
|
||
"message": "Experiment created successfully",
|
||
"id": str(result.inserted_id)
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error creating experiment: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to create experiment: {str(e)}")
|
||
|
||
@app.get("/lab/experiments")
|
||
async def get_experiments(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取项目下的所有实验"""
|
||
db = await get_database()
|
||
|
||
# 验证项目是否存在且属于当前用户
|
||
project = await db.projects.find_one({
|
||
"_id": ObjectId(project_id),
|
||
"user_id": current_user.id
|
||
})
|
||
if not project:
|
||
raise HTTPException(status_code=404, detail="Project not found or unauthorized access")
|
||
|
||
# 查询实验列表
|
||
experiments = []
|
||
async for experiment in db.experiments.find({"project_id": ObjectId(project_id)}):
|
||
experiments.append(ExperimentModel(**experiment))
|
||
|
||
return experiments
|
||
|
||
|
||
class ExperimentDeviceModel(BaseModel):
|
||
"""实验-设备关联模型"""
|
||
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
||
experiment_id: PyObjectId
|
||
user_device_id: PyObjectId
|
||
|
||
class Config:
|
||
populate_by_name = True
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {ObjectId: str}
|
||
|
||
@app.post("/lab/experiments/{experiment_id}/devices/{user_device_id}")
|
||
async def add_device_to_experiment(
|
||
experiment_id: str,
|
||
user_device_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""将设备添加到特定实验中"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证实验是否存在且属于当前用户
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
if not experiment:
|
||
raise HTTPException(status_code=404, detail="Experiment not found")
|
||
|
||
# 验证项目所有权
|
||
project = await db.projects.find_one({
|
||
"_id": experiment["project_id"],
|
||
"user_id": current_user.id
|
||
})
|
||
if not project:
|
||
raise HTTPException(status_code=403, detail="Unauthorized access to this experiment")
|
||
|
||
# 验证用户设备是否存在且状态为active
|
||
user_device = await db.user_devices.find_one({
|
||
"_id": ObjectId(user_device_id),
|
||
"user_id": current_user.id,
|
||
"status": "active" # 添加状态检查
|
||
})
|
||
if not user_device:
|
||
raise HTTPException(status_code=404, detail="Active device not found or unauthorized access")
|
||
|
||
# 检查是否已经添加过这个设备
|
||
existing = await db.experiment_devices.find_one({
|
||
"experiment_id": ObjectId(experiment_id),
|
||
"user_device_id": ObjectId(user_device_id)
|
||
})
|
||
if existing:
|
||
raise HTTPException(status_code=400, detail="This device has already been added to the experiment")
|
||
|
||
# 创建关联记录,直接复制用户设备的所有信息
|
||
experiment_device_dict = {
|
||
"experiment_id": ObjectId(experiment_id),
|
||
"user_device_id": ObjectId(user_device_id),
|
||
**{k:v for k,v in user_device.items() if k not in ['_id', 'user_id']}
|
||
}
|
||
|
||
result = await db.experiment_devices.insert_one(experiment_device_dict)
|
||
|
||
return {
|
||
"message": "Device added successfully",
|
||
"experiment_device_id": str(result.inserted_id)
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error adding device to experiment: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to add device: {str(e)}")
|
||
|
||
@app.get("/lab/experiments/{experiment_id}/devices")
|
||
async def get_experiment_devices(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取实验中的所有设备信息"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证实验是否存在且属于当前用户
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
if not experiment:
|
||
raise HTTPException(status_code=404, detail="Experiment not found")
|
||
|
||
# 验证项目所有权
|
||
project = await db.projects.find_one({
|
||
"_id": experiment["project_id"],
|
||
"user_id": current_user.id
|
||
})
|
||
if not project:
|
||
raise HTTPException(status_code=403, detail="Unauthorized access to this experiment")
|
||
|
||
# 获取实验关联的所有设备
|
||
devices = []
|
||
async for device in db.experiment_devices.find({"experiment_id": ObjectId(experiment_id)}):
|
||
# 确保返回传感器的公式信息
|
||
device_info = {
|
||
"_id": str(device["_id"]),
|
||
"experiment_id": str(device["experiment_id"]),
|
||
"user_device_id": str(device["user_device_id"]),
|
||
"device_id": str(device["device_id"]),
|
||
"serial_number": device["serial_number"],
|
||
"device_name": device["device_name"],
|
||
"device_type": device["device_type"],
|
||
"device_number": device["device_number"],
|
||
"sensors": []
|
||
}
|
||
|
||
# 保留完整的传感器信息，包括 index
|
||
for sensor in device.get("sensors", []):
|
||
sensor_info = {
|
||
"index": sensor["index"], # 保留 index
|
||
"sensor_name": sensor["sensor_name"],
|
||
"sensor_type": sensor["sensor_type"],
|
||
"unit": sensor["unit"]
|
||
}
|
||
# 如果有公式信息，也添加进去
|
||
if "formulas" in sensor:
|
||
sensor_info["formulas"] = sensor["formulas"]
|
||
|
||
device_info["sensors"].append(sensor_info)
|
||
|
||
devices.append(device_info)
|
||
|
||
return devices
|
||
|
||
except Exception as e:
|
||
print(f"Error getting experiment devices: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to get experiment devices: {str(e)}")
|
||
|
||
@app.get("/lab/experiments/{experiment_id}/devices/public")
|
||
async def get_experiment_devices_public(
|
||
experiment_id: str
|
||
):
|
||
"""获取实验中的所有设备信息(公开接口)"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证实验是否存在
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
if not experiment:
|
||
raise HTTPException(status_code=404, detail="Experiment not found")
|
||
|
||
# 获取实验关联的所有设备
|
||
devices = []
|
||
async for device in db.experiment_devices.find({"experiment_id": ObjectId(experiment_id)}):
|
||
# 确保返回传感器的公式信息
|
||
device_info = {
|
||
"_id": str(device["_id"]),
|
||
"experiment_id": str(device["experiment_id"]),
|
||
"user_device_id": str(device["user_device_id"]),
|
||
"device_id": str(device["device_id"]),
|
||
"serial_number": device["serial_number"],
|
||
"device_name": device["device_name"],
|
||
"device_type": device["device_type"],
|
||
"device_number": device["device_number"],
|
||
"sensors": []
|
||
}
|
||
|
||
# 保留完整的传感器信息，包括 index
|
||
for sensor in device.get("sensors", []):
|
||
sensor_info = {
|
||
"index": sensor["index"], # 保留 index
|
||
"sensor_name": sensor["sensor_name"],
|
||
"sensor_type": sensor["sensor_type"],
|
||
"unit": sensor["unit"]
|
||
}
|
||
# 如果有公式信息，也添加进去
|
||
if "formulas" in sensor:
|
||
sensor_info["formulas"] = sensor["formulas"]
|
||
|
||
device_info["sensors"].append(sensor_info)
|
||
|
||
devices.append(device_info)
|
||
|
||
return devices
|
||
|
||
except Exception as e:
|
||
print(f"Error getting experiment devices: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to get experiment devices: {str(e)}")
|
||
|
||
@app.delete("/lab/experiments/{experiment_id}/devices/{user_device_id}")
|
||
async def remove_device_from_experiment(
|
||
experiment_id: str,
|
||
user_device_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""
|
||
从实验中移除特定设备
|
||
|
||
参数:
|
||
experiment_id: 实验ID
|
||
user_device_id: 用户设备ID
|
||
current_user: 当前登录用户
|
||
"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证实验是否存在且属于当前用户
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
if not experiment:
|
||
raise HTTPException(status_code=404, detail="Experiment not found")
|
||
|
||
# 验证项目所有权
|
||
project = await db.projects.find_one({
|
||
"_id": experiment["project_id"],
|
||
"user_id": current_user.id
|
||
})
|
||
if not project:
|
||
raise HTTPException(status_code=403, detail="Unauthorized to delete this experiment")
|
||
|
||
# 删除实验-设备关联
|
||
result = await db.experiment_devices.delete_one({
|
||
"experiment_id": ObjectId(experiment_id),
|
||
"user_device_id": ObjectId(user_device_id)
|
||
})
|
||
|
||
if result.deleted_count == 0:
|
||
raise HTTPException(status_code=404, detail="Device association not found")
|
||
|
||
return {
|
||
"message": "Device removed from experiment",
|
||
"experiment_id": experiment_id,
|
||
"user_device_id": user_device_id
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error removing device from experiment: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to remove device: {str(e)}")
|
||
|
||
@app.delete("/lab/user/devices/{device_id}")
|
||
async def remove_user_device(
|
||
device_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""
|
||
将设备标记为非活动状态
|
||
|
||
参数:
|
||
device_id: 用户设备的ID
|
||
current_user: 当前登录用户
|
||
"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证设备是否存在且属于当前用户，且状态为active
|
||
device = await db.user_devices.find_one({
|
||
"_id": ObjectId(device_id),
|
||
"user_id": current_user.id,
|
||
"status": "active" # 只能停用活动状态的设备
|
||
})
|
||
if not device:
|
||
raise HTTPException(status_code=404, detail="Active device not found or unauthorized access")
|
||
|
||
# 更新设备状态为非活动
|
||
result = await db.user_devices.update_one(
|
||
{
|
||
"_id": ObjectId(device_id),
|
||
"user_id": current_user.id
|
||
},
|
||
{
|
||
"$set": {
|
||
"status": "inactive",
|
||
"deactivate_time": datetime.now(timezone.utc) # 记录停用时间
|
||
}
|
||
}
|
||
)
|
||
|
||
if result.modified_count == 0:
|
||
raise HTTPException(status_code=500, detail="Failed to deactivate device")
|
||
|
||
return {
|
||
"message": "Device deactivated successfully",
|
||
"device_id": device_id,
|
||
"status": "inactive", # 添加状态字段
|
||
"deactivate_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error deactivating device: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to deactivate device: {str(e)}")
|
||
|
||
# Connection Manager for WebSocket clients
|
||
class ConnectionManager:
|
||
def __init__(self):
|
||
self.active_connections: Dict[str, Dict[str, Set[WebSocket]]] = {}
|
||
self._lock = asyncio.Lock()
|
||
# 添加消息队列
|
||
self.message_queues: Dict[str, asyncio.Queue] = {}
|
||
self.broadcast_tasks: Dict[str, asyncio.Task] = {}
|
||
|
||
async def connect(self, websocket: WebSocket, experiment_id: str, serial_number: str):
|
||
"""添加新的WebSocket连接"""
|
||
async with self._lock:
|
||
if experiment_id not in self.active_connections:
|
||
self.active_connections[experiment_id] = {}
|
||
if serial_number not in self.active_connections[experiment_id]:
|
||
self.active_connections[experiment_id][serial_number] = set()
|
||
self.active_connections[experiment_id][serial_number].add(websocket)
|
||
print(f"新连接已添加到管理器 - 实验ID: {experiment_id}, 设备: {serial_number}")
|
||
|
||
async def disconnect(self, websocket: WebSocket, experiment_id: str, serial_number: str):
|
||
"""移除WebSocket连接"""
|
||
async with self._lock:
|
||
try:
|
||
if (experiment_id in self.active_connections and
|
||
serial_number in self.active_connections[experiment_id]):
|
||
self.active_connections[experiment_id][serial_number].remove(websocket)
|
||
# 只在成功移除连接时打印一次日志
|
||
print(f"WebSocket连接已断开 - 实验ID: {experiment_id}, 设备: {serial_number}")
|
||
|
||
# 清理空集合
|
||
if not self.active_connections[experiment_id][serial_number]:
|
||
del self.active_connections[experiment_id][serial_number]
|
||
if not self.active_connections[experiment_id]:
|
||
del self.active_connections[experiment_id]
|
||
except KeyError:
|
||
pass
|
||
except Exception as e:
|
||
print(f"断开连接时出错: {e}")
|
||
|
||
async def broadcast_worker(self, experiment_id: str, serial_number: str):
|
||
"""单独的广播工作器"""
|
||
queue = self.message_queues[f"{experiment_id}:{serial_number}"]
|
||
while True:
|
||
try:
|
||
message = await queue.get()
|
||
disconnected = set()
|
||
|
||
async with self._lock:
|
||
connections = self.active_connections[experiment_id][serial_number].copy()
|
||
|
||
await asyncio.gather(*[
|
||
self.send_message(ws, message, disconnected)
|
||
for ws in connections
|
||
], return_exceptions=True)
|
||
|
||
queue.task_done()
|
||
|
||
# 清理断开的连接
|
||
for websocket in disconnected:
|
||
await self.disconnect(websocket, experiment_id, serial_number)
|
||
except Exception as e:
|
||
print(f"广播工作器错误: {e}")
|
||
|
||
async def broadcast(self, message: str, experiment_id: str, serial_number: str):
|
||
"""使用消息队列进行广播"""
|
||
queue_key = f"{experiment_id}:{serial_number}"
|
||
if queue_key not in self.message_queues:
|
||
self.message_queues[queue_key] = asyncio.Queue()
|
||
self.broadcast_tasks[queue_key] = asyncio.create_task(
|
||
self.broadcast_worker(experiment_id, serial_number)
|
||
)
|
||
|
||
await self.message_queues[queue_key].put(message)
|
||
|
||
async def send_message(self, websocket: WebSocket, message: str, disconnected: set):
|
||
"""处理单个连接的消息发送"""
|
||
try:
|
||
await websocket.send_text(message)
|
||
except Exception as e:
|
||
disconnected.add(websocket)
|
||
|
||
# 创建连接管理器实例
|
||
manager = ConnectionManager()
|
||
|
||
@app.websocket("/lab/ws/{experiment_id}/{serial_number}")
|
||
async def websocket_endpoint(
|
||
websocket: WebSocket,
|
||
experiment_id: str,
|
||
serial_number: str
|
||
):
|
||
try:
|
||
await websocket.accept()
|
||
print(f"WebSocket连接已建立 - 实验ID: {experiment_id}, 设备: {serial_number}")
|
||
|
||
# 添加到连接管理器
|
||
await manager.connect(websocket, experiment_id, serial_number)
|
||
|
||
# 发送初始连接响应
|
||
response = {
|
||
"type": "connect_response",
|
||
"status": "success",
|
||
"message": "连接成功",
|
||
"experiment_id": experiment_id,
|
||
"serial_number": serial_number
|
||
}
|
||
await websocket.send_text(json.dumps(response))
|
||
|
||
try:
|
||
# 保持连接并持续接收消息
|
||
while True:
|
||
try:
|
||
data = await websocket.receive_text()
|
||
# 只处理非心跳消息
|
||
if data != "ping":
|
||
message = json.loads(data)
|
||
await manager.broadcast(data, experiment_id, serial_number)
|
||
except WebSocketDisconnect:
|
||
break
|
||
except json.JSONDecodeError:
|
||
continue # 忽略无效的JSON数据
|
||
|
||
except Exception as e:
|
||
print(f"处理WebSocket消息时出错: {e}")
|
||
|
||
finally:
|
||
await manager.disconnect(websocket, experiment_id, serial_number)
|
||
|
||
# Status endpoint
|
||
@app.get("/lab/status")
|
||
async def get_status():
|
||
return {
|
||
"status": "running",
|
||
"timestamp": time.time(),
|
||
"active_connections": len(manager.active_connections),
|
||
"last_data": manager.last_data
|
||
}
|
||
|
||
# Health check endpoint
|
||
@app.get("/health")
|
||
async def health_check():
|
||
return {"status": "healthy"}
|
||
|
||
# 添加新的数据模型
|
||
class ExperimentSession(BaseModel):
|
||
"""实会话模型，记录每次实验的开始和结束时间"""
|
||
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
||
experiment_id: PyObjectId
|
||
user_id: PyObjectId
|
||
start_time: datetime
|
||
end_time: Optional[datetime] = None
|
||
duration: Optional[float] = None # 持续时间(秒)
|
||
|
||
class Config:
|
||
populate_by_name = True
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {ObjectId: str}
|
||
|
||
class ExperimentData(BaseModel):
|
||
"""实验数据模型"""
|
||
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
||
experiment_id: PyObjectId
|
||
session_ids: List[PyObjectId] # 修改为session_ids数组
|
||
user_id: PyObjectId
|
||
device_id: str
|
||
sensor_name: str
|
||
last_update: datetime # 添加最后更新时间
|
||
|
||
class Config:
|
||
populate_by_name = True
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {ObjectId: str}
|
||
|
||
# 添加实验会话管理路由
|
||
@app.post("/lab/experiments/{experiment_id}/start")
|
||
async def start_experiment_session(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""开始新的实验会话"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 首先获取该实验关联的所有设备
|
||
experiment_devices = []
|
||
async for device in db.experiment_devices.find({"experiment_id": ObjectId(experiment_id)}):
|
||
experiment_devices.append({
|
||
"user_device_id": str(device["user_device_id"]),
|
||
"device_name": device["device_name"],
|
||
"serial_number": device["serial_number"],
|
||
"sensors": device.get("sensors", [])
|
||
})
|
||
|
||
if not experiment_devices:
|
||
raise HTTPException(status_code=400, detail="No devices added to the experiment")
|
||
|
||
# 更新实验状态到Redis (新增)
|
||
await redis.select(199)
|
||
for device in experiment_devices:
|
||
status_key = f"experiment_status:{experiment_id}:{device['serial_number']}"
|
||
await redis.set(status_key, "active")
|
||
|
||
# 创建实验会话记录
|
||
session = {
|
||
"experiment_id": ObjectId(experiment_id),
|
||
"user_id": current_user.id,
|
||
"start_time": datetime.now(timezone.utc),
|
||
"devices": experiment_devices
|
||
}
|
||
|
||
result = await db.experiment_sessions.insert_one(session)
|
||
session_id = str(result.inserted_id)
|
||
|
||
return {
|
||
"message": "Experiment session started",
|
||
"session_id": session_id,
|
||
"devices": experiment_devices
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error starting experiment session: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to start experiment session: {str(e)}")
|
||
finally:
|
||
if redis:
|
||
await redis.aclose()
|
||
|
||
|
||
@app.post("/lab/experiments/{experiment_id}/stop")
|
||
async def stop_experiment_session(
|
||
experiment_id: str,
|
||
session_data: dict,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""停止实验会话"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
session_id = session_data.get("session_id")
|
||
if not session_id:
|
||
raise HTTPException(status_code=422, detail="Missing session_id parameter")
|
||
|
||
# 验证session是否存在且未结束
|
||
session = await db.experiment_sessions.find_one({
|
||
"_id": ObjectId(session_id),
|
||
"experiment_id": ObjectId(experiment_id),
|
||
"end_time": None
|
||
})
|
||
|
||
if not session:
|
||
raise HTTPException(status_code=404, detail="Active experiment session not found")
|
||
|
||
# 更新实验状态到Redis (新增)
|
||
await redis.select(199) # 使用同一个db
|
||
for device in session.get("devices", []):
|
||
status_key = f"experiment_status:{experiment_id}:{device['serial_number']}"
|
||
await redis.set(status_key, "inactive")
|
||
|
||
# 确保start_time是带时区的
|
||
start_time = session["start_time"]
|
||
if start_time.tzinfo is None:
|
||
start_time = start_time.replace(tzinfo=timezone.utc)
|
||
|
||
end_time = datetime.now(timezone.utc) # 确保end_time带有UTC时区
|
||
duration = (end_time - start_time).total_seconds()
|
||
|
||
# 更新会话状态
|
||
result = await db.experiment_sessions.update_one(
|
||
{"_id": ObjectId(session_id)},
|
||
{
|
||
"$set": {
|
||
"end_time": end_time,
|
||
"duration": duration
|
||
}
|
||
}
|
||
)
|
||
|
||
if result.modified_count == 0:
|
||
raise HTTPException(status_code=400, detail="Failed to update session status")
|
||
|
||
return {
|
||
"message": "Experiment session stopped",
|
||
"session_id": session_id,
|
||
"duration": duration
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error stopping experiment session: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to stop experiment session: {str(e)}")
|
||
finally:
|
||
if redis:
|
||
await redis.aclose()
|
||
|
||
# 获取实验会话历史的路由
|
||
@app.get("/lab/experiments/{experiment_id}/sessions")
|
||
async def get_experiment_sessions(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取实验的所有会话记录"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
sessions = []
|
||
async for session in db.experiment_sessions.find({
|
||
"experiment_id": ObjectId(experiment_id)
|
||
}).sort("start_time", -1):
|
||
sessions.append({
|
||
"session_id": str(session["_id"]),
|
||
"start_time": session["start_time"],
|
||
"end_time": session.get("end_time"),
|
||
"duration": session.get("duration")
|
||
})
|
||
|
||
return sessions
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"Failed to get experiment session records: {str(e)}")
|
||
|
||
|
||
@app.get("/lab/experiments/{experiment_id}/export")
|
||
async def export_experiment_data(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""导出实验的所有数据为ZIP文件(包含多个CSV)"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
zip_buffer = BytesIO()
|
||
|
||
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
|
||
# 获取实验的所有会话
|
||
sessions = []
|
||
async for session in db.experiment_sessions.find({
|
||
"experiment_id": ObjectId(experiment_id)
|
||
}):
|
||
end_time = session.get("end_time") or datetime.now(timezone.utc)
|
||
duration = (end_time - session["start_time"]).total_seconds()
|
||
sessions.append({
|
||
"id": str(session["_id"]),
|
||
"start_time": session["start_time"],
|
||
"end_time": end_time,
|
||
"devices": session.get("devices", []),
|
||
"duration": duration
|
||
})
|
||
|
||
if not sessions:
|
||
raise HTTPException(status_code=404, detail="No experiment session data found")
|
||
|
||
# 写入会话信息到sessions.csv
|
||
sessions_output = StringIO()
|
||
sessions_writer = csv.writer(sessions_output)
|
||
sessions_writer.writerow(['Session ID', 'Start Time', 'End Time', 'Duration (seconds)'])
|
||
|
||
for session in sessions:
|
||
sessions_writer.writerow([
|
||
session["id"],
|
||
session["start_time"].isoformat(),
|
||
session["end_time"].isoformat() if isinstance(session["end_time"], datetime) else "Active",
|
||
session["duration"]
|
||
])
|
||
|
||
zip_file.writestr('sessions.csv', sessions_output.getvalue())
|
||
|
||
# 创建README.txt
|
||
readme_content = """Experiment Data Export Instructions:
|
||
1. sessions.csv: Contains basic information for all experiment sessions
|
||
2. Data for each device is stored in folders named by device serial number
|
||
3. Each device folder contains multiple sensor_data_XXX.csv files, each file containing sensor data for a specific time period
|
||
4. All timestamps are in UTC timezone"""
|
||
|
||
zip_file.writestr('README.txt', readme_content)
|
||
|
||
# 选择db200获取原始数据
|
||
await redis.select(200)
|
||
|
||
# 按设备处理数据
|
||
for session in sessions:
|
||
for device in session["devices"]:
|
||
serial_number = device["serial_number"]
|
||
stream_key = f"experiment:{experiment_id}:{serial_number}"
|
||
|
||
try:
|
||
start_ms = int(session["start_time"].timestamp() * 1000)
|
||
end_ms = int(session["end_time"].timestamp() * 1000)
|
||
|
||
stream_data = await redis.xrange(
|
||
stream_key,
|
||
min=str(start_ms),
|
||
max=str(end_ms)
|
||
)
|
||
|
||
# 将数据分批处理，每个文件最多包含100000行数据
|
||
MAX_ROWS_PER_FILE = 100000
|
||
current_file_rows = 0
|
||
file_counter = 1
|
||
current_output = StringIO()
|
||
current_writer = csv.writer(current_output)
|
||
current_writer.writerow(['Session ID', 'Device Time', 'Sensor', 'Value'])
|
||
|
||
for entry_id, data in stream_data:
|
||
stream_timestamp = int(entry_id.split('-')[0])
|
||
device_time = datetime.fromtimestamp(
|
||
stream_timestamp / 1000,
|
||
tz=timezone.utc
|
||
)
|
||
|
||
data_str = data[b'data'] if isinstance(data.get('data'), bytes) else data['data']
|
||
if isinstance(data_str, bytes):
|
||
data_str = data_str.decode('utf-8')
|
||
|
||
point_data = json.loads(data_str)
|
||
channel_data = point_data["channel_data"]
|
||
|
||
for sensor_name, values in channel_data.items():
|
||
if isinstance(values, list):
|
||
for value in values:
|
||
current_writer.writerow([
|
||
session["id"],
|
||
device_time.isoformat(),
|
||
sensor_name,
|
||
value
|
||
])
|
||
current_file_rows += 1
|
||
else:
|
||
current_writer.writerow([
|
||
session["id"],
|
||
device_time.isoformat(),
|
||
sensor_name,
|
||
values
|
||
])
|
||
current_file_rows += 1
|
||
|
||
# 如果当前文件达到行数限制，保存并创建新文件
|
||
if current_file_rows >= MAX_ROWS_PER_FILE:
|
||
file_name = f"{serial_number}/session_{session['id']}/sensor_data_{file_counter:03d}.csv"
|
||
zip_file.writestr(file_name, current_output.getvalue())
|
||
|
||
current_output = StringIO()
|
||
current_writer = csv.writer(current_output)
|
||
current_writer.writerow(['Session ID', 'Device Time', 'Sensor', 'Value'])
|
||
current_file_rows = 0
|
||
file_counter += 1
|
||
|
||
# 保存最后一个文件
|
||
if current_file_rows > 0:
|
||
file_name = f"{serial_number}/session_{session['id']}/sensor_data_{file_counter:03d}.csv"
|
||
zip_file.writestr(file_name, current_output.getvalue())
|
||
|
||
except Exception as e:
|
||
print(f"Error processing stream data: {str(e)}")
|
||
continue
|
||
|
||
# 获取实验信息用于文件命名
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
zip_filename = f"experiment_{experiment['experiment_name']}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip"
|
||
|
||
# 准备ZIP文<50><E69687>下载
|
||
zip_buffer.seek(0)
|
||
|
||
return StreamingResponse(
|
||
iter([zip_buffer.getvalue()]),
|
||
media_type="application/zip",
|
||
headers={
|
||
'Content-Disposition': f'attachment; filename="{zip_filename}"'
|
||
}
|
||
)
|
||
|
||
except Exception as e:
|
||
print(f"Error exporting data: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to export data: {str(e)}")
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
# 添加完成实验的路由
|
||
@app.post("/lab/experiments/{experiment_id}/complete")
|
||
async def complete_experiment(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""完成实验，将实验状态标记为已完成"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证实验是否存在
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
if not experiment:
|
||
raise HTTPException(status_code=404, detail="Experiment not found")
|
||
|
||
# 检查是否有正在进行的会话
|
||
active_session = await db.experiment_sessions.find_one({
|
||
"experiment_id": ObjectId(experiment_id),
|
||
"end_time": None
|
||
})
|
||
|
||
if active_session:
|
||
raise HTTPException(status_code=400, detail="Please stop all ongoing experiment sessions first")
|
||
|
||
# 更新实验状态为已完成
|
||
result = await db.experiments.update_one(
|
||
{"_id": ObjectId(experiment_id)},
|
||
{"$set": {"status": ExperimentStatus.COMPLETED}}
|
||
)
|
||
|
||
if result.modified_count == 0:
|
||
raise HTTPException(status_code=400, detail="Failed to update experiment status")
|
||
|
||
return {"message": "Experiment completed"}
|
||
|
||
except Exception as e:
|
||
print(f"Error completing experiment: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to complete experiment: {str(e)}")
|
||
|
||
|
||
# 修改进入实验的路由，添加状态检查
|
||
@app.get("/lab/experiments/{experiment_id}")
|
||
async def get_experiment(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取实验详情，包括状态"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
if not experiment:
|
||
raise HTTPException(status_code=404, detail="Experiment not found")
|
||
|
||
# 创建一个新的字典来存储处理后的数据
|
||
experiment_dict = {
|
||
"_id": str(experiment["_id"]),
|
||
"project_id": str(experiment["project_id"]),
|
||
"experiment_name": experiment["experiment_name"],
|
||
"create_time": experiment["create_time"],
|
||
"description": experiment["description"],
|
||
"status": experiment.get("status", ExperimentStatus.ACTIVE)
|
||
}
|
||
|
||
# 获取实验会话历史
|
||
sessions = []
|
||
async for session in db.experiment_sessions.find({"experiment_id": ObjectId(experiment_id)}):
|
||
session_dict = {
|
||
"id": str(session["_id"]),
|
||
"experiment_id": str(session["experiment_id"]),
|
||
"start_time": session["start_time"],
|
||
"end_time": session.get("end_time"),
|
||
"duration": session.get("duration")
|
||
}
|
||
sessions.append(session_dict)
|
||
|
||
experiment_dict["sessions"] = sessions
|
||
|
||
return experiment_dict
|
||
|
||
except Exception as e:
|
||
print(f"Error getting experiment: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to get experiment: {str(e)}")
|
||
|
||
# DeepSeek API Configuration
|
||
client = OpenAI(
|
||
base_url="https://api.deepseek.com/v1",
|
||
api_key="sk-3027fb3c810b4e17985fa397d41250b9"
|
||
)
|
||
# 修改函数定义为异步函数
|
||
async def analyze_experiment_data(experiment_info):
|
||
system_prompt = """
|
||
You are an AI assistant tasked with analyzing experimental data.
|
||
Generate a comprehensive experiment analysis report in JSON format.
|
||
The JSON structure must strictly follow the provided template.
|
||
"""
|
||
|
||
user_prompt = f"""Analyze the experiment data based on the following information:
|
||
Experiment data: {json.dumps(experiment_info['sessions'], ensure_ascii=False)}
|
||
|
||
Generate a JSON response with the following structure:
|
||
|
||
{{
|
||
"Experiment Analysis Report": {{
|
||
"1. Basic Information": {{
|
||
"Total Sessions": "{experiment_info['total_sessions']}",
|
||
"Total Duration": "{experiment_info['total_duration']} seconds",
|
||
"Data Points": "{experiment_info['total_points']}",
|
||
"Devices number": "{experiment_info['device_stats']['total_devices']}",
|
||
"Sensors number": "{experiment_info['device_stats']['total_sensors']}"
|
||
}},
|
||
"2. Session Data Analysis": {{
|
||
"[Session ID]": {{
|
||
"Duration": "[Duration] seconds",
|
||
"Data Points": "[data_points]"
|
||
}},
|
||
// Repeat for each session dynamically
|
||
}},
|
||
"3. Key Findings": [
|
||
"[Finding 1]",
|
||
"[Finding 2]",
|
||
"[Finding 3]"
|
||
],
|
||
"4. Recommendations": [
|
||
"[Recommendation 1]",
|
||
"[Recommendation 2]",
|
||
"[Recommendation 3]"
|
||
]
|
||
}}
|
||
}}
|
||
"""
|
||
|
||
messages = [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
]
|
||
|
||
try:
|
||
response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=messages,
|
||
response_format={'type': 'json_object'}
|
||
)
|
||
return json.loads(response.choices[0].message.content)
|
||
except Exception as e:
|
||
print(f"Error calling DeepSeek API: {e}")
|
||
return None
|
||
|
||
|
||
# 创建线程池
|
||
exp_analysis_thread_pool = ThreadPoolExecutor(max_workers=3)
|
||
pro_analysis_thread_pool = ThreadPoolExecutor(max_workers=3)
|
||
def run_experiment_in_thread(experiment_id: str):
|
||
"""在独立线程中运行分析任务"""
|
||
# 创建新的事件循环
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
|
||
try:
|
||
# 在新的事件循环中运行异步任务
|
||
loop.run_until_complete(process_experiment_analysis(experiment_id))
|
||
except Exception as e:
|
||
print(f"Error in analysis thread: {str(e)}")
|
||
finally:
|
||
try:
|
||
# 清理所有待处理的任务
|
||
pending = asyncio.all_tasks(loop)
|
||
for task in pending:
|
||
task.cancel()
|
||
# 运行直到所有任务完成
|
||
if pending:
|
||
loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
|
||
except Exception as e:
|
||
print(f"Error cleaning up tasks: {str(e)}")
|
||
finally:
|
||
loop.close()
|
||
|
||
@app.get("/lab/experiments/{experiment_id}/analyze")
|
||
async def analyze_data(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""启动实验数据分析"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
await redis.select(201)
|
||
status_key = f"experiment_analysis_status:{experiment_id}"
|
||
current_status = await redis.get(status_key)
|
||
|
||
if current_status:
|
||
status_data = json.loads(current_status)
|
||
if status_data.get("status") == "processing":
|
||
return {
|
||
"message": "实验分析任务正在进行中",
|
||
"status": "processing",
|
||
"experiment_id": experiment_id,
|
||
"start_time": status_data.get("start_time")
|
||
}
|
||
|
||
# 记录分析开始状态
|
||
status_data = {
|
||
"status": "processing",
|
||
"start_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
|
||
# 使用 loop.call_soon_threadsafe 来确保线程安全
|
||
# 创建并启动新线程
|
||
analysis_thread = threading.Thread(
|
||
target=run_experiment_in_thread,
|
||
args=(experiment_id,),
|
||
daemon=True
|
||
)
|
||
analysis_thread.start()
|
||
|
||
return {
|
||
"message": "实验分析任务已启动",
|
||
"status": "processing",
|
||
"experiment_id": experiment_id,
|
||
"start_time": status_data["start_time"]
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error starting experiment analysis: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
async def process_experiment_analysis(experiment_id: str):
|
||
"""后台处理实验分析任务"""
|
||
print(f"\n=== 开始实验分析 ===")
|
||
print(f"实验ID: {experiment_id}")
|
||
|
||
# 创建新的数据库和Redis连接
|
||
mongo_client = AsyncIOMotorClient(MONGODB_URL)
|
||
db = mongo_client["lab"]
|
||
redis = await aioredis.from_url(REDIS_URL, encoding="utf-8", decode_responses=True)
|
||
|
||
try:
|
||
await redis.select(201)
|
||
status_key = f"experiment_analysis_status:{experiment_id}"
|
||
|
||
# 更新状态为进行中
|
||
status_data = {
|
||
"status": "processing",
|
||
"start_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
|
||
# 获取实验基本信息
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
if not experiment:
|
||
raise ValueError("实验不存在")
|
||
|
||
# 收集会话数据
|
||
sessions = []
|
||
total_points = 0
|
||
total_duration = 0
|
||
devices_set = set()
|
||
sensors_count = 0
|
||
|
||
async for session in db.experiment_sessions.find({
|
||
"experiment_id": ObjectId(experiment_id)
|
||
}):
|
||
# 计算会话持续时间
|
||
end_time = session.get("end_time") or datetime.now(timezone.utc)
|
||
duration = (end_time - session["start_time"]).total_seconds()
|
||
total_duration += duration
|
||
|
||
# 转换时间戳为毫秒
|
||
start_ms = int(session["start_time"].timestamp() * 1000)
|
||
end_ms = int(end_time.timestamp() * 1000)
|
||
|
||
# 统计设备和传感器数量
|
||
session_devices = session.get("devices", [])
|
||
for device in session_devices:
|
||
devices_set.add(device["serial_number"])
|
||
sensors_count += len(device.get("sensors", []))
|
||
|
||
# 获取会话数据点数
|
||
session_points = 0
|
||
for device in session_devices:
|
||
stream_key = f"experiment:{experiment_id}:{device['serial_number']}"
|
||
await redis.select(200)
|
||
data_points = await redis.xrange(
|
||
stream_key,
|
||
min=str(start_ms),
|
||
max=str(end_ms)
|
||
)
|
||
session_points += len(data_points)
|
||
|
||
total_points += session_points
|
||
sessions.append({
|
||
"session_id": str(session["_id"]),
|
||
"duration": duration,
|
||
"data_points": session_points
|
||
})
|
||
|
||
if not sessions:
|
||
raise ValueError("没有找到实验会话数据")
|
||
|
||
experiment_info = {
|
||
"experiment_name": experiment["experiment_name"],
|
||
"total_sessions": len(sessions),
|
||
"total_duration": total_duration,
|
||
"total_points": total_points,
|
||
"device_stats": {
|
||
"total_devices": len(devices_set),
|
||
"total_sensors": sensors_count
|
||
},
|
||
"sessions": sessions
|
||
}
|
||
|
||
# 执行分析
|
||
analysis_result = await analyze_experiment_data(experiment_info)
|
||
|
||
if analysis_result:
|
||
# 保存分析结果
|
||
await redis.select(201)
|
||
report_key = f"experiment_report:{experiment_id}"
|
||
await redis.set(report_key, json.dumps(analysis_result))
|
||
|
||
# 更新状态为完成
|
||
status_data = {
|
||
"status": "completed",
|
||
"completion_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
print("分析报告已保存")
|
||
else:
|
||
print("分析失败")
|
||
status_data = {
|
||
"status": "failed",
|
||
"error": "Failed to generate analysis result",
|
||
"completion_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
|
||
except Exception as e:
|
||
print(f"Error in experiment analysis: {str(e)}")
|
||
try:
|
||
await redis.select(201)
|
||
status_key = f"experiment_analysis_status:{experiment_id}"
|
||
status_data = {
|
||
"status": "failed",
|
||
"error": str(e),
|
||
"completion_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
except Exception as redis_error:
|
||
print(f"Error updating Redis status: {redis_error}")
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
mongo_client.close()
|
||
except Exception as e:
|
||
print(f"Error closing connections: {e}")
|
||
|
||
@app.get("/lab/experiments/{experiment_id}/analysis_status")
|
||
async def get_experiment_analysis_status(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取实验分析任务的状态"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
await redis.select(201)
|
||
status_key = f"experiment_analysis_status:{experiment_id}"
|
||
status_data = await redis.get(status_key)
|
||
|
||
if not status_data:
|
||
return {
|
||
"status": "not_started",
|
||
"experiment_id": experiment_id
|
||
}
|
||
|
||
return json.loads(status_data)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting analysis status: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.get("/lab/experiments/{experiment_id}/report")
|
||
async def get_saved_report(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""从 Redis db201 读取已保存的实验报告"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 选择 db201
|
||
await redis.select(201)
|
||
report_key = f"experiment_report:{experiment_id}"
|
||
|
||
# 获取已保存的报告
|
||
existing_report = await redis.get(report_key)
|
||
|
||
if not existing_report:
|
||
raise HTTPException(status_code=404, detail="No saved experiment report found")
|
||
|
||
# 返回报告
|
||
return json.loads(existing_report)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting experiment report: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
|
||
# 修改函数定义为异步函数
|
||
async def analyze_project_data(project_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||
system_prompt = """
|
||
You are an AI assistant responsible for analyzing lab reports.
|
||
You will summarize and analyze all lab reports and generate a comprehensive analysis report in JSON format.
|
||
The JSON structure must strictly follow the provided template.
|
||
"""
|
||
|
||
user_prompt = f"""The project report is summarized based on the following experimental analysis reports:
|
||
Experimental analysis reports: {json.dumps(project_data, ensure_ascii=False)}
|
||
|
||
Generate a JSON response with the following structure:
|
||
|
||
{{
|
||
"Project Analysis Report": {{
|
||
"1. Project Overview": {{
|
||
"Project Name": "{project_data['project_stats']['project_name']}",
|
||
"Total Experiments": "{project_data['project_stats']['total_experiments']}",
|
||
"Total Data Points": "{project_data['project_stats']['total_data_points']}",
|
||
}},
|
||
"2. Aggregated Statistics": {{
|
||
"Total Sessions": {project_data['project_stats']['total_sessions']},
|
||
"Total Duration": {project_data['project_stats']['total_duration']},
|
||
"Total Data Points": {project_data['project_stats']['total_data_points']},
|
||
"Average Session Duration": {project_data['project_stats']['avg_session_duration']},
|
||
"Average Data Points per Session": {project_data['project_stats']['avg_data_points_per_session']}
|
||
}},
|
||
"3. Performance Analysis": {{
|
||
"Best Performing Sessions": [
|
||
{{
|
||
"Session ID": "session_id",
|
||
"experiment_id": "experiment_id",
|
||
"experiment_name": "experiment_name",
|
||
"Duration": "duration",
|
||
"Data Points": "data_points",
|
||
"Success Factors": "success_factors"
|
||
}}
|
||
],
|
||
"Problematic Sessions": [
|
||
{{
|
||
"Session ID": "session_id",
|
||
"experiment_id": "experiment_id",
|
||
"experiment_name": "experiment_name",
|
||
"Issues": "specific_issues",
|
||
"Possible Causes": "possible_causes"
|
||
}}
|
||
]
|
||
}},
|
||
"4. Common Findings": {{
|
||
"Recurring Issues": "common_failure_modes",
|
||
"Equipment performance": "equipment_performance",
|
||
"Sensor reliability analysis": "sensor_reliability_analysis"
|
||
}},
|
||
"5. Recommendations": {{
|
||
"Equipment optimization suggestions": "equipment_optimization_suggestions",
|
||
"Experiment process improvement suggestions": "experiment_process_improvement_suggestions",
|
||
"Data collection strategy adjustment": "data_collection_strategy_adjustment"
|
||
}}
|
||
}}
|
||
}}
|
||
"""
|
||
|
||
messages = [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
]
|
||
|
||
try:
|
||
response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=messages,
|
||
response_format={'type': 'json_object'}
|
||
)
|
||
return json.loads(response.choices[0].message.content)
|
||
except Exception as e:
|
||
print(f"Error calling DeepSeek API: {e}")
|
||
return None
|
||
# 项目分析的异步实现类似：
|
||
@app.get("/lab/projects/{project_id}/analyze")
|
||
async def analyze_project_data_endpoint(
|
||
project_id: str,
|
||
force: bool = False,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""启动项目数据分析"""
|
||
print(f"\n=== analyze_project_data_endpoint 开始 ===")
|
||
print(f"项目ID: {project_id}")
|
||
print(f"Force: {force}")
|
||
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 检查是否已经有正在进行的分析任务
|
||
await redis.select(202)
|
||
status_key = f"project_analysis_status:{project_id}"
|
||
current_status = await redis.get(status_key)
|
||
|
||
if current_status and not force:
|
||
status_data = json.loads(current_status)
|
||
if status_data.get("status") == "processing":
|
||
print("已有分析任务正在进行中")
|
||
return {
|
||
"message": "项目分析任务正在进行中",
|
||
"status": "processing",
|
||
"project_id": project_id,
|
||
"start_time": status_data.get("start_time")
|
||
}
|
||
|
||
# 记录分析开始状态
|
||
status_data = {
|
||
"status": "processing",
|
||
"start_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
|
||
print("启动分析线程...")
|
||
# 创建并启动新线程
|
||
import threading
|
||
analysis_thread = threading.Thread(
|
||
target=run_project_in_thread,
|
||
args=(project_id,),
|
||
daemon=True
|
||
)
|
||
analysis_thread.start()
|
||
print("分析线程已启动")
|
||
|
||
return {
|
||
"message": "项目分析任务已启动",
|
||
"status": "processing",
|
||
"project_id": project_id,
|
||
"start_time": status_data["start_time"]
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error starting project analysis: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
def run_project_in_thread(project_id: str):
|
||
"""在独立线程中运行分析任务"""
|
||
# 创建新的事件循环
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
|
||
try:
|
||
# 在新的事件循环中运行异步任务
|
||
loop.run_until_complete(process_project_analysis(project_id))
|
||
except Exception as e:
|
||
print(f"Error in analysis thread: {str(e)}")
|
||
finally:
|
||
loop.close()
|
||
|
||
async def process_project_analysis(project_id: str):
|
||
"""后台处理项目分析任务"""
|
||
print(f"\n=== 开始项目分析 ===")
|
||
print(f"项目ID: {project_id}")
|
||
|
||
# 创建新的数据库和Redis连接
|
||
mongo_client = AsyncIOMotorClient(MONGODB_URL)
|
||
db = mongo_client["lab"]
|
||
redis = await aioredis.from_url(REDIS_URL, encoding="utf-8", decode_responses=True)
|
||
|
||
try:
|
||
await redis.select(202)
|
||
status_key = f"project_analysis_status:{project_id}"
|
||
|
||
# 更新状态为进行中
|
||
status_data = {
|
||
"status": "processing",
|
||
"start_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
|
||
# 获取项目名称
|
||
project = await db.projects.find_one({"_id": ObjectId(project_id)})
|
||
project_name = project["project_name"] if project else "Unknown Project"
|
||
|
||
# 从MongoDB查找项目下的所有实验
|
||
experiment_cursor = db.experiments.find({
|
||
"project_id": ObjectId(project_id)
|
||
})
|
||
|
||
experiment_ids = []
|
||
experiment_names = {} # 存储实验ID和名称的映射
|
||
async for experiment in experiment_cursor:
|
||
exp_id = str(experiment["_id"])
|
||
experiment_ids.append(exp_id)
|
||
experiment_names[exp_id] = experiment["experiment_name"]
|
||
|
||
if not experiment_ids:
|
||
print("错误: 未找到任何实验")
|
||
raise Exception("No experiments found in this project")
|
||
|
||
# 从Redis db201获取所有实验报告
|
||
await redis.select(201)
|
||
all_experiment_reports = {}
|
||
|
||
# 统计数据初始化
|
||
total_sessions = 0
|
||
total_duration = 0
|
||
total_data_points = 0
|
||
total_devices = set()
|
||
total_sensors = 0
|
||
|
||
for exp_id in experiment_ids:
|
||
report_key = f"experiment_report:{exp_id}"
|
||
report_data = await redis.get(report_key)
|
||
|
||
if report_data:
|
||
try:
|
||
parsed_report = json.loads(report_data)
|
||
all_experiment_reports[exp_id] = parsed_report
|
||
|
||
# 从Basic Information中提取数据
|
||
basic_info = parsed_report["Experiment Analysis Report"]["1. Basic Information"]
|
||
total_sessions += int(basic_info["Total Sessions"])
|
||
total_duration += float(basic_info["Total Duration"].split()[0]) # 去掉"seconds"
|
||
total_data_points += int(basic_info["Data Points"])
|
||
total_devices.update([str(i) for i in range(int(basic_info["Devices number"]))])
|
||
total_sensors += int(basic_info["Sensors number"])
|
||
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
continue
|
||
|
||
if not all_experiment_reports:
|
||
raise Exception("No experiment reports found")
|
||
|
||
# 计算平均值
|
||
avg_session_duration = total_duration / total_sessions if total_sessions > 0 else 0
|
||
avg_data_points_per_session = total_data_points / total_sessions if total_sessions > 0 else 0
|
||
|
||
|
||
# 添加项目级统计数据
|
||
project_stats = {
|
||
"project_name": project_name,
|
||
"total_experiments": len(experiment_ids),
|
||
"total_sessions": total_sessions,
|
||
"total_duration": total_duration,
|
||
"total_data_points": total_data_points,
|
||
"total_devices": len(total_devices),
|
||
"total_sensors": total_sensors,
|
||
"avg_session_duration": avg_session_duration,
|
||
"avg_data_points_per_session": avg_data_points_per_session,
|
||
"experiment_names": experiment_names
|
||
}
|
||
|
||
# 将统计数据和原始报告一起发送给分析函数
|
||
project_data = {
|
||
"project_stats": project_stats,
|
||
"experiment_reports": all_experiment_reports
|
||
}
|
||
|
||
# 执行分析
|
||
analysis_result = await analyze_project_data(project_data)
|
||
|
||
if analysis_result:
|
||
# 保存分析结果
|
||
await redis.select(202)
|
||
report_key = f"project_report:{project_id}"
|
||
await redis.set(report_key, json.dumps(analysis_result))
|
||
|
||
# 更新状态为完成
|
||
status_key = f"project_analysis_status:{project_id}"
|
||
status_data = {
|
||
"status": "completed",
|
||
"completion_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
print("分析报告已保存")
|
||
else:
|
||
print("\n7. 分析失败")
|
||
# 更新失败状态
|
||
status_data = {
|
||
"status": "failed",
|
||
"error": "Failed to generate analysis result",
|
||
"completion_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
|
||
except Exception as e:
|
||
try:
|
||
await redis.select(202)
|
||
status_key = f"project_analysis_status:{project_id}"
|
||
status_data = {
|
||
"status": "failed",
|
||
"error": str(e),
|
||
"completion_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
except Exception as redis_error:
|
||
print(f"Error updating Redis status: {redis_error}")
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
mongo_client.close()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
@app.get("/lab/projects/{project_id}/analysis_status")
|
||
async def get_project_analysis_status(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取项目分析任务的状态"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
await redis.select(202)
|
||
status_key = f"project_analysis_status:{project_id}"
|
||
status_data = await redis.get(status_key)
|
||
|
||
if not status_data:
|
||
return {
|
||
"status": "not_started",
|
||
"project_id": project_id
|
||
}
|
||
|
||
return json.loads(status_data)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting analysis status: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.get("/lab/projects/{project_id}/report")
|
||
async def get_project_saved_report(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""从 Redis db202 读取已保存的项目报告"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 选择 db202
|
||
await redis.select(202)
|
||
report_key = f"project_report:{project_id}"
|
||
|
||
# 获取已保存的报告
|
||
existing_report = await redis.get(report_key)
|
||
|
||
if not existing_report:
|
||
raise HTTPException(status_code=404, detail="No saved project report found")
|
||
|
||
# 返回报告
|
||
return json.loads(existing_report)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting project report: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.get("/lab/projects/{project_id}")
|
||
async def get_project_detail(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""
|
||
获取单个项目的详细信息
|
||
|
||
参数:
|
||
project_id: 项目ID
|
||
current_user: 当前登录用户（通过token验证）
|
||
|
||
返回:
|
||
project: 项目详细信息，包括基本信息和统计数据
|
||
"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证项目ID格式
|
||
if not ObjectId.is_valid(project_id):
|
||
raise HTTPException(status_code=400, detail="Invalid project ID format")
|
||
|
||
# 查询项目信息
|
||
project = await db.projects.find_one({
|
||
"_id": ObjectId(project_id),
|
||
"user_id": current_user.id # 确保只能访问自己的项目
|
||
})
|
||
|
||
if not project:
|
||
raise HTTPException(status_code=404, detail="Project not found or unauthorized access")
|
||
|
||
# 获取项目下的实验数量
|
||
experiment_count = await db.experiments.count_documents({
|
||
"project_id": ObjectId(project_id)
|
||
})
|
||
|
||
# 构造响应数据
|
||
project_data = {
|
||
"_id": str(project["_id"]),
|
||
"project_name": project["project_name"],
|
||
"description": project["description"],
|
||
"create_time": project["create_time"],
|
||
"experiment_count": experiment_count,
|
||
"user_id": str(project["user_id"])
|
||
}
|
||
|
||
return project_data
|
||
|
||
except Exception as e:
|
||
print(f"Error getting project detail: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to get project detail: {str(e)}")
|
||
|
||
# 添加删除实验的路由
|
||
@app.delete("/lab/experiments/{experiment_id}")
|
||
async def delete_experiment(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""删除实验及其相关数据"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 验证实验是否存在且属于当前用户的项目
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
if not experiment:
|
||
raise HTTPException(status_code=404, detail="Experiment not found")
|
||
|
||
project = await db.projects.find_one({
|
||
"_id": experiment["project_id"],
|
||
"user_id": current_user.id
|
||
})
|
||
if not project:
|
||
raise HTTPException(status_code=403, detail="Unauthorized to delete this experiment")
|
||
|
||
# 删除实验相关数据
|
||
await db.experiment_sessions.delete_many({"experiment_id": ObjectId(experiment_id)})
|
||
await db.experiment_devices.delete_many({"experiment_id": ObjectId(experiment_id)})
|
||
await db.experiments.delete_one({"_id": ObjectId(experiment_id)})
|
||
|
||
# 删除Redis中的实验报告
|
||
await redis.select(201)
|
||
await redis.delete(f"experiment_report:{experiment_id}")
|
||
|
||
return {"message": "Experiment successfully deleted"}
|
||
|
||
except Exception as e:
|
||
print(f"Error deleting experiment: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to delete experiment: {str(e)}")
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
# 添加删除项目的路由
|
||
@app.delete("/lab/projects/{project_id}")
|
||
async def delete_project(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""删除项目及其所有相关数据"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 验证项目是否存在且属于当前用户
|
||
project = await db.projects.find_one({
|
||
"_id": ObjectId(project_id),
|
||
"user_id": current_user.id
|
||
})
|
||
if not project:
|
||
raise HTTPException(status_code=404, detail="Project not found or unauthorized access")
|
||
|
||
# 获取项目下所有实验的ID
|
||
experiment_ids = []
|
||
async for exp in db.experiments.find({"project_id": ObjectId(project_id)}):
|
||
experiment_ids.append(str(exp["_id"]))
|
||
|
||
# 删除所有相关数据
|
||
for exp_id in experiment_ids:
|
||
await db.experiment_sessions.delete_many({"experiment_id": ObjectId(exp_id)})
|
||
await db.experiment_devices.delete_many({"experiment_id": ObjectId(exp_id)})
|
||
|
||
# 删除Redis中的实验报告
|
||
await redis.select(201)
|
||
await redis.delete(f"experiment_report:{exp_id}")
|
||
|
||
# 删除所有实验
|
||
await db.experiments.delete_many({"project_id": ObjectId(project_id)})
|
||
|
||
# 删除项目
|
||
await db.projects.delete_one({"_id": ObjectId(project_id)})
|
||
|
||
# 删除Redis中的项目报告
|
||
await redis.select(202)
|
||
await redis.delete(f"project_report:{project_id}")
|
||
|
||
return {"message": "Project successfully deleted"}
|
||
|
||
except Exception as e:
|
||
print(f"Error deleting project: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to delete project: {str(e)}")
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
|
||
# 在其他 Pydantic 模型后添加
|
||
class ReferenceModel(BaseModel):
|
||
"""文献引用模型"""
|
||
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
||
project_id: PyObjectId
|
||
reference_link: str
|
||
reference_title: str
|
||
upload_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||
|
||
class Config:
|
||
populate_by_name = True
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {ObjectId: str}
|
||
|
||
|
||
# 删除文献
|
||
@app.delete("/lab/projects/{project_id}/references/{reference_id}")
|
||
async def delete_reference(
|
||
project_id: str,
|
||
reference_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""删除项目文献"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证项目所有权
|
||
project = await db.projects.find_one({
|
||
"_id": ObjectId(project_id),
|
||
"user_id": current_user.id
|
||
})
|
||
if not project:
|
||
raise HTTPException(status_code=403, detail="Unauthorized to access this project")
|
||
|
||
# 获取文献信息
|
||
reference = await db.references.find_one({"_id": ObjectId(reference_id)})
|
||
if not reference:
|
||
raise HTTPException(status_code=404, detail="Reference not found")
|
||
|
||
# 删除文件
|
||
if os.path.exists(reference["reference_link"]):
|
||
os.remove(reference["reference_link"])
|
||
|
||
# 删除数据库记录
|
||
await db.references.delete_one({"_id": ObjectId(reference_id)})
|
||
|
||
return {"message": "Reference successfully deleted"}
|
||
|
||
except Exception as e:
|
||
print(f"Error deleting reference: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to delete reference: {str(e)}")
|
||
|
||
@app.get("/lab/projects/{project_id}/references")
|
||
async def get_project_references(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取项目的文献列表"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 查找项目下的所有文献
|
||
references_cursor = db.references.find({
|
||
"project_id": ObjectId(project_id)
|
||
})
|
||
|
||
references = []
|
||
async for ref in references_cursor:
|
||
references.append({
|
||
"_id": str(ref["_id"]),
|
||
"project_id": str(ref["project_id"]),
|
||
"reference_link": ref["reference_link"],
|
||
"reference_title": ref["reference_title"],
|
||
"upload_time": ref["upload_time"]
|
||
})
|
||
return references
|
||
|
||
except Exception as e:
|
||
print(f"Error getting project references: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.get("/lab/references/{reference_id}/report")
|
||
async def get_reference_report(
|
||
reference_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""从 Redis db203 读取已保存的文献报告"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 选择 db203
|
||
await redis.select(203)
|
||
report_key = f"reference_report:{reference_id}"
|
||
|
||
# 获取已保存的报告
|
||
existing_report = await redis.get(report_key)
|
||
|
||
if not existing_report:
|
||
raise HTTPException(status_code=404, detail="No saved reference report found")
|
||
|
||
# 返回报告
|
||
return json.loads(existing_report)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting reference report: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
|
||
|
||
# 修改函数定义为异步函数
|
||
async def analyze_reference_summary(reference_data):
|
||
system_prompt = """
|
||
You are an AI assistant responsible for analyzing paper reports.
|
||
You will summarize and analyze all paper reports and generate a comprehensive analysis report in JSON format.
|
||
The JSON structure must strictly follow the provided template.
|
||
"""
|
||
|
||
user_prompt = f"""Analyze the following paper reports:
|
||
Paper reports: {json.dumps(reference_data, ensure_ascii=False)}
|
||
|
||
Generate a JSON response with the following structure:
|
||
|
||
{{
|
||
"Paper Summary Report": {{
|
||
"overview": {{
|
||
"total_papers": "[Number of papers]",
|
||
"time_range": {{"start_year": "[Start year]", "end_year": "[End year]"}},
|
||
"main_research_areas": "[Main research areas of the papers]"
|
||
}},
|
||
"research_trends": {{
|
||
"major_themes": "[Major themes areas of the papers]",
|
||
"common_methodologies": "[Common methodologies used in the papers]",
|
||
"emerging_topics": "[Emerging topics in the papers]"
|
||
}},
|
||
"key_findings": {{
|
||
"theoretical_advances": "[Theoretical advances in the papers]",
|
||
"experimental_results": "[Experimental results in the papers]",
|
||
"common_conclusions": "[Common conclusions in the papers]"
|
||
}},
|
||
"research_gaps": {{
|
||
"current_limitations": "[Current limitations]",
|
||
"unexplored_areas": "[Unexplored areas]",
|
||
"technical_challenges": "[Technical challenges"
|
||
}},
|
||
"future_directions": {{
|
||
"potential_applications": "[Potential future applications]",
|
||
"methodological_suggestions": "[Methodological suggestions based on paper summary]"
|
||
}},
|
||
"impact_assessment": {{
|
||
"academic_influence": "[Summarize the academic influence of the paper]",
|
||
"practical_value": "[Summarize the practical value of the paper]"
|
||
}}
|
||
}}
|
||
}}
|
||
"""
|
||
|
||
messages = [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
]
|
||
|
||
try:
|
||
response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=messages,
|
||
response_format={'type': 'json_object'}
|
||
)
|
||
return json.loads(response.choices[0].message.content)
|
||
except Exception as e:
|
||
print(f"Error calling DeepSeek API: {e}")
|
||
return None
|
||
|
||
|
||
# 创建线程池
|
||
analysis_thread_pool = ThreadPoolExecutor(max_workers=3)
|
||
|
||
@app.get("/lab/references/{project_id}/analyze_report")
|
||
async def analyze_reference_summary_report(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""分析文献数据"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 检查是否已经有正在进行的分析任务
|
||
await redis.select(204)
|
||
status_key = f"reference_analysis_status:{project_id}"
|
||
current_status = await redis.get(status_key)
|
||
|
||
if current_status:
|
||
status_data = json.loads(current_status)
|
||
if status_data.get("status") == "processing":
|
||
return {
|
||
"message": "文献分析任务正在进行中",
|
||
"status": "processing",
|
||
"project_id": project_id,
|
||
"start_time": status_data.get("start_time")
|
||
}
|
||
|
||
# 从MongoDB查找项目下的所有实验ID
|
||
reference_cursor = db.references.find({
|
||
"project_id": ObjectId(project_id)
|
||
})
|
||
|
||
reference_ids = []
|
||
async for reference in reference_cursor:
|
||
reference_ids.append(str(reference["_id"]))
|
||
|
||
if not reference_ids:
|
||
raise HTTPException(status_code=404, detail="No references found for this project")
|
||
|
||
# 记录分析开始状态
|
||
status_data = {
|
||
"status": "processing",
|
||
"start_time": datetime.now(timezone.utc).isoformat(),
|
||
"total_references": len(reference_ids),
|
||
"completed_references": 0
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
|
||
# 在新线程中运行分析任务
|
||
analysis_thread_pool.submit(run_analysis_in_thread, project_id, reference_ids)
|
||
|
||
return {
|
||
"message": "文献分析任务已启动",
|
||
"status": "processing",
|
||
"project_id": project_id,
|
||
"start_time": status_data["start_time"],
|
||
"total_references": len(reference_ids)
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error starting reference analysis: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
def run_analysis_in_thread(project_id: str, reference_ids: List[str]):
|
||
"""在独立线程中运行分析任务"""
|
||
# 创建新的事件循环
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
|
||
try:
|
||
# 在新的事件循环中运行异步任务
|
||
loop.run_until_complete(process_reference_analysis(project_id, reference_ids))
|
||
except Exception as e:
|
||
print(f"Error in analysis thread: {str(e)}")
|
||
finally:
|
||
loop.close()
|
||
|
||
async def process_reference_analysis(project_id: str, reference_ids: List[str]):
|
||
"""后台处理文献分析任务"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 更新状态键
|
||
await redis.select(204)
|
||
status_key = f"reference_analysis_status:{project_id}"
|
||
|
||
# 从Redis db203获取所有实验报告
|
||
await redis.select(203)
|
||
all_reference_reports = {}
|
||
completed_count = 0
|
||
|
||
for ref_id in reference_ids:
|
||
report_key = f"reference_report:{ref_id}"
|
||
report_data = await redis.get(report_key)
|
||
|
||
if report_data:
|
||
try:
|
||
parsed_report = json.loads(report_data)
|
||
all_reference_reports[ref_id] = parsed_report
|
||
completed_count += 1
|
||
|
||
# 更新进度
|
||
await redis.select(204)
|
||
current_status = await redis.get(status_key)
|
||
if current_status:
|
||
status_data = json.loads(current_status)
|
||
status_data["completed_references"] = completed_count
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
await redis.select(203)
|
||
|
||
except json.JSONDecodeError:
|
||
print(f"Error parsing reference {ref_id} report")
|
||
continue
|
||
|
||
if all_reference_reports:
|
||
analysis_result = await analyze_reference_summary(all_reference_reports)
|
||
|
||
if analysis_result:
|
||
await redis.select(204)
|
||
report_key = f"reference_summary_report:{project_id}"
|
||
await redis.set(report_key, json.dumps(analysis_result))
|
||
|
||
# 更新最终状态
|
||
status_data = {
|
||
"status": "completed",
|
||
"completion_time": datetime.now(timezone.utc).isoformat(),
|
||
"total_references": len(reference_ids),
|
||
"completed_references": completed_count
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
|
||
print(f"Reference analysis completed for project {project_id}")
|
||
else:
|
||
# 更新失败状态
|
||
status_data = {
|
||
"status": "failed",
|
||
"error": "Failed to generate analysis result",
|
||
"completion_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
print(f"Failed to generate analysis for project {project_id}")
|
||
else:
|
||
# 更新失败状态
|
||
status_data = {
|
||
"status": "failed",
|
||
"error": "No valid reference reports found",
|
||
"completion_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
print(f"No valid reference reports found for project {project_id}")
|
||
|
||
except Exception as e:
|
||
print(f"Error in background reference analysis: {str(e)}")
|
||
try:
|
||
await redis.select(204)
|
||
status_data = {
|
||
"status": "failed",
|
||
"error": str(e),
|
||
"completion_time": datetime.now(timezone.utc).isoformat()
|
||
}
|
||
await redis.set(status_key, json.dumps(status_data))
|
||
except Exception as redis_error:
|
||
print(f"Error updating Redis status: {redis_error}")
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
# 添加新的状态检查路由
|
||
@app.get("/lab/references/{project_id}/analysis_status")
|
||
async def get_reference_analysis_status(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取文献分析任务的状态"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
await redis.select(204)
|
||
status_key = f"reference_analysis_status:{project_id}"
|
||
status_data = await redis.get(status_key)
|
||
|
||
if not status_data:
|
||
return {
|
||
"status": "not_started",
|
||
"project_id": project_id
|
||
}
|
||
|
||
return json.loads(status_data)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting analysis status: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.get("/lab/references/{project_id}/summary_report")
|
||
async def get_reference_summary_report(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""从 Redis db204 读取已保存的文献报告"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 选择 db204
|
||
await redis.select(204)
|
||
report_key = f"reference_summary_report:{project_id}"
|
||
|
||
# 获取已保存的报告
|
||
existing_report = await redis.get(report_key)
|
||
|
||
if not existing_report:
|
||
raise HTTPException(status_code=404, detail="No saved reference report found")
|
||
|
||
# 返回报告
|
||
return json.loads(existing_report)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting reference report: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
# 在现有路由之前添加新的Pydantic模型
|
||
class MemoModel(BaseModel):
|
||
"""项目备忘录模型"""
|
||
content: str
|
||
create_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||
|
||
# 添加保存备忘录的路由
|
||
@app.post("/lab/projects/{project_id}/memo")
|
||
async def save_project_memo(
|
||
project_id: str,
|
||
memo: MemoModel,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""保存项目备忘录"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 选择 db205
|
||
await redis.select(205)
|
||
memo_key = f"memo:{project_id}"
|
||
|
||
# 保存备忘录内容和创建时间
|
||
memo_data = {
|
||
"content": memo.content,
|
||
"create_time": memo.create_time.isoformat()
|
||
}
|
||
|
||
await redis.set(memo_key, json.dumps(memo_data))
|
||
return {"message": "备忘录保存成功"}
|
||
|
||
except Exception as e:
|
||
print(f"Error saving memo: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
# 添加获取备忘录的路由
|
||
@app.get("/lab/projects/{project_id}/memo")
|
||
async def get_project_memo(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取项目备忘录"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 选择 db205
|
||
await redis.select(205)
|
||
memo_key = f"memo:{project_id}"
|
||
|
||
# 获取备忘录
|
||
memo_data = await redis.get(memo_key)
|
||
|
||
if not memo_data:
|
||
return {"content": "", "create_time": None}
|
||
|
||
return json.loads(memo_data)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting memo: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
# 添加保存实验备忘录的路由
|
||
@app.post("/lab/experiments/{experiment_id}/memo")
|
||
async def save_experiment_memo(
|
||
experiment_id: str,
|
||
memo: MemoModel,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""保存实验备忘录"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 选择 db206
|
||
await redis.select(206)
|
||
memo_key = f"memo:{experiment_id}"
|
||
|
||
# 保存备忘录内容和创建时间
|
||
memo_data = {
|
||
"content": memo.content,
|
||
"create_time": memo.create_time.isoformat()
|
||
}
|
||
|
||
await redis.set(memo_key, json.dumps(memo_data))
|
||
return {"message": "Experiment memo saved successfully"}
|
||
|
||
except Exception as e:
|
||
print(f"Error saving experiment memo: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
# 添加获取实验备忘录的路由
|
||
@app.get("/lab/experiments/{experiment_id}/memo")
|
||
async def get_experiment_memo(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取实验备忘录"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 选择 db206
|
||
await redis.select(206)
|
||
memo_key = f"memo:{experiment_id}"
|
||
|
||
# 获取备忘录
|
||
memo_data = await redis.get(memo_key)
|
||
|
||
if not memo_data:
|
||
return {"content": "", "create_time": None}
|
||
|
||
return json.loads(memo_data)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting experiment memo: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
# 添加公式模型
|
||
class FormulaModel(BaseModel):
|
||
data_name: str
|
||
data_unit: str
|
||
formula: str
|
||
|
||
# 添加公式请求模型
|
||
class FormulaCreate(BaseModel):
|
||
sensor_name: str
|
||
data_name: str
|
||
data_unit: str
|
||
formula: str
|
||
|
||
@app.post("/lab/experiments/{experiment_id}/devices/{device_id}/formulas")
|
||
async def add_sensor_formula(
|
||
experiment_id: str,
|
||
device_id: str,
|
||
formula: FormulaCreate,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""为实验设备的传感器添加计算公式"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证实验设备是否存在，使用 user_device_id 查询
|
||
device = await db.experiment_devices.find_one({
|
||
"user_device_id": ObjectId(device_id),
|
||
"experiment_id": ObjectId(experiment_id)
|
||
})
|
||
|
||
if not device:
|
||
raise HTTPException(status_code=404, detail="Experiment device not found")
|
||
|
||
# 查找对应的传感器
|
||
sensor_found = False
|
||
sensors = device.get("sensors", [])
|
||
for sensor in sensors:
|
||
if sensor["sensor_name"] == formula.sensor_name:
|
||
# 初始化或获取现有公式列表
|
||
if "formulas" not in sensor:
|
||
sensor["formulas"] = []
|
||
# 添加新公式
|
||
sensor["formulas"].append({
|
||
"data_name": formula.data_name,
|
||
"data_unit": formula.data_unit,
|
||
"formula": formula.formula
|
||
})
|
||
sensor_found = True
|
||
break
|
||
|
||
if not sensor_found:
|
||
raise HTTPException(status_code=404, detail="Sensor not found")
|
||
|
||
# 更新设备文档，使用 user_device_id 更新
|
||
result = await db.experiment_devices.update_one(
|
||
{
|
||
"user_device_id": ObjectId(device_id),
|
||
"experiment_id": ObjectId(experiment_id)
|
||
},
|
||
{"$set": {"sensors": sensors}}
|
||
)
|
||
|
||
if result.modified_count == 0:
|
||
raise HTTPException(status_code=400, detail="Failed to add formula")
|
||
|
||
return {"message": "Formula added successfully"}
|
||
|
||
except Exception as e:
|
||
print(f"Error adding formula: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=f"Failed to add formula: {str(e)}")
|
||
|
||
# 添加问答相关的模型
|
||
class QuestionModel(BaseModel):
|
||
"""问题模型"""
|
||
question: str
|
||
|
||
class TaskStatus:
|
||
PENDING = "pending"
|
||
PROCESSING = "processing"
|
||
COMPLETED = "completed"
|
||
FAILED = "failed"
|
||
|
||
@app.post("/lab/references/{reference_id}/qa")
|
||
async def ask_reference_question(
|
||
reference_id: str,
|
||
question: QuestionModel,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""向文献提问(异步)"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 验证文献是否存在
|
||
reference = await db.references.find_one({"_id": ObjectId(reference_id)})
|
||
if not reference:
|
||
raise HTTPException(status_code=404, detail="文献不存在")
|
||
|
||
# 生成任务ID
|
||
task_id = str(ObjectId())
|
||
|
||
# 创建后台任务
|
||
asyncio.create_task(process_reference_question(
|
||
task_id=task_id,
|
||
reference_id=reference_id,
|
||
question=question.question,
|
||
reference=reference
|
||
))
|
||
|
||
return {
|
||
"task_id": task_id,
|
||
"status": TaskStatus.PENDING,
|
||
"message": "问题已提交，正在处理中"
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error in ask_reference_question: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
async def process_reference_question(task_id: str, reference_id: str, question: str, reference: dict):
|
||
"""处理文献问答的后台任务"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 更新任务状态为处理中
|
||
await redis.select(208) # 使用db208存储任务状态
|
||
await redis.hset(
|
||
f"task:{task_id}",
|
||
mapping={
|
||
"status": TaskStatus.PROCESSING,
|
||
"reference_id": reference_id,
|
||
"question": question
|
||
}
|
||
)
|
||
|
||
# 从Redis获取分析报告
|
||
await redis.select(203)
|
||
report_key = f"reference_report:{reference_id}"
|
||
report_data = await redis.get(report_key)
|
||
|
||
if not report_data:
|
||
raise Exception("文献分析报告不存在，请先进行分析")
|
||
|
||
# 读取PDF文件内容
|
||
file_path = reference.get("reference_link")
|
||
if not file_path or not os.path.exists(file_path):
|
||
raise Exception("文献文件不存在")
|
||
|
||
# 读取PDF内容
|
||
pdf_content = ""
|
||
with open(file_path, 'rb') as file:
|
||
pdf_reader = PyPDF2.PdfReader(file)
|
||
for page in pdf_reader.pages:
|
||
content = page.extract_text()
|
||
pdf_content += content
|
||
|
||
# 限制内容长度
|
||
MAX_CHARS = 180000
|
||
pdf_content = pdf_content[:MAX_CHARS]
|
||
|
||
# 构建系统提示和用户提示
|
||
system_prompt = """
|
||
你是一个专业的学术助手，负责回答关于学术文献的问题。
|
||
你应该基于文献内容和分析报告提供准确、专业的回答。
|
||
回答应当简洁明了，并尽可能引用文献中的具体内容。
|
||
"""
|
||
|
||
# 构建上下文
|
||
context = {
|
||
"文献内容": pdf_content,
|
||
"分析报告": json.loads(report_data)
|
||
}
|
||
|
||
messages = [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": f"基于以下文献内容和分析报告回答问题：\n\n文献信息：{json.dumps(context, ensure_ascii=False)}\n\n问题：{question}"}
|
||
]
|
||
|
||
# 调用DeepSeek API获取回答
|
||
response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=messages
|
||
)
|
||
answer = response.choices[0].message.content
|
||
|
||
# 保存对话历史到Redis db207
|
||
await redis.select(207)
|
||
chat_history_key = f"chat_history:{reference_id}"
|
||
|
||
# 获取现有历史记录
|
||
existing_history = await redis.get(chat_history_key)
|
||
history = json.loads(existing_history) if existing_history else []
|
||
|
||
# 添加新的对话
|
||
history.append({
|
||
"question": question,
|
||
"answer": answer,
|
||
"timestamp": datetime.now(timezone.utc).isoformat()
|
||
})
|
||
|
||
# 保存更新后的历史记录
|
||
await redis.set(chat_history_key, json.dumps(history))
|
||
|
||
# 更新任务状态为完成
|
||
await redis.select(208)
|
||
await redis.hset(
|
||
f"task:{task_id}",
|
||
mapping={
|
||
"status": TaskStatus.COMPLETED,
|
||
"answer": answer,
|
||
"reference_title": reference.get("reference_title")
|
||
}
|
||
)
|
||
|
||
except Exception as e:
|
||
print(f"Error processing question: {e}")
|
||
# 更新任务状态为失败
|
||
await redis.select(208)
|
||
await redis.hset(
|
||
f"task:{task_id}",
|
||
mapping={
|
||
"status": TaskStatus.FAILED,
|
||
"error": str(e)
|
||
}
|
||
)
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.get("/lab/task/{task_id}")
|
||
async def get_task_status(task_id: str):
|
||
"""获取任务状态和结果"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
await redis.select(208)
|
||
task_data = await redis.hgetall(f"task:{task_id}")
|
||
|
||
if not task_data:
|
||
raise HTTPException(status_code=404, detail="任务不存在")
|
||
|
||
response = {
|
||
"task_id": task_id,
|
||
"status": task_data.get("status", TaskStatus.PENDING)
|
||
}
|
||
|
||
# 如果任务完成，添加结果
|
||
if task_data.get("status") == TaskStatus.COMPLETED:
|
||
response.update({
|
||
"answer": task_data.get("answer"),
|
||
"reference_title": task_data.get("reference_title")
|
||
})
|
||
# 如果任务失败，添加错误信息
|
||
elif task_data.get("status") == TaskStatus.FAILED:
|
||
response.update({
|
||
"error": task_data.get("error")
|
||
})
|
||
|
||
return response
|
||
|
||
except Exception as e:
|
||
print(f"Error getting task status: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.get("/lab/references/{reference_id}/qa/history")
|
||
async def get_reference_qa_history(
|
||
reference_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取文献问答历史记录"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 从Redis db207获取对话历史
|
||
await redis.select(207)
|
||
chat_history_key = f"chat_history:{reference_id}"
|
||
|
||
history_data = await redis.get(chat_history_key)
|
||
if not history_data:
|
||
return []
|
||
|
||
return json.loads(history_data)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting QA history: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.post("/lab/projects/{project_id}/references/batch")
|
||
async def batch_upload_references(
|
||
project_id: str,
|
||
files: List[UploadFile] = File(...),
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""批量上传项目相关文献"""
|
||
db = await get_database()
|
||
|
||
try:
|
||
# 验证项目是否存在且属于当前用户
|
||
project = await db.projects.find_one({
|
||
"_id": ObjectId(project_id),
|
||
"user_id": current_user.id
|
||
})
|
||
if not project:
|
||
raise HTTPException(status_code=404, detail="Project not found or unauthorized access")
|
||
|
||
uploaded_references = []
|
||
|
||
# 批量上传文件
|
||
for file in files:
|
||
# 验证文件类型
|
||
allowed_types = ["application/pdf", "application/msword",
|
||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document"]
|
||
if file.content_type not in allowed_types:
|
||
continue # 跳过不支持的文件类型
|
||
|
||
# 确保上传目录存在
|
||
os.makedirs(upload_path, exist_ok=True)
|
||
|
||
# 生成安全的文件名
|
||
file_extension = os.path.splitext(file.filename)[1]
|
||
safe_filename = f"{project_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}{file_extension}"
|
||
file_path = os.path.join(upload_path, safe_filename)
|
||
|
||
# 保存文件
|
||
with open(file_path, "wb") as buffer:
|
||
content = await file.read()
|
||
buffer.write(content)
|
||
|
||
# 创建引用记录
|
||
reference = {
|
||
"project_id": ObjectId(project_id),
|
||
"reference_link": file_path,
|
||
"reference_title": file.filename,
|
||
"upload_time": datetime.now(timezone.utc)
|
||
}
|
||
|
||
result = await db.references.insert_one(reference)
|
||
reference_info = {
|
||
"reference_id": str(result.inserted_id),
|
||
"file_path": file_path,
|
||
"reference_title": reference["reference_title"]
|
||
}
|
||
uploaded_references.append(reference_info)
|
||
|
||
# 为每个文献创建初始状态
|
||
redis = await get_redis()
|
||
try:
|
||
await redis.select(203)
|
||
report_key = f"reference_report:{str(result.inserted_id)}"
|
||
initial_status = {
|
||
"status": "processing",
|
||
"message": "Analysis in progress"
|
||
}
|
||
await redis.set(report_key, json.dumps(initial_status))
|
||
finally:
|
||
await redis.aclose()
|
||
|
||
# 在后台启动分析任务
|
||
if uploaded_references:
|
||
asyncio.create_task(process_batch_analysis(uploaded_references))
|
||
|
||
return {
|
||
"message": f"Successfully uploaded {len(uploaded_references)} files",
|
||
"uploaded_files": uploaded_references
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Batch upload error: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
async def process_batch_analysis(references: List[dict]):
|
||
"""批量处理文献分析的后台任务"""
|
||
redis = await get_redis()
|
||
|
||
# 限制并发数量
|
||
semaphore = asyncio.Semaphore(3)
|
||
|
||
async def process_single_reference(ref: dict):
|
||
async with semaphore:
|
||
try:
|
||
reference_id = ref["reference_id"]
|
||
file_path = ref["file_path"]
|
||
|
||
if not os.path.exists(file_path):
|
||
print(f"Reference file not found: {file_path}")
|
||
return
|
||
|
||
print(f"\n开始处理文献 {ref.get('reference_title', '未知标题')}")
|
||
print(f"文献ID: {reference_id}")
|
||
print(f"文件路径: {file_path}")
|
||
|
||
# 异步读取PDF
|
||
print("\n=== 步骤1: 读取PDF文件 ===")
|
||
pdf_content = await read_pdf_async(file_path)
|
||
if not pdf_content:
|
||
raise Exception("Failed to read PDF content")
|
||
|
||
# 打印字符数
|
||
content_length = len(pdf_content)
|
||
print(f"\n=== 步骤2: 内容长度检查 ===")
|
||
print(f"PDF内容总字符数: {content_length}")
|
||
|
||
# 根据内容长度选择不同的处理方式
|
||
if content_length <= 200000:
|
||
print(f"\n=== 步骤3A: 使用直接分析方式 ===")
|
||
print(f"文档长度在处理范围内 ({content_length} <= 200000)")
|
||
# 直接分析文档内容
|
||
print("开始分析文档内容...")
|
||
document_analysis = await analyze_reference_document_async(pdf_content[:180000])
|
||
if not document_analysis:
|
||
raise Exception("Failed to analyze document")
|
||
print("文档分析完成")
|
||
else:
|
||
print(f"\n=== 步骤3B: 使用分段分析方式 ===")
|
||
print(f"文档超过200000字符 ({content_length} > 200000)")
|
||
# 分段分析长文档
|
||
print("\n--- 开始分段分析 ---")
|
||
print("正在调用 analyze_long_document_async...")
|
||
try:
|
||
analysis_results = await analyze_long_document_async(pdf_content)
|
||
except Exception as e:
|
||
print(f"分段分析过程中出错: {str(e)}")
|
||
raise
|
||
|
||
if not analysis_results:
|
||
raise Exception("Failed to analyze document in segments")
|
||
print(f"分段分析完成，共分析了 {len(analysis_results)} 个段落")
|
||
|
||
# 合并分析结果
|
||
try:
|
||
document_analysis = await merge_analysis_results(analysis_results)
|
||
except Exception as e:
|
||
raise
|
||
|
||
if not document_analysis:
|
||
raise Exception("Failed to merge analysis results")
|
||
|
||
# 等待一小段时间避免API限制
|
||
await asyncio.sleep(1)
|
||
|
||
# 异步分析文献价值
|
||
try:
|
||
value_evaluation = await analyze_reference_value_async(document_analysis)
|
||
print(f"analyze_reference_value_async 返回结果类型: {type(value_evaluation)}")
|
||
except Exception as e:
|
||
raise
|
||
|
||
if not value_evaluation:
|
||
raise Exception("Failed to evaluate value")
|
||
print("文献价值分析完成")
|
||
|
||
# 合并结果
|
||
print("\n=== 步骤5: 保存最终结果 ===")
|
||
analysis_result = {
|
||
**document_analysis,
|
||
**value_evaluation,
|
||
"status": "completed"
|
||
}
|
||
|
||
# 保存结果
|
||
await redis.select(203)
|
||
report_key = f"reference_report:{reference_id}"
|
||
await redis.set(report_key, json.dumps(analysis_result))
|
||
|
||
except Exception as e:
|
||
try:
|
||
await redis.select(203)
|
||
report_key = f"reference_report:{ref['reference_id']}"
|
||
error_status = {
|
||
"status": "failed",
|
||
"message": str(e)
|
||
}
|
||
await redis.set(report_key, json.dumps(error_status))
|
||
except Exception as redis_error:
|
||
print(f"Error updating Redis status: {redis_error}")
|
||
|
||
try:
|
||
# 并发处理所有引用
|
||
await asyncio.gather(
|
||
*(process_single_reference(ref) for ref in references)
|
||
)
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
# 在全局范围创建线程池
|
||
pdf_thread_pool = ThreadPoolExecutor(max_workers=3) # 限制并发PDF处理数量
|
||
|
||
# 创建异步HTTP客户端会话
|
||
async def get_aiohttp_session():
|
||
return aiohttp.ClientSession(
|
||
base_url="https://api.deepseek.com/v1/", # 添加了末尾的斜杠
|
||
headers={"Authorization": f"Bearer sk-3027fb3c810b4e17985fa397d41250b9"}
|
||
)
|
||
|
||
async def read_pdf_async(file_path: str) -> str:
|
||
"""在线程池中异步读取PDF"""
|
||
def read_pdf():
|
||
try:
|
||
pdf_content = ""
|
||
with open(file_path, 'rb') as file:
|
||
pdf_reader = PyPDF2.PdfReader(file)
|
||
for page in pdf_reader.pages:
|
||
content = page.extract_text()
|
||
pdf_content += content
|
||
|
||
# 根据内容长度返回不同的结果
|
||
content_length = len(pdf_content)
|
||
print(f"\nPDF原始内容字符数: {content_length}")
|
||
|
||
if content_length <= 180000:
|
||
print("文档长度 ≤ 180000，返回完整内容")
|
||
return pdf_content
|
||
elif content_length <= 200000:
|
||
print("文档长度在 180000-200000 之间，截取前 180000 个字符")
|
||
return pdf_content[:180000]
|
||
else:
|
||
print("文档长度 > 200000，返回完整内容供分段处理")
|
||
return pdf_content # 返回完整内容，由调用者处理分段
|
||
|
||
except Exception as e:
|
||
print(f"Error reading PDF: {e}")
|
||
return ""
|
||
|
||
loop = asyncio.get_event_loop()
|
||
return await loop.run_in_executor(pdf_thread_pool, read_pdf)
|
||
|
||
async def call_deepseek_api_async(messages: list) -> dict:
|
||
"""异步调用DeepSeek API"""
|
||
async with await get_aiohttp_session() as session:
|
||
async with session.post("/chat/completions", json={
|
||
"model": "deepseek-chat",
|
||
"messages": messages,
|
||
"response_format": {"type": "json_object"}
|
||
}) as response:
|
||
if response.status == 200:
|
||
data = await response.json()
|
||
return json.loads(data["choices"][0]["message"]["content"])
|
||
else:
|
||
raise Exception(f"API调用失败: {await response.text()}")
|
||
|
||
async def analyze_long_document_async(content: str) -> List[dict]:
|
||
"""分段分析长文档"""
|
||
# 将内容分成多个段落，每段约60000个字符（估算后约50000 tokens）
|
||
segments = []
|
||
content_length = len(content)
|
||
segment_size = 60000 # 减小段落大小
|
||
|
||
print(f"\n开始分段处理，总字符数: {content_length}")
|
||
print(f"每段大小: {segment_size} 字符")
|
||
|
||
for i in range(0, content_length, segment_size):
|
||
segment = content[i:i + segment_size]
|
||
segments.append(segment)
|
||
|
||
print(f"文档已分段，共 {len(segments)} 个段落")
|
||
|
||
# 对每个段落进行分析
|
||
analysis_results = []
|
||
for i, segment in enumerate(segments):
|
||
print(f"\n开始分析第 {i+1}/{len(segments)} 个段落...")
|
||
|
||
system_prompt = f"""
|
||
You are an AI assistant tasked with analyzing part {i+1} of {len(segments)} of an academic paper.
|
||
Generate a comprehensive analysis in JSON format covering both basic information and content analysis.
|
||
Note that this is part {i+1} of a longer document, so focus on the content provided.
|
||
The JSON structure must strictly follow the provided template.
|
||
Also generate a Mermaid flowchart code to visualize the research methodology if this segment contains methodology information.
|
||
Create a detailed and comprehensive flowchart that accurately represents the paper's research methodology.
|
||
"""
|
||
|
||
user_prompt = f"""Analyze the following paper segment and extract all relevant information:
|
||
Content: {segment}
|
||
|
||
Generate a JSON response with the following structure:
|
||
|
||
{{
|
||
"1. Basic Information": {{
|
||
"author": "[Author name(s) and affiliations]",
|
||
"publication_date": "[Publication date in YYYY-MM format]",
|
||
"title": "[Full title of the document]",
|
||
"journal_publisher": "[Journal name or publisher details]",
|
||
"document_type": "[Type: journal article/book/conference paper etc.]"
|
||
}},
|
||
"2. Content Analysis": {{
|
||
"abstract": "[Paper abstract]",
|
||
"research_purpose": "[Main objectives and research questions]",
|
||
"methodology": "[Research methods, data collection and analysis approaches]",
|
||
"main_arguments": "[Key theoretical frameworks and arguments]",
|
||
"conclusions": "[Complete findings and conclusions]",
|
||
"innovations": "[Novel contributions and original aspects]"
|
||
}},
|
||
"flowchart": "[If this segment contains methodology information, generate a Mermaid flowchart code that visualizes the research methodology. Follow these rules:
|
||
1. Use 'graph TD' for top-down flow
|
||
2. Each node should be in format: id[text] where:
|
||
- id is a unique identifier (A, B1, B2, etc.)
|
||
- text should be simple and clear, using only letters, numbers, and spaces
|
||
- DO NOT use any special characters including parentheses, colons, commas
|
||
- abbreviations should be written without parentheses, e.g., 'DNN' not '(DNN)'
|
||
- use space instead of special characters, e.g., 'Deep Learning Model' not 'Deep-Learning/Model'
|
||
3. Connections use '-->' between nodes
|
||
4. Ensure each line ends with a proper node reference
|
||
|
||
Create a detailed flowchart that shows:
|
||
- Research objectives and questions
|
||
- All major research methods used
|
||
- Data collection and analysis processes
|
||
- Key experimental or analytical steps
|
||
- Result synthesis and conclusion formation
|
||
|
||
Make the flowchart as detailed as possible while maintaining clarity.
|
||
If this segment does not contain methodology information, set this field to null.]"
|
||
}}
|
||
"""
|
||
|
||
try:
|
||
result = await call_deepseek_api_async([
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
])
|
||
|
||
if result:
|
||
print(f"第 {i+1} 个段落分析完成")
|
||
analysis_results.append(result)
|
||
# 等待一小段时间避免API限制
|
||
await asyncio.sleep(1)
|
||
else:
|
||
print(f"第 {i+1} 个段落分析失败")
|
||
except Exception as e:
|
||
print(f"分析段落时出错: {str(e)}")
|
||
continue
|
||
|
||
return analysis_results
|
||
|
||
async def merge_analysis_results(results: List[dict]) -> dict:
|
||
"""合并多个分析结果"""
|
||
if not results:
|
||
return {}
|
||
|
||
print(f"开始合并 {len(results)} 个分析结果...")
|
||
|
||
system_prompt = """
|
||
You are an AI assistant tasked with merging multiple analysis results of different parts of the same academic paper.
|
||
Generate a comprehensive merged analysis in JSON format.
|
||
The JSON structure must strictly follow the provided template.
|
||
Ensure the merged result is coherent and eliminates redundancy.
|
||
For the flowchart, select the most comprehensive one from the input results, or combine multiple flowcharts if they contain complementary information.
|
||
When combining flowcharts, ensure the result follows Mermaid syntax rules and avoids special characters.
|
||
Create a detailed and comprehensive flowchart that accurately represents the paper's complete research methodology.
|
||
"""
|
||
|
||
user_prompt = f"""Merge the following analysis results into a single coherent analysis:
|
||
Analysis Results: {json.dumps(results, ensure_ascii=False)}
|
||
|
||
Generate a JSON response with the following structure:
|
||
|
||
{{
|
||
"1. Basic Information": {{
|
||
"author": "[Merged author information]",
|
||
"publication_date": "[Publication date]",
|
||
"title": "[Complete title]",
|
||
"journal_publisher": "[Journal/publisher information]",
|
||
"document_type": "[Document type]"
|
||
}},
|
||
"2. Content Analysis": {{
|
||
"abstract": "[Complete abstract]",
|
||
"research_purpose": "[Comprehensive research objectives]",
|
||
"methodology": "[Complete methodology description]",
|
||
"main_arguments": "[Comprehensive theoretical frameworks and arguments]",
|
||
"conclusions": "[Complete findings and conclusions]",
|
||
"innovations": "[Complete list of innovations]"
|
||
}},
|
||
"flowchart": "[Select or combine the flowcharts following these rules:
|
||
1. Use 'graph TD' for top-down flow
|
||
2. Each node should be in format: id[text] where:
|
||
- id is a unique identifier (A, B1, B2, etc.)
|
||
- text should be simple and clear, using only letters, numbers, and spaces
|
||
- DO NOT use any special characters including parentheses, colons, commas
|
||
- abbreviations should be written without parentheses, e.g., 'DNN' not '(DNN)'
|
||
- use space instead of special characters, e.g., 'Deep Learning Model' not 'Deep-Learning/Model'
|
||
3. Connections use '-->' between nodes
|
||
4. Ensure each line ends with a proper node reference
|
||
|
||
Create a detailed flowchart that shows:
|
||
- Research objectives and questions
|
||
- All major research methods used
|
||
- Data collection and analysis processes
|
||
- Key experimental or analytical steps
|
||
- Result synthesis and conclusion formation
|
||
|
||
Make the flowchart as detailed as possible while maintaining clarity.
|
||
If no valid flowchart is found in any segment, set this field to null.]"
|
||
}}
|
||
"""
|
||
|
||
result = await call_deepseek_api_async([
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
])
|
||
|
||
if result:
|
||
print("分析结果合并完成")
|
||
else:
|
||
print("分析结果合并失败")
|
||
|
||
return result
|
||
async def analyze_reference_document_async(content: str):
|
||
"""分析文献的基本信息和内容"""
|
||
system_prompt = """
|
||
You are an AI assistant tasked with analyzing academic paper.
|
||
Generate a comprehensive analysis in JSON format covering both basic information and content analysis.
|
||
The JSON structure must strictly follow the provided template.
|
||
Also generate a Mermaid flowchart code to visualize the research methodology.
|
||
Ensure the flowchart follows strict formatting rules to avoid parsing errors.
|
||
Create a detailed and comprehensive flowchart that accurately represents the paper's research methodology.
|
||
"""
|
||
|
||
user_prompt = f"""Analyze the following paper and extract all relevant information:
|
||
Content: {content}
|
||
|
||
Generate a JSON response with the following structure:
|
||
|
||
{{
|
||
"1. Basic Information": {{
|
||
"author": "[Author name(s) and affiliations]",
|
||
"publication_date": "[Publication date in YYYY-MM format]",
|
||
"title": "[Full title of the document]",
|
||
"journal_publisher": "[Journal name or publisher details]",
|
||
"document_type": "[Type: journal article/book/conference paper etc.]"
|
||
}},
|
||
"2. Content Analysis": {{
|
||
"abstract": "[Paper abstract]",
|
||
"research_purpose": "[Main objectives and research questions]",
|
||
"methodology": "[Research methods, data collection and analysis approaches]",
|
||
"main_arguments": "[Key theoretical frameworks and arguments]",
|
||
"conclusions": "[Primary findings and conclusions]",
|
||
"innovations": "[Novel contributions and original aspects]"
|
||
}},
|
||
"flowchart": "[Generate a Mermaid flowchart code that visualizes the research methodology. Follow these rules:
|
||
1. Use 'graph TD' for top-down flow
|
||
2. Each node should be in format: id[text] where:
|
||
- id is a unique identifier (A, B1, B2, etc.)
|
||
- text should be simple and clear, using only letters, numbers, and spaces
|
||
- DO NOT use any special characters including parentheses, colons, commas
|
||
- abbreviations should be written without parentheses, e.g., 'DNN' not '(DNN)'
|
||
- use space instead of special characters, e.g., 'Deep Learning Model' not 'Deep-Learning/Model'
|
||
3. Connections use '-->' between nodes
|
||
4. Ensure each line ends with a proper node reference
|
||
|
||
Create a detailed flowchart that shows:
|
||
- Research objectives and questions
|
||
- All major research methods used
|
||
- Data collection and analysis processes
|
||
- Key experimental or analytical steps
|
||
- Result synthesis and conclusion formation
|
||
|
||
Make the flowchart as detailed as possible while maintaining clarity.
|
||
If the paper does not contain clear methodology information, set this field to null.]"
|
||
}}
|
||
"""
|
||
|
||
return await call_deepseek_api_async([
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
])
|
||
|
||
async def analyze_reference_value_async(content_analysis: dict):
|
||
"""基于内容分析结果评估文献的价值"""
|
||
system_prompt = """
|
||
You are an AI assistant tasked with evaluating the value of academic paper based on its content analysis.
|
||
Generate a comprehensive value evaluation in JSON format.
|
||
The JSON structure must strictly follow the provided template.
|
||
"""
|
||
|
||
user_prompt = f"""Based on the following content analysis, evaluate the paper's value:
|
||
Content Analysis: {json.dumps(content_analysis, ensure_ascii=False)}
|
||
|
||
Generate a JSON response with the following structure:
|
||
|
||
{{
|
||
"3. Value Evaluation": {{
|
||
"academic_contribution": "[Significance to the field of study]",
|
||
"practical_significance": "[Real-world applications and implications]",
|
||
"limitations": "[Research constraints and weaknesses]",
|
||
"implications": "[Suggestions for future research and practice]"
|
||
}}
|
||
}}
|
||
"""
|
||
|
||
return await call_deepseek_api_async([
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
])
|
||
|
||
@app.post("/lab/experiments/{experiment_id}/qa")
|
||
async def ask_experiment_question(
|
||
experiment_id: str,
|
||
question: QuestionModel,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""向实验报告提问(异步)"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 验证实验是否存在
|
||
experiment = await db.experiments.find_one({"_id": ObjectId(experiment_id)})
|
||
if not experiment:
|
||
raise HTTPException(status_code=404, detail="实验不存在")
|
||
|
||
# 生成任务ID
|
||
task_id = str(ObjectId())
|
||
|
||
# 创建后台任务
|
||
asyncio.create_task(process_experiment_question(
|
||
task_id=task_id,
|
||
experiment_id=experiment_id,
|
||
question=question.question,
|
||
experiment=experiment
|
||
))
|
||
|
||
return {
|
||
"task_id": task_id,
|
||
"status": TaskStatus.PENDING,
|
||
"message": "问题已提交，正在处理中"
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error in ask_experiment_question: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
async def process_experiment_question(task_id: str, experiment_id: str, question: str, experiment: dict):
|
||
"""处理实验问答的后台任务"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 更新任务状态为处理中
|
||
await redis.select(208) # 使用db208存储任务状态
|
||
await redis.hset(
|
||
f"task:{task_id}",
|
||
mapping={
|
||
"status": TaskStatus.PROCESSING,
|
||
"experiment_id": experiment_id,
|
||
"question": question
|
||
}
|
||
)
|
||
|
||
# 从Redis获取分析报告
|
||
await redis.select(201)
|
||
report_key = f"experiment_report:{experiment_id}"
|
||
report_data = await redis.get(report_key)
|
||
|
||
if not report_data:
|
||
raise Exception("实验分析报告不存在，请先进行分析")
|
||
|
||
# 构建系统提示和用户提示
|
||
system_prompt = """
|
||
你是一个专业的实验助手，负责回答关于实验报告的问题。
|
||
你应该基于实验报告提供准确、专业的回答。
|
||
回答应当简洁明了，并尽可能引用报告中的具体内容。
|
||
"""
|
||
|
||
# 构建上下文
|
||
context = {
|
||
"实验名称": experiment.get("experiment_name", "未知实验"),
|
||
"分析报告": json.loads(report_data)
|
||
}
|
||
|
||
messages = [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": f"基于以下实验报告回答问题：\n\n实验信息：{json.dumps(context, ensure_ascii=False)}\n\n问题：{question}"}
|
||
]
|
||
|
||
# 调用DeepSeek API获取回答
|
||
response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=messages
|
||
)
|
||
answer = response.choices[0].message.content
|
||
|
||
# 保存对话历史到Redis db207
|
||
await redis.select(207)
|
||
chat_history_key = f"experiment_chat_history:{experiment_id}"
|
||
|
||
# 获取现有历史记录
|
||
existing_history = await redis.get(chat_history_key)
|
||
history = json.loads(existing_history) if existing_history else []
|
||
|
||
# 添加新的对话
|
||
history.append({
|
||
"question": question,
|
||
"answer": answer,
|
||
"timestamp": datetime.now(timezone.utc).isoformat()
|
||
})
|
||
|
||
# 保存更新后的历史记录
|
||
await redis.set(chat_history_key, json.dumps(history))
|
||
|
||
# 更新任务状态为完成
|
||
await redis.select(208)
|
||
await redis.hset(
|
||
f"task:{task_id}",
|
||
mapping={
|
||
"status": TaskStatus.COMPLETED,
|
||
"answer": answer,
|
||
"experiment_name": experiment.get("experiment_name")
|
||
}
|
||
)
|
||
|
||
except Exception as e:
|
||
print(f"Error processing question: {e}")
|
||
# 更新任务状态为失败
|
||
await redis.select(208)
|
||
await redis.hset(
|
||
f"task:{task_id}",
|
||
mapping={
|
||
"status": TaskStatus.FAILED,
|
||
"error": str(e)
|
||
}
|
||
)
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.get("/lab/experiments/{experiment_id}/qa/history")
|
||
async def get_experiment_qa_history(
|
||
experiment_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取实验问答历史记录"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 从Redis db207获取对话历史
|
||
await redis.select(207)
|
||
chat_history_key = f"experiment_chat_history:{experiment_id}"
|
||
|
||
history_data = await redis.get(chat_history_key)
|
||
if not history_data:
|
||
return []
|
||
|
||
return json.loads(history_data)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting QA history: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.post("/lab/projects/{project_id}/qa")
|
||
async def ask_project_question(
|
||
project_id: str,
|
||
question: QuestionModel,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""向项目报告提问(异步)"""
|
||
db = await get_database()
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 验证项目是否存在
|
||
project = await db.projects.find_one({"_id": ObjectId(project_id)})
|
||
if not project:
|
||
raise HTTPException(status_code=404, detail="项目不存在")
|
||
|
||
# 生成任务ID
|
||
task_id = str(ObjectId())
|
||
|
||
# 创建后台任务
|
||
asyncio.create_task(process_project_question(
|
||
task_id=task_id,
|
||
project_id=project_id,
|
||
question=question.question,
|
||
project=project
|
||
))
|
||
|
||
return {
|
||
"task_id": task_id,
|
||
"status": TaskStatus.PENDING,
|
||
"message": "问题已提交，正在处理中"
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"Error in ask_project_question: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
async def process_project_question(task_id: str, project_id: str, question: str, project: dict):
|
||
"""处理项目问答的后台任务"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 更新任务状态为处理中
|
||
await redis.select(208) # 使用db208存储任务状态
|
||
await redis.hset(
|
||
f"task:{task_id}",
|
||
mapping={
|
||
"status": TaskStatus.PROCESSING,
|
||
"project_id": project_id,
|
||
"question": question
|
||
}
|
||
)
|
||
|
||
# 从Redis获取分析报告
|
||
await redis.select(202)
|
||
report_key = f"project_report:{project_id}"
|
||
report_data = await redis.get(report_key)
|
||
|
||
if not report_data:
|
||
raise Exception("项目分析报告不存在，请先进行分析")
|
||
|
||
# 构建系统提示和用户提示
|
||
system_prompt = """
|
||
你是一个专业的项目助手，负责回答关于项目报告的问题。
|
||
你应该基于项目报告提供准确、专业的回答。
|
||
回答应当简洁明了，并尽可能引用报告中的具体内容。
|
||
"""
|
||
|
||
# 构建上下文
|
||
context = {
|
||
"项目名称": project.get("project_name", "未知项目"),
|
||
"分析报告": json.loads(report_data)
|
||
}
|
||
|
||
messages = [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": f"基于以下项目报告回答问题：\n\n项目信息：{json.dumps(context, ensure_ascii=False)}\n\n问题：{question}"}
|
||
]
|
||
|
||
# 调用DeepSeek API获取回答
|
||
response = client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=messages
|
||
)
|
||
answer = response.choices[0].message.content
|
||
|
||
# 保存对话历史到Redis db207
|
||
await redis.select(207)
|
||
chat_history_key = f"project_chat_history:{project_id}"
|
||
|
||
# 获取现有历史记录
|
||
existing_history = await redis.get(chat_history_key)
|
||
history = json.loads(existing_history) if existing_history else []
|
||
|
||
# 添加新的对话
|
||
history.append({
|
||
"question": question,
|
||
"answer": answer,
|
||
"timestamp": datetime.now(timezone.utc).isoformat()
|
||
})
|
||
|
||
# 保存更新后的历史记录
|
||
await redis.set(chat_history_key, json.dumps(history))
|
||
|
||
# 更新任务状态为完成
|
||
await redis.select(208)
|
||
await redis.hset(
|
||
f"task:{task_id}",
|
||
mapping={
|
||
"status": TaskStatus.COMPLETED,
|
||
"answer": answer,
|
||
"project_name": project.get("project_name")
|
||
}
|
||
)
|
||
|
||
except Exception as e:
|
||
print(f"Error processing question: {e}")
|
||
# 更新任务状态为失败
|
||
await redis.select(208)
|
||
await redis.hset(
|
||
f"task:{task_id}",
|
||
mapping={
|
||
"status": TaskStatus.FAILED,
|
||
"error": str(e)
|
||
}
|
||
)
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
@app.get("/lab/projects/{project_id}/qa/history")
|
||
async def get_project_qa_history(
|
||
project_id: str,
|
||
current_user: UserModel = Depends(get_current_user)
|
||
):
|
||
"""获取项目问答历史记录"""
|
||
redis = await get_redis()
|
||
|
||
try:
|
||
# 从Redis db207获取对话历史
|
||
await redis.select(207)
|
||
chat_history_key = f"project_chat_history:{project_id}"
|
||
|
||
history_data = await redis.get(chat_history_key)
|
||
if not history_data:
|
||
return []
|
||
|
||
return json.loads(history_data)
|
||
|
||
except Exception as e:
|
||
print(f"Error getting QA history: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
finally:
|
||
try:
|
||
await redis.aclose()
|
||
except Exception as e:
|
||
print(f"Error closing Redis connection: {e}")
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
uvicorn.run(app, host="0.0.0.0", port=6000)
|