125 lines
5.1 KiB
Python
125 lines
5.1 KiB
Python
import ast
|
|
import os
|
|
from datetime import datetime as dt, timezone as tz
|
|
|
|
import chardet
|
|
from flask import request, redirect, flash, render_template, Blueprint, session, current_app
|
|
from flask_security import roles_accepted, current_user
|
|
from sqlalchemy import desc
|
|
from sqlalchemy.orm import joinedload
|
|
from werkzeug.datastructures import FileStorage
|
|
from werkzeug.utils import secure_filename
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
import requests
|
|
from requests.exceptions import SSLError
|
|
from urllib.parse import urlparse
|
|
import io
|
|
|
|
from common.models.document import Embedding, DocumentVersion
|
|
from common.models.interaction import ChatSession, Interaction, InteractionEmbedding
|
|
from common.extensions import db
|
|
from .document_forms import AddDocumentForm, AddURLForm, EditDocumentForm, EditDocumentVersionForm
|
|
from common.utils.middleware import mw_before_request
|
|
from common.utils.celery_utils import current_celery
|
|
from common.utils.nginx_utils import prefixed_url_for
|
|
from common.utils.view_assistants import form_validation_failed, prepare_table_for_macro
|
|
|
|
interaction_bp = Blueprint('interaction_bp', __name__, url_prefix='/interaction')
|
|
|
|
|
|
@interaction_bp.before_request
|
|
def log_before_request():
|
|
current_app.logger.debug(f"Before request (interaction_bp): {request.method} {request.url}")
|
|
|
|
|
|
@interaction_bp.after_request
|
|
def log_after_request(response):
|
|
current_app.logger.debug(
|
|
f"After request (interaction_bp): {request.method} {request.url} - Status: {response.status}")
|
|
return response
|
|
|
|
|
|
@interaction_bp.before_request
|
|
def before_request():
|
|
try:
|
|
mw_before_request()
|
|
except Exception as e:
|
|
current_app.logger.error(f'Error switching schema in Interaction Blueprint: {e}')
|
|
for role in current_user.roles:
|
|
current_app.logger.debug(f'User {current_user.email} has role {role.name}')
|
|
raise
|
|
|
|
|
|
@interaction_bp.route('/chat_sessions', methods=['GET', 'POST'])
|
|
def chat_sessions():
|
|
page = request.args.get('page', 1, type=int)
|
|
per_page = request.args.get('per_page', 10, type=int)
|
|
|
|
query = ChatSession.query.order_by(desc(ChatSession.session_start))
|
|
|
|
pagination = query.paginate(page=page, per_page=per_page, error_out=False)
|
|
docs = pagination.items
|
|
|
|
rows = prepare_table_for_macro(docs, [('id', ''), ('session_id', ''), ('session_start', ''), ('session_end', '')])
|
|
|
|
return render_template('interaction/chat_sessions.html', rows=rows, pagination=pagination)
|
|
|
|
|
|
@interaction_bp.route('/handle_chat_session_selection', methods=['POST'])
|
|
@roles_accepted('Super User', 'Tenant Admin')
|
|
def handle_chat_session_selection():
|
|
chat_session_identification = request.form['selected_row']
|
|
cs_id = ast.literal_eval(chat_session_identification).get('value')
|
|
|
|
action = request.form['action']
|
|
|
|
match action:
|
|
case 'view_chat_session':
|
|
return redirect(prefixed_url_for('interaction_bp.view_chat_session', chat_session_id=cs_id))
|
|
|
|
# Add more conditions for other actions
|
|
return redirect(prefixed_url_for('interaction_bp.chat_sessions'))
|
|
|
|
|
|
@interaction_bp.route('/view_chat_session/<int:chat_session_id>', methods=['GET'])
|
|
@roles_accepted('Super User', 'Tenant Admin')
|
|
def view_chat_session(chat_session_id):
|
|
chat_session = ChatSession.query.get_or_404(chat_session_id)
|
|
interactions = (Interaction.query
|
|
.filter_by(chat_session_id=chat_session.id)
|
|
.order_by(Interaction.question_at)
|
|
.all())
|
|
|
|
# Fetch all related embeddings for the interactions in this session
|
|
embedding_query = (db.session.query(InteractionEmbedding.interaction_id,
|
|
DocumentVersion.url,
|
|
DocumentVersion.object_name)
|
|
.join(Embedding, InteractionEmbedding.embedding_id == Embedding.id)
|
|
.join(DocumentVersion, Embedding.doc_vers_id == DocumentVersion.id)
|
|
.filter(InteractionEmbedding.interaction_id.in_([i.id for i in interactions])))
|
|
|
|
# Create a dictionary to store embeddings for each interaction
|
|
embeddings_dict = {}
|
|
for interaction_id, url, object_name in embedding_query:
|
|
if interaction_id not in embeddings_dict:
|
|
embeddings_dict[interaction_id] = []
|
|
embeddings_dict[interaction_id].append({'url': url, 'object_name': object_name})
|
|
|
|
return render_template('interaction/view_chat_session.html',
|
|
chat_session=chat_session,
|
|
interactions=interactions,
|
|
embeddings_dict=embeddings_dict)
|
|
|
|
|
|
@interaction_bp.route('/view_chat_session_by_session_id/<session_id>', methods=['GET'])
|
|
@roles_accepted('Super User', 'Tenant Admin')
|
|
def view_chat_session_by_session_id(session_id):
|
|
chat_session = ChatSession.query.filter_by(session_id=session_id).first_or_404()
|
|
show_chat_session(chat_session)
|
|
|
|
|
|
def show_chat_session(chat_session):
|
|
interactions = Interaction.query.filter_by(chat_session_id=chat_session.id).all()
|
|
return render_template('interaction/view_chat_session.html', chat_session=chat_session, interactions=interactions)
|
|
|