from src.extensions import db from sqlalchemy import case, func from flask_login import current_user from sqlalchemy.orm import joinedload from datetime import datetime, timedelta from src.models import Chat, Message, Agente class ChatService: @staticmethod def get_conversation_history(agent_id=None): """ Busca o histórico de conversas para o usuário logado e agente específico (se fornecido), estruturado para o frontend. """ try: order_criterion = Chat.updated_at.desc() if hasattr(Chat, 'updated_at') else Chat.created_at.desc() query = db.session.query(Chat).filter(Chat.user_id == current_user.id) if agent_id is not None: query = query.filter(Chat.agent_id == agent_id) user_chats = query.options(joinedload(Chat.messages)).order_by(order_criterion).all() history_list = [] for chat in user_chats: messages_sorted = sorted(chat.messages, key=lambda m: m.created_at) first_user_message_obj = next((m for m in messages_sorted if m.is_user), None) first_agent_message_obj = next((m for m in messages_sorted if not m.is_user), None) preview_parts = [] if first_user_message_obj and first_user_message_obj.message: preview_parts.append(f"Você: {first_user_message_obj.message}") if first_agent_message_obj and first_agent_message_obj.message: preview_parts.append(f"Agente: {first_agent_message_obj.message}") if not preview_parts: preview = messages_sorted[0].message if messages_sorted else "Nenhuma mensagem." else: preview = " ... ".join(preview_parts) preview_text = (preview[:150] + '...') if len(preview) > 150 else preview actual_first_message_for_duration = messages_sorted[0].created_at if messages_sorted else None last_message_for_duration = messages_sorted[-1].created_at if messages_sorted else None message_count = len(messages_sorted) duration_str = "N/A" if actual_first_message_for_duration and last_message_for_duration: ts_inicio = actual_first_message_for_duration ts_fim = last_message_for_duration if ts_inicio and ts_fim: duration_seconds = (ts_fim - ts_inicio).total_seconds() if duration_seconds < 0: duration_seconds = 0 if duration_seconds < 60: duration_str = f"{int(duration_seconds)}s" else: duration_str = f"{int(duration_seconds // 60)}m" model_display_name = "Agente Desconhecido" if chat.agent_id: agent = db.session.get(Agente, chat.agent_id) if agent: model_display_name = agent.name if hasattr(agent, 'name') and agent.name else f"Agente {agent.id}" chat_date = chat.updated_at if hasattr(Chat, 'updated_at') and chat.updated_at else chat.created_at history_item = { "id": str(chat.id), "title": chat.title or f"Conversa {chat.id}", "model": model_display_name, "date": chat_date.isoformat() if chat_date else datetime.now().isoformat(), "preview": preview_text, "messageCount": message_count or 0, "duration": duration_str } history_list.append(history_item) return history_list, None except Exception as e: print(f"Erro crítico ao buscar histórico de conversas: {str(e)}") return [], f"Erro interno ao processar o histórico: {str(e)}" @staticmethod def create_chat(agent_id): try: print(f'[DEBUG] ChatService.create_chat: Criando chat para agent_id={agent_id}') print(f'[DEBUG] ChatService.create_chat: current_user.id={current_user.id}') chat = Chat( user_id=current_user.id, agent_id=agent_id, created_at=datetime.now() ) db.session.add(chat) db.session.flush() print(f'[DEBUG] ChatService.create_chat: Chat criado com ID={chat.id}') agente = Agente.query.get(agent_id) print(f'[DEBUG] ChatService.create_chat: Agente encontrado: {agente.nome if agente else None}') if agente and agente.welcome_message and agente.welcome_message.strip(): print(f'[DEBUG] ChatService.create_chat: Agente possui mensagem de boas-vindas: {repr(agente.welcome_message[:50])}...') print(f'[DEBUG] ChatService.create_chat: Mensagem será mostrada apenas no frontend') else: print(f'[DEBUG] ChatService.create_chat: Agente não possui mensagem de boas-vindas') db.session.commit() print(f'[DEBUG] ChatService.create_chat: Chat criado com sucesso - Commit realizado') return chat, None except Exception as e: print(f'[ERROR] ChatService.create_chat: Erro ao criar chat: {str(e)}') import traceback print(f'[ERROR] ChatService.create_chat: Traceback: {traceback.format_exc()}') db.session.rollback() return None, str(e) @staticmethod def add_message(chat_id, message, is_user=True): try: chat_message = Message( chat_id=chat_id, message=message, is_user=is_user, created_at=datetime.now() ) db.session.add(chat_message) db.session.commit() if is_user: from src.security.audit import log_audit log_audit( action="Mensagem do usuario:", object_type="chat", object_id=chat_id, details=f"Mensagem enviada: {message[:10000]}" ) chat = Chat.query.get(chat_id) if not is_user and chat and (not chat.title or chat.title.strip() == ""): from src.services.gemini_service import GeminiService messages = Message.query.filter_by(chat_id=chat_id).order_by(Message.created_at).limit(10).all() first_user = next((m for m in messages if m.is_user), None) first_ia = next((m for m in messages if not m.is_user), None) if first_user and first_ia: context = f"Usuário: {first_user.message}\nIA: {first_ia.message}" else: context = "\n".join([f"Usuário: {m.message}" if m.is_user else f"IA: {m.message}" for m in messages]) prompt = f"Dê um título curto e descritivo em português para esta conversa, considerando a mensagem do usuário e a resposta da IA:\n{context}" try: gemini_instance = GeminiService() resposta_ia = gemini_instance.generate_response(prompt, max_tokens=40, model="gemini-2.5-flash") titulo_ia = resposta_ia["response"] if isinstance(resposta_ia, dict) and "response" in resposta_ia else None if titulo_ia: chat.title = titulo_ia.strip().replace('\n', ' ') db.session.commit() else: words = message.split() title = " ".join(words[:10]) + ("..." if len(words) > 10 else "") chat.title = title db.session.commit() except Exception: words = message.split() title = " ".join(words[:10]) + ("..." if len(words) > 10 else "") chat.title = title db.session.commit() return chat_message, None except Exception as e: db.session.rollback() return None, str(e) @staticmethod def get_user_chats(): return Chat.query.filter_by(user_id=current_user.id)\ .options(joinedload(Chat.messages))\ .order_by(Chat.created_at.desc()).all() @staticmethod def get_chat(chat_id): chat = Chat.query.filter_by(id=chat_id, user_id=current_user.id)\ .options(joinedload(Chat.messages))\ .first() if not chat: from src.discord.discord_webhook import send_discord_audit_log send_discord_audit_log( action="Acesso não autorizado", user_id=current_user.id, object_type="Chat", object_id=chat_id, details="Tentativa de acesso ao chat sem permissão", ip_address=None ) return chat @staticmethod def delete_chat(chat_id): try: chat = Chat.query.filter_by(id=chat_id, user_id=current_user.id).first() if chat: db.session.delete(chat) db.session.commit() return True, None return False, "Chat não encontrado" except Exception as e: db.session.rollback() return False, str(e) def get_dashboard_stats(period_days: int = 7): """ Calcula as estatísticas para o dashboard. period_days: O número de dias para considerar no período (ex: 7, 30). """ try: start_date = datetime.now() - timedelta(days=period_days) total_conversations = db.session.query(func.count(Chat.id))\ .filter(Chat.user_id == current_user.id, Chat.created_at >= start_date)\ .scalar() or 0 total_messages = db.session.query(func.count(Message.id))\ .join(Chat, Message.chat_id == Chat.id)\ .filter(Chat.user_id == current_user.id, Message.created_at >= start_date)\ .scalar() or 0 avg_response_time_query = db.session.query(func.avg(Message.response_time))\ .join(Chat, Message.chat_id == Chat.id)\ .filter( Chat.user_id == current_user.id, Message.created_at >= start_date, Message.is_user == False, Message.response_time.isnot(None) )\ .scalar() avg_response_time = round(avg_response_time_query, 2) if avg_response_time_query else 0.0 total_tokens = 0 messages_subquery = db.session.query( Message.chat_id, func.count(Message.id).label('message_count_for_chat'), func.avg(case((Message.is_user == False, Message.response_time), else_=None)).label('avg_response_time_for_chat') ).filter(Message.created_at >= start_date).group_by(Message.chat_id).subquery() model_usage_query = db.session.query( Agente.name.label('model_name'), func.count(Chat.id).label('conversations'), func.sum(messages_subquery.c.message_count_for_chat).label('messages'), func.avg(messages_subquery.c.avg_response_time_for_chat).label('avg_time') ).join(Agente, Chat.agent_id == Agente.id)\ .outerjoin(messages_subquery, Chat.id == messages_subquery.c.chat_id)\ .filter(Chat.user_id == current_user.id, Chat.created_at >= start_date)\ .group_by(Agente.name)\ .all() model_usage_dict = {} for row in model_usage_query: model_usage_dict[row.model_name or "Desconhecido"] = { "conversations": row.conversations or 0, "messages": int(row.messages or 0), "tokens": 0, "avg_time": round(row.avg_time, 2) if row.avg_time else 0.0 } activity_data = {} daily_conversations = db.session.query( func.date(Chat.created_at).label('day'), func.count(Chat.id).label('num_conversations') ).filter( Chat.user_id == current_user.id, Chat.created_at >= start_date ).group_by(func.date(Chat.created_at)).order_by(func.date(Chat.created_at)).all() for row in daily_conversations: day_str = row.day.isoformat() if hasattr(row.day, 'isoformat') else str(row.day) if day_str not in activity_data: activity_data[day_str] = {'conversations': 0, 'messages': 0} activity_data[day_str]['conversations'] = row.num_conversations daily_messages = db.session.query( func.date(Message.created_at).label('day'), func.count(Message.id).label('num_messages') ).join(Chat, Message.chat_id == Chat.id)\ .filter( Chat.user_id == current_user.id, Message.created_at >= start_date ).group_by(func.date(Message.created_at)).order_by(func.date(Message.created_at)).all() for row in daily_messages: day_str = row.day.isoformat() if hasattr(row.day, 'isoformat') else str(row.day) if day_str not in activity_data: activity_data[day_str] = {'conversations': 0, 'messages': 0} activity_data[day_str]['messages'] = row.num_messages stats_result = { "total_conversations": total_conversations, "total_messages": total_messages, "avg_response_time": avg_response_time, "total_tokens": total_tokens, "model_usage": model_usage_dict, "activity_data": activity_data, "period_days": period_days } return stats_result, None except Exception as e: print(f"Erro ao calcular estatísticas: {str(e)}") return None, str(e)