from __future__ import annotations import difflib from flask import current_app import openai import os import base64 import requests import tempfile import re from DocumentManager.Constants import NET_ACCESS, TEMP_CHAT_RESPONSE, TEMP_SUMMARIZE, TEMP_TITLE from DocumentManager.DocumentsManager import DocumentManager from models.doc_manager import DocConfig from models.ai_model import Balance, ModelTokenUsage, OpenAIModel, PromptType, SystemPrompt from services.prompt import * from extensions import db from models.chats import ChatSession from utils.services import count_tokens # NEW key = "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" # Initialize OpenAI client client = openai.OpenAI(api_key=key) class ChatService: """ChatService handles AI chat, TTS, STT, and streaming logic.""" @staticmethod def summarize_history(history: list, new_user_message: str, summary_model: str = "gpt-5-mini") -> str: """ Summarize the chat history based on relevance to the new user message. Returns a concise text summary of the history that is relevant to the new message. """ if not NET_ACCESS: return f'{TEMP_SUMMARIZE}' if not history: return "" # Convert history to plain text plain_history = "" for msg in history: role = "User" if msg.sender_type == "user" else "Assistant" plain_history += f"{role}: {msg.message}\n" try: # Prompt emphasizes relevance to the new user message summary_prompt = ( "You are an assistant that summarizes conversations.\n" "Do NOT answer the user's message. Only provide a short, concise summary " "of the conversation relevant to the user's message.\n\n" f"Conversation history:\n{plain_history}\n\n" f"New user message:\n{new_user_message}\n\n" "Relevant summary for responding:" ) completion = client.responses.create( model=summary_model, input=[{"role": "user", "content": summary_prompt}], ) summary = completion.output_text.strip() OpenAIModel.update_tokens_in_db(model_name=summary_model, response=completion) return summary except Exception as e: print("Error summarizing history:", e) return "" @staticmethod def ask_openai(user_message: str, history: list, image_urls=None, resources:str="",text_model_name:str="gpt-5.1", summarize_history=True, system_prompt=None, enable_web_search=False): """ image_urls: list of image URLs Message format for images: { "role": "user", "content": [ {"type": "text", "text": "متن پیام"}, {"type": "image_url", "image_url": {"url": "DATA_URI_HERE"}} ] } """ request_options = {} if enable_web_search: request_options["tools"] = [ { "type": "web_search_preview" } ] if system_prompt is None: system_prompt = get_prompt(PromptType.TEXT) system_prompt = system_prompt + get_prompt(PromptType.ANSWER_STRUCT) # Initialize messages with system prompt (kept as existing behavior) messages = [] # Convert chat history to structured format if summarize_history: summary_text = ChatService.summarize_history(history, user_message) if summary_text: messages.append({"role": "system", "content": summary_text}) else: for msg in history: role = "user" if msg.sender_type == "user" else "assistant" messages.append({"role": role, "content": msg.message}) # Prepare user message content_list = [{"type": "input_text", "text": user_message}] # Handle multiple images if image_urls: for img_url in image_urls: relative_path = img_url.lstrip('/') abs_path = os.path.normpath(os.path.join(current_app.root_path, relative_path)) try: with open(abs_path, "rb") as img_file: b64_image = base64.b64encode(img_file.read()).decode("utf-8") content_list.append({ "type": "input_image", "image_url": f"data:image/png;base64,{b64_image}", }) except Exception as e: print(f"Error reading image {img_url}: {e}") messages.append({"role": "user", "content": content_list}) final_system_prompt = system_prompt + resources try: if NET_ACCESS: completion = client.responses.create( model=text_model_name, input=messages, instructions=final_system_prompt, **request_options # temperature=0.2 ) OpenAIModel.update_tokens_in_db(model_name=text_model_name,response=completion) reply = completion.output_text.strip() return reply, True else: return TEMP_CHAT_RESPONSE,True except Exception as e: print("OpenAI error:", e) return "", False # ---------------------------- # NEW: Session title generation # ---------------------------- @staticmethod def generate_session_title(first_user_message: str) -> str: """ Generate a short session title similar to OpenAI ChatGPT. Falls back to a trimmed version of the message if API fails. """ if not NET_ACCESS: return TEMP_SUMMARIZE text = (first_user_message or "").strip() if not text: return "New chat" # Safe fallback fallback = text.replace("\n", " ").strip() fallback = fallback[:60] + ("…" if len(fallback) > 60 else "") try: prompt = ( "Create a very short chat title (2 to 6 words). " "No quotes. No punctuation at the end. " "Use the same language as the user's message." ) title_model = "gpt-4o-mini" resp = client.responses.create( model=title_model, input=[{"role": "user", "content": text}], instructions=prompt, temperature=0.2 ) OpenAIModel.update_tokens_in_db(model_name=title_model,response=resp) out = (resp.output_text or "").strip() out = re.sub(r'[\r\n]+', ' ', out).strip() out = out.strip('"“”').strip() out = out[:80] return out or fallback except Exception: return fallback @staticmethod def generate_session_title_llm(first_user_message: str, assistant_reply: str | None = None,text_model_name:str='') -> str | None: """ Generate a short chat title using the LLM. Rules: - 3 to 7 words (or short phrase) - No quotes, no markdown, no emojis - Output ONLY the title text """ try: if not NET_ACCESS: return TEMP_TITLE user_text = (first_user_message or "").strip() bot_text = (assistant_reply or "").strip() if not user_text: return None # Minimal context: first user message (and optionally the first assistant reply) context = f"User: {user_text}" if bot_text: context += f"\nAssistant: {bot_text}" # System + user prompt designed for strict, short output title_prompt = ( "You are a title generator.\n" "Create a short, meaningful chat title.\n" "Constraints:\n" "- 3 to 7 words (short phrase)\n" "- No quotes\n" "- No emojis\n" "- No markdown\n" "- Output ONLY the title text\n\n" f"{context}\n\n" "Title:" ) # IMPORTANT: reuse your existing OpenAI call style # If you have a low-cost model config, use it here. # This assumes you have an internal helper to call the model. title, ok = ChatService._ask_openai_for_title(title_prompt,text_model_name=text_model_name) if not ok: return None title = (title or "").strip() # Hard safety: keep it short and clean title = title.replace("\n", " ").strip() if len(title) > 80: title = title[:80].rstrip() # Avoid empty/garbage if not title or len(title) < 3: return None return title except Exception: return None @staticmethod def _ask_openai_for_title(prompt: str,text_model_name:str=''): """ A small wrapper for title generation. Keep it separated so you can control model, tokens, temperature, etc. Returns: (text, success) """ try: # NOTE: adapt this part to your existing OpenAI client code inside ask_openai. # The key is: low tokens, low temperature. # # Example pseudo: # resp = client.responses.create( # model=..., input=[{"role":"user","content":prompt}], # max_output_tokens=30, temperature=0.2 # ) # text = resp.output_text # # Since I don't see your OpenAI client here, I keep it as a call you should map. text, success = ChatService.ask_openai( user_message=prompt, history=[], image_urls=[], text_model_name=text_model_name # If your ask_openai supports knobs, pass: # max_output_tokens=30, # temperature=0.2, ) # If your ask_openai returns (reply, success) already, this matches. return text, success except Exception: return None, False @staticmethod def maybe_update_session_title(session_id: int, first_user_message: str, assistant_reply: str | None = None,text_model_name:str=''): """ Update session title only if it is still default. Title is generated by LLM (not user raw text). """ if not session_id: return sess = ChatSession.query.filter_by(id=session_id).first() if not sess: return # Titles that are considered "default" and can be overwritten default_titles = { "New chat", "New Chat", "Chat", "گفت و گوی جدید", "گفتگوی جدید", "گفتگو جدید", } current_title = (sess.title or "").strip() # If already customized, do nothing if current_title and current_title not in default_titles: return # Generate a concise title using the model title = ChatService.generate_session_title_llm(first_user_message, assistant_reply,text_model_name=text_model_name) if not title: return sess.title = title db.session.commit() @staticmethod def ask_openai_stream(user_message: str, history: list, resources:str="", voice_prompt: bool = False,model_name:str='', summarize_history=True): """ Stream AI response for real-time applications. Yields: (text_chunk, success) """ # SYSTEM_PROMPT prompt_text = get_prompt(PromptType.TEXT) prompt_struct = get_prompt(PromptType.ANSWER_STRUCT) system_prompt = prompt_text + prompt_struct if voice_prompt: prompt_conv = get_prompt(PromptType.CONVERSATION) system_prompt = system_prompt + prompt_conv total_input_text = "" messages = [] if summarize_history: summary_text = ChatService.summarize_history(history, user_message) if summary_text: messages.append({"role": "system", "content": summary_text}) total_input_text += summary_text else: for msg in history: role = "user" if msg.sender_type == "user" else "assistant" messages.append({"role": role, "content": msg.message}) total_input_text += msg.message messages.append({"role": "user", "content": user_message}) total_input_text += user_message # Stream from OpenAI final_system_prompt = system_prompt + resources total_input_text += final_system_prompt try: stream = client.responses.create( model=model_name, input=messages, instructions=final_system_prompt, stream=True ) input_tokens = count_tokens(text=total_input_text,model=model_name) OpenAIModel.update_tokens_in_db(model_name=model_name,input_tokens=input_tokens) for event in stream: delta = getattr(event, "delta", None) if delta: yield delta, True yield None, True except Exception as e: print("OpenAI stream error:", e) yield "", False @staticmethod def text_to_speech(text: str, voice="sage",model_name:str='gpt-4o-mini-tts'): """Convert text to base64 audio""" try: prompt = get_prompt(PromptType.TEXT_TO_VOICE) tts_response = client.audio.speech.create( model=model_name, voice=voice, input=text, instructions=prompt, ) # OpenAIModel.update_tokens_in_db(model_name=text_to_speech_model_name,response=tts_response) input_tokens = count_tokens(prompt+text,model=model_name) output_tokens = 0 # for audio this price is zero OpenAIModel.update_tokens_in_db(model_name=model_name,input_tokens=input_tokens) audio_base64 = base64.b64encode(tts_response.content).decode("utf-8") return audio_base64, True except Exception as e: print("TTS error:", e) return None, False @staticmethod def speech_to_text(base64_audio: str,speech_text_to_model_name:str=''): """ Universal STT pipeline - supports: caf, m4a, mp4, webm, wav, ogg, flac - works on Linux + Windows - no need for external ffmpeg installed """ try: audio_bytes = base64.b64decode(base64_audio) header = audio_bytes[:12] if header.startswith(b"RIFF"): suffix = ".wav" elif header.startswith(b"ID3"): suffix = ".mp3" elif header.startswith(b"ftyp"): suffix = ".m4a" elif header.startswith(b"OggS"): suffix = ".ogg" elif header.startswith(b"fLaC"): suffix = ".flac" elif header.startswith(b"\x1aE\xdf\xa3"): suffix = ".webm" elif header.startswith(b"caff"): suffix = ".caf" elif b"ftyp" in header: suffix = ".m4a" else: suffix = ".wav" temp_path = "" with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: tmp.write(audio_bytes) tmp.flush() temp_path = tmp.name try: with open(temp_path, "rb") as f: result = client.audio.transcriptions.create( model=speech_text_to_model_name, file=f, response_format="json" ) OpenAIModel.update_tokens_in_db(model_name=speech_text_to_model_name,response=result) if temp_path: os.remove(temp_path) return result.text.strip(), True except: os.remove(temp_path) return "", False except Exception as e: print("STT error:", e) return "", False @staticmethod def sentence_split(text: str, min_length=20): """Split text at the last sentence boundary if longer than min_length""" cut_idx = -1 for char in ".!?؟": idx = text.rfind(char) if idx > cut_idx: cut_idx = idx if cut_idx > min_length: return text[:cut_idx + 1], text[cut_idx + 1:] return "", text @staticmethod def openai_check_connection(timeout=3): try: if NET_ACCESS: api_key = current_app.config.get("SECOND_OPENAI_API_KEY") header ={"Authorization": f"Bearer {api_key}"} # client.models.list(timeout=timeout) requests.get("https://api.openai.com", headers=header, timeout=timeout) return True else: return False except: return False @staticmethod def sentence_spilit(text:str, min_length=20): cut_idx = -1 for endpoint_char in ".!?؟!.": if endpoint_char not in text: continue _idx = text.rfind(endpoint_char) if _idx > cut_idx: cut_idx = _idx if cut_idx > 0 and cut_idx > min_length: sentence = text[:cut_idx + 1] remain = text[cut_idx + 1:] return sentence, remain else: return "", text @staticmethod def detect_language(text): rtl_chars = re.compile(r'[\u0600-\u06FF]') return 'rtl' if rtl_chars.search(text) else 'ltr' @staticmethod def chunks_to_prompt( chunks:list[str]): if not chunks: return "" prompt = get_prompt(PromptType.REFERENCES) Seperator = '\n' prompt = prompt + Seperator.join(chunks) return prompt @staticmethod def remove_harakat(text): harakat_pattern = re.compile(r'[\u064B-\u0652]') return harakat_pattern.sub('', text) def clean_text_for_voice(text: str): """ Remove common Markdown syntax for clean TTS. """ text = re.sub(r'(\*\*|__)(.*?)\1', r'\2', text) text = re.sub(r'(\*|_)(.*?)\1', r'\2', text) text = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', text) text = re.sub(r'!\[([^\]]*)\]\([^\)]+\)', r'\1', text) text = re.sub(r'^\s*#+\s*(.*)', r'\1', text, flags=re.MULTILINE) text = re.sub(r'^\s*[-*+]\s+', '', text, flags=re.MULTILINE) text = re.sub(r'`(.+?)`', r'\1', text) text = re.sub(r'~~', '', text) text = re.sub(r'>\s?', '', text) text = text.replace('\u200c', ' ') text = re.sub(r'\s+', ' ', text).strip() return text def get_prompt(prompt_type: PromptType) -> str: prompt = SystemPrompt.query.filter_by( type=prompt_type, is_active=True ).first() return prompt.content if prompt else "" #------------------------------------------------------------------------------------------------- # Define Global Doc Manager #-------------------------------------------------------------------------------------------------