mirror of
https://github.com/patw/AudioSumma.git
synced 2026-10-19 07:14:03 +00:00
9cf3f879d0
Replace FFmpeg silence removal with dependency-free PCM processing, add in-app activity and status indicators, extract pipeline orchestration, prevent duplicate daily summaries, remove obsolete image assets, and add audio, pipeline, and Qt tests.
95 lines
3.8 KiB
Python
95 lines
3.8 KiB
Python
"""Transcription, summarization, and cleanup orchestration for AudioSumma."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import os
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
class ProcessingPipeline:
|
|
"""Run the file-processing workflow independently of the Qt UI."""
|
|
|
|
def __init__(
|
|
self,
|
|
config: dict[str, Any],
|
|
*,
|
|
log: Callable[[str], None] = lambda _message: None,
|
|
trim_silence: Callable[[str], Any] | None = None,
|
|
transcribe: Callable[[Any], str] | None = None,
|
|
summarize: Callable[[str], str] | None = None,
|
|
) -> None:
|
|
self.config = config
|
|
self.log = log
|
|
self.trim_silence = trim_silence
|
|
self.transcribe = transcribe
|
|
self.summarize = summarize
|
|
|
|
@property
|
|
def output_dir(self) -> Path:
|
|
return Path(self.config["OUTPUT_DIR"])
|
|
|
|
def process_wav_files(self) -> None:
|
|
"""Trim and transcribe WAV files without an existing transcript."""
|
|
self.output_dir.mkdir(parents=True, exist_ok=True)
|
|
for wav_path in sorted(self.output_dir.glob("*.wav")):
|
|
transcript_path = wav_path.with_suffix(".tns")
|
|
if transcript_path.exists():
|
|
self.log(f"Skipping {wav_path.name}, transcript already exists")
|
|
continue
|
|
if self.trim_silence is None or self.transcribe is None:
|
|
raise RuntimeError("transcription services are not configured")
|
|
|
|
self.log(f"Trimming silence: {wav_path}")
|
|
self.trim_silence(str(wav_path))
|
|
self.log(f"Transcribing: {wav_path}")
|
|
with wav_path.open("rb") as audio, transcript_path.open("w") as output:
|
|
output.write(self.transcribe(audio))
|
|
|
|
def summarize_transcripts(self) -> None:
|
|
"""Summarize every transcript into today's Markdown summary."""
|
|
if self.summarize is None:
|
|
raise RuntimeError("summarization service is not configured")
|
|
|
|
summary_path = self.output_dir / (
|
|
f"summary-{datetime.datetime.now():%Y%m%d}.md"
|
|
)
|
|
existing_summary = summary_path.read_text() if summary_path.exists() else ""
|
|
for transcript_path in sorted(self.output_dir.glob("*.tns")):
|
|
heading = f"# Call Transcript - {transcript_path.name}\n"
|
|
if heading in existing_summary:
|
|
self.log(
|
|
f"Skipping {transcript_path.name}, already summarized today"
|
|
)
|
|
continue
|
|
|
|
self.log(f"Summarizing: {transcript_path}")
|
|
transcript = transcript_path.read_text()
|
|
with summary_path.open("a") as summary:
|
|
summary.write(f"# Call Transcript - {transcript_path.name}\n\n")
|
|
for step in self.config["SUMMARIZATION_STEPS"]:
|
|
self.log(f" Running step: {step['name']}")
|
|
result = self.summarize(step["prompt"].format(chunk=transcript))
|
|
summary.write(f"## {step['name']}\n\n{result}\n\n")
|
|
summary.write("---\n")
|
|
self.log("Summarizing complete")
|
|
|
|
def run(self) -> None:
|
|
self.log("Starting transcription and summarization")
|
|
self.process_wav_files()
|
|
self.summarize_transcripts()
|
|
self.log("Processing complete")
|
|
|
|
def clean(self) -> None:
|
|
"""Delete intermediate WAV and transcript files, retaining Markdown."""
|
|
self.log("Cleaning files...")
|
|
try:
|
|
for path in self.output_dir.iterdir():
|
|
if path.suffix in {".wav", ".tns"}:
|
|
path.unlink()
|
|
self.log(f"Deleted: {path}")
|
|
self.log("Cleaning complete.")
|
|
except Exception as error:
|
|
self.log(f"An error occurred while cleaning files: {error}")
|