75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
"""Analysis service for VoIdea."""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.agents.models import AgentReport
|
|
from app.schemas.idea import AI_AGENT_ROLES
|
|
from app.tasks.analysis import analyze_idea
|
|
|
|
|
|
class AnalysisService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def start_analysis(self, idea_id: str) -> dict:
|
|
"""Start full analysis of an idea across all AI agent roles.
|
|
|
|
Args:
|
|
idea_id: UUID of the idea to analyze
|
|
|
|
Returns:
|
|
Dict with idea_id, task_count, and list of celery task_ids
|
|
"""
|
|
tasks = []
|
|
for role in AI_AGENT_ROLES:
|
|
task = analyze_idea.delay(idea_id, role)
|
|
tasks.append({
|
|
"role": role,
|
|
"task_id": task.id,
|
|
})
|
|
|
|
return {
|
|
"idea_id": idea_id,
|
|
"task_count": len(tasks),
|
|
"tasks": tasks,
|
|
}
|
|
|
|
async def get_analysis_results(
|
|
self, idea_id: str, role: str | None = None,
|
|
) -> list[dict]:
|
|
"""Get analysis results for an idea.
|
|
|
|
Args:
|
|
idea_id: Filter by idea
|
|
role: Optional filter by agent role
|
|
|
|
Returns:
|
|
List of analysis reports
|
|
"""
|
|
query = select(AgentReport).where(
|
|
AgentReport.details["idea_id"].as_string() == idea_id
|
|
)
|
|
|
|
if role:
|
|
query = query.where(AgentReport.agent_id == f"ai_{role}")
|
|
|
|
query = query.order_by(AgentReport.created_at.desc())
|
|
result = await self.db.execute(query)
|
|
reports = result.scalars().all()
|
|
|
|
return [
|
|
{
|
|
"id": str(r.id),
|
|
"role": r.agent_id.replace("ai_", ""),
|
|
"status": r.status,
|
|
"message": r.message,
|
|
"success": r.success,
|
|
"duration_ms": r.duration_ms,
|
|
"created_at": r.created_at.isoformat() if r.created_at else None,
|
|
}
|
|
for r in reports
|
|
]
|