diff --git a/src/agents/metadata-extractor.py b/src/agents/metadata-extractor.py index 1963ada0..ce766ee9 100644 --- a/src/agents/metadata-extractor.py +++ b/src/agents/metadata-extractor.py @@ -1,14 +1,12 @@ -"""Visible room join agent (for connection/testing) + JSON speaker intervals per participant.""" +"""Metadata agent that tracks active speakers and their speaking intervals.""" +import json import logging import os -from datetime import datetime, timezone -from typing import Dict, List, Optional -import json import pathlib +from datetime import datetime, timezone from io import BytesIO -from minio import Minio -from minio.error import S3Error +from typing import Dict, List, Optional from dotenv import load_dotenv from livekit import api, rtc @@ -20,6 +18,8 @@ from livekit.agents import ( WorkerPermissions, cli, ) +from minio import Minio +from minio.error import S3Error load_dotenv() @@ -29,13 +29,18 @@ logger.propagate = False if not logger.handlers: _h = logging.StreamHandler() - _h.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s %(name)s - %(message)s")) + _h.setFormatter( + logging.Formatter("%(asctime)s - %(levelname)s %(name)s - %(message)s") + ) logger.addHandler(_h) VISIBLE_AGENT_NAME = os.getenv("VISIBLE_AGENT_NAME", "visible-joiner") class SpeakerTracker: + """Track active speakers and their speaking intervals.""" + def __init__(self, room_name: str, write_json: bool = True): + """Track active speakers and their speaking intervals.""" self.room_name = room_name self.active_since: Dict[str, datetime] = {} self.by_participant: Dict[str, List[dict]] = {} @@ -44,7 +49,10 @@ class SpeakerTracker: ts = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') outdir = pathlib.Path("./speaker_logs") outdir.mkdir(parents=True, exist_ok=True) - self.json_path: Optional[pathlib.Path] = outdir / f"speakers_{room_name}_{ts}.json" if write_json else None + self.json_path: Optional[pathlib.Path] = ( + outdir / f"speakers_{room_name}_{ts}.json" + if write_json else None +) def _now(self) -> datetime: return datetime.now(timezone.utc) @@ -59,6 +67,7 @@ class SpeakerTracker: self.by_participant.setdefault(identity, []).append(seg) def update_active_speakers(self, current_identities: List[str]): + """Update the list of currently active speakers.""" now = self._now() current = set(current_identities) before = set(self.active_since.keys()) @@ -72,18 +81,21 @@ class SpeakerTracker: self._emit_interval(ident, start, now) def on_participant_disconnected(self, identity: str): + """Handle participant disconnection by finalizing their active interval.""" now = self._now() start = self.active_since.pop(identity, None) if start: self._emit_interval(identity, start, now) def flush_all(self): + """Flush all active speakers as ended now.""" now = self._now() for ident, start in list(self.active_since.items()): self._emit_interval(ident, start, now) self.active_since.clear() def build_json(self) -> dict: + """Build the JSON structure for the collected speaker intervals.""" return { "room": self.room_name, "generated_at": self._now().isoformat(), @@ -92,15 +104,16 @@ class SpeakerTracker: def write_json(self): + """Write the collected speaker intervals to a JSON file and upload to MinIO.""" def _as_bool(v: str, default=False): if v is None: return default return v.strip().lower() in ("1", "true", "yes", "y") - + if not self.write_json_flag: return payload = self.build_json() - + minio_client = Minio( endpoint=os.getenv("AWS_S3_ENDPOINT_URL"), access_key=os.getenv("AWS_S3_ACCESS_KEY_ID"), @@ -110,9 +123,11 @@ class SpeakerTracker: bucket = "meet-media-storage" ts = self._now().strftime("%Y%m%dT%H%M%SZ") - object_name = f"speaker_logs/{self.room_name}/speakers_{self.room_name}_{ts}.json" + prefix = f"speaker_logs/{self.room_name}" + object_name = f"{prefix}/speakers_{self.room_name}_{ts}.json" data = json.dumps(payload, indent=2).encode("utf-8") + stream = BytesIO(data) try: @@ -123,12 +138,17 @@ class SpeakerTracker: length=len(data), content_type="application/json", ) - logger.info("Uploaded speaker intervals JSON to s3://%s/%s", bucket, object_name) + logger.info("Uploaded speaker intervals JSON to s3://%s/%s", + bucket, + object_name) except S3Error: - logger.exception("Failed to upload JSON to bucket=%s object=%s", bucket, object_name) + logger.exception("Failed to upload JSON to bucket=%s object=%s", + bucket, + object_name) async def entrypoint(ctx: JobContext): + """Main entrypoint for the metadata extractor agent.""" await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY) lp = ctx.room.local_participant @@ -173,6 +193,7 @@ async def entrypoint(ctx: JobContext): async def handle_job_request(job_req: JobRequest) -> None: + """Accept or reject the job request based on agent presence in the room.""" room_name = job_req.room.name agent_identity = f"{VISIBLE_AGENT_NAME}-{room_name}" @@ -190,7 +211,9 @@ async def handle_job_request(job_req: JobRequest) -> None: logger.info("Agent already in the room '%s' — reject", room_name) await job_req.reject() else: - logger.info("Accept job for '%s' — identity=%s", room_name, agent_identity) + logger.info("Accept job for '%s' — identity=%s", + room_name, + agent_identity) await job_req.accept(identity=agent_identity) except Exception: logger.exception("Error treating the job for '%s'", room_name) @@ -205,4 +228,4 @@ if __name__ == "__main__": agent_name=VISIBLE_AGENT_NAME, permissions=WorkerPermissions(), ) - ) \ No newline at end of file + ) diff --git a/src/summary/summary/core/celery_worker.py b/src/summary/summary/core/celery_worker.py index d838eb30..313fff76 100644 --- a/src/summary/summary/core/celery_worker.py +++ b/src/summary/summary/core/celery_worker.py @@ -6,6 +6,7 @@ import json import os import tempfile import time +from datetime import datetime, timedelta from pathlib import Path from typing import Optional @@ -30,6 +31,70 @@ from summary.core.prompt import ( PROMPT_USER_PART, ) + +def parse_iso(s: str) -> datetime: + """Convert ISO 8601 string to datetime object.""" + return datetime.fromisoformat(s.replace("Z", "+00:00")) + +def overlap(a, b): + """Calculate overlap duration between two segments.""" + return max(0.0, min(a["end"], b["end"]) - max(a["start"], b["start"])) + +def total_duration(segs): + """Total duration of segments.""" + return sum(s["end"] - s["start"] for s in segs) + +def total_overlap(segs_a, segs_b): + """Calculate total overlap duration between two sets of segments.""" + tot = 0.0 + for a in segs_a: + for b in segs_b: + tot += overlap(a, b) + return tot + +def jaccard_score(ov, dur_a, dur_b): + """Calculate the Jaccard similarity score.""" + denom = dur_a + dur_b - ov + return ov / denom if denom > 0 else 0.0 + +def align_participants_seconds(participants, speakers): + """Align participant segments to speaker segments in seconds.""" + min_speaker_start = min(seg["start"] for segs in speakers.values() for seg in segs) + min_participant_iso = min( + parse_iso(seg["start_iso"]) + for plist in participants["by_participant"].values() + for seg in plist + ) + t0_base = min_participant_iso - timedelta(seconds=min_speaker_start) + + out = {} + for pid, segs in participants["by_participant"].items(): + out[pid] = [ + { + "start": (parse_iso(s["start_iso"]) - t0_base).total_seconds(), + "end": (parse_iso(s["end_iso"]) - t0_base).total_seconds(), + } + for s in segs + ] + return out + +def map_speakers_to_participants(speakers, participants): + """Main function to map speakers to participants.""" + participants_seconds = align_participants_seconds(participants, speakers) + + mapping = {} + for spk, s_segs in speakers.items(): + s_dur = total_duration(s_segs) + best_pid, best_score = None, -1.0 + for pid, p_segs in participants_seconds.items(): + p_dur = total_duration(p_segs) + ov = total_overlap(s_segs, p_segs) + score = jaccard_score(ov, s_dur, p_dur) + if score > best_score: + best_score, best_pid = score, pid + mapping[spk] = best_pid + return mapping + settings = get_settings() analytics = get_analytics() @@ -286,10 +351,15 @@ def process_audio_transcribe_summarize_v2( os.remove(temp_file_path) logger.debug("Temporary file removed: %s", temp_file_path) + metadata_name = minio_client.get_object( + settings.aws_storage_bucket_name, + object_name=settings.metadata_file.format(filename=filename) + ) + formatted_transcription = ( DEFAULT_EMPTY_TRANSCRIPTION if not transcription.segments - else format_segments(transcription) + else format_segments(transcription,metadata_name) ) metadata_manager.track_transcription_metadata(task_id, transcription) diff --git a/src/summary/summary/core/config.py b/src/summary/summary/core/config.py index 77dc7aa3..5a05e789 100644 --- a/src/summary/summary/core/config.py +++ b/src/summary/summary/core/config.py @@ -59,6 +59,7 @@ class Settings(BaseSettings): # Summary related settings is_summary_enabled: bool = True + metadata_file: Optional[str] = "{filename}_metadata.json" # Sentry sentry_is_enabled: bool = False