mirror of
https://github.com/aljazceru/vibeline.git
synced 2026-01-13 11:34:37 +01:00
chore: move python files to src/
This commit is contained in:
Gigi
@@ -1,101 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import ollama
|
||||
import time
|
||||
|
||||
def read_transcript(transcript_file: Path) -> str:
|
||||
"""Read the content of a transcript file."""
|
||||
with open(transcript_file, 'r', encoding='utf-8') as f:
|
||||
return f.read()
|
||||
|
||||
def load_prompt_template(transcript_text: str) -> str:
|
||||
"""Load the appropriate prompt template based on transcript content."""
|
||||
prompt_dir = Path("prompts")
|
||||
|
||||
# Check if transcript contains app-related content
|
||||
if "idea" in transcript_text.lower() and "app" in transcript_text.lower():
|
||||
prompt_file = prompt_dir / "idea_app.md"
|
||||
else:
|
||||
prompt_file = prompt_dir / "default.md"
|
||||
|
||||
with open(prompt_file, 'r', encoding='utf-8') as f:
|
||||
return f.read()
|
||||
|
||||
def process_transcript(transcript_text: str) -> str:
|
||||
"""Process a transcript using LLaMA to generate a summary."""
|
||||
# Load the appropriate prompt template
|
||||
prompt_template = load_prompt_template(transcript_text)
|
||||
|
||||
# Format the prompt with the transcript
|
||||
prompt = prompt_template.format(transcript=transcript_text)
|
||||
|
||||
# Use Ollama to generate the summary
|
||||
response = ollama.chat(model='llama2', messages=[
|
||||
{
|
||||
'role': 'user',
|
||||
'content': prompt
|
||||
}
|
||||
])
|
||||
|
||||
# Debug print
|
||||
print("Response structure:", response)
|
||||
|
||||
# Extract the content from the response
|
||||
return response['message']['content'].strip()
|
||||
|
||||
def save_summary(summary: str, output_file: Path) -> None:
|
||||
"""Save the summary to a file."""
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
f.write(summary)
|
||||
|
||||
def main():
|
||||
transcript_dir = Path("VoiceMemos/transcripts")
|
||||
summary_dir = Path("VoiceMemos/summaries")
|
||||
|
||||
# Create summaries directory if it doesn't exist
|
||||
summary_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Get list of all transcript files
|
||||
transcript_files = list(transcript_dir.glob("*.txt"))
|
||||
total_files = len(transcript_files)
|
||||
|
||||
print(f"Found {total_files} transcript(s) to process")
|
||||
|
||||
# Process all transcript files
|
||||
for idx, transcript_file in enumerate(transcript_files, 1):
|
||||
print(f"\nProcessing {transcript_file.name} ({idx}/{total_files})...")
|
||||
|
||||
# Skip if summary already exists
|
||||
summary_file = summary_dir / f"{transcript_file.stem}_summary.txt"
|
||||
if summary_file.exists():
|
||||
print(" Summary already exists, skipping...")
|
||||
continue
|
||||
|
||||
try:
|
||||
# Read transcript
|
||||
transcript_text = read_transcript(transcript_file)
|
||||
print(f" Read transcript ({len(transcript_text)} characters)")
|
||||
|
||||
# Generate summary
|
||||
summary = process_transcript(transcript_text)
|
||||
|
||||
# Save summary
|
||||
save_summary(summary, summary_file)
|
||||
print(f" Summary saved to {summary_file}")
|
||||
|
||||
# Add a small delay between files to avoid overloading
|
||||
if idx < total_files:
|
||||
time.sleep(1)
|
||||
|
||||
except Exception as e:
|
||||
print(f" Failed to process {transcript_file.name}")
|
||||
print(f" Error: {str(e)}")
|
||||
continue
|
||||
|
||||
print("\nDone! All transcripts processed.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,105 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import sys
|
||||
import time
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from watchdog.observers import Observer
|
||||
from watchdog.events import FileSystemEventHandler
|
||||
|
||||
class VoiceMemoHandler(FileSystemEventHandler):
|
||||
def __init__(self, voice_memo_dir: Path, transcript_dir: Path):
|
||||
self.voice_memo_dir = voice_memo_dir
|
||||
self.transcript_dir = transcript_dir
|
||||
self.processing_lock = False
|
||||
|
||||
def on_created(self, event):
|
||||
if self.processing_lock:
|
||||
return
|
||||
|
||||
if not event.is_directory:
|
||||
file_path = Path(event.src_path)
|
||||
|
||||
# Handle new voice memo
|
||||
if file_path.parent == self.voice_memo_dir and file_path.suffix.lower() == '.m4a':
|
||||
print(f"\nNew voice memo detected: {file_path.name}")
|
||||
self.process_voice_memo()
|
||||
|
||||
# Handle new transcript
|
||||
elif file_path.parent == self.transcript_dir and file_path.suffix.lower() == '.txt':
|
||||
print(f"\nNew transcript detected: {file_path.name}")
|
||||
self.process_transcript()
|
||||
|
||||
def process_voice_memo(self):
|
||||
"""Run the voice memo processing script"""
|
||||
try:
|
||||
self.processing_lock = True
|
||||
print("Processing voice memo...")
|
||||
result = subprocess.run(['./process_voice_memos.sh'],
|
||||
capture_output=True,
|
||||
text=True)
|
||||
if result.returncode == 0:
|
||||
print("Voice memo processing completed successfully")
|
||||
else:
|
||||
print(f"Error processing voice memo: {result.stderr}")
|
||||
except Exception as e:
|
||||
print(f"Error running voice memo script: {str(e)}")
|
||||
finally:
|
||||
self.processing_lock = False
|
||||
|
||||
def process_transcript(self):
|
||||
"""Run the transcript summarization script"""
|
||||
try:
|
||||
self.processing_lock = True
|
||||
print("Processing transcript...")
|
||||
result = subprocess.run(['./summarize_transcripts.py'],
|
||||
capture_output=True,
|
||||
text=True)
|
||||
if result.returncode == 0:
|
||||
print("Transcript processing completed successfully")
|
||||
else:
|
||||
print(f"Error processing transcript: {result.stderr}")
|
||||
except Exception as e:
|
||||
print(f"Error running summarization script: {str(e)}")
|
||||
finally:
|
||||
self.processing_lock = False
|
||||
|
||||
def main():
|
||||
# Set up directory paths
|
||||
voice_memo_dir = Path("VoiceMemos")
|
||||
transcript_dir = voice_memo_dir / "transcripts"
|
||||
|
||||
# Verify directories exist
|
||||
if not voice_memo_dir.exists():
|
||||
print(f"Error: {voice_memo_dir} directory does not exist")
|
||||
sys.exit(1)
|
||||
if not transcript_dir.exists():
|
||||
print(f"Error: {transcript_dir} directory does not exist")
|
||||
sys.exit(1)
|
||||
|
||||
# Set up event handler and observer
|
||||
event_handler = VoiceMemoHandler(voice_memo_dir, transcript_dir)
|
||||
observer = Observer()
|
||||
|
||||
# Watch both directories
|
||||
observer.schedule(event_handler, str(voice_memo_dir), recursive=False)
|
||||
observer.schedule(event_handler, str(transcript_dir), recursive=False)
|
||||
|
||||
# Start the observer
|
||||
observer.start()
|
||||
print(f"\nWatching for changes in:")
|
||||
print(f"- Voice memos: {voice_memo_dir}")
|
||||
print(f"- Transcripts: {transcript_dir}")
|
||||
print("\nPress Ctrl+C to stop...")
|
||||
|
||||
try:
|
||||
while True:
|
||||
time.sleep(1)
|
||||
except KeyboardInterrupt:
|
||||
observer.stop()
|
||||
print("\nStopping file watcher...")
|
||||
|
||||
observer.join()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user