""" 이벤트 자동 생성기 — 분석 결과 → prediction_events INSERT. 매 분석 사이클마다 vessel_analysis_results를 스캔하여 룰 기반으로 Event 객체를 생성합니다. dedup: 동일 mmsi + category + 윈도우 내 중복 방지. """ import hashlib import logging from datetime import datetime, timedelta, timezone from typing import Optional from psycopg2.extras import execute_values from config import qualified_table, settings from db.kcgdb import get_conn logger = logging.getLogger(__name__) EVENTS_TABLE = qualified_table('prediction_events') # 카테고리별 dedup 윈도우 (분) DEDUP_WINDOWS = { 'EEZ_INTRUSION': 30, 'DARK_VESSEL': 120, 'FLEET_CLUSTER': 360, 'ILLEGAL_TRANSSHIP': 60, 'MMSI_TAMPERING': 30, 'AIS_LOSS': 120, 'SPEED_ANOMALY': 60, 'ZONE_DEPARTURE': 120, 'GEAR_ILLEGAL': 360, 'AIS_RESUME': 60, } # 이벤트 생성 룰 RULES = [ { 'name': 'critical_risk', 'condition': lambda r: r.get('risk_score', 0) >= 90, 'level': 'CRITICAL', 'category': 'EEZ_INTRUSION', 'title_fn': lambda r: f"고위험 선박 탐지 (위험도 {r.get('risk_score', 0)})", }, { 'name': 'eez_violation', 'condition': lambda r: r.get('zone_code', '') in ('NLL', 'SPECIAL_FISHING_1', 'SPECIAL_FISHING_2') and r.get('risk_score', 0) >= 70, 'level': 'CRITICAL', 'category': 'EEZ_INTRUSION', 'title_fn': lambda r: f"EEZ 침범 탐지 ({r.get('zone_code', '')})", }, { 'name': 'dark_vessel_long', 'condition': lambda r: r.get('is_dark') and (r.get('gap_duration_min', 0) or 0) > 60, 'level': 'HIGH', 'category': 'DARK_VESSEL', 'title_fn': lambda r: f"다크베셀 장기 소실 ({r.get('gap_duration_min', 0)}분)", }, { 'name': 'spoofing', 'condition': lambda r: (r.get('spoofing_score', 0) or 0) > 0.7, 'level': 'HIGH', 'category': 'MMSI_TAMPERING', 'title_fn': lambda r: f"GPS/MMSI 조작 의심 (점수 {r.get('spoofing_score', 0):.2f})", }, { 'name': 'transship', 'condition': lambda r: r.get('transship_suspect'), 'level': 'HIGH', 'category': 'ILLEGAL_TRANSSHIP', 'title_fn': lambda r: f"환적 의심 (상대: {r.get('transship_pair_mmsi', '미상')})", }, { 'name': 'fleet_cluster', 'condition': lambda r: r.get('fleet_is_leader') and (r.get('fleet_cluster_id') is not None), 'level': 'MEDIUM', 'category': 'FLEET_CLUSTER', 'title_fn': lambda r: f"선단 밀집 감지 (클러스터 {r.get('fleet_cluster_id')})", }, { 'name': 'high_risk', 'condition': lambda r: r.get('risk_level') == 'HIGH' and r.get('risk_score', 0) >= 60, 'level': 'MEDIUM', 'category': 'ZONE_DEPARTURE', 'title_fn': lambda r: f"위험 행동 패턴 (위험도 {r.get('risk_score', 0)})", }, ] def _make_dedup_key(mmsi: str, category: str) -> str: return f"{mmsi}:{category}" def _make_event_uid(now: datetime, seq: int) -> str: date_str = now.strftime('%Y%m%d') return f"EVT-{date_str}-{seq:04d}" def _get_next_seq(conn, date_str: str) -> int: cur = conn.cursor() cur.execute( f"SELECT COUNT(*) FROM {EVENTS_TABLE} WHERE event_uid LIKE %s", (f'EVT-{date_str}-%',) ) return cur.fetchone()[0] + 1 def _check_dedup(conn, dedup_key: str, category: str, now: datetime) -> bool: """중복 이벤트 존재 여부 확인.""" window_min = DEDUP_WINDOWS.get(category, 60) cutoff = now - timedelta(minutes=window_min) cur = conn.cursor() cur.execute( f"SELECT 1 FROM {EVENTS_TABLE} WHERE dedup_key = %s AND occurred_at > %s LIMIT 1", (dedup_key, cutoff) ) return cur.fetchone() is not None def run_event_generator(analysis_results: list[dict]) -> dict: """ 분석 결과 리스트를 스캔하여 이벤트 생성. Args: analysis_results: vessel_analysis_results 행 딕셔너리 리스트 (mmsi, risk_score, zone_code, is_dark, gap_duration_min, spoofing_score, ...) Returns: { 'generated': int, 'skipped_dedup': int } """ now = datetime.now(timezone.utc) generated = 0 skipped_dedup = 0 events_to_insert = [] with get_conn() as conn: date_str = now.strftime('%Y%m%d') seq = _get_next_seq(conn, date_str) for result in analysis_results: mmsi = result.get('mmsi', '') if not mmsi: continue for rule in RULES: try: if not rule['condition'](result): continue except Exception: continue category = rule['category'] dedup_key = _make_dedup_key(mmsi, category) if _check_dedup(conn, dedup_key, category, now): skipped_dedup += 1 continue event_uid = _make_event_uid(now, seq) seq += 1 events_to_insert.append(( event_uid, now, # occurred_at rule['level'], category, rule['title_fn'](result), # title None, # detail mmsi, result.get('vessel_name'), result.get('zone_code'), # area_name (zone으로 대체) result.get('zone_code'), result.get('lat'), result.get('lon'), result.get('speed_kn'), 'VESSEL_ANALYSIS', # source_type result.get('id'), # source_ref_id result.get('confidence') or result.get('risk_score', 0) / 100.0, 'NEW', # status dedup_key, )) generated += 1 break # 한 분석결과당 최고 우선순위 룰 1개만 if events_to_insert: execute_values( conn.cursor(), f"""INSERT INTO {EVENTS_TABLE} (event_uid, occurred_at, level, category, title, detail, vessel_mmsi, vessel_name, area_name, zone_code, lat, lon, speed_kn, source_type, source_ref_id, ai_confidence, status, dedup_key) VALUES %s ON CONFLICT (event_uid) DO NOTHING""", events_to_insert, ) conn.commit() logger.info(f'event_generator: generated={generated}, skipped_dedup={skipped_dedup}') return {'generated': generated, 'skipped_dedup': skipped_dedup}