Spaces:
Running
Running
| from datetime import datetime, timezone | |
| from typing import Optional, Dict, Any | |
| from yake import KeywordExtractor | |
| from app.services.chathistory import ChatSession | |
| from app.services.llm_model import Model | |
| from app.services.environmental_condition import EnvironmentalData | |
| from app.services.prompts import * | |
| from app.services.MagicConvert import MagicConvert | |
| class Report: | |
| def __init__(self, token: str, session_id: Optional[str] = None): | |
| self.token = token | |
| self.session_id = session_id | |
| self.chat_session = ChatSession(token, session_id) | |
| self.user_city = self.chat_session.get_city() | |
| city = self.user_city if self.user_city else '' | |
| self.environment_data = EnvironmentalData(city) | |
| self.markitdown = MagicConvert() | |
| def extract_keywords_yake(self, text: str, max_ngram_size: int = 2, num_keywords: int = 4) -> list: | |
| kw_extractor = KeywordExtractor( | |
| lan="en", | |
| n=max_ngram_size, | |
| top=num_keywords, | |
| features=None | |
| ) | |
| keywords = kw_extractor.extract_keywords(text) | |
| return [kw[0] for kw in keywords] | |
| def ensure_valid_session(self, title: str = None) -> str: | |
| if not self.session_id or not self.session_id.strip(): | |
| self.chat_session.create_new_session(title=title) | |
| self.session_id = self.chat_session.session_id | |
| else: | |
| try: | |
| if not self.chat_session.validate_session(self.session_id, title=title): | |
| self.chat_session.create_new_session(title=title) | |
| self.session_id = self.chat_session.session_id | |
| except ValueError: | |
| self.chat_session.create_new_session(title=title) | |
| self.session_id = self.chat_session.session_id | |
| return self.session_id | |
| def process_chat(self, query: str, report_file: str, file_type: Optional[str] = None) -> Dict[str, Any]: | |
| try: | |
| profile = self.chat_session.get_name_and_age() | |
| self.chat_session.update_title(self.session_id, query) | |
| self.session_id = self.ensure_valid_session(title=query) | |
| language = self.chat_session.get_language().lower() | |
| language_prompt = LANGUAGE_RESPONSE_PROMPT.format(language=language) | |
| if not report_file or not file_type: | |
| return { | |
| "error": "Report file or file type missing", | |
| "query": query, | |
| "response": "Sorry, report file or file type is missing.", | |
| "timestamp": datetime.now(timezone.utc).isoformat() | |
| } | |
| report_file_name = report_file + " (File Uploaded)" | |
| conversion_result = self.markitdown.magic(report_file) | |
| report_text = conversion_result.text_content | |
| prompt = MEDICAL_REPORT_ANALYSIS_PROMPT.format( | |
| report=report_text, | |
| current_query=query | |
| ) | |
| response = Model().response = Model().llm(prompt + "\n" + language_prompt , query) | |
| keywords = self.extract_keywords_yake(response) | |
| chat_data = { | |
| "query": report_file_name + "\n" +query, | |
| "response": response, | |
| "references": "", | |
| "page_no": "", | |
| "keywords": keywords, | |
| "images": "", | |
| "context": report_text, | |
| "timestamp": datetime.now(timezone.utc).isoformat(), | |
| "session_id": self.chat_session.session_id | |
| } | |
| if not self.chat_session.save_chat(chat_data): | |
| raise ValueError("Failed to save chat message") | |
| return chat_data | |
| except Exception as e: | |
| return { | |
| "error": str(e), | |
| "query": query, | |
| "response": "Sorry, there was an error processing your request.", | |
| "timestamp": datetime.now(timezone.utc).isoformat() | |
| } |