658 lines
21 KiB
Python
658 lines
21 KiB
Python
"""
|
|
Celery tasks for async transcription, diarization, and email notifications.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
import tempfile
|
|
import re
|
|
from datetime import datetime
|
|
|
|
from .celery_app import celery_app
|
|
from .autotranscript import Scraibe
|
|
from .summarizer import SummarizerClient, SummarizerError
|
|
from .misc import setup_logging
|
|
from .email_sender import send_email, EmailError, load_template
|
|
from .email_sender import create_transcript_docx, create_summary_docx
|
|
|
|
logger = logging.getLogger("scraibe.tasks")
|
|
|
|
|
|
def _local_part(email: str) -> str:
|
|
"""
|
|
Extract the part before '@' from an email, sanitized for filenames.
|
|
"""
|
|
local = (email or "").split("@")[0].strip()
|
|
local = "".join(ch if ch.isalnum() or ch in ("-", "_", ".") else "_" for ch in local)
|
|
return local or "user"
|
|
|
|
|
|
def _date_tag() -> str:
|
|
"""
|
|
Date tag in DD-MON-YYYY format (e.g. 01-JAN-2025).
|
|
"""
|
|
return datetime.utcnow().strftime("%d-%b-%Y").upper()
|
|
|
|
|
|
def _safe_filename(base: str, local: str, date_tag: str, ext: str) -> str:
|
|
"""
|
|
Create a temp file with the requested logical name.
|
|
Uses mktemp for uniqueness but keeps the desired name pattern.
|
|
"""
|
|
name = f"{base}-{local}-{date_tag}{ext}"
|
|
return tempfile.mktemp(prefix=name.replace(".", ""), suffix=ext)
|
|
|
|
|
|
def _remove_file(path: str):
|
|
"""
|
|
Remove a file if it exists. Best-effort; logs but never raises.
|
|
"""
|
|
if not path:
|
|
return
|
|
try:
|
|
if os.path.exists(path):
|
|
os.remove(path)
|
|
except Exception as e:
|
|
logger.warning("Failed to remove file %s: %s", path, e)
|
|
|
|
|
|
def _get_subject(env_var: str, default: str) -> str:
|
|
"""
|
|
Safely read an email subject from an environment variable.
|
|
Uses default if unset or blank. Logs the final value.
|
|
"""
|
|
value = (os.getenv(env_var) or "").strip()
|
|
subject = value or default
|
|
logger.info("Email subject [%s] = %s", env_var, subject)
|
|
return subject
|
|
|
|
|
|
def get_queue_position(task_id: str) -> int:
|
|
"""
|
|
Estimate the job's position in the queue.
|
|
Returns:
|
|
- A positive int if we can estimate (1 = first in line).
|
|
- 0 if we cannot reliably determine position.
|
|
"""
|
|
try:
|
|
inspect = celery_app.control.inspect()
|
|
reserved = inspect.reserved() or {} # queued but not yet running
|
|
active = inspect.active() or {} # currently running
|
|
|
|
# Count tasks ahead of this one in the reserved (waiting) queue
|
|
ahead = 0
|
|
found = False
|
|
for _, tasks in list(reserved.values()):
|
|
for t in tasks:
|
|
tid = t.get("id")
|
|
if tid == task_id:
|
|
found = True
|
|
break
|
|
ahead += 1
|
|
if found:
|
|
break
|
|
|
|
# If not found in reserved, it may already be active or not yet visible.
|
|
# In that case, treat it as position 1.
|
|
if found:
|
|
return max(ahead + 1, 1)
|
|
else:
|
|
return 1
|
|
except Exception:
|
|
# If inspection fails, don't guess; caller should use a safe message.
|
|
return 0
|
|
|
|
|
|
def send_initial_email(to: str, queue_pos: int):
|
|
"""
|
|
Send initial confirmation email with queue position.
|
|
Subject is customizable via EMAIL_SUBJECT_UPLOAD.
|
|
"""
|
|
subject = _get_subject(
|
|
"EMAIL_SUBJECT_UPLOAD",
|
|
"ScrAIbe: Your transcription request has been received",
|
|
)
|
|
|
|
body = (
|
|
"Hello,\n\n"
|
|
"We have received your audio file for transcription.\n"
|
|
)
|
|
|
|
if queue_pos > 0:
|
|
body += f"Your request is currently number {queue_pos} in the queue.\n"
|
|
queue_position_display = (
|
|
f'<span style="color:{_accent_color()}; font-weight:bold;">{queue_pos}</span>'
|
|
)
|
|
else:
|
|
body += "Your request has been queued for processing.\n"
|
|
queue_position_display = "the queue"
|
|
|
|
body += (
|
|
"\n"
|
|
"You will receive an email with your transcript (and summary, if requested) "
|
|
"once processing is complete.\n\n"
|
|
"If you have any questions, contact us at "
|
|
f"{os.getenv('EMAIL_CONTACT_ADDRESS', 'support@example.com')}.\n\n"
|
|
"This is an automated message from ScrAIbe.\n"
|
|
)
|
|
|
|
html = None
|
|
try:
|
|
html = load_template(
|
|
"upload_notification_template.html",
|
|
queue_position_text=queue_position_display,
|
|
)
|
|
except EmailError as e:
|
|
logger.warning("Failed to render upload notification template: %s", e)
|
|
|
|
try:
|
|
send_email(to=to, subject=subject, body=body, html=html, attachments=[])
|
|
logger.info("Initial confirmation email sent to %s", to)
|
|
except EmailError as e:
|
|
logger.error("Failed to send initial email to %s: %s", to, e)
|
|
|
|
|
|
def send_success_email(
|
|
to: str,
|
|
transcript_text: str,
|
|
summary_text: str,
|
|
attachments: list,
|
|
task_id: str,
|
|
):
|
|
"""
|
|
Send final email with transcript and attachments.
|
|
Subject is customizable via EMAIL_SUBJECT_SUCCESS.
|
|
Falls back to a safe default if the env var is missing or blank.
|
|
"""
|
|
subject = _get_subject(
|
|
"EMAIL_SUBJECT_SUCCESS",
|
|
"ScrAIbe: Your transcript is ready",
|
|
)
|
|
|
|
body = (
|
|
"Hello,\n\n"
|
|
"Your transcription is ready.\n\n"
|
|
"Please find the transcript and JSON files attached.\n"
|
|
)
|
|
|
|
if summary_text:
|
|
body += (
|
|
"\n"
|
|
"SUMMARY\n"
|
|
"-------\n"
|
|
f"{summary_text}\n"
|
|
)
|
|
|
|
body += (
|
|
"\n"
|
|
"Job ID: " + str(task_id) + "\n\n"
|
|
"If you have any questions, contact us at "
|
|
f"{os.getenv('EMAIL_CONTACT_ADDRESS', 'support@example.com')}.\n\n"
|
|
"This is an automated message from ScrAIbe.\n"
|
|
)
|
|
|
|
html = None
|
|
try:
|
|
html = load_template("success_template.html")
|
|
except EmailError as e:
|
|
logger.warning("Failed to render success template: %s", e)
|
|
|
|
try:
|
|
send_email(
|
|
to=to,
|
|
subject=subject,
|
|
body=body,
|
|
html=html,
|
|
attachments=attachments,
|
|
)
|
|
logger.info("Success email sent to %s for job %s with subject: %s", to, task_id, subject)
|
|
except EmailError as e:
|
|
logger.error("Failed to send success email to %s for job %s: %s", to, task_id, e)
|
|
|
|
|
|
def send_error_email(to: str, error_message: str, task_id: str):
|
|
"""
|
|
Send error notification email.
|
|
Subject is customizable via EMAIL_SUBJECT_ERROR.
|
|
"""
|
|
subject = _get_subject(
|
|
"EMAIL_SUBJECT_ERROR",
|
|
"ScrAIbe: Error with your transcription request",
|
|
)
|
|
|
|
body = (
|
|
"Hello,\n\n"
|
|
"We encountered an error while processing your transcription request.\n\n"
|
|
f"Details: {error_message}\n\n"
|
|
"Job ID: " + str(task_id) + "\n\n"
|
|
"Please contact your administrator if the problem persists.\n\n"
|
|
"If you have any questions, contact us at "
|
|
f"{os.getenv('EMAIL_CONTACT_ADDRESS', 'support@example.com')}.\n\n"
|
|
"This is an automated message from ScrAIbe.\n"
|
|
)
|
|
|
|
html = None
|
|
try:
|
|
html = load_template(
|
|
"error_notification_template.html",
|
|
exception=str(error_message),
|
|
)
|
|
except EmailError as e:
|
|
logger.warning("Failed to render error template: %s", e)
|
|
|
|
try:
|
|
send_email(to=to, subject=subject, body=body, html=html, attachments=[])
|
|
logger.info("Error email sent to %s for job %s", to, task_id)
|
|
except EmailError as e:
|
|
logger.error("Failed to send error email to %s for job %s: %s", to, task_id, e)
|
|
|
|
|
|
def _accent_color() -> str:
|
|
"""
|
|
Return the accent color used in emails/templates.
|
|
"""
|
|
return (os.getenv("EMAIL_ACCENT_COLOR") or "#7C6DA0").strip()
|
|
|
|
|
|
def _identify_speakers_via_llm(scraibe, transcript_text: str) -> dict:
|
|
"""
|
|
Use the summarizer LLM to identify speakers in the transcript.
|
|
|
|
Returns: dict mapping normalized speaker labels to name/role, e.g.
|
|
{"SPEAKER 1": "JUDGE MARTINEZ", "SPEAKER 2": "DEFENSE COUNSEL"}
|
|
"""
|
|
try:
|
|
scraibe._ensure_summarizer()
|
|
summarizer = scraibe._summarizer
|
|
except Exception as e:
|
|
logger.warning("Failed to initialize summarizer for speaker identification: %s", e)
|
|
return {}
|
|
|
|
prompt = (
|
|
"Below is a transcript with speaker labels like 'SPEAKER 1', 'SPEAKER 2', etc. "
|
|
"Based on the context and how each speaker talks, identify each speaker as:\n"
|
|
"- Their real name, if it is clearly mentioned or strongly implied, OR\n"
|
|
"- A concise role/position (e.g., Judge, Doctor, Manager, Interviewer, Client, Witness), "
|
|
"if their identity is not clear.\n"
|
|
"Do not invent random personal names. "
|
|
"Do not add extra commentary. Output ONLY a mapping in this exact format, one per line:\n"
|
|
"SPEAKER 1: Name or Role\n"
|
|
"SPEAKER 2: Name or Role\n"
|
|
"SPEAKER 3: Name or Role\n"
|
|
"\n"
|
|
"Transcript:\n"
|
|
+ transcript_text
|
|
)
|
|
|
|
try:
|
|
reply = summarizer._chat_completion(
|
|
system_prompt="You are an expert legal and business meeting analyst.",
|
|
user_prompt=prompt,
|
|
)
|
|
except Exception as e:
|
|
logger.warning("LLM call failed during speaker identification: %s", e)
|
|
return {}
|
|
|
|
speaker_map = {}
|
|
for m in re.finditer(r"SPEAKER\s+(\d+)\s*:\s*(.+)", reply, re.IGNORECASE):
|
|
spk = f"SPEAKER {m.group(1).strip()}"
|
|
name = m.group(2).strip().rstrip(".").upper()
|
|
if name:
|
|
speaker_map[spk] = name
|
|
|
|
logger.info("Speaker identification mapping: %s", speaker_map)
|
|
return speaker_map
|
|
|
|
|
|
def _apply_speaker_map(transcript_text: str, segments: list, speaker_map: dict) -> tuple:
|
|
"""
|
|
Apply speaker_map to:
|
|
- transcript_text lines like "[00:12] SPEAKER 1: ..."
|
|
- segment speaker fields.
|
|
|
|
Returns (updated_transcript_text, updated_segments).
|
|
"""
|
|
if not speaker_map:
|
|
return transcript_text, segments
|
|
|
|
# Normalize function: e.g. "SPEAKER 1" -> "SPEAKER 1"
|
|
def normalize_label(label: str) -> str:
|
|
return re.sub(r"\s+", " ", re.sub(r"[^A-Z0-9\s]", "", label.upper())).strip()
|
|
|
|
# Replace in transcript lines
|
|
def replace_in_line(line: str) -> str:
|
|
# Match: [MM:SS] or [HH:MM:SS] then speaker label then colon
|
|
# Use a lazy match for the label up to the first colon
|
|
def repl(m):
|
|
prefix = m.group(1) # e.g. "[00:12] "
|
|
label = m.group(2).strip()
|
|
normalized = normalize_label(label)
|
|
new_label = speaker_map.get(normalized, label)
|
|
return f"{prefix}{new_label}: "
|
|
return re.sub(
|
|
r"(\[\d+:\d+(?::\d+)?\]\s*)(.+?):\s*",
|
|
repl,
|
|
line,
|
|
count=1,
|
|
)
|
|
|
|
updated_transcript = "\n".join(
|
|
replace_in_line(line) for line in transcript_text.splitlines()
|
|
)
|
|
|
|
# Update segments
|
|
updated_segments = []
|
|
for seg in segments:
|
|
sp = (seg.get("speaker") or "").strip()
|
|
sp_norm = normalize_label(sp)
|
|
sp_new = speaker_map.get(sp_norm, sp)
|
|
seg = dict(seg)
|
|
seg["speaker"] = sp_new
|
|
updated_segments.append(seg)
|
|
|
|
return updated_transcript, updated_segments
|
|
|
|
|
|
@celery_app.task(
|
|
name="scraibe.tasks.process_transcription_task",
|
|
bind=True,
|
|
max_retries=1,
|
|
task_time_limit=14400, # 4 hours
|
|
task_soft_time_limit=13500, # warn at 3h45m
|
|
)
|
|
def process_transcription_task(
|
|
self,
|
|
audio_path: str,
|
|
task_type: str,
|
|
language: str,
|
|
num_speakers: int,
|
|
email_to: str,
|
|
email_cc: str,
|
|
include_summary: bool,
|
|
identify_speakers: bool = False,
|
|
):
|
|
"""
|
|
Async task: transcribe audio, optionally identify speakers, optionally summarize, then email results.
|
|
Cleans up temporary files after completion.
|
|
"""
|
|
task_id = self.request.id
|
|
|
|
log_level = os.getenv("LOG_LEVEL", "INFO")
|
|
setup_logging(level=log_level)
|
|
|
|
temp_files = []
|
|
local = _local_part(email_to)
|
|
date_tag = _date_tag()
|
|
|
|
try:
|
|
# 1) Queue position and initial email
|
|
queue_pos = get_queue_position(task_id)
|
|
send_initial_email(to=email_to, queue_pos=queue_pos)
|
|
|
|
# 2) Initialize Scraibe
|
|
try:
|
|
scraibe = Scraibe(verbose=True)
|
|
except Exception as e:
|
|
send_error_email(
|
|
to=email_to,
|
|
error_message=f"Failed to initialize transcription service: {e}",
|
|
task_id=task_id,
|
|
)
|
|
raise
|
|
|
|
# 3) Transcription (always first, without summarization)
|
|
result = scraibe.transcribe(
|
|
audio_file=audio_path,
|
|
language=language or None,
|
|
num_speakers=int(num_speakers) if num_speakers else None,
|
|
verbose=True,
|
|
for_export=True,
|
|
)
|
|
transcript_text = result.get("transcript", "")
|
|
segments = result.get("segments", [])
|
|
raw_result = result.get("raw_result")
|
|
summary_text = ""
|
|
|
|
# 4) Optional speaker identification (before summarization)
|
|
if identify_speakers:
|
|
try:
|
|
speaker_map = _identify_speakers_via_llm(scraibe, transcript_text)
|
|
if speaker_map:
|
|
transcript_text, segments = _apply_speaker_map(
|
|
transcript_text, segments, speaker_map
|
|
)
|
|
logger.info("Applied speaker identification to transcript and segments.")
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Speaker identification failed; continuing with original labels: %s", e
|
|
)
|
|
|
|
# 5) Summarization (if requested) using the (possibly identified) transcript
|
|
if include_summary:
|
|
try:
|
|
summary_text = scraibe._summarizer.summarize_transcript(transcript_text)
|
|
except Exception as e:
|
|
logger.warning("Summarization failed; continuing without summary: %s", e)
|
|
summary_text = ""
|
|
|
|
# 6) Prepare files
|
|
|
|
# Transcript .md
|
|
md_transcript_path = _safe_filename("TRANSCRIPT", local, date_tag, ".md")
|
|
with open(md_transcript_path, "w", encoding="utf-8") as f:
|
|
f.write("# Transcript\n\n")
|
|
f.write(transcript_text)
|
|
temp_files.append(md_transcript_path)
|
|
|
|
# Transcript .docx (with summary appended if present)
|
|
docx_transcript_path = _safe_filename("TRANSCRIPT", local, date_tag, ".docx")
|
|
if summary_text:
|
|
create_transcript_docx(
|
|
transcript_text,
|
|
docx_transcript_path,
|
|
summary_text=summary_text,
|
|
)
|
|
else:
|
|
create_transcript_docx(
|
|
transcript_text,
|
|
docx_transcript_path,
|
|
)
|
|
temp_files.append(docx_transcript_path)
|
|
|
|
# JSON as SOURCE
|
|
json_data = {
|
|
"task": task_type,
|
|
"transcript": transcript_text,
|
|
"segments": segments,
|
|
"metadata": {
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"job_id": task_id,
|
|
},
|
|
}
|
|
if summary_text:
|
|
json_data["summary"] = summary_text
|
|
if raw_result is not None:
|
|
json_data["raw_result"] = raw_result
|
|
|
|
json_path = _safe_filename("SOURCE", local, date_tag, ".json")
|
|
with open(json_path, "w", encoding="utf-8") as f:
|
|
json.dump(json_data, f, indent=2, ensure_ascii=False)
|
|
temp_files.append(json_path)
|
|
|
|
# 7) Build attachments list
|
|
attachments = [
|
|
md_transcript_path,
|
|
docx_transcript_path,
|
|
json_path,
|
|
]
|
|
|
|
# 8) Send success email
|
|
send_success_email(
|
|
to=email_to,
|
|
transcript_text=transcript_text,
|
|
summary_text=summary_text if include_summary else "",
|
|
attachments=attachments,
|
|
task_id=task_id,
|
|
)
|
|
|
|
logger.info("Job %s completed successfully.", task_id)
|
|
|
|
except Exception as e:
|
|
logger.error("Error processing job %s: %s", task_id, e, exc_info=True)
|
|
send_error_email(
|
|
to=email_to,
|
|
error_message=str(e),
|
|
task_id=task_id,
|
|
)
|
|
raise e
|
|
finally:
|
|
# 9) Cleanup
|
|
for path in temp_files:
|
|
_remove_file(path)
|
|
if audio_path:
|
|
_remove_file(audio_path)
|
|
logger.info("Cleanup completed for job %s.", task_id)
|
|
|
|
|
|
@celery_app.task(
|
|
name="scraibe.tasks.process_watch_file_task",
|
|
bind=True,
|
|
max_retries=1,
|
|
task_time_limit=14400,
|
|
task_soft_time_limit=13500,
|
|
)
|
|
def process_watch_file_task(
|
|
self,
|
|
file_path: str,
|
|
):
|
|
"""
|
|
Async task for watch-folder mode:
|
|
- Transcribe + summarize
|
|
- Email results
|
|
- Optionally delete source file
|
|
"""
|
|
task_id = self.request.id
|
|
|
|
log_level = os.getenv("LOG_LEVEL", "INFO")
|
|
setup_logging(level=log_level)
|
|
|
|
email_to = os.getenv("WATCH_EMAIL_TO") or os.getenv("EMAIL_DEFAULT_TO")
|
|
if not email_to:
|
|
logger.error("No email address configured for watch-folder mode.")
|
|
raise RuntimeError("WATCH_EMAIL_TO or EMAIL_DEFAULT_TO not set.")
|
|
|
|
delete_on_success = os.getenv("WATCH_DELETE_ON_SUCCESS", "true").strip().lower() in ("true", "1", "yes")
|
|
|
|
temp_files = []
|
|
local = "watch"
|
|
date_tag = _date_tag()
|
|
|
|
try:
|
|
scraibe = Scraibe(verbose=True)
|
|
|
|
result = scraibe.transcript_and_summarize(
|
|
audio_file=file_path,
|
|
language=None,
|
|
num_speakers=None,
|
|
verbose=True,
|
|
for_export=True,
|
|
)
|
|
|
|
transcript_text = result.get("transcript", "")
|
|
summary_text = result.get("summary", "")
|
|
segments = result.get("segments", [])
|
|
raw_result = result.get("raw_result")
|
|
|
|
# Transcript .md
|
|
md_transcript_path = _safe_filename("TRANSCRIPT", local, date_tag, ".md")
|
|
with open(md_transcript_path, "w", encoding="utf-8") as f:
|
|
f.write("# Transcript\n\n")
|
|
f.write(transcript_text)
|
|
temp_files.append(md_transcript_path)
|
|
|
|
# Transcript .docx
|
|
docx_transcript_path = _safe_filename("TRANSCRIPT", local, date_tag, ".docx")
|
|
create_transcript_docx(
|
|
transcript_text,
|
|
docx_transcript_path,
|
|
)
|
|
temp_files.append(docx_transcript_path)
|
|
|
|
# Summary .md
|
|
md_summary_path = _safe_filename("SUMMARY", local, date_tag, ".md")
|
|
with open(md_summary_path, "w", encoding="utf-8") as f:
|
|
f.write("# Summary\n\n")
|
|
f.write(summary_text)
|
|
temp_files.append(md_summary_path)
|
|
|
|
# Summary .docx
|
|
docx_summary_path = _safe_filename("SUMMARY", local, date_tag, ".docx")
|
|
create_summary_docx(
|
|
summary_text,
|
|
docx_summary_path,
|
|
)
|
|
temp_files.append(docx_summary_path)
|
|
|
|
# JSON as SOURCE
|
|
json_data = {
|
|
"task": "watch_transcript_and_summarize",
|
|
"transcript": transcript_text,
|
|
"summary": summary_text,
|
|
"segments": segments,
|
|
"metadata": {
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
"job_id": task_id,
|
|
"source_file": file_path,
|
|
},
|
|
}
|
|
if raw_result is not None:
|
|
json_data["raw_result"] = raw_result
|
|
|
|
json_path = _safe_filename("SOURCE", local, date_tag, ".json")
|
|
with open(json_path, "w", encoding="utf-8") as f:
|
|
json.dump(json_data, f, indent=2, ensure_ascii=False)
|
|
temp_files.append(json_path)
|
|
|
|
# Attachments
|
|
attachments = [
|
|
md_transcript_path,
|
|
docx_transcript_path,
|
|
md_summary_path,
|
|
docx_summary_path,
|
|
json_path,
|
|
]
|
|
|
|
# Send email
|
|
send_success_email(
|
|
to=email_to,
|
|
transcript_text=transcript_text,
|
|
summary_text=summary_text,
|
|
attachments=attachments,
|
|
task_id=task_id,
|
|
)
|
|
|
|
logger.info("Watch-folder job %s completed for %s.", task_id, file_path)
|
|
|
|
# Delete source file if configured
|
|
if delete_on_success and os.path.exists(file_path):
|
|
try:
|
|
os.remove(file_path)
|
|
logger.info("Deleted source file: %s", file_path)
|
|
except Exception as e:
|
|
logger.warning("Failed to delete source file %s: %s", file_path, e)
|
|
|
|
except Exception as e:
|
|
logger.error("Error processing watch file %s: %s", file_path, e, exc_info=True)
|
|
send_error_email(
|
|
to=email_to,
|
|
error_message=str(e),
|
|
task_id=task_id,
|
|
)
|
|
raise e
|
|
finally:
|
|
# Cleanup temp files
|
|
for path in temp_files:
|
|
_remove_file(path)
|
|
logger.info("Watch-folder job %s cleanup completed.", task_id)
|