Create main.py

main.py

@@ -0,0 +1,163 @@
+import whisper
+
+def transcribe_audio(audio_path):
+    model = whisper.load_model("base")
+    result = model.transcribe(audio_path)
+    return result["text"]
+from pyannote.audio import Pipeline
+
+def perform_speaker_diarization(audio_path):
+    pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization@2.1", use_auth_token="YOUR_HUGGINGFACE_TOKEN")
+    diarization = pipeline(audio_path)
+    
+    speaker_segments = []
+    for turn, _, speaker in diarization.itertracks(yield_label=True):
+        speaker_segments.append({
+            "start": turn.start,
+            "end": turn.end,
+            "speaker": speaker
+        })
+    return speaker_segments
+
+from textblob import TextBlob
+from sklearn.feature_extraction.text import CountVectorizer
+from sklearn.decomposition import LatentDirichletAllocation
+from collections import Counter
+import nltk
+from nltk.corpus import stopwords
+import spacy
+
+nltk.download('stopwords')
+nltk.download('punkt')
+
+# Load spaCy model for NER
+nlp = spacy.load("en_core_web_sm")
+
+def analyze_sentiment(text):
+    blob = TextBlob(text)
+    return blob.sentiment.polarity, blob.sentiment.subjectivity
+
+def extract_keywords(text, top_n=5):
+    stop_words = set(stopwords.words("english"))
+    words = nltk.word_tokenize(text.lower())
+    filtered_words = [word for word in words if word.isalnum() and word not in stop_words]
+    word_counts = Counter(filtered_words)
+    return word_counts.most_common(top_n)
+
+def perform_topic_modeling(text, num_topics=5, num_words=10):
+    vectorizer = CountVectorizer(stop_words="english", max_features=1000)
+    X = vectorizer.fit_transform([text])
+    lda = LatentDirichletAllocation(n_components=num_topics, random_state=42)
+    lda.fit(X)
+
+    topics = []
+    for idx, topic in enumerate(lda.components_):
+        top_words = [vectorizer.get_feature_names_out()[i] for i in topic.argsort()[:-num_words - 1:-1]]
+        topics.append(f"Topic {idx + 1}: {' '.join(top_words)}")
+    return topics
+
+def extract_entities(text):
+    doc = nlp(text)
+    entities = [(ent.text, ent.label_) for ent in doc.ents]
+    return entities
+
+def parse_query(query):
+    doc = nlp(query)
+    keywords = [token.text.lower() for token in doc if token.is_alpha and not token.is_stop]
+    intent = None
+    
+    if any(word in ["how many", "count"] for word in keywords):
+        intent = "count"
+    elif any(word in ["list", "show me"] for word in keywords):
+        intent = "list"
+    elif any(word in ["sentiment", "polarity", "subjectivity"] for word in keywords):
+        intent = "sentiment"
+    elif any(word in ["theme", "topic", "main"] for word in keywords):
+        intent = "topic"
+    elif any(word in ["keyword", "common"] for word in keywords):
+        intent = "keyword"
+    elif any(word in ["entity", "name", "person", "organization"] for word in keywords):
+        intent = "ner"
+    return intent, keywords
+
+def answer_question(query, qa_df):
+    intent, keywords = parse_query(query)
+    
+    if intent == "count":
+        filtered = qa_df[qa_df["Transcript"].str.contains("|".join(keywords), case=False)]
+        return f"{len(filtered)} responses contain the keywords: {', '.join(keywords)}."
+    
+    elif intent == "list":
+        filtered = qa_df[qa_df["Transcript"].str.contains("|".join(keywords), case=False)]["Transcript"].tolist()
+        return "\n".join(filtered) if filtered else "No matching responses found."
+    
+    elif intent == "sentiment":
+        avg_polarity = qa_df["Sentiment_Polarity"].mean()
+        avg_subjectivity = qa_df["Sentiment_Subjectivity"].mean()
+        return f"Average Polarity: {avg_polarity:.2f}, Average Subjectivity: {avg_subjectivity:.2f}"
+    
+    elif intent == "topic":
+        all_text = " ".join(qa_df["Transcript"])
+        topics = perform_topic_modeling(all_text)
+        return "\n".join(topics)
+    
+    elif intent == "keyword":
+        all_text = " ".join(qa_df["Transcript"])
+        keywords = extract_keywords(all_text)
+        return ", ".join([word for word, count in keywords])
+    
+    elif intent == "ner":
+        all_text = " ".join(qa_df["Transcript"])
+        entities = extract_entities(all_text)
+        return "\n".join([f"{entity} ({label})" for entity, label in entities])
+    
+    else:
+        return "I'm not sure how to answer that. Try asking about counts, lists, sentiment, topics, keywords, or entities."
+
+import gradio as gr
+
+# Global variables to store processed data
+qa_df = None
+
+def process_audio(audio_path):
+    global qa_df
+    
+    # Step 1: Transcribe audio
+    transcription = transcribe_audio(audio_path)
+    
+    # Step 2: Perform speaker diarization
+    speaker_segments = perform_speaker_diarization(audio_path)
+    
+    # Step 3: Analyze text
+    sentiment_polarity, sentiment_subjectivity = analyze_sentiment(transcription)
+    topics = perform_topic_modeling(transcription)
+    keywords = extract_keywords(transcription)
+    entities = extract_entities(transcription)
+    
+    # Create a DataFrame
+    qa_df = pd.DataFrame({
+        "Speaker": [seg["speaker"] for seg in speaker_segments],
+        "Transcript": [transcription],
+        "Sentiment_Polarity": [sentiment_polarity],
+        "Sentiment_Subjectivity": [sentiment_subjectivity],
+        "Topics": [topics],
+        "Keywords": [keywords],
+        "Entities": [entities]
+    })
+    
+    return "Audio processed successfully!"
+
+# Gradio Interface
+with gr.Blocks() as demo:
+    gr.Markdown("# Advanced Audio Analysis App")
+    audio_input = gr.Audio(label="Upload Audio File")
+    process_button = gr.Button("Process Audio")
+    status_output = gr.Textbox(label="Status")
+    
+    question_input = gr.Textbox(label="Ask a Question")
+    answer_output = gr.Textbox(label="Answer")
+    
+    process_button.click(process_audio, inputs=audio_input, outputs=status_output)
+    question_input.submit(answer_question, inputs=[question_input], outputs=answer_output)
+
+demo.launch()