133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
import logging
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, List
|
|
import openai
|
|
from app.core.config import settings
|
|
from app.models.task import TaskPriority
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AIService:
|
|
def __init__(self):
|
|
if settings.OPENAI_API_KEY:
|
|
openai.api_key = settings.OPENAI_API_KEY
|
|
self.client = openai.OpenAI(api_key=settings.OPENAI_API_KEY)
|
|
else:
|
|
self.client = None
|
|
logger.warning("OpenAI API key not configured. AI functionality disabled.")
|
|
|
|
def suggest_optimal_time(self, due_date: datetime, priority: TaskPriority) -> Optional[datetime]:
|
|
if not self.client:
|
|
return self._fallback_scheduling(due_date, priority)
|
|
|
|
try:
|
|
current_time = datetime.utcnow()
|
|
time_until_due = due_date - current_time
|
|
|
|
if time_until_due.total_seconds() <= 0:
|
|
return current_time + timedelta(hours=1)
|
|
|
|
prompt = f"""
|
|
Based on task management best practices, suggest an optimal time to work on a task with the following details:
|
|
|
|
Current time: {current_time.strftime('%Y-%m-%d %H:%M:%S')}
|
|
Due date: {due_date.strftime('%Y-%m-%d %H:%M:%S')}
|
|
Priority: {priority.value}
|
|
Time available: {time_until_due.days} days, {time_until_due.seconds // 3600} hours
|
|
|
|
Consider:
|
|
- Task priority (urgent tasks should be scheduled sooner)
|
|
- Optimal work hours (9 AM - 5 PM on weekdays)
|
|
- Buffer time before due date
|
|
- Work-life balance
|
|
|
|
Respond with only the suggested datetime in format: YYYY-MM-DD HH:MM:SS
|
|
"""
|
|
|
|
response = self.client.chat.completions.create(
|
|
model="gpt-3.5-turbo",
|
|
messages=[
|
|
{"role": "system", "content": "You are a task scheduling AI assistant."},
|
|
{"role": "user", "content": prompt}
|
|
],
|
|
max_tokens=50,
|
|
temperature=0.3
|
|
)
|
|
|
|
suggested_time_str = response.choices[0].message.content.strip()
|
|
suggested_time = datetime.strptime(suggested_time_str, '%Y-%m-%d %H:%M:%S')
|
|
|
|
if suggested_time < current_time:
|
|
suggested_time = current_time + timedelta(hours=1)
|
|
elif suggested_time > due_date:
|
|
suggested_time = due_date - timedelta(hours=2)
|
|
|
|
return suggested_time
|
|
|
|
except Exception as e:
|
|
logger.error(f"AI scheduling failed: {str(e)}")
|
|
return self._fallback_scheduling(due_date, priority)
|
|
|
|
def _fallback_scheduling(self, due_date: datetime, priority: TaskPriority) -> datetime:
|
|
current_time = datetime.utcnow()
|
|
time_until_due = due_date - current_time
|
|
|
|
if priority == TaskPriority.URGENT:
|
|
return current_time + timedelta(hours=1)
|
|
elif priority == TaskPriority.HIGH:
|
|
return current_time + timedelta(hours=min(6, time_until_due.total_seconds() // 3600 // 4))
|
|
elif priority == TaskPriority.MEDIUM:
|
|
return current_time + timedelta(hours=min(24, time_until_due.total_seconds() // 3600 // 2))
|
|
else: # LOW priority
|
|
return current_time + timedelta(hours=min(48, time_until_due.total_seconds() // 3600 * 0.75))
|
|
|
|
def generate_task_insights(self, tasks: List) -> str:
|
|
if not self.client:
|
|
return "AI insights not available. Please configure OpenAI API key."
|
|
|
|
try:
|
|
task_summary = []
|
|
for task in tasks:
|
|
task_summary.append(f"- {task.title} (Priority: {task.priority.value}, Status: {task.status.value})")
|
|
|
|
prompt = f"""
|
|
Analyze the following task list and provide productivity insights:
|
|
|
|
{chr(10).join(task_summary)}
|
|
|
|
Provide insights on:
|
|
1. Task distribution by priority
|
|
2. Potential bottlenecks
|
|
3. Productivity recommendations
|
|
4. Time management suggestions
|
|
|
|
Keep the response concise and actionable (max 200 words).
|
|
"""
|
|
|
|
response = self.client.chat.completions.create(
|
|
model="gpt-3.5-turbo",
|
|
messages=[
|
|
{"role": "system", "content": "You are a productivity analysis AI assistant."},
|
|
{"role": "user", "content": prompt}
|
|
],
|
|
max_tokens=250,
|
|
temperature=0.7
|
|
)
|
|
|
|
return response.choices[0].message.content.strip()
|
|
|
|
except Exception as e:
|
|
logger.error(f"AI insights generation failed: {str(e)}")
|
|
return "Unable to generate insights at this time."
|
|
|
|
|
|
ai_service = AIService()
|
|
|
|
|
|
def suggest_optimal_time(due_date: datetime, priority: TaskPriority) -> Optional[datetime]:
|
|
return ai_service.suggest_optimal_time(due_date, priority)
|
|
|
|
|
|
def generate_task_insights(tasks: List) -> str:
|
|
return ai_service.generate_task_insights(tasks) |