Create processing.py

processing.py

@@ -0,0 +1,109 @@
+import os
+import time
+import shutil
+from langdetect import detect
+from transformers import AutoTokenizer
+from langchain.chains import RetrievalQA
+from diarization import process_video as diarize_video
+
+# Include the necessary imports and lazy loading classes here
+
+def detect_language(text):
+    try:
+        return detect(text)
+    except:
+        return "en"  # default to English if detection fails
+
+def count_words_and_tokens(text):
+    words = len(text.split())
+    tokens = len(AutoTokenizer.from_pretrained("mistralai/Mistral-7B-Instruct-v0.3").tokenize(text))
+    return words, tokens
+
+def process_text(input_file):
+    with open(input_file.name, 'r', encoding='utf-8') as file:
+        content = file.read()
+    words, tokens = count_words_and_tokens(content)
+    input_info = f"Text file processed. Words: {words}, Tokens: {tokens}"
+    return content, input_info
+
+def process_pdf(input_file):
+    loader = PyPDFLoader(input_file.name)
+    pages = loader.load_and_split()
+    content = '\n'.join([page.page_content for page in pages])
+    words, tokens = count_words_and_tokens(content)
+    input_info = f"PDF file processed. Words: {words}, Tokens: {tokens}"
+    return content, input_info
+
+def process_video(input_file, progress):
+    file_extension = os.path.splitext(input_file.name)[1].lower()
+    temp_video_path = "temp_video" + file_extension
+    shutil.copy2(input_file.name, temp_video_path)
+    
+    if progress:
+        progress(0.2, desc="Transcribing video...")
+    
+    language = "en"  # Default to English for video files
+    diarize_video(temp_video_path, os.environ.get('hf_secret'), language)
+    
+    srt_path = temp_video_path.replace(file_extension, "_combined.srt")
+    with open(srt_path, 'r', encoding='utf-8') as file:
+        content = file.read()
+    words, tokens = count_words_and_tokens(content)
+    input_info = f"Input Words: {words} / Input Tokens: {tokens}"
+    
+    return content, input_info
+
+def analyze_content(content, progress):
+    attachments_chain, bigfive_chain, personalities_chain = lazy_chains.get_chains()
+    
+    if progress:
+        progress(0.6, desc="Analyzing attachments...")
+    attachments_result = attachments_chain({"query": content})
+    attachments_answer = attachments_result['result'].split("-----------\n\nAnswer:")[-1].strip()
+    
+    if progress:
+        progress(0.7, desc="Analyzing Big Five traits...")
+    bigfive_result = bigfive_chain({"query": content})
+    bigfive_answer = bigfive_result['result'].split("-----------\n\nAnswer:")[-1].strip()
+    
+    if progress:
+        progress(0.8, desc="Analyzing personalities...")
+    personalities_result = personalities_chain({"query": content})
+    personalities_answer = personalities_result['result'].split("-----------\n\nAnswer:")[-1].strip()
+    
+    return attachments_answer, bigfive_answer, personalities_answer
+
+def process_input(input_file, progress=None):
+    start_time = time.time()
+    
+    if progress:
+        progress(0, desc="Processing file...")
+    
+    file_extension = os.path.splitext(input_file.name)[1].lower()
+    
+    if file_extension == '.txt':
+        content, input_info = process_text(input_file)
+    elif file_extension == '.pdf':
+        content, input_info = process_pdf(input_file)
+    elif file_extension in ['.mp4', '.avi', '.mov']:
+        content, input_info = process_video(input_file, progress)
+    else:
+        return "Unsupported file format. Please upload a TXT, PDF, or video file.", None, None, None, None, None, None
+
+    detected_language = detect_language(content)
+    
+    if progress:
+        progress(0.4, desc="Analyzing content...")
+    
+    attachments_answer, bigfive_answer, personalities_answer = analyze_content(content, progress)
+    
+    end_time = time.time()
+    execution_time = end_time - start_time
+    
+    execution_info = f"{execution_time:.2f} seconds"
+    
+    if progress:
+        progress(1.0, desc="Analysis complete!")
+
+    return ("Analysis complete!", execution_info, detected_language, input_info, 
+            attachments_answer, bigfive_answer, personalities_answer)