Ruben/MeetingTranskrip
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added self sign up and email verification

Browse Source
This commit is contained in:
RubenRWU
2026-03-09 13:31:30 +01:00
commit f38dde8188
6 changed files with 2119 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

594
main.py Normal file
View File

@@ -0,0 +1,594 @@
import io
import os
import re
import smtplib
import subprocess
import uuid
import warnings
from contextlib import asynccontextmanager
from datetime import datetime
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from pathlib import Path
from dotenv import load_dotenv
load_dotenv()
os.environ["TORCH_CODEC_ENABLED"] = "0"
warnings.filterwarnings("ignore", category=UserWarning, module="pyannote.audio")
warnings.filterwarnings("ignore", category=UserWarning, module="torchaudio")
from fastapi import FastAPI, UploadFile, File, Form, Body, HTTPException
from fastapi.responses import HTMLResponse
import whisperx
import torch
MAX_FILE_SIZE = 25 * 1024 * 1024
ALLOWED_CONTENT_TYPES = {
"audio/mpeg",
"audio/mp3",
"audio/webm",
"audio/ogg",
"audio/wav",
"video/webm",
}
DATA_DIR = Path("data")
DATA_DIR.mkdir(exist_ok=True)
OUTPUT_DIR = Path("output")
OUTPUT_DIR.mkdir(exist_ok=True)
device = "cuda" if torch.cuda.is_available() else "cpu"
compute_type = "float32" if device == "cpu" else "float16"
_whisper_model = None
_align_model = None
_align_metadata = None
# ── Model loading ──────────────────────────────────────────────────────────────
def get_whisper_model():
global _whisper_model
if _whisper_model is None:
print(f"Lade WhisperX Modell large-v3 auf {device} ({compute_type})...")
_whisper_model = whisperx.load_model(
"large-v3", device=device, compute_type=compute_type
)
print("WhisperX Modell geladen.")
return _whisper_model
def get_align_model():
global _align_model, _align_metadata
if _align_model is None:
print("Lade Alignment Modell (de)...")
_align_model, _align_metadata = whisperx.load_align_model(
language_code="de", device=device
)
print("Alignment Modell geladen.")
return _align_model, _align_metadata
@asynccontextmanager
async def lifespan(app: FastAPI):
print(f"Server startet – lade Modelle auf {device}...")
get_whisper_model()
get_align_model()
print("Alle Modelle geladen – Server bereit!")
yield
print("Server fährt herunter.")
app = FastAPI(lifespan=lifespan)
# ── Audio helpers ──────────────────────────────────────────────────────────────
def convert_to_mp3(input_path: Path) -> Path:
output_path = input_path.with_suffix(".converted.mp3")
r = subprocess.run(
[
"ffmpeg",
"-y",
"-i",
str(input_path),
"-acodec",
"libmp3lame",
"-q:a",
"2",
str(output_path),
],
capture_output=True,
text=True,
)
if r.returncode != 0:
raise RuntimeError(f"ffmpeg fehlgeschlagen:\n{r.stderr}")
return output_path
def detect_audio_format(data: bytes) -> str:
if data[:3] == b"ID3" or data[:2] in (b"\xff\xfb", b"\xff\xf3"):
return "mp3"
if data[:4] == b"fLaC":
return "flac"
if data[:4] == b"OggS":
return "ogg"
if data[:4] == b"RIFF":
return "wav"
if data[:4] == b"\x1aE\xdf\xa3":
return "webm"
return "unknown"
# ── Transcription ──────────────────────────────────────────────────────────────
def transcribe_audio(file_path: Path) -> str:
whisper_model = get_whisper_model()
align_model, align_metadata = get_align_model()
fmt = detect_audio_format(file_path.read_bytes())
needs_conversion = (
fmt in ("webm", "ogg", "flac") or file_path.suffix.lower() != ".mp3"
)
audio_path = convert_to_mp3(file_path) if needs_conversion else file_path
audio_str = str(audio_path)
print("Transkription...")
result = whisper_model.transcribe(audio_str, language="de")
print("Alignment...")
result = whisperx.align(
result["segments"], align_model, align_metadata, audio_str, device=device
)
if os.getenv("ENABLE_SPEAKER_DIARIZATION", "true").lower() == "true":
try:
print("Diarization...")
from whisperx.diarize import DiarizationPipeline
pipeline = DiarizationPipeline(token=os.getenv("HF_TOKEN"), device=device)
diarize_segments = pipeline(audio_str)
result = whisperx.assign_word_speakers(diarize_segments, result)
print("Diarization fertig.")
except Exception as e:
print(f"Diarization fehlgeschlagen: {e}")
if needs_conversion:
audio_path.unlink(missing_ok=True)
return "\n".join(
f"{seg.get('speaker', 'SPRECHER')}: {seg['text'].strip()}"
for seg in result["segments"]
if seg["text"].strip()
)
# ── AI helpers ─────────────────────────────────────────────────────────────────
def _openai_client():
from openai import OpenAI
key = os.getenv("OPENAI_API_KEY")
if not key:
raise RuntimeError("OPENAI_API_KEY nicht gesetzt")
return OpenAI(api_key=key)
def process_with_ai(text: str) -> str:
"""Grammatik- und Interpunktionskorrektur, Sprecher-Labels bleiben."""
client = _openai_client()
r = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": (
"Du bist ein Assistent zur Nachbearbeitung von Sprachtranskripten. "
"Verbessere Grammatik, Rechtschreibung und Interpunktion. "
"Behalte die Sprecher-Labels (z.B. 'Max Mustermann:') exakt bei. "
"Verändere die inhaltliche Bedeutung NICHT. "
"Format: Sprecher: Text"
),
},
{"role": "user", "content": text},
],
)
return r.choices[0].message.content or ""
def summarize_with_ai(transcript: str, title: str) -> str:
"""Erstellt eine strukturierte Meeting-Zusammenfassung ohne Markdown."""
client = _openai_client()
r = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": (
"Du bist ein Assistent für Meeting-Protokolle. "
"Erstelle eine strukturierte Zusammenfassung des Meeting-Transkripts. "
"Gliedere immer in genau diese drei Abschnitte, auch wenn das Transkript kurz oder unvollständig ist – "
"fasse dann zusammen was vorhanden ist:\n\n"
"Kernthemen:\n[Was wurde besprochen]\n\n"
"Entscheidungen:\n[Getroffene Entscheidungen, falls keine: 'Keine Entscheidungen getroffen.']\n\n"
"Nächste Schritte:\n[Offene Punkte und Aufgaben, falls keine: 'Keine Folgeaufgaben definiert.']\n\n"
"Wichtige Regeln:\n"
"- Kein Markdown, keine Sterne, keine Rauten, keine Aufzählungszeichen mit Bindestrich\n"
"- Nur Fließtext und Zeilenumbrüche\n"
"- Professionelles Deutsch\n"
"- Immer alle drei Abschnitte ausgeben, egal wie kurz das Transkript ist"
),
},
{
"role": "user",
"content": f"Meeting: {title}\n\nTranskript:\n{transcript}",
},
],
)
return r.choices[0].message.content or ""
# ── PDF generation ─────────────────────────────────────────────────────────────
def _safe_filename(title: str) -> str:
"""Erstellt einen sicheren Dateinamen aus dem Meeting-Titel."""
safe = re.sub(r"[^\w\-äöüÄÖÜß ]", "", title).strip()
safe = re.sub(r"\s+", "_", safe)
return safe or "Transkript"
def build_pdf(title: str, summary: str, transcript: str) -> bytes:
"""Professionelles PDF: heller Hintergrund, Noto Sans, saubere Typografie."""
from fpdf import FPDF
# Font paths
FONT_DIR = Path("/usr/share/fonts/truetype/noto")
F_REG = str(FONT_DIR / "NotoSans-Regular.ttf")
F_BOLD = str(FONT_DIR / "NotoSans-Bold.ttf")
F_ITAL = str(FONT_DIR / "NotoSans-Italic.ttf")
# Colour palette (fully print-friendly, dark on white)
C_HEADER_BG = (248, 248, 250) # very light grey header band
C_HEADER_FG = (20, 20, 30) # near-black title
C_HEADER_SUB = (120, 120, 135) # label / date in header
C_HEADER_LINE = (200, 200, 210) # bottom border of header band
C_LABEL = (130, 130, 145) # section labels
C_RULE = (220, 220, 228) # horizontal rules
C_BODY = (30, 30, 40) # main body text
C_SUMMARY = (55, 55, 70) # summary text
C_FOOTER = (160, 160, 175)
# Speaker accent colours (left border + name) — muted, print-safe
SPEAKER_PALETTE = [
(51, 102, 187), # blue
(180, 60, 60), # red
(40, 140, 80), # green
(140, 70, 170), # purple
(190, 120, 20), # amber
(20, 150, 160), # teal
(190, 80, 120), # rose
(80, 100, 180), # indigo
]
date_str = datetime.now().strftime("%d. %B %Y")
class PDF(FPDF):
def __init__(self):
super().__init__()
self.add_font("Noto", "", F_REG, uni=True)
self.add_font("Noto", "B", F_BOLD, uni=True)
self.add_font("Noto", "I", F_ITAL, uni=True)
def header(self):
if self.page_no() == 1:
return # cover drawn manually
# Running header on pages 2+
self.set_font("Noto", "", 8)
self.set_text_color(*C_LABEL)
self.cell(0, 6, title, align="L")
self.cell(0, 6, date_str, align="R", new_x="LMARGIN", new_y="NEXT")
self.set_draw_color(*C_RULE)
self.set_line_width(0.3)
self.line(self.l_margin, self.get_y(), self.w - self.r_margin, self.get_y())
self.ln(5)
def footer(self):
self.set_y(-14)
self.set_font("Noto", "", 8)
self.set_text_color(*C_FOOTER)
self.cell(0, 8, f"{self.page_no()}", align="C")
pdf = PDF()
pdf.set_margins(22, 28, 22)
pdf.set_auto_page_break(auto=True, margin=20)
# ── PAGE 1: Cover block ───────────────────────────────────────────────────
pdf.add_page()
# Light header band
pdf.set_fill_color(*C_HEADER_BG)
pdf.rect(0, 0, pdf.w, 52, style="F")
# Bottom border line
pdf.set_draw_color(*C_HEADER_LINE)
pdf.set_line_width(0.4)
pdf.line(0, 52, pdf.w, 52)
# "MEETING-PROTOKOLL" label
pdf.set_xy(22, 13)
pdf.set_font("Noto", "", 8)
pdf.set_text_color(*C_HEADER_SUB)
pdf.cell(0, 5, "MEETING-PROTOKOLL")
# Title
pdf.set_xy(22, 21)
pdf.set_font("Noto", "B", 20)
pdf.set_text_color(*C_HEADER_FG)
pdf.cell(0, 10, title)
# Date
pdf.set_xy(22, 38)
pdf.set_font("Noto", "", 9)
pdf.set_text_color(*C_HEADER_SUB)
pdf.cell(0, 5, date_str)
pdf.set_y(62) # below header band
# ── SUMMARY SECTION ───────────────────────────────────────────────────────
if summary:
# Section label
pdf.set_font("Noto", "B", 8)
pdf.set_text_color(*C_LABEL)
pdf.set_x(pdf.l_margin)
pdf.cell(0, 5, "ZUSAMMENFASSUNG", new_x="LMARGIN", new_y="NEXT")
# Rule
pdf.set_draw_color(*C_RULE)
pdf.set_line_width(0.3)
pdf.line(pdf.l_margin, pdf.get_y() + 1, pdf.w - pdf.r_margin, pdf.get_y() + 1)
pdf.ln(5)
# Summary text — render each paragraph
body_w = pdf.w - pdf.l_margin - pdf.r_margin
pdf.set_font("Noto", "", 10)
pdf.set_text_color(*C_SUMMARY)
for para in summary.strip().split("\n"):
para = para.strip()
if not para:
pdf.ln(3)
continue
pdf.set_x(pdf.l_margin)
# Section headers inside summary (e.g. "Kernthemen:")
if para.endswith(":") and len(para) < 50:
pdf.ln(2)
pdf.set_font("Noto", "B", 10)
pdf.set_text_color(*C_BODY)
pdf.multi_cell(body_w, 6, para)
pdf.set_font("Noto", "", 10)
pdf.set_text_color(*C_SUMMARY)
else:
pdf.multi_cell(body_w, 6, para)
pdf.ln(10)
# ── TRANSCRIPT SECTION ────────────────────────────────────────────────────
pdf.set_font("Noto", "B", 8)
pdf.set_text_color(*C_LABEL)
pdf.set_x(pdf.l_margin)
pdf.cell(0, 5, "TRANSKRIPT", new_x="LMARGIN", new_y="NEXT")
pdf.set_draw_color(*C_RULE)
pdf.set_line_width(0.3)
pdf.line(pdf.l_margin, pdf.get_y() + 1, pdf.w - pdf.r_margin, pdf.get_y() + 1)
pdf.ln(6)
for line in transcript.split("\n"):
if not line.strip():
pdf.ln(2)
continue
colon = line.find(":")
if colon == -1:
pdf.set_font("Noto", "", 10)
pdf.set_text_color(*C_BODY)
pdf.set_x(pdf.l_margin)
pdf.multi_cell(pdf.w - pdf.l_margin - pdf.r_margin, 6, line.strip())
pdf.ln(1)
continue
speaker = line[:colon].strip()
text = line[colon + 1 :].strip()
if not text:
continue
# Speaker name bold, then text inline — no colour, no background
pdf.set_x(pdf.l_margin)
pdf.set_font("Noto", "B", 10)
pdf.set_text_color(*C_BODY)
name_label = speaker + ": "
name_w = pdf.get_string_width(name_label)
y = pdf.get_y()
pdf.set_xy(pdf.l_margin, y)
pdf.cell(name_w, 5, name_label)
text_x = pdf.l_margin + name_w
text_w = pdf.w - pdf.r_margin - text_x
pdf.set_font("Noto", "", 10)
pdf.set_xy(text_x, y)
pdf.multi_cell(text_w, 5, text)
pdf.ln(0.5)
return bytes(pdf.output())
# ── Email / file output ────────────────────────────────────────────────────────
def deliver(
to_email: str,
title: str,
summary: str,
transcript: str,
) -> bool:
"""Erstellt PDF und sendet es per E-Mail oder speichert es als Datei."""
pdf_bytes = build_pdf(title, summary, transcript)
safe_title = _safe_filename(title)
date_str = datetime.now().strftime("%Y-%m-%d")
pdf_name = f"{date_str}_{safe_title}.pdf"
smtp_host = os.getenv("SMTP_HOST")
smtp_port = int(os.getenv("SMTP_PORT", 587))
smtp_user = os.getenv("SMTP_USER")
smtp_password = os.getenv("SMTP_PASSWORD")
# SMTP_FROM kann ein Anzeigename sein — wir stellen sicher dass eine
# gültige Absenderadresse verwendet wird: "Name <email>" oder nur die E-Mail
smtp_from_raw = os.getenv("SMTP_FROM", "").strip()
if smtp_from_raw and "@" not in smtp_from_raw and smtp_user:
smtp_from = f"{smtp_from_raw} <{smtp_user}>"
elif smtp_from_raw:
smtp_from = smtp_from_raw
else:
smtp_from = smtp_user or ""
if not all([smtp_host, smtp_user, smtp_password]):
# Fallback: Datei speichern
out = OUTPUT_DIR / pdf_name
out.write_bytes(pdf_bytes)
print(f"SMTP nicht konfiguriert – PDF gespeichert: {out}")
return True
body = (
f"Meeting: {title}\n"
f"Datum: {datetime.now().strftime('%d.%m.%Y')}\n\n"
f"--- Zusammenfassung ---\n\n{summary}\n\n"
f"Das vollständige Transkript ist als PDF angehängt."
)
msg = MIMEMultipart()
msg["Subject"] = f"Meeting-Protokoll: {title}"
msg["From"] = smtp_from
msg["To"] = to_email
msg.attach(MIMEText(body, "plain", "utf-8"))
attachment = MIMEApplication(pdf_bytes, _subtype="pdf")
attachment.add_header("Content-Disposition", "attachment", filename=pdf_name)
msg.attach(attachment)
with smtplib.SMTP(str(smtp_host), smtp_port) as server:
server.starttls()
server.login(str(smtp_user), str(smtp_password))
server.send_message(msg)
return True
# ── API Endpoints ──────────────────────────────────────────────────────────────
@app.get("/", response_class=HTMLResponse)
async def get_index():
return Path(__file__).parent.joinpath("index.html").read_text(encoding="utf-8")
@app.post("/upload")
async def upload_audio(file: UploadFile = File(...)):
content_type = file.content_type or ""
if content_type not in ALLOWED_CONTENT_TYPES and not content_type.startswith(
"audio/"
):
raise HTTPException(
status_code=415, detail=f"Nicht unterstützter Typ: {content_type}"
)
content = await file.read()
if len(content) > MAX_FILE_SIZE:
raise HTTPException(
status_code=413,
detail=f"Datei zu groß ({len(content) // 1024 // 1024} MB). Max 25 MB",
)
if len(content) == 0:
raise HTTPException(status_code=400, detail="Leere Datei")
ext_map = {
"audio/mpeg": ".mp3",
"audio/mp3": ".mp3",
"audio/webm": ".webm",
"video/webm": ".webm",
"audio/ogg": ".ogg",
"audio/wav": ".wav",
"audio/flac": ".flac",
}
ext = ext_map.get(content_type, ".audio")
file_id = str(uuid.uuid4())
file_path = DATA_DIR / f"{file_id}{ext}"
file_path.write_bytes(content)
return {"file_id": file_id, "filename": file.filename, "size": len(content)}
@app.post("/transcribe")
async def transcribe(body: dict = Body(...)):
file_id = body.get("file_id")
if not file_id:
raise HTTPException(status_code=400, detail="file_id required")
matches = [m for m in DATA_DIR.glob(f"{file_id}.*") if ".converted." not in m.name]
if not matches:
raise HTTPException(status_code=404, detail="Datei nicht gefunden")
try:
text = transcribe_audio(matches[0])
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Transkription fehlgeschlagen: {e}"
)
return {"transcription": text}
@app.post("/process")
async def process_text(text: str = Form(...)):
if not text.strip():
raise HTTPException(status_code=400, detail="Kein Text")
try:
return {"processed": process_with_ai(text)}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/summarize")
async def summarize(transcript: str = Form(...), title: str = Form(...)):
"""Erstellt eine KI-Zusammenfassung des finalen Transkripts."""
if not transcript.strip():
raise HTTPException(status_code=400, detail="Kein Transkript")
try:
return {"summary": summarize_with_ai(transcript, title or "Meeting")}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/send")
async def send(
email: str = Form(...),
title: str = Form(...),
summary: str = Form(...),
transcript: str = Form(...),
):
"""Generiert PDF und sendet/speichert es."""
if not email or "@" not in email:
raise HTTPException(status_code=400, detail="Ungültige E-Mail")
try:
deliver(email, title or "Meeting", summary, transcript)
return {"success": True}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)