commit 2cc58c18c4d476aae1938c45dd2a25c275167e76 Author: Gnill82 Date: Wed Aug 27 01:10:50 2025 +0900 분석 스크립트 초안 작성 diff --git a/ds_new_user_analy.py b/ds_new_user_analy.py new file mode 100644 index 0000000..79baeef --- /dev/null +++ b/ds_new_user_analy.py @@ -0,0 +1,803 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +던전 스토커즈 신규 유저 리텐션 분석 스크립트 +hack-detector의 고급 최적화 기법을 적용한 대용량 데이터 처리 + +주요 최적화 기법: +1. Composite Aggregation - 메모리 효율적 코호트 선정 +2. Streaming Pattern - 활동 시간 계산 최적화 +3. NDJSON + 백오프 재시도 - msearch 배치 처리 안정화 +4. Future Pattern - 비동기 병렬 처리 +5. Memory Optimization - 스트리밍 CSV 작성 + +작성자: Claude Code +기획서 기반: DS-new_users-analy.md +""" + +import os +import csv +import json +import time +import argparse +import threading +from datetime import datetime, timedelta, timezone +from collections import defaultdict +from typing import Dict, List, Optional, Tuple, Generator, Any +from concurrent.futures import ThreadPoolExecutor, as_completed, Future + +import pandas as pd +from tqdm import tqdm +from opensearchpy import OpenSearch + + +# ============================================================================== +# 1. 설정 및 상수 +# ============================================================================== + +# OpenSearch 연결 설정 (기획서에서 업데이트됨) +OPENSEARCH_CONFIG = { + "host": "ds-opensearch.oneunivrs.com", + "port": 9200, + "auth": { + "username": "admin", + "password": "DHp5#r#GYQ9d" + }, + "use_ssl": True, + "verify_certs": False, + "timeout": 60, + "max_retries": 3, + "headers": {"Connection": "close"} +} + +# 한국 표준시 +KST = timezone(timedelta(hours=9)) + +# 성능 최적화 설정 +DEFAULT_BATCH_SIZE = 1000 +DEFAULT_MAX_WORKERS = 6 +DEFAULT_COMPOSITE_SIZE = 1000 # composite aggregation 페이지 크기 +DEFAULT_TIMEOUT = 120 +SCROLL_TIMEOUT = "5m" +SESSION_GAP_MINUTES = 5 # 세션 분리 기준 (5분) +MAX_SESSION_HOURS = 3 # 최대 세션 길이 (3시간) + +# 출력 파일 설정 +OUTPUT_DIR = r"E:\DS_Git\DS_data_center\DS Log 분석" + +# 전역 타이머 제어 +stop_timer_event = threading.Event() + + +# ============================================================================== +# 2. OpenSearch 연결 및 유틸리티 +# ============================================================================== + +def create_opensearch_client() -> Optional[OpenSearch]: + """OpenSearch 클라이언트 생성 (hack-detector 방식)""" + print("[INFO] OpenSearch 클러스터에 연결 중...") + + try: + client = OpenSearch( + hosts=[{ + "host": OPENSEARCH_CONFIG['host'], + "port": OPENSEARCH_CONFIG['port'], + "scheme": "https" if OPENSEARCH_CONFIG['use_ssl'] else "http" + }], + http_auth=( + OPENSEARCH_CONFIG['auth']['username'], + OPENSEARCH_CONFIG['auth']['password'] + ), + use_ssl=OPENSEARCH_CONFIG['use_ssl'], + verify_certs=OPENSEARCH_CONFIG['verify_certs'], + ssl_show_warn=False, + timeout=OPENSEARCH_CONFIG['timeout'], + max_retries=OPENSEARCH_CONFIG['max_retries'], + retry_on_timeout=True, + headers=OPENSEARCH_CONFIG['headers'] + ) + + if not client.ping(): + raise ConnectionError("클러스터에 PING을 보낼 수 없습니다.") + + print("[SUCCESS] OpenSearch 연결 성공!") + return client + + except Exception as e: + print(f"[ERROR] OpenSearch 연결 실패: {e}") + return None + + +def exponential_backoff_retry(func, *args, **kwargs) -> Any: + """지수 백오프 재시도 패턴 (hack-detector 기법)""" + for delay in [1, 2, 4, 8, 16]: + try: + return func(*args, **kwargs) + except Exception as e: + if delay == 16: # 마지막 시도 + raise e + print(f"[WARNING] 재시도 중... {delay}초 대기 (오류: {str(e)[:100]})") + time.sleep(delay) + + +# ============================================================================== +# 3. 핵심 알고리즘 - Composite Aggregation을 활용한 코호트 선정 +# ============================================================================== + +def get_new_user_cohort_optimized( + client: OpenSearch, + start_time: str, + end_time: str, + page_size: int = DEFAULT_COMPOSITE_SIZE +) -> Dict[str, Dict]: + """ + Composite Aggregation을 활용한 메모리 효율적 신규 유저 코호트 선정 + hack-detector의 고급 기법 적용 + """ + print(f"\n[1단계] 신규 유저 코호트 선정 (Composite Aggregation)") + print(f" - 분석 기간: {start_time} ~ {end_time}") + + cohort = {} + after_key = None + + # Composite aggregation 쿼리 + base_query = { + "size": 0, + "query": { + "bool": { + "filter": [ + {"range": {"@timestamp": {"gte": start_time, "lt": end_time}}} + ] + } + }, + "aggs": { + "new_users": { + "composite": { + "size": page_size, + "sources": [ + {"auth_id": {"terms": {"field": "auth.id.keyword"}}}, + {"uid": {"terms": {"field": "uid.keyword"}}} + ] + }, + "aggs": { + "first_login": {"min": {"field": "@timestamp"}}, + "user_info": { + "top_hits": { + "size": 1, + "sort": [{"@timestamp": {"order": "asc"}}], + "_source": ["country", "body.device_mod"] + } + } + } + } + } + } + + total_users = 0 + + while True: + query = base_query.copy() + if after_key: + query["aggs"]["new_users"]["composite"]["after"] = after_key + + try: + response = exponential_backoff_retry( + client.search, + index="ds-logs-live-login_comp", + body=query, + request_timeout=DEFAULT_TIMEOUT, + track_total_hits=False # 성능 최적화 + ) + + buckets = response["aggregations"]["new_users"]["buckets"] + if not buckets: + break + + for bucket in buckets: + auth_id = bucket["key"]["auth_id"] + uid = bucket["key"]["uid"] + first_login_utc = bucket["first_login"]["value_as_string"] + + # 사용자 정보 추출 + user_hit = bucket["user_info"]["hits"]["hits"][0]["_source"] + + cohort[uid] = { + 'auth_id': auth_id, + 'first_login_utc': first_login_utc, + 'first_login_dt': datetime.fromisoformat(first_login_utc.replace('Z', '+00:00')), + 'country': user_hit.get('country', 'N/A'), + 'device': user_hit.get('body', {}).get('device_mod', 'N/A') + } + total_users += 1 + + # 다음 페이지 키 확인 + after_key = response["aggregations"]["new_users"].get("after_key") + if not after_key: + break + + except Exception as e: + print(f"[ERROR] 코호트 선정 중 오류: {e}") + break + + print(f" - [SUCCESS] 총 {total_users}명의 신규 유저 코호트 확정") + return cohort + + +# ============================================================================== +# 4. Active Hours 계산 - 스트리밍 방식 (Generator Pattern) +# ============================================================================== + +def calculate_active_hours_streaming( + client: OpenSearch, + uids: List[str], + cohort_data: Dict[str, Dict] +) -> Dict[str, int]: + """ + 스트리밍 방식으로 활동 시간 계산 + Generator 패턴으로 메모리 사용량 최소화 + """ + + def stream_user_events(uid: str) -> Generator[Dict, None, None]: + """개별 유저의 이벤트를 스트리밍으로 처리""" + user_info = cohort_data.get(uid) + if not user_info: + return + + first_login_dt = user_info['first_login_dt'] + d0_end_dt = first_login_dt + timedelta(hours=24) + + # 해당 유저의 D+0 이벤트만 스캔 + query = { + "query": { + "bool": { + "filter": [ + {"term": {"uid.keyword": uid}}, + {"range": {"@timestamp": { + "gte": user_info['first_login_utc'], + "lt": d0_end_dt.strftime('%Y-%m-%dT%H:%M:%SZ') + }}} + ] + } + } + } + + try: + from opensearchpy.helpers import scan + for doc in scan( + client, + query=query, + index="ds-logs-live-*", + scroll=SCROLL_TIMEOUT, + _source=["@timestamp", "type"] + ): + source = doc['_source'] + event_dt = datetime.fromisoformat(source['@timestamp'].replace('Z', '+00:00')) + + if first_login_dt <= event_dt < d0_end_dt: + yield { + "time": event_dt, + "type": source.get('type', '').lower() + } + except Exception: + # 오류 시 빈 generator 반환 + pass + + results = {} + + for uid in uids: + events = list(stream_user_events(uid)) + + if len(events) < 2: + results[uid] = 0 + continue + + # 세션 기반 활동 시간 계산 + events.sort(key=lambda x: x['time']) + total_active_seconds = 0 + + i = 0 + while i < len(events) - 1: + current_event = events[i] + next_event = events[i + 1] + + # 세션 간격 체크 (5분 이상 차이나면 다른 세션) + time_diff = next_event['time'] - current_event['time'] + + if time_diff <= timedelta(minutes=SESSION_GAP_MINUTES): + # 최대 세션 길이 제한 + session_duration = min( + time_diff.total_seconds(), + MAX_SESSION_HOURS * 3600 + ) + total_active_seconds += session_duration + + i += 1 + + results[uid] = int(total_active_seconds) + + return results + + +# ============================================================================== +# 5. NDJSON + 백오프 재시도를 활용한 msearch 배치 처리 +# ============================================================================== + +def build_msearch_queries(uids: List[str], cohort: Dict[str, Dict]) -> List[str]: + """ + msearch용 NDJSON 쿼리 생성 + hack-detector의 NDJSON 직접 생성 기법 적용 + """ + queries = [] + + # 분석할 지표 정의 (기획서 기반) + metrics_config = { + "retention_d1": { + "index": "ds-logs-live-login_comp", + "time_range": "d1", # 24-48시간 + "filters": [] + }, + "tutorial_entry": { + "index": "ds-logs-live-tutorial_entry", + "time_range": "d0", + "filters": [{"nested": {"path": "body", "query": {"term": {"body.action.keyword": "Start"}}}}] + }, + "tutorial_completed": { + "index": "ds-logs-live-log_tutorial", + "time_range": "d0", + "filters": [ + {"nested": {"path": "body", "query": {"bool": {"must": [ + {"term": {"body.action_type.keyword": "Complet"}}, + {"term": {"body.stage_type.keyword": "result"}} + ]}}}} + ] + }, + "dungeon_entry_count": { + "index": "ds-logs-live-survival_sta", + "time_range": "d0", + "filters": [] + }, + "dungeon_escape_count": { + "index": "ds-logs-live-survival_end", + "time_range": "d0", + "filters": [{"nested": {"path": "body", "query": {"term": {"body.result": 1}}}}] + }, + "monster_kill_count": { + "index": "ds-logs-live-survival_end", + "time_range": "d0", + "agg_field": "body.play_stats.monster_kill_cnt" + }, + "player_kill_count": { + "index": "ds-logs-live-player_kill", + "time_range": "d0", + "target_field": "body.instigator_uid" + }, + "matching_complete_count": { + "index": "ds-logs-live-matching_complete", + "time_range": "d0", + "filters": [] + }, + "friend_add_count": { + "index": "ds-logs-live-friend", + "time_range": "d0", + "filters": [{"nested": {"path": "body", "query": {"bool": {"must": [ + {"term": {"body.oper_type": 0}}, + {"term": {"body.friend_type": 0}} + ]}}}}] + } + } + + for uid in uids: + user_data = cohort[uid] + first_login_dt = user_data['first_login_dt'] + + # 시간 범위 정의 + d0_start = user_data['first_login_utc'] + d0_end = (first_login_dt + timedelta(hours=24)).strftime('%Y-%m-%dT%H:%M:%SZ') + d1_start = d0_end + d1_end = (first_login_dt + timedelta(hours=48)).strftime('%Y-%m-%dT%H:%M:%SZ') + + for metric_name, config in metrics_config.items(): + # 시간 범위 선택 + if config["time_range"] == "d0": + time_filter = {"range": {"@timestamp": {"gte": d0_start, "lt": d0_end}}} + else: # d1 + time_filter = {"range": {"@timestamp": {"gte": d1_start, "lt": d1_end}}} + + # 사용자 식별 필터 + user_filter = {"bool": {"should": [ + {"term": {"uid.keyword": uid}}, + {"term": {"auth.id.keyword": user_data['auth_id']}} + ], "minimum_should_match": 1}} + + # 쿼리 구성 + query_filters = [user_filter, time_filter] + + # 추가 필터 적용 + if "filters" in config: + query_filters.extend(config["filters"]) + + # 특별한 필드 처리 (player_kill의 경우) + if "target_field" in config and config["target_field"] == "body.instigator_uid": + query_filters.append({"nested": {"path": "body", "query": {"term": {"body.instigator_uid.keyword": uid}}}}) + + query_body = { + "size": 0 if "agg_field" not in config else 1000, + "query": {"bool": {"filter": query_filters}}, + "track_total_hits": False + } + + # Aggregation이 필요한 경우 + if "agg_field" in config: + query_body["aggs"] = { + "total": {"sum": {"field": config["agg_field"]}} + } + + # NDJSON 형태로 추가 + queries.append(json.dumps({"index": config["index"]}, ensure_ascii=False)) + queries.append(json.dumps(query_body, ensure_ascii=False)) + + return queries + + +def execute_msearch_with_backoff(client: OpenSearch, queries: List[str]) -> List[Dict]: + """ + NDJSON + 지수 백오프 재시도로 msearch 실행 + hack-detector의 안정화 기법 적용 + """ + # NDJSON 문자열 생성 + body_ndjson = "\n".join(queries) + "\n" + + # 지수 백오프로 재시도 + response = exponential_backoff_retry( + client.msearch, + body=body_ndjson, + request_timeout=60 + ) + + return response.get('responses', []) + + +# ============================================================================== +# 6. Future Pattern을 활용한 병렬 처리 최적화 +# ============================================================================== + +def process_user_batch_optimized( + client: OpenSearch, + batch_uids: List[str], + cohort: Dict[str, Dict] +) -> List[Dict]: + """ + 최적화된 배치 처리 함수 + Future Pattern + 스트리밍 + NDJSON 기법 결합 + """ + + # 1. 활동 시간 계산 (스트리밍 방식) + active_hours_map = calculate_active_hours_streaming(client, batch_uids, cohort) + + # 2. msearch 쿼리 생성 및 실행 (NDJSON + 백오프) + msearch_queries = build_msearch_queries(batch_uids, cohort) + msearch_responses = execute_msearch_with_backoff(client, msearch_queries) + + # 3. 결과 집계 + batch_results = [] + metrics_per_user = 9 # 정의된 지표 수 + + for idx, uid in enumerate(batch_uids): + try: + user_data = cohort[uid] + user_responses = msearch_responses[idx * metrics_per_user : (idx + 1) * metrics_per_user] + + # 기본 정보 + result = { + 'uid': uid, + 'auth_id': user_data['auth_id'], + 'nickname': 'N/A', # TODO: nickname 조회 로직 추가 + 'first_login_time': user_data['first_login_utc'], + 'retention_status': 'Retained_d0', # 기본값 + 'country': user_data['country'], + 'device': user_data['device'], + 'active_seconds': active_hours_map.get(uid, 0) + } + + # msearch 결과 파싱 + metrics = [ + 'retention_d1', 'tutorial_entry', 'tutorial_completed', + 'dungeon_entry_count', 'dungeon_escape_count', 'monster_kill_count', + 'player_kill_count', 'matching_complete_count', 'friend_add_count' + ] + + for i, metric in enumerate(metrics): + response = user_responses[i] + + if 'error' in response: + result[metric] = 0 + continue + + hits_total = response.get('hits', {}).get('total', {}).get('value', 0) + + if metric == 'retention_d1': + result['retention_status'] = 'Retained_d1' if hits_total > 0 else 'Retained_d0' + result[metric] = 1 if hits_total > 0 else 0 + elif metric == 'monster_kill_count': + agg_value = response.get('aggregations', {}).get('total', {}).get('value', 0) + result[metric] = int(agg_value) if agg_value else 0 + else: + result[metric] = hits_total + + batch_results.append(result) + + except Exception as e: + print(f" - ⚠️ UID '{uid}' 처리 중 오류: {e}") + + return batch_results + + +def process_cohort_parallel( + client: OpenSearch, + cohort: Dict[str, Dict], + batch_size: int, + max_workers: int +) -> List[Dict]: + """ + Future Pattern을 활용한 병렬 처리 + 동적 청크 크기 조정 + 실패 재처리 + """ + + # 동적 배치 크기 조정 (사용자 수 기반) + user_count = len(cohort) + if user_count < 1000: + adjusted_batch_size = min(batch_size, 100) + else: + adjusted_batch_size = batch_size + + print(f"\n[2단계] 병렬 배치 처리 시작 (배치크기: {adjusted_batch_size}, 워커: {max_workers})") + + # UID 리스트를 청크로 분할 + uid_list = list(cohort.keys()) + chunks = [uid_list[i:i + adjusted_batch_size] for i in range(0, len(uid_list), adjusted_batch_size)] + + all_results = [] + failed_chunks = [] + + with ThreadPoolExecutor(max_workers=max_workers) as executor: + # Future 객체 생성 + future_to_chunk = { + executor.submit(process_user_batch_optimized, client, chunk, cohort): chunk + for chunk in chunks + } + + # 진행률 표시 + with tqdm(total=len(chunks), desc=" - 배치 처리 진행률") as pbar: + for future in as_completed(future_to_chunk): + chunk = future_to_chunk[future] + try: + batch_results = future.result(timeout=300) # 5분 타임아웃 + if batch_results: + all_results.extend(batch_results) + else: + failed_chunks.append(chunk) + except Exception as e: + print(f" - ⚠️ 배치 처리 실패: {e}") + failed_chunks.append(chunk) + finally: + pbar.update(1) + + # 실패한 청크 재처리 (단일 스레드) + if failed_chunks: + print(f"\n - 실패한 {len(failed_chunks)}개 배치 재처리 중...") + for chunk in failed_chunks: + try: + batch_results = process_user_batch_optimized(client, chunk, cohort) + all_results.extend(batch_results) + except Exception as e: + print(f" - ❌ 재처리 실패: {e}") + + return all_results + + +# ============================================================================== +# 7. 스트리밍 CSV 작성 (메모리 최적화) +# ============================================================================== + +def write_results_streaming(results: List[Dict], output_path: str) -> None: + """ + 스트리밍 방식으로 CSV 작성 + 메모리에 모든 데이터를 올리지 않고 직접 파일에 쓰기 + """ + + if not results: + print(" - [ERROR] 저장할 결과 데이터가 없습니다.") + return + + # CSV 헤더 정의 (기획서 기반) + headers = [ + 'uid', 'auth_id', 'nickname', 'first_login_time', 'retention_status', + 'country', 'device', 'active_seconds', 'retention_d1', + 'tutorial_entry', 'tutorial_completed', 'dungeon_entry_count', + 'dungeon_escape_count', 'monster_kill_count', 'player_kill_count', + 'matching_complete_count', 'friend_add_count' + ] + + try: + with open(output_path, 'w', newline='', encoding='utf-8-sig') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=headers, extrasaction='ignore') + writer.writeheader() + + # 스트리밍으로 한 줄씩 작성 + for result in results: + writer.writerow(result) + + print(f" - [SUCCESS] 결과 파일 저장 완료: {output_path}") + print(f" - [INFO] 총 {len(results)}명의 데이터가 저장되었습니다.") + + except Exception as e: + print(f" - [ERROR] CSV 파일 저장 실패: {e}") + + +# ============================================================================== +# 8. 실시간 타이머 (유틸리티) +# ============================================================================== + +def live_timer(start_time: float, pbar: tqdm) -> None: + """실시간 경과 시간 표시""" + while not stop_timer_event.is_set(): + elapsed = str(timedelta(seconds=int(time.time() - start_time))) + pbar.set_postfix_str(f"경과 시간: {elapsed}") + time.sleep(1) + + +# ============================================================================== +# 9. 메인 함수 및 명령줄 인터페이스 +# ============================================================================== + +def parse_arguments() -> argparse.Namespace: + """명령줄 인자 파싱""" + parser = argparse.ArgumentParser( + description="던전 스토커즈 신규 유저 리텐션 분석 스크립트", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +예시: + python ds_new_user_analy.py --start-time "2025-08-22T12:00:00+09:00" --end-time "2025-08-25T12:00:00+09:00" + python ds_new_user_analy.py --start-time "2025-08-22T12:00:00+09:00" --end-time "2025-08-22T13:00:00+09:00" --sample-size 100 + """ + ) + + parser.add_argument( + '--start-time', + required=True, + help='분석 시작 시간 (KST, ISO 형식): "2025-08-22T12:00:00+09:00"' + ) + + parser.add_argument( + '--end-time', + required=True, + help='분석 종료 시간 (KST, ISO 형식): "2025-08-25T12:00:00+09:00"' + ) + + parser.add_argument( + '--output-dir', + default=OUTPUT_DIR, + help=f'결과 파일 저장 경로 (기본값: {OUTPUT_DIR})' + ) + + parser.add_argument( + '--batch-size', + type=int, + default=DEFAULT_BATCH_SIZE, + help=f'배치 처리 크기 (기본값: {DEFAULT_BATCH_SIZE})' + ) + + parser.add_argument( + '--max-workers', + type=int, + default=DEFAULT_MAX_WORKERS, + help=f'병렬 처리 스레드 수 (기본값: {DEFAULT_MAX_WORKERS})' + ) + + parser.add_argument( + '--sample-size', + type=int, + help='샘플 분석 크기 (None이면 전체 분석)' + ) + + return parser.parse_args() + + +def main(): + """메인 실행 함수""" + + # 시작 시간 기록 + overall_start_time = time.time() + + print("=" * 80) + print("던전 스토커즈 신규 유저 리텐션 분석 v2.0 (Claude Code)") + print("hack-detector 고급 최적화 기법 적용") + print("=" * 80) + + # 명령줄 인자 파싱 + args = parse_arguments() + + # 시간 변환 (KST -> UTC) + try: + start_kst = datetime.fromisoformat(args.start_time) + end_kst = datetime.fromisoformat(args.end_time) + + start_utc = start_kst.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') + end_utc = end_kst.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') + + print(f"[INFO] 분석 기간: {args.start_time} ~ {args.end_time}") + print(f"[INFO] UTC 변환: {start_utc} ~ {end_utc}") + + except Exception as e: + print(f"[ERROR] 시간 형식 오류: {e}") + return + + # OpenSearch 클라이언트 생성 + client = create_opensearch_client() + if not client: + return + + try: + # 1단계: 신규 유저 코호트 선정 (Composite Aggregation) + cohort = get_new_user_cohort_optimized(client, start_utc, end_utc) + + if not cohort: + print("\n[ERROR] 분석할 신규 유저가 없습니다.") + return + + # 샘플링 모드 + if args.sample_size and args.sample_size < len(cohort): + uid_list = list(cohort.keys()) + sampled_uids = uid_list[:args.sample_size] + cohort = {uid: cohort[uid] for uid in sampled_uids} + print(f"[WARNING] 샘플링 모드: {args.sample_size}명만 분석합니다.") + + # 2단계: 병렬 배치 처리 (Future Pattern) + results = process_cohort_parallel( + client, + cohort, + args.batch_size, + args.max_workers + ) + + # 3단계: 결과 저장 (스트리밍 CSV) + print(f"\n[3단계] 결과 저장") + + timestamp = datetime.now(KST).strftime('%Y%m%d_%H%M%S') + filename = f"ds_new_users_analy_{timestamp}.csv" + output_path = os.path.join(args.output_dir, filename) + + write_results_streaming(results, output_path) + + # 통계 요약 + if results: + retained_d1 = sum(1 for r in results if r.get('retention_status') == 'Retained_d1') + retention_rate = (retained_d1 / len(results)) * 100 + + print(f"\n[SUMMARY] 분석 요약:") + print(f" - 총 신규 유저: {len(results)}명") + print(f" - D+1 리텐션: {retained_d1}명 ({retention_rate:.1f}%)") + print(f" - 평균 활동 시간: {sum(r.get('active_seconds', 0) for r in results) / len(results) / 60:.1f}분") + + except KeyboardInterrupt: + print(f"\n[WARNING] 사용자에 의해 중단되었습니다.") + + except Exception as e: + print(f"\n[ERROR] 예상치 못한 오류: {e}") + import traceback + traceback.print_exc() + + finally: + # 타이머 정지 + stop_timer_event.set() + + # 총 소요 시간 + end_time = time.time() + total_time = str(timedelta(seconds=int(end_time - overall_start_time))) + print(f"\n[INFO] 총 소요 시간: {total_time}") + print("\n[SUCCESS] 분석 완료!") + + +if __name__ == "__main__": + main() \ No newline at end of file