""" Contextra AI Analytics — PostgreSQL + Redis Metadata Storage (v2.0) MLOps 베이스 레이어: 탐지 결과를 PostgreSQL에 저장 + Redis Streams 발행 역할: - frames 테이블: 모든 디코딩 프레임 메타데이터 INSERT - detections 테이블: RF-DETR 탐지 결과 INSERT - events 테이블: 시나리오 조건 충족 시 이벤트 트리거 INSERT - Redis Streams: 실시간 메타데이터 큐 (Metadata Publisher) 설계 원칙: - batch insert + flush interval로 DB 부하 최소화 - private_key 기반 고유 식별 (클라이언트 RTSP 동기화용) - 비동기 INSERT (추론 파이프라인 블로킹 방지) - Redis Streams 발행은 fire-and-forget — 실패해도 PostgreSQL 저장 영향 없음 """ import os import time import json from datetime import datetime, timezone from dataclasses import dataclass, field from typing import Optional, List from contextlib import contextmanager try: import psycopg2 from psycopg2.extras import execute_values except ImportError: psycopg2 = None # --------------------------------------------------------------------------- # Redis 설정 (Metadata Publisher) # --------------------------------------------------------------------------- REDIS_URL = os.environ.get("REDIS_URL", "redis://redis:6379/0") REDIS_STREAMS_ENABLED = os.environ.get("REDIS_STREAMS_ENABLED", "true").lower() == "true" try: import redis as redis_lib except ImportError: redis_lib = None # --------------------------------------------------------------------------- # PostgreSQL 연결 설정 (환경변수에서 읽음) # --------------------------------------------------------------------------- DATABASE_URL = str(os.environ.get("DATABASE_URL", "") or "").strip() if not DATABASE_URL: raise RuntimeError("DATABASE_URL environment variable is required") DB_POOL_SIZE = int(os.environ.get("DB_POOL_SIZE", "10")) DB_MAX_OVERFLOW = int(os.environ.get("DB_MAX_OVERFLOW", "5")) def _parse_db_url(url: str): """Parse DATABASE_URL into connection params.""" # postgresql://user:***@host:port/dbname _, _, db = url.partition("//") at_idx = db.index("@") creds = db[:at_idx] hostdb = db[at_idx + 1:] # Split user:pass — password may contain special chars, split on FIRST colon only colon_idx = creds.index(":") user = creds[:colon_idx] password = creds[colon_idx + 1:] # Split host:port/dbname slash_idx = hostdb.index("/") host_port = hostdb[:slash_idx] # 'contextra-postgres:5432' dbname = hostdb[slash_idx + 1:] # 'contextra' if ":" in host_port: host, port = host_port.rsplit(":", 1) else: host = host_port port = "5432" return { "dbname": dbname, "user": user, "password": password, "host": host or "localhost", "port": int(port), } # --------------------------------------------------------------------------- # Dataclasses # --------------------------------------------------------------------------- @dataclass class FrameMeta: """프레임 메타데이터""" private_key: str stream_id: str = "CAM_01" site_id: str = "SITE01" timestamp_utc_ms: int = 0 frame_sequence_id: int = 0 inference_fps: float = 0.0 width: int = 1920 height: int = 1080 # 이벤트 게이트드 윈도우 프레임만 보유: {"event_id": ..., "event_type": ...} event_context: Optional[dict] = None @dataclass class DetectionMeta: """탐지 결과 메타데이터""" private_key: str track_id: Optional[int] = None class_label: str = "person" confidence: float = 0.0 bbox: tuple = () # (x_min, y_min, x_max, y_max) pixel coords bbox_normalized: tuple = () # (x_min/w, y_min/h, ...) normalized keypoints: Optional[list] = None # pose: [[x,y,conf]x17] normalized (det rows: None) source: str = "det" # det | pose | seg (행 유형 구분) seg: Optional[dict] = None # seg: {fg_px, fg_ratio, mask_w, mask_h} 요약 (seg rows only) @dataclass class EventMeta: """이벤트 메타데이터 (시나리오 조건 충족 시)""" private_key: str stream_id: str = "CAM_01" site_id: str = "SITE01" timestamp_utc_ms: int = 0 event_type: str = "person_detected" severity: str = "INFO" # INFO, WARN, CRITICAL detection_result: dict = field(default_factory=dict) metadata: dict = field(default_factory=dict) # --------------------------------------------------------------------------- # Redis Streams Metadata Publisher (fire-and-forget) # --------------------------------------------------------------------------- class RedisMetadataPublisher: """ Redis Streams 기반 메타데이터 발행기. 스트림 구조: contextra:stream:frames — 프레임 메타데이터 contextra:stream:detections — 탐지 결과 contextra:stream:events — 이벤트 트리거 용도: 실시간 큐 (클라이언트 WebSocket 브로드캐스팅, 다른 마이크로서비스 구독) PostgreSQL는 영구 저장용, Redis Streams는 실시간 배포용. 실패 시 silently pass — analytics 파이프라인이 차단되지 않도록 함. """ STREAM_FRAMES = "contextra:stream:frames" STREAM_DETECTIONS = "contextra:stream:detections" STREAM_EVENTS = "contextra:stream:events" # 스트림 트림 정책: 최신 N개만 유지(approximate) — 무한 성장 방지 (2026-08-17, 237만 엔트리 백로그 발견) STREAM_MAXLEN = int(os.environ.get("REDIS_STREAM_MAXLEN", "100000")) def __init__(self, redis_url: str = REDIS_URL): self._client = None if redis_lib is None or not REDIS_STREAMS_ENABLED: return try: self._client = redis_lib.from_url(redis_url, decode_responses=True) # 연결 테스트 (시작 시 한 번만) self._client.ping() print("[RedisPublisher] Connected") except Exception as e: # 인증 URL과 예외 원문은 로그에 남기지 않는다. kind = type(e).__name__ if kind not in {"AuthenticationError", "ConnectionError", "TimeoutError"}: kind = "RedisError" print(f"[RedisPublisher] Connection failed ({kind}) — streams disabled") self._client = None def publish_frame(self, frame: FrameMeta): """프레임 메타데이터를 Redis Streams에 발행""" if not self._client: return try: message = { "private_key": frame.private_key, "stream_id": frame.stream_id, "site_id": frame.site_id, "timestamp_utc_ms": str(frame.timestamp_utc_ms), "frame_sequence_id": str(frame.frame_sequence_id), "inference_fps": str(frame.inference_fps), "width": str(frame.width), "height": str(frame.height), } self._client.xadd(self.STREAM_FRAMES, message, maxlen=self.STREAM_MAXLEN, approximate=True) except Exception: pass # fire-and-forget def publish_detections(self, detections: List[DetectionMeta]): """탐지 결과를 Redis Streams에 발행 (배치)""" if not self._client or not detections: return try: messages = [] for d in detections: msg = { "private_key": d.private_key, "track_id": str(d.track_id or ""), "class_label": d.class_label, "confidence": str(round(d.confidence, 4)), "bbox": json.dumps(list(d.bbox)), "bbox_normalized": json.dumps([round(x, 4) for x in d.bbox_normalized]), "keypoints": json.dumps(d.keypoints) if d.keypoints is not None else "", "source": d.source, "seg": json.dumps(d.seg) if d.seg is not None else "", } messages.append(msg) # 배치 발행 — 각 메시지를 개별 entry로 (순서 보장 필요 시 single xadd) pipe = self._client.pipeline(transaction=False) for msg in messages: pipe.xadd(self.STREAM_DETECTIONS, msg, maxlen=self.STREAM_MAXLEN, approximate=True) pipe.execute() except Exception: pass # fire-and-forget def publish_event(self, event: EventMeta): """이벤트 메타데이터를 Redis Streams에 발행 (실시간 우선)""" if not self._client: return try: message = { "private_key": event.private_key, "stream_id": event.stream_id, "site_id": event.site_id, "timestamp_utc_ms": str(event.timestamp_utc_ms), "event_type": event.event_type, "severity": event.severity, "detection_result": json.dumps(event.detection_result), "metadata": json.dumps(event.metadata), } self._client.xadd(self.STREAM_EVENTS, message, maxlen=self.STREAM_MAXLEN, approximate=True) except Exception: pass # fire-and-forget def close(self): if self._client: try: self._client.close() except Exception: pass def ping(self) -> bool: """Return whether Redis responds to a bounded PING.""" return bool(self._client and self._client.ping()) # --------------------------------------------------------------------------- # PostgreSQL Metadata Storage (싱글톤 패턴) — 기존 로직 유지 # --------------------------------------------------------------------------- class MetadataStorage: """PostgreSQL 메타데이터 저장소 (싱글톤 패턴)""" def __init__(self, db_url: str = DATABASE_URL): if psycopg2 is None: raise ImportError("psycopg2 is not installed. Install with: pip install psycopg2-binary") self._conn_params = _parse_db_url(db_url) self._connection = None def _get_connection(self): """커넥션 풀에서 연결 획득 (thread-unsafe, caller가 관리 필요)""" if self._connection is None: self._connection = psycopg2.connect(**self._conn_params) self._connection.autocommit = False return self._connection def close(self): """연결 닫기""" if self._connection: try: self._connection.close() except Exception: pass finally: self._connection = None def ping(self) -> bool: """Return whether PostgreSQL accepts a minimal query.""" connection = psycopg2.connect(**self._conn_params) try: connection.autocommit = True with connection.cursor() as cursor: cursor.execute("SELECT 1") return cursor.fetchone() == (1,) finally: connection.close() @contextmanager def get_cursor(self, autocommit=False): """커서 컨텍스트 매니저""" conn = self._get_connection() cur = conn.cursor() try: yield cur if not autocommit: conn.commit() except Exception: conn.rollback() raise def insert_frame(self, frame: FrameMeta): """프레임 메타데이터 INSERT""" with self.get_cursor() as cur: execute_values( cur, """INSERT INTO frames (private_key, stream_id, site_id, timestamp_utc_ms, frame_sequence_id, inference_fps, width, height, event_context) VALUES %s ON CONFLICT (private_key) DO NOTHING""", [(frame.private_key, frame.stream_id, frame.site_id, frame.timestamp_utc_ms, frame.frame_sequence_id, frame.inference_fps, frame.width, frame.height, json.dumps(frame.event_context) if frame.event_context is not None else None)] ) def insert_detections(self, detections: List[DetectionMeta]): """탐지 결과 배치 INSERT""" if not detections: return with self.get_cursor() as cur: # 먼저 frames 테이블에서 private_key로 frame_id 조회 unique_keys = list(set(d.private_key for d in detections)) placeholders = ",".join(["%s"] * len(unique_keys)) cur.execute(f"SELECT id, private_key FROM frames WHERE private_key IN ({placeholders})", unique_keys) key_to_frame_id = {row[1]: row[0] for row in cur.fetchall()} values = [] for d in detections: frame_id = key_to_frame_id.get(d.private_key) if frame_id is None: # frames에 해당 private_key가 없으면 SKIP (应先插入frame) continue values.append(( frame_id, d.private_key, d.track_id, d.class_label, d.confidence, json.dumps(list(d.bbox)), json.dumps([round(x, 4) for x in d.bbox_normalized]), json.dumps(d.keypoints) if d.keypoints is not None else None, d.source, json.dumps(d.seg) if d.seg is not None else None )) if values: execute_values( cur, """INSERT INTO detections (frame_id, private_key, track_id, class_label, confidence, bbox, bbox_normalized, keypoints, source, seg) VALUES %s""", values ) def insert_event(self, event: EventMeta): """이벤트 INSERT (커밋 포함 — autocommit=True는 커넥션이 autocommit=False라 커밋이 건너뛰어져 롤백되던 버그 수정 2026-08-17)""" with self.get_cursor() as cur: execute_values( cur, """INSERT INTO events (private_key, stream_id, site_id, timestamp_utc_ms, event_type, severity, detection_result, metadata) VALUES %s""", [(event.private_key, event.stream_id, event.site_id, event.timestamp_utc_ms, event.event_type, event.severity, json.dumps(event.detection_result), json.dumps(event.metadata))] ) def insert_events(self, events): """이벤트 배치 INSERT — 이벤트 게이트드 프레임 태깅용 row id 목록 반환""" if not events: return [] # autocommit=True 제거 — 커넥션 autocommit=False라 커밋이 건너뛰어져 # INSERT가 롤백되던 버그 (insert_event와 동일, 2026-08-17) with self.get_cursor() as cur: rows = [(e.private_key, e.stream_id, e.site_id, e.timestamp_utc_ms, e.event_type, e.severity, json.dumps(e.detection_result), json.dumps(e.metadata)) for e in events] # fetch=True는 execute_values가 결과를 내부적으로 소비해 이후 # cur.fetchall()이 빈 리스트를 반환한다 (2026-08-18 e2e에서 발견) — # fetch=False로 두고 명시적 fetchall. execute_values(cur, """INSERT INTO events (private_key, stream_id, site_id, timestamp_utc_ms, event_type, severity, detection_result, metadata) VALUES %s RETURNING id""", rows, fetch=False) # 이벤트 게이트드 프레임 태깅용: 삽입된 row id 목록 반환 return [r[0] for r in cur.fetchall()] def insert_event_with_id(self, event: EventMeta): """단일 이벤트 INSERT ... RETURNING id — 새 row id 반환 (실패 시 None, non-fatal)""" try: with self.get_cursor() as cur: cur.execute( """INSERT INTO events (private_key, stream_id, site_id, timestamp_utc_ms, event_type, severity, detection_result, metadata) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) RETURNING id""", (event.private_key, event.stream_id, event.site_id, event.timestamp_utc_ms, event.event_type, event.severity, json.dumps(event.detection_result), json.dumps(event.metadata)) ) return cur.fetchone()[0] except Exception as e: print(f"[MetadataStorage] insert_event_with_id failed: {e}") return None def _row_to_event_dict(self, row): """events row -> dict (SELECT 순서: id, private_key, stream_id, site_id, timestamp_utc_ms, event_type, severity, detection_result, metadata, is_verified, verified_by, verified_at, created_at, context_frames)""" # JSONB 컬럼은 psycopg2가 dict/list로 역직렬화 — str guard (레거시 데이터 대비) def _j(v): if isinstance(v, str): try: return json.loads(v) except Exception: return None return v if isinstance(v, (dict, list)) else None return { "id": str(row[0]), "private_key": row[1], "stream_id": row[2], "site_id": row[3], "timestamp_utc_ms": int(row[4]), "event_type": row[5], "severity": row[6], "detection_result": _j(row[7]), "metadata": _j(row[8]), "is_verified": bool(row[9]) if row[9] is not None else None, "verified_by": row[10], "verified_at": row[11].isoformat() if row[11] is not None else None, "created_at": row[12].isoformat() if row[12] is not None else None, "context_frames": int(row[13]), } def list_events(self, stream_id=None, event_type=None, is_verified=None, limit=50, offset=0): """이벤트 목록 조회 — 컨텍스트 프레임 수 포함 (파라미터화된 동적 WHERE)""" query = """ SELECT e.id, e.private_key, e.stream_id, e.site_id, e.timestamp_utc_ms, e.event_type, e.severity, e.detection_result, e.metadata, e.is_verified, e.verified_by, e.verified_at, e.created_at, (SELECT count(*) FROM frames f WHERE f.event_context->>'event_id' = e.id::text) AS context_frames FROM events e """ where = [] params = [] if stream_id is not None: where.append("e.stream_id = %s") params.append(stream_id) if event_type is not None: where.append("e.event_type = %s") params.append(event_type) if is_verified is not None: where.append("e.is_verified = %s") params.append(is_verified) if where: query += " WHERE " + " AND ".join(where) query += " ORDER BY e.timestamp_utc_ms DESC LIMIT %s OFFSET %s" params.extend([limit, offset]) with self.get_cursor() as cur: cur.execute(query, params) return [self._row_to_event_dict(row) for row in cur.fetchall()] def get_event(self, event_id: str): """단일 이벤트 조회 (id::uuid) — 없으면 None""" with self.get_cursor() as cur: cur.execute(""" SELECT e.id, e.private_key, e.stream_id, e.site_id, e.timestamp_utc_ms, e.event_type, e.severity, e.detection_result, e.metadata, e.is_verified, e.verified_by, e.verified_at, e.created_at, (SELECT count(*) FROM frames f WHERE f.event_context->>'event_id' = e.id::text) AS context_frames FROM events e WHERE e.id::uuid = %s """, (event_id,)) row = cur.fetchone() return self._row_to_event_dict(row) if row else None def get_event_frames(self, event_id: str, limit=100): """이벤트 컨텍스트 프레임 조회 (frames.event_context->>'event_id' 매칭)""" with self.get_cursor() as cur: cur.execute(""" SELECT f.private_key, f.timestamp_utc_ms, f.frame_sequence_id, f.width, f.height, (SELECT count(*) FROM detections d WHERE d.frame_id = f.id) AS n_detections FROM frames f WHERE f.event_context->>'event_id' = %s ORDER BY f.timestamp_utc_ms ASC LIMIT %s """, (event_id, limit)) return [{ "private_key": row[0], "timestamp_utc_ms": row[1], "frame_sequence_id": row[2], "width": row[3], "height": row[4], "n_detections": int(row[5]), } for row in cur.fetchall()] def verify_event(self, event_id: str, is_verified: bool, verified_by="api"): """이벤트 검증 상태 UPDATE — rowcount > 0 반환""" with self.get_cursor() as cur: cur.execute( """UPDATE events SET is_verified = %s, verified_by = %s, verified_at = now() WHERE id::text = %s""", (is_verified, verified_by, event_id) ) return cur.rowcount > 0 def prune_old_frames(self, retention_days: int = 30): """retention: N일 초과 비증거 프레임 삭제 (event_context IS NULL). - 이벤트 증거 프레임(event_context 태그)은 무기한 보존. - detections는 ON DELETE CASCADE로 자동 정리, events는 FK 없음(무영향). - 공유 INSERT 커넥션과 충돌 방지를 위해 전용 커넥션을 사용. Returns: 삭제된 frames 행 수 (실패 시 0 — non-fatal). """ if psycopg2 is None: return 0 cutoff_ms = int((time.time() - retention_days * 86400) * 1000) conn = None try: conn = psycopg2.connect(**self._conn_params) conn.autocommit = True cur = conn.cursor() cur.execute( "DELETE FROM frames WHERE event_context IS NULL AND timestamp_utc_ms < %s", (cutoff_ms,)) return int(cur.rowcount or 0) except Exception as e: print(f"[MetadataStorage] prune_old_frames failed (non-fatal): {e}") return 0 finally: if conn: try: conn.close() except Exception: pass def get_frames_by_stream(self, stream_id: str, limit: int = 30): """클라이언트용: 최근 프레임 조회""" with self.get_cursor() as cur: cur.execute(""" SELECT f.private_key, f.timestamp_utc_ms, f.inference_fps, d.track_id, d.class_label, d.confidence, d.bbox_normalized FROM frames f LEFT JOIN detections d ON d.frame_id = f.id WHERE f.stream_id = %s ORDER BY f.timestamp_utc_ms DESC LIMIT %s """, (stream_id, limit)) rows = cur.fetchall() results = [] for row in rows: # JSONB 컬럼은 psycopg2가 이미 dict/list로 역직렬화함 bn = row[6] if isinstance(bn, str): bn = json.loads(bn) results.append({ "private_key": row[0], "timestamp_utc_ms": row[1], "inference_fps": row[2], "track_id": row[3], "class_label": row[4], "confidence": row[5], "bbox_normalized": bn if isinstance(bn, (list, dict)) else None }) return results # --------------------------------------------------------------------------- # 전역 인스턴스 (싱글톤) # --------------------------------------------------------------------------- storage = MetadataStorage() redis_publisher = RedisMetadataPublisher()