Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| from dotenv import load_dotenv | |
| from azure.cosmos import CosmosClient | |
| from typing import List | |
| import os | |
| import requests | |
| import json | |
| import re | |
| load_dotenv() | |
| def get_db_items(offset: int, database_name: str, container_name: str, num_items: int = 10) -> List[dict]: | |
| try: | |
| client: CosmosClient = st.session_state["db_items"] | |
| database = client.get_database_client(database_name) | |
| container = database.get_container_client(container_name) | |
| query = f"SELECT * FROM c ORDER BY c._ts DESC OFFSET {offset} LIMIT {num_items}" | |
| items = list(container.query_items(query=query, enable_cross_partition_query=True)) | |
| return items | |
| except Exception as e: | |
| print(f"Fehler beim Abrufen der Daten aus der Cosmos DB: {e}") | |
| st.error("Someting went wrong. Please try again later.", icon="🚨") | |
| return [] | |
| def update_db_item(item: dict, database_name: str, container_name: str): | |
| try: | |
| database = st.session_state["db_items"].get_database_client(database_name) | |
| container = database.get_container_client(container_name) | |
| container.upsert_item(item) | |
| except Exception as e: | |
| print(f"Fehler beim Updaten des Items: {e}") | |
| st.error("Someting went wrong. Please try again later.", icon="🚨") | |
| def get_summary(input_params) -> dict: | |
| try: | |
| url = f"{os.environ.get('BASE_API')}/api/create-interview-summary" | |
| response = requests.post(url=url,data=input_params) | |
| response.raise_for_status() | |
| return response.json() | |
| except Exception as e: | |
| print(f"Fehler beim erstellen des Termins: {str(e)}") | |
| #Mache abfrage auf den HTTP Status Code | |
| if response.status_code == 500: | |
| st.info("There is currently no recording available from Zoom. Please try again in a few minutes.", icon="ℹ️") | |
| else: | |
| st.error(f"Something went wrong: {str(e)}", icon="🚨") | |
| def change_meeting_items(): | |
| st.session_state["detailed_view"] = False | |
| st.session_state["selected_item"] = None | |
| if st.session_state["meeting_type"] == "Candidate interview": | |
| st.session_state["items_to_show"] = st.session_state["interview_items"] | |
| else: | |
| st.session_state["items_to_show"] = st.session_state["assessment_items"] | |
| def back_to_overview(): | |
| st.session_state["detailed_view"] = False | |
| st.session_state["selected_item"] = None | |
| st.session_state["created_question_evaluations"] = [] | |
| def summary_button_clicked(interview_object: dict): | |
| st.session_state["detailed_view"] = True | |
| st.session_state["selected_item"] = interview_object | |
| def update_assessment_db_item(): | |
| try: | |
| client: CosmosClient = st.session_state["db_items"] | |
| database = client.get_database_client("assessment-database") | |
| container = database.get_container_client("assessments") | |
| assessment_item = st.session_state["selected_item"] | |
| assessment_item["process_status"] = "evaluations_created" | |
| for i, question in enumerate(assessment_item["questions"]): | |
| question["score"] = st.session_state[f"rating_question_{i}"] | |
| question["evaluation"] = st.session_state[f"evaluation_question_{i}"] | |
| for i, tasks in enumerate(assessment_item["coding_tasks"]): | |
| tasks["score"] = st.session_state[f"rating_coding_task_{i}"] | |
| tasks["evaluation"] = st.session_state[f"evaluation_coding_task_{i}"] | |
| container.upsert_item(assessment_item) | |
| except Exception as e: | |
| print(f"Fehler beim Updaten des Items: {e}") | |
| st.error("Someting went wrong. Please try again later.", icon="🚨") | |
| def remove_html_tags(input_string: str): | |
| clean = re.compile('<.*?>') | |
| return re.sub(clean, '', input_string) | |
| if "db_items" not in st.session_state: | |
| endpoint = os.getenv("COSMOS_DB_ENDPOINT") | |
| key = os.getenv("COSMOS_DB_KEY") | |
| client = CosmosClient(endpoint, key) | |
| st.session_state["db_items"] = client | |
| if "interview_items" not in st.session_state: | |
| initial_items = get_db_items(0, "appointment-database", "appointments", 50) | |
| st.session_state["interview_items"] = initial_items | |
| if "assessment_items" not in st.session_state: | |
| initial_items = get_db_items(0, "assessment-database", "assessments", 50) | |
| st.session_state["assessment_items"] = initial_items | |
| if "items_to_show" not in st.session_state: | |
| st.session_state["items_to_show"] = st.session_state["interview_items"] | |
| if "detailed_view" not in st.session_state: | |
| st.session_state["detailed_view"] = False | |
| if "selected_item" not in st.session_state: | |
| st.session_state["selected_item"] = None | |
| col1, col2 = st.columns([2, 1]) | |
| col1.title("Meeting Manager") | |
| col2.image("https://www.workgenius.com/wp-content/uploads/2023/03/WorkGenius_navy-1.svg") | |
| st.radio("Select a meeting type", ("Candidate interview", "Peer to peer assessment"), key="meeting_type", on_change=change_meeting_items) | |
| st.divider() | |
| if st.session_state["detailed_view"]: | |
| if st.session_state["meeting_type"] == "Candidate interview": | |
| st.subheader("Summary for meeting: "+st.session_state["selected_item"]["job_title"]) | |
| print(isinstance(st.session_state["selected_item"]["summary_recruiter"],str)) | |
| if isinstance(st.session_state["selected_item"]["summary_recruiter"],str): | |
| with st.spinner("Creating summary..."): | |
| input_params = { | |
| "meeting_id": st.session_state["selected_item"]["zoom_meeting_id"], | |
| } | |
| summary = get_summary(json.dumps(input_params)) | |
| copied_db_item = st.session_state["selected_item"].copy() | |
| if summary: | |
| if st.session_state["meeting_type"] == "Candidate interview": | |
| #Operations for candidate interviews | |
| copied_db_item["process_status"] = "summaries_created" | |
| copied_db_item["summary_recruiter"] = summary | |
| update_db_item(copied_db_item, "appointment-database", "appointments") | |
| else: | |
| #TODO: Operations for peer to peer assessments | |
| print("To implement") | |
| st.write(summary) | |
| else: | |
| st.write(st.session_state["selected_item"]["summary_recruiter"]) | |
| st.button("Back to overview", on_click=back_to_overview) | |
| else: | |
| st.subheader("Evaluation for meeting: "+st.session_state["selected_item"]["assessment_title"]) | |
| with st.expander("Interview transcript"): | |
| st.write(st.session_state["selected_item"]["interview_transcript"]) | |
| for i, question in enumerate(st.session_state["selected_item"]["questions"]): | |
| with st.expander(f"Question {i+1}: {remove_html_tags(question['question'])}"): | |
| st.number_input("Score (1-10)", key=f"rating_question_{i}", min_value=1, max_value=10) | |
| st.text_area("Your evaluation:", key=f"evaluation_question_{i}") | |
| st.write("AI based score: "+str(question["score"])) | |
| st.write("AI based evalutaion: "+question["evaluation"]) | |
| for i, tasks in enumerate(st.session_state["selected_item"]["coding_tasks"]): | |
| with st.expander(f"Coding task {i+1}: {tasks['coding_task']}"): | |
| st.number_input("Score (1-10)", key=f"rating_coding_task_{i}", min_value=1, max_value=10) | |
| st.text_area("Your evaluation:", key=f"evaluation_coding_task_{i}") | |
| st.write("AI based score: "+str(tasks["score"])) | |
| st.write("AI based evalutaion: "+tasks["evaluation"]) | |
| col_evaluations = st.columns([1,2,1]) | |
| with col_evaluations[0]: | |
| st.button("Back to overview", on_click=back_to_overview, use_container_width=True) | |
| with col_evaluations[2]: | |
| st.button("Save evaluation", on_click=update_assessment_db_item, use_container_width=True) | |
| else: | |
| st.subheader("Candidate interview:" if st.session_state["meeting_type"] == "Candidate interview" else "Peer to peer assessments:") | |
| # st.write(st.session_state["interview_items"]) | |
| meeting_title = st.columns([6,2]) | |
| meeting_title[0].write("**Meeting title:**") | |
| for i, interview in enumerate(st.session_state["items_to_show"]): | |
| cols = st.columns([6,2]) | |
| if st.session_state["meeting_type"] == "Candidate interview": | |
| cols[0].write(interview["job_title"]) | |
| cols[1].button("Summary", key=f"button_{i}", on_click=summary_button_clicked, args=(interview,)) | |
| else: | |
| cols[0].write(interview["assessment_title"]) | |
| cols[1].button("Evaluation", key=f"button_{i}", on_click=summary_button_clicked, args=(interview,)) |