🔨(agents) change json metadata dump to bucket_s3

Change the saving of the json metadata to the minio bucket_s3
to prepare the changes to map the name in the diarization

src/agents/metadata-extracter.py

@@ -0,0 +1,211 @@
+"""Visible room join agent (for connection/testing) + JSON speaker intervals per participant."""
+
+import logging
+import os
+from datetime import datetime, timezone
+from typing import Dict, List, Optional
+import json
+import pathlib
+
+from io import BytesIO
+from minio import Minio
+from minio.error import S3Error
+
+from dotenv import load_dotenv
+from livekit import api, rtc
+from livekit.agents import (
+    AutoSubscribe,
+    JobContext,
+    JobRequest,
+    WorkerOptions,
+    WorkerPermissions,
+    cli,
+)
+
+load_dotenv()
+
+logger = logging.getLogger("visible-joiner")
+logger.setLevel(logging.INFO)
+logger.propagate = False
+
+if not logger.handlers:
+    _h = logging.StreamHandler()
+    _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:
+    def __init__(self, room_name: str, write_json: bool = True):
+        self.room_name = room_name
+        self.active_since: Dict[str, datetime] = {}
+        self.by_participant: Dict[str, List[dict]] = {}
+        self.write_json_flag = write_json
+
+        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
+
+    def _now(self) -> datetime:
+        return datetime.now(timezone.utc)
+
+    def _emit_interval(self, identity: str, start: datetime, end: datetime):
+        dur = max(0.0, (end - start).total_seconds())
+        seg = {
+            "start_iso": start.isoformat(),
+            "end_iso": end.isoformat(),
+            "duration_sec": round(dur, 3),
+        }
+        self.by_participant.setdefault(identity, []).append(seg)
+
+    def update_active_speakers(self, current_identities: List[str]):
+        now = self._now()
+        current = set(current_identities)
+        before = set(self.active_since.keys())
+
+        for ident in current - before:
+            self.active_since[ident] = now
+
+        for ident in before - current:
+            start = self.active_since.pop(ident, None)
+            if start:
+                self._emit_interval(ident, start, now)
+
+    def on_participant_disconnected(self, identity: str):
+        now = self._now()
+        start = self.active_since.pop(identity, None)
+        if start:
+            self._emit_interval(identity, start, now)
+
+    def flush_all(self):
+        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:
+        return {
+            "room": self.room_name,
+            "generated_at": self._now().isoformat(),
+            "by_participant": self.by_participant,
+        }
+
+    
+
+    def write_json(self):
+        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"),
+            secret_key=os.getenv("AWS_S3_SECRET_ACCESS_KEY"),
+            secure= _as_bool(os.getenv("AWS_S3_SECURE_ACCESS", "false"))
+        )
+
+        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"
+
+        data = json.dumps(payload, indent=2).encode("utf-8")
+        stream = BytesIO(data)
+
+        try:
+            minio_client.put_object(
+                bucket,
+                object_name,
+                stream,
+                length=len(data),
+                content_type="application/json",
+            )
+            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)
+
+
+
+async def entrypoint(ctx: JobContext):
+    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
+
+    lp = ctx.room.local_participant
+    logger.info(
+        "Connected to SFU; room=%s identity=%s sid=%s",
+        ctx.room.name,
+        getattr(lp, "identity", "<unknown>"),
+        getattr(lp, "sid", "<unknown>"),
+    )
+
+    tracker = SpeakerTracker(ctx.room.name, write_json=True)
+
+    def _on_participant_connected(p: rtc.RemoteParticipant):
+        logger.info("remote participant connected: %s (%s)", p.identity, p.sid)
+
+    def _on_active_speakers_changed(speakers: list[rtc.Participant]):
+        idents = [p.identity for p in speakers]
+        logger.info("active speakers changed: %s", idents)
+        tracker.update_active_speakers(idents)
+
+    def _on_participant_disconnected(p: rtc.RemoteParticipant):
+        logger.info("remote participant disconnected: %s", p.identity)
+        tracker.on_participant_disconnected(p.identity)
+
+    if not ctx.proc.userdata.get("events_registered"):
+        ctx.room.on("participant_connected", _on_participant_connected)
+        ctx.room.on("active_speakers_changed", _on_active_speakers_changed)
+        ctx.room.on("participant_disconnected", _on_participant_disconnected)
+        ctx.proc.userdata["events_registered"] = True
+
+    async def cleanup():
+        if ctx.proc.userdata.get("events_registered"):
+            ctx.room.off("participant_connected", _on_participant_connected)
+            ctx.room.off("active_speakers_changed", _on_active_speakers_changed)
+            ctx.room.off("participant_disconnected", _on_participant_disconnected)
+            ctx.proc.userdata["events_registered"] = False
+        tracker.flush_all()
+        tracker.write_json()
+
+    ctx.add_shutdown_callback(cleanup)
+
+
+
+async def handle_job_request(job_req: JobRequest) -> None:
+    room_name = job_req.room.name
+    agent_identity = f"{VISIBLE_AGENT_NAME}-{room_name}"
+
+    async with api.LiveKitAPI() as lk:
+        try:
+            resp = await lk.room.list_participants(
+                list=api.ListParticipantsRequest(room=room_name)
+            )
+            already_present = any(
+                p.kind == rtc.ParticipantKind.PARTICIPANT_KIND_AGENT
+                and p.identity == agent_identity
+                for p in resp.participants
+            )
+            if already_present:
+                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)
+                await job_req.accept(identity=agent_identity)
+        except Exception:
+            logger.exception("Error treating the job for '%s'", room_name)
+            await job_req.reject()
+
+
+if __name__ == "__main__":
+    cli.run_app(
+        WorkerOptions(
+            entrypoint_fnc=entrypoint,
+            request_fnc=handle_job_request,
+            agent_name=VISIBLE_AGENT_NAME,
+            permissions=WorkerPermissions(),
+        )
+    )