import asyncio import logging import os from dataclasses import dataclass from datetime import datetime from decimal import Decimal from typing import Any import aiomysql from persistence import DB_CONFIG logger = logging.getLogger(__name__) def _env(name: str, fallback: str = "") -> str: value = str(os.environ.get(name, "")).strip() return value or fallback APP_MASTER_DB_CONFIG = { "host": _env("APP_MASTER_DB_HOST", str(DB_CONFIG["host"])), "port": int(_env("APP_MASTER_DB_PORT", str(DB_CONFIG["port"]))), "user": _env("APP_MASTER_DB_USER", str(DB_CONFIG["user"])), "password": _env("APP_MASTER_DB_PASSWORD", str(DB_CONFIG["password"])), "db": _env("APP_MASTER_DB_NAME", "app_master"), "charset": "utf8mb4", "autocommit": True, } APP_MASTER_ADMIN_URL = _env( "APP_MASTER_ADMIN_URL", "https://app-master.officialsite.kr/admin", ).rstrip("/") OFFICIAL_MAIL_ADMIN_URL = _env( "OFFICIAL_MAIL_ADMIN_URL", "https://mail.officialsite.kr/kavenix?view=domain-assistance", ) OFFICIAL_MAIL_CHARGES_URL = _env( "OFFICIAL_MAIL_CHARGES_URL", "https://mail.officialsite.kr/kavenix?view=charges", ) @dataclass(frozen=True) class Metric: label: str table: str expression: str = "COUNT(*)" where: str = "" value_format: str = "number" @dataclass(frozen=True) class Activity: label: str table: str date_column: str where: str = "" @dataclass(frozen=True) class Purchase: table: str paid_statuses: tuple[str, ...] date_column: str = "purchased_at" APP_LABELS = { "babynote": "아가노트", "balancepick": "밸런스픽", "randomquestion": "랜덤질문", "tapcounter": "탭카운터", "stonemaster": "StoneMaster", "qnote": "QNote", "calmoa": "캘모아", "chromakey": "크로마키 스크린", "onetakeduo": "가세로캠", "wallet_keeper": "지갑지켜", "tiply": "팁리", "ramen-timer": "꼬들 타이머", "tag-shuttle": "태그 셔틀", "kaomoji": "카오모지", "focus-interval": "포커스 인터벌", "water-reminder": "물마시기 알리미", "pill-reminder": "약 알리미", "standup-reminder": "기상 리마인더", "sleep-sound": "슬립 사운드", "sound-meter": "소음 측정기", "plant-reminder": "물주기 알리미", "named": "네임드", "evidence-note": "증거노트", "eye-break": "눈 휴식 알리미", "neck-relief": "넥 릴리프", "ghost-running": "고스트러닝", "merge-snack-factory": "Merge Snack Factory", } USER_APP_NAMES = { "babynote": "babynote", "balancepick": "balancepick", "stonemaster": "stonemaster", "calmoa": "calmoa", "onetakeduo": "onetakeduo", "wallet_keeper": "wallet_keeper", "plant-reminder": "plant_reminder", "evidence-note": "evidence-note", "ghost-running": "ghost-running", } INQUIRY_APPS = {"babynote", "wallet_keeper"} AD_SETTING_TABLES = { "tiply": "tiply_ad_settings", "kaomoji": "kaomoji_ad_settings", "focus-interval": "focus_interval_ad_settings", "water-reminder": "water_reminder_ad_settings", "pill-reminder": "pill_reminder_ad_settings", "standup-reminder": "standup_reminder_ad_settings", "sleep-sound": "sleep_sound_ad_settings", "sound-meter": "soundmeter_ad_settings", "eye-break": "eye_break_ad_settings", "neck-relief": "neck_relief_ad_settings", "merge-snack-factory": "merge_snack_factory_ad_settings", } CORE_METRICS = { "babynote": [ Metric("전체 가족", "babynote_families"), Metric("전체 일기", "babynote_diaries"), Metric("전체 댓글", "babynote_comments"), ], "balancepick": [ Metric("전체 게임", "balancepick_games"), Metric("공개 게임", "balancepick_games", where="is_public = 1 AND status = 'approved'"), Metric("전체 플레이", "balancepick_game_plays"), Metric( "활성 프리미엄", "balancepick_purchases", where=( "status = 'completed' AND verified_at IS NOT NULL " "AND purchase_state = 0 AND is_test_purchase = 0" ), ), ], "randomquestion": [ Metric("활성 카테고리", "randomquestion_categories", where="is_active = 1"), Metric("활성 질문 세트", "randomquestion_game_sets", where="is_active = 1"), Metric("전체 질문 카드", "randomquestion_game_items"), ], "tapcounter": [ Metric("광고 운영 모드", "tapcounter_ad_settings", "MAX(ad_mode)", value_format="text"), Metric( "일일 무료 사용", "app_settings", "MAX(CAST(setting_value AS UNSIGNED))", "app_name = 'tapcounter' AND setting_key = 'daily_free_limit'", ), Metric( "광고 시청 보상", "app_settings", "MAX(CAST(setting_value AS UNSIGNED))", "app_name = 'tapcounter' AND setting_key = 'ad_reward_count'", ), Metric("활성 광고 제거", "tapcounter_ad_removal", where="is_active = 1"), ], "stonemaster": [ Metric("클라우드 저장", "stonemaster_cloud_saves"), Metric("광고 이벤트", "stonemaster_ad_events"), Metric("리워드 완료", "stonemaster_ad_events", where="event_type = 'completed'"), ], "qnote": [ Metric("전체 QR", "qr_codes"), Metric("즐겨찾기 QR", "qr_codes", where="is_favorite = 1"), Metric("생성 기기", "qr_codes", "COUNT(DISTINCT device_serial)"), ], "calmoa": [ Metric("전체 약속", "calmoa_appointments"), Metric("진행 중 약속", "calmoa_appointments", where="is_closed = 0"), Metric("전체 응답", "calmoa_responses"), Metric("발송 알림", "calmoa_notifications"), ], "chromakey": [ Metric("전체 구매", "chromakey_purchases"), Metric("활성 구매", "chromakey_purchases", where="is_active = 1 AND (expiry_time IS NULL OR expiry_time > NOW())"), Metric("누적 세션", "chromakey_stats", "COALESCE(SUM(total_sessions), 0)"), Metric("광고 노출", "chromakey_stats", "COALESCE(SUM(ad_views), 0)"), ], "onetakeduo": [ Metric("저장 완료", "onetakeduo_save_sessions", where="status = 'saved'"), Metric("워터마크 저장", "onetakeduo_save_sessions", where="status = 'saved' AND watermark_applied = 1"), Metric("광고 보상", "onetakeduo_reward_claims", where="status = 'completed'"), Metric("Pro 사용자", "onetakeduo_user_state", "COUNT(*)", "is_pro = 1"), ], "wallet_keeper": [ Metric("클라우드 저장 사용자", "wallet_keeper_cloud_saves"), Metric("문자 자산 수집", "wallet_keeper_sms_reports"), Metric("관리자 알림 발송", "wallet_keeper_customer_reminders"), ], "ramen-timer": [ Metric("전체 라면", "ramen_timer_ramens"), Metric("활성 라면", "ramen_timer_ramens", where="is_active = 1"), Metric("커스텀 라면", "ramen_timer_ramens", where="is_custom = 1"), ], "tag-shuttle": [ Metric("활성 카테고리", "tag_shuttle_categories", where="is_active = 1"), Metric("활성 태그", "tag_shuttle_tags", where="is_active = 1"), Metric("사용자 저장 태그", "tag_shuttle_my_tags"), Metric("누적 복사", "tag_shuttle_tags", "COALESCE(SUM(copy_count), 0)"), ], "plant-reminder": [ Metric("클라우드 식물 데이터", "plant_reminder_user_data"), Metric("활성 식물 프리셋", "plant_reminder_presets", where="is_active = 1"), Metric("푸시 발송 기록", "plant_reminder_push_history"), ], "named": [ Metric("생성 명함", "business_cards"), Metric("전자서명", "electronic_signatures"), Metric("명함 생성 기기", "business_cards", "COUNT(DISTINCT device_serial)"), Metric("활성 템플릿", "named_card_templates", where="is_active = 1"), ], "evidence-note": [ Metric("증거 기록", "evidence_note_records"), Metric("PDF 내보내기", "evidence_note_pdf_exports"), Metric("첨부 업로드", "evidence_note_attachment_uploads"), Metric("업로드 완료", "evidence_note_attachment_uploads", where="status = 'completed'"), ], "ghost-running": [ Metric("전체 러닝", "ghost_running_runs"), Metric("완료 러닝", "ghost_running_runs", where="status = 'completed'"), Metric("생성 고스트", "ghost_running_ghosts"), Metric("누적 거리", "ghost_running_runs", "COALESCE(SUM(distance_m), 0)", value_format="distance"), ], } for _app_key, _table in AD_SETTING_TABLES.items(): CORE_METRICS.setdefault( _app_key, [Metric("광고 운영 모드", _table, "MAX(ad_mode)", value_format="text")], ) CORE_METRICS["merge-snack-factory"].append( Metric( "활성 광고 유형", "merge_snack_factory_ad_settings", "MAX(banner_enabled + interstitial_enabled + rewarded_enabled)", ) ) ACTIVITIES = { "babynote": [ Activity("일기 작성", "babynote_diaries", "created_at"), Activity("댓글 작성", "babynote_comments", "created_at"), ], "balancepick": [ Activity("게임 플레이", "balancepick_game_plays", "created_at"), Activity("게임 생성", "balancepick_games", "created_at"), Activity( "프리미엄 구매", "balancepick_purchases", "purchased_at", "status = 'completed' AND verified_at IS NOT NULL AND purchase_state = 0 AND is_test_purchase = 0", ), ], "randomquestion": [ Activity("질문 카드 등록", "randomquestion_game_items", "created_at"), Activity("질문 세트 등록", "randomquestion_game_sets", "created_at"), ], "tapcounter": [ Activity("결제 완료", "tapcounter_purchases", "purchased_at", "status = 'completed'"), ], "stonemaster": [ Activity("광고 이벤트", "stonemaster_ad_events", "created_at"), Activity("클라우드 저장", "stonemaster_cloud_saves", "updated_at"), ], "qnote": [Activity("QR 생성", "qr_codes", "created_at")], "calmoa": [ Activity("약속 생성", "calmoa_appointments", "created_at"), Activity("응답 제출", "calmoa_responses", "submitted_at"), ], "chromakey": [Activity("구매 등록", "chromakey_purchases", "purchase_time")], "onetakeduo": [ Activity("저장 완료", "onetakeduo_save_sessions", "created_at", "status = 'saved'"), Activity("광고 보상", "onetakeduo_reward_claims", "created_at", "status = 'completed'"), ], "wallet_keeper": [ Activity("클라우드 저장", "wallet_keeper_cloud_saves", "updated_at"), Activity("문자 자산 수집", "wallet_keeper_sms_reports", "created_at"), ], "ramen-timer": [Activity("라면 등록", "ramen_timer_ramens", "created_at")], "tag-shuttle": [ Activity("태그 저장", "tag_shuttle_my_tags", "created_at"), Activity("태그 등록", "tag_shuttle_tags", "created_at"), ], "plant-reminder": [ Activity("식물 데이터 저장", "plant_reminder_user_data", "updated_at"), Activity("푸시 발송", "plant_reminder_push_history", "created_at"), ], "named": [ Activity("명함 생성", "business_cards", "created_at"), Activity("전자서명 생성", "electronic_signatures", "created_at"), ], "evidence-note": [ Activity("증거 기록", "evidence_note_records", "created_at"), Activity("PDF 내보내기", "evidence_note_pdf_exports", "created_at"), ], "ghost-running": [ Activity("러닝 기록", "ghost_running_runs", "created_at"), Activity("실시간 세션", "ghost_running_live_sessions", "created_at"), ], } PURCHASES = { "babynote": Purchase("babynote_purchases", ("completed",)), "tapcounter": Purchase("tapcounter_purchases", ("completed",)), "calmoa": Purchase("calmoa_purchases", ("completed", "restored")), "onetakeduo": Purchase("onetakeduo_purchases", ("completed", "restored")), } EVENT_SOURCES = ( { "table": "app_inquiries", "source": "app_inquiries", "event_type": "inquiry", "sql": """ SELECT id AS event_id, app_name AS app_key, subject AS title, content AS detail, COALESCE(NULLIF(user_name, ''), NULLIF(user_email, ''), NULLIF(user_id, ''), '익명') AS actor, NULL AS amount, NULL AS currency, NULL AS platform, status, created_at AS event_time FROM app_inquiries WHERE created_at > %s AND created_at <= %s ORDER BY created_at ASC """, }, { "table": "babynote_purchases", "source": "babynote_purchases", "event_type": "purchase", "sql": """ SELECT id AS event_id, 'babynote' AS app_key, product_id AS title, product_type AS detail, user_id AS actor, price AS amount, currency, platform, status, COALESCE(purchased_at, created_at) AS event_time FROM babynote_purchases WHERE status = 'completed' AND COALESCE(purchased_at, created_at) > %s AND COALESCE(purchased_at, created_at) <= %s ORDER BY event_time ASC """, }, { "table": "tapcounter_purchases", "source": "tapcounter_purchases", "event_type": "purchase", "sql": """ SELECT id AS event_id, 'tapcounter' AS app_key, product_id AS title, product_type AS detail, user_id AS actor, price AS amount, currency, platform, status, COALESCE(purchased_at, created_at) AS event_time FROM tapcounter_purchases WHERE status = 'completed' AND COALESCE(purchased_at, created_at) > %s AND COALESCE(purchased_at, created_at) <= %s ORDER BY event_time ASC """, }, { "table": "calmoa_purchases", "source": "calmoa_purchases", "event_type": "purchase", "sql": """ SELECT id AS event_id, 'calmoa' AS app_key, product_id AS title, NULL AS detail, COALESCE(user_id, device_serial) AS actor, price AS amount, currency, platform, status, COALESCE(purchased_at, created_at) AS event_time FROM calmoa_purchases WHERE status = 'completed' AND COALESCE(purchased_at, created_at) > %s AND COALESCE(purchased_at, created_at) <= %s ORDER BY event_time ASC """, }, { "table": "onetakeduo_purchases", "source": "onetakeduo_purchases", "event_type": "purchase", "sql": """ SELECT id AS event_id, 'onetakeduo' AS app_key, product_id AS title, NULL AS detail, COALESCE(user_id, guest_serial) AS actor, price AS amount, currency, platform, status, COALESCE(verified_at, purchased_at, created_at) AS event_time FROM onetakeduo_purchases WHERE status = 'completed' AND COALESCE(verified_at, purchased_at, created_at) > %s AND COALESCE(verified_at, purchased_at, created_at) <= %s ORDER BY event_time ASC """, }, { "table": "chromakey_purchases", "source": "chromakey_purchases", "event_type": "purchase", "sql": """ SELECT id AS event_id, 'chromakey' AS app_key, product_id AS title, '광고 제거 활성화' AS detail, device_id AS actor, NULL AS amount, NULL AS currency, platform, 'completed' AS status, COALESCE(purchase_time, created_at) AS event_time FROM chromakey_purchases WHERE is_active = 1 AND COALESCE(purchase_time, created_at) > %s AND COALESCE(purchase_time, created_at) <= %s ORDER BY event_time ASC """, }, { "table": "balancepick_purchases", "source": "balancepick_purchases", "event_type": "purchase", "sql": """ SELECT id AS event_id, 'balancepick' AS app_key, product_id AS title, 'lifetime 프리미엄' AS detail, device_serial AS actor, NULL AS amount, NULL AS currency, platform, status, purchased_at AS event_time FROM balancepick_purchases WHERE status = 'completed' AND verified_at IS NOT NULL AND purchase_state = 0 AND is_test_purchase = 0 AND purchased_at > %s AND purchased_at <= %s ORDER BY event_time ASC """, }, { "table": "babynote_kidsnote_payment_orders", "source": "babynote_kidsnote_payment_orders", "event_type": "purchase", "sql": """ SELECT id AS event_id, 'babynote' AS app_key, order_name AS title, '키즈노트 포토북' AS detail, user_id AS actor, total_price AS amount, 'KRW' AS currency, 'toss' AS platform, status, COALESCE(approved_at, created_at) AS event_time FROM babynote_kidsnote_payment_orders WHERE status = 'paid' AND COALESCE(approved_at, created_at) > %s AND COALESCE(approved_at, created_at) <= %s ORDER BY event_time ASC """, }, ) def list_apps() -> list[dict[str, str]]: return [{"key": key, "label": label} for key, label in APP_LABELS.items()] def resolve_app_key(value: str | None) -> str | None: normalized = str(value or "").strip().lower() if not normalized: return None if normalized in APP_LABELS: return normalized compact = normalized.replace(" ", "").replace("_", "").replace("-", "") for key, label in APP_LABELS.items(): if compact in { key.replace("_", "").replace("-", "").lower(), label.replace(" ", "").lower(), }: return key return None def app_stat_categories(app_key: str) -> list[tuple[str, str]]: categories = [("overview", "요약")] if app_key in USER_APP_NAMES: categories.append(("users", "사용자")) if CORE_METRICS.get(app_key) or ACTIVITIES.get(app_key): categories.append(("activity", "핵심 활동")) if app_key in PURCHASES or app_key in {"balancepick", "chromakey"}: categories.append(("purchases", "결제")) if app_key in INQUIRY_APPS: categories.append(("inquiries", "문의")) return categories def _number(value: Any) -> int | float: if isinstance(value, Decimal): value = float(value) if isinstance(value, bool): return int(value) if isinstance(value, (int, float)): return value try: parsed = float(value or 0) return int(parsed) if parsed.is_integer() else parsed except (TypeError, ValueError): return 0 def _format_number(value: Any) -> str: number = _number(value) if isinstance(number, float) and not number.is_integer(): return f"{number:,.2f}".rstrip("0").rstrip(".") return f"{int(number):,}" def _format_metric_value(value: Any, value_format: str = "number") -> str: if value_format == "text": text = str(value or "-") if text.lower() in {"release", "production"}: return "릴리즈" if text.lower() == "test": return "테스트" return text if value_format == "distance": return f"{_number(value) / 1000:,.1f}km" return _format_number(value) def _format_currency(amount: Any, currency: Any) -> str: code = str(currency or "KRW").upper() value = _number(amount) if code == "KRW": return f"₩{value:,.0f}" return f"{code} {value:,.2f}" def _short(value: Any, limit: int = 260) -> str: text = " ".join(str(value or "").split()) if len(text) <= limit: return text return f"{text[:limit - 1]}…" def format_app_event_message(event: dict[str, Any]) -> str: app_key = str(event.get("app_key") or "") app_label = APP_LABELS.get(app_key, app_key or "알 수 없는 앱") if event.get("event_type") == "domain_assistance": return "\n".join( [ "[도메인 연결 설정 대행 신청]", "서비스: 오피셜메일", f"신청 번호: #{event.get('event_id')}", f"도메인: {_short(event.get('title'), 253) or '-'}", f"신청자: {_short(event.get('actor'), 191) or '-'}", f"도메인 구매처: {_short(event.get('detail'), 80) or '-'}", f"완료 알림: {_short(event.get('notification_email'), 191) or '-'}", f"완료 후 결제 예정: {_format_currency(event.get('amount'), event.get('currency'))}", f"관리: {OFFICIAL_MAIL_ADMIN_URL}", ] ) if event.get("event_type") == "polar_payment": succeeded = event.get("status") == "succeeded" trigger_labels = { "purchase": "최초 결제", "subscription_create": "구독 시작", "subscription_cycle": "정기 갱신", "subscription_update": "구독 변경", "retry_dunning": "자동 재시도", "retry_customer": "고객 재시도", "retry_payment_method_update": "결제수단 변경 후 재시도", "retry_admin": "관리자 재시도", } lines = [ "[해외 결제 완료]" if succeeded else "[해외 결제 실패]", "서비스: 오피셜메일", f"상품: {_short(event.get('title'), 120) or '-'}", f"금액: {_format_currency(event.get('amount'), event.get('currency'))}", f"사용자: {_short(event.get('actor'), 191) or '-'}", f"결제 유형: {trigger_labels.get(event.get('payment_trigger'), '해외 결제')}", ] card_label = " ".join( value for value in [ str(event.get("card_brand") or "").upper(), f"•••• {event.get('card_last_four')}" if event.get("card_last_four") else "", ] if value ) if card_label: lines.append(f"카드: {card_label}") if not succeeded: decline = _short(event.get("detail"), 220) reason = _short(event.get("decline_reason"), 80) lines.append(f"실패 사유: {decline or reason or 'Polar에서 사유 확인 필요'}") if reason and reason not in decline: lines.append(f"오류 코드: {reason}") lines.extend( [ f"결제 ID: {_short(event.get('polar_payment_id'), 80) or '-'}", f"관리: {OFFICIAL_MAIL_CHARGES_URL}", ] ) return "\n".join(lines) if event.get("event_type") == "payment_method_registered": overseas = event.get("region") == "overseas" lines = [ ( "[해외 결제수단 등록 및 정기결제 활성화]" if overseas else "[국내 결제수단 등록 완료]" ), "서비스: 오피셜메일", f"사용자: {_short(event.get('actor'), 191) or '-'}", f"제공자: {'Polar' if overseas else 'Toss Pay'}", ] card_label = " ".join( value for value in [ str(event.get("card_brand") or "").upper(), f"•••• {event.get('card_last_four')}" if event.get("card_last_four") else "", ] if value ) if card_label: lines.append(f"결제수단: {card_label}") elif event.get("account_bank_name"): lines.append( f"결제수단: {_short(event.get('account_bank_name'), 80)}" ) elif event.get("payment_method"): lines.append( f"결제수단: {_short(event.get('payment_method'), 32)}" ) lines.append(f"관리: {OFFICIAL_MAIL_CHARGES_URL}") return "\n".join(lines) if event.get("event_type") == "toss_payment": succeeded = event.get("status") == "success" initial_activation = ( succeeded and bool(event.get("is_initial_subscription_charge")) ) if initial_activation: heading = "[국내 정기결제 활성화]" elif event.get("payment_trigger") == "cycle": heading = ( "[국내 정기결제 완료]" if succeeded else "[국내 정기결제 실패]" ) else: heading = "[국내 결제 완료]" if succeeded else "[국내 결제 실패]" lines = [ heading, "서비스: 오피셜메일", f"상품: {_short(event.get('title'), 120) or '-'}", f"금액: {_format_currency(event.get('amount'), event.get('currency'))}", f"사용자: {_short(event.get('actor'), 191) or '-'}", ] card_label = " ".join( value for value in [ str(event.get("card_brand") or "").upper(), f"•••• {event.get('card_last_four')}" if event.get("card_last_four") else "", ] if value ) if card_label: lines.append(f"결제수단: {card_label}") if not succeeded: decline = _short(event.get("detail"), 220) reason = _short(event.get("decline_reason"), 80) lines.append( f"실패 사유: {decline or reason or 'Toss Pay에서 사유 확인 필요'}" ) if reason and reason not in decline: lines.append(f"오류 코드: {reason}") lines.extend( [ f"주문번호: {_short(event.get('payment_id'), 80) or '-'}", f"관리: {OFFICIAL_MAIL_CHARGES_URL}", ] ) return "\n".join(lines) if event.get("event_type") == "inquiry": lines = [ "[새 문의]", f"앱: {app_label}", f"제목: {_short(event.get('title'), 120) or '-'}", f"작성자: {_short(event.get('actor'), 80) or '익명'}", f"내용: {_short(event.get('detail')) or '-'}", f"관리: {APP_MASTER_ADMIN_URL}/app/{app_key}/inquiries", ] return "\n".join(lines) lines = [ "[결제 완료]", f"앱: {app_label}", f"상품: {_short(event.get('title'), 120) or '-'}", ] if event.get("amount") is not None: lines.append(f"금액: {_format_currency(event.get('amount'), event.get('currency'))}") if event.get("platform"): lines.append(f"플랫폼: {str(event['platform']).upper()}") if event.get("actor"): lines.append(f"사용자: {_short(event.get('actor'), 80)}") if event.get("detail"): lines.append(f"상세: {_short(event.get('detail'), 120)}") lines.append(f"관리: {APP_MASTER_ADMIN_URL}/app/{app_key}/dashboard") return "\n".join(lines) def format_app_stats_message(app_key: str, category: str, data: dict[str, Any]) -> str: app_label = APP_LABELS.get(app_key, app_key) title_by_category = { "overview": "요약", "users": "사용자", "activity": "핵심 활동", "purchases": "결제", "inquiries": "문의", } lines = [f"[{app_label} · {title_by_category.get(category, '통계')}]"] if category == "users": if not data.get("tracked"): lines.append("로그인 사용자 추적을 사용하지 않는 앱입니다.") else: lines.extend( [ f"전체 사용자: {_format_number(data.get('total'))}명", f"오늘 가입: {_format_number(data.get('new_today'))}명", f"최근 7일 가입: {_format_number(data.get('new_7d'))}명", f"오늘 접속: {_format_number(data.get('active_today'))}명", f"최근 7일 접속: {_format_number(data.get('active_7d'))}명", ] ) elif category == "activity": metrics = data.get("metrics") or [] trends = data.get("trends") or [] if metrics: lines.append("\n현재 누적") lines.extend( f"• {item['label']}: {_format_metric_value(item.get('value'), item.get('format', 'number'))}" for item in metrics ) if trends: lines.append("\n최근 7일") lines.extend( f"• {item['label']}: {_format_number(item.get('value'))}건" for item in trends ) if not metrics and not trends: lines.append("연결된 서버 활동 데이터가 없습니다.") elif category == "purchases": if not data.get("supported"): lines.append("서버에 기록되는 결제 데이터가 없는 앱입니다.") else: lines.extend( [ f"전체 결제 기록: {_format_number(data.get('total'))}건", f"결제 완료: {_format_number(data.get('paid'))}건", f"오늘 결제: {_format_number(data.get('today'))}건", f"최근 7일 결제: {_format_number(data.get('last_7d'))}건", ] ) if data.get("has_revenue"): currencies = data.get("revenue_by_currency") or [] if currencies: lines.append( "누적 결제 금액: " + ", ".join( _format_currency(item.get("amount"), item.get("currency")) for item in currencies ) ) if data.get("extra"): lines.extend(f"• {item['label']}: {_format_number(item['value'])}건" for item in data["extra"]) elif category == "inquiries": if not data.get("supported"): lines.append("문의 기능을 제공하지 않는 앱입니다.") else: lines.extend( [ f"전체 문의: {_format_number(data.get('total'))}건", f"미처리 문의: {_format_number(data.get('open'))}건", f"오늘 문의: {_format_number(data.get('today'))}건", f"최근 7일 문의: {_format_number(data.get('last_7d'))}건", ] ) else: users = data.get("users") or {} if users.get("tracked"): lines.extend( [ f"전체 사용자: {_format_number(users.get('total'))}명", f"최근 7일 가입/접속: {_format_number(users.get('new_7d'))}명 / {_format_number(users.get('active_7d'))}명", ] ) else: lines.append("사용자: 로그인 사용자 추적 안 함") for item in (data.get("metrics") or [])[:4]: lines.append( f"{item['label']}: {_format_metric_value(item.get('value'), item.get('format', 'number'))}" ) purchases = data.get("purchases") or {} if purchases.get("supported"): lines.append(f"결제 완료: {_format_number(purchases.get('paid'))}건") inquiries = data.get("inquiries") or {} if inquiries.get("supported"): lines.append(f"미처리 문의: {_format_number(inquiries.get('open'))}건") admin_page = "inquiries" if category == "inquiries" else "dashboard" lines.append(f"\n관리: {APP_MASTER_ADMIN_URL}/app/{app_key}/{admin_page}") return "\n".join(lines) class AppMasterService: def __init__(self) -> None: self._pool: aiomysql.Pool | None = None self._available_tables: set[str] = set() async def initialize(self) -> None: if self._pool: return try: self._pool = await aiomysql.create_pool( minsize=1, maxsize=4, connect_timeout=5, **APP_MASTER_DB_CONFIG, ) await self.refresh_schema() except Exception: await self.close() raise logger.info( "App Master DB connected: %s (%d tables)", APP_MASTER_DB_CONFIG["db"], len(self._available_tables), ) async def close(self) -> None: if self._pool: self._pool.close() await self._pool.wait_closed() self._pool = None self._available_tables.clear() async def refresh_schema(self) -> None: rows = await self._fetch_all( "SELECT TABLE_NAME AS table_name FROM information_schema.tables WHERE table_schema = %s", (APP_MASTER_DB_CONFIG["db"],), ) self._available_tables = { str(row.get("table_name") or row.get("TABLE_NAME")) for row in rows if row.get("table_name") or row.get("TABLE_NAME") } async def database_now(self) -> datetime: row = await self._fetch_one("SELECT NOW() AS value") return row["value"] def has_table(self, table: str) -> bool: return table in self._available_tables async def _fetch_all(self, sql: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]: if not self._pool: raise RuntimeError("App Master DB is not initialized") async with self._pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cur: await cur.execute(sql, params) return list(await cur.fetchall()) async def _fetch_one(self, sql: str, params: tuple[Any, ...] = ()) -> dict[str, Any]: rows = await self._fetch_all(sql, params) return rows[0] if rows else {} async def user_stats(self, app_key: str) -> dict[str, Any]: app_name = USER_APP_NAMES.get(app_key) if not app_name or not self.has_table("users"): return {"tracked": False} row = await self._fetch_one( """ SELECT COUNT(*) AS total, SUM(CASE WHEN created_at >= CURDATE() THEN 1 ELSE 0 END) AS new_today, SUM(CASE WHEN created_at >= CURDATE() - INTERVAL 6 DAY THEN 1 ELSE 0 END) AS new_7d, SUM(CASE WHEN last_login_at >= CURDATE() THEN 1 ELSE 0 END) AS active_today, SUM(CASE WHEN last_login_at >= CURDATE() - INTERVAL 6 DAY THEN 1 ELSE 0 END) AS active_7d FROM users WHERE app_name = %s AND (is_del = 0 OR is_del IS NULL) """, (app_name,), ) return { "tracked": True, "total": _number(row.get("total")), "new_today": _number(row.get("new_today")), "new_7d": _number(row.get("new_7d")), "active_today": _number(row.get("active_today")), "active_7d": _number(row.get("active_7d")), } async def core_stats(self, app_key: str) -> dict[str, Any]: metrics: list[dict[str, Any]] = [] for item in CORE_METRICS.get(app_key, []): if not self.has_table(item.table): continue where_sql = f" WHERE {item.where}" if item.where else "" row = await self._fetch_one( f"SELECT {item.expression} AS value FROM `{item.table}`{where_sql}" ) metrics.append( { "label": item.label, "value": row.get("value"), "format": item.value_format, } ) trends: list[dict[str, Any]] = [] for item in ACTIVITIES.get(app_key, []): if not self.has_table(item.table): continue conditions = [f"`{item.date_column}` >= CURDATE() - INTERVAL 6 DAY"] if item.where: conditions.append(item.where) row = await self._fetch_one( f"SELECT COUNT(*) AS value FROM `{item.table}` WHERE {' AND '.join(conditions)}" ) trends.append({"label": item.label, "value": _number(row.get("value"))}) return {"metrics": metrics, "trends": trends} async def inquiry_stats(self, app_key: str) -> dict[str, Any]: if app_key not in INQUIRY_APPS or not self.has_table("app_inquiries"): return {"supported": False} row = await self._fetch_one( """ SELECT COUNT(*) AS total, SUM(CASE WHEN status IN ('pending', 'in_progress') THEN 1 ELSE 0 END) AS open_count, SUM(CASE WHEN created_at >= CURDATE() THEN 1 ELSE 0 END) AS today, SUM(CASE WHEN created_at >= CURDATE() - INTERVAL 6 DAY THEN 1 ELSE 0 END) AS last_7d FROM app_inquiries WHERE app_name = %s """, (app_key,), ) return { "supported": True, "total": _number(row.get("total")), "open": _number(row.get("open_count")), "today": _number(row.get("today")), "last_7d": _number(row.get("last_7d")), } async def purchase_stats(self, app_key: str) -> dict[str, Any]: if app_key == "balancepick": if not self.has_table("balancepick_purchases"): return {"supported": False} row = await self._fetch_one( """ SELECT COUNT(*) AS total, SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) AS paid, SUM(CASE WHEN status = 'completed' AND purchased_at >= CURDATE() THEN 1 ELSE 0 END) AS today, SUM(CASE WHEN status = 'completed' AND purchased_at >= CURDATE() - INTERVAL 6 DAY THEN 1 ELSE 0 END) AS last_7d FROM balancepick_purchases WHERE verified_at IS NOT NULL AND purchase_state = 0 AND is_test_purchase = 0 """ ) return { "supported": True, "total": _number(row.get("total")), "paid": _number(row.get("paid")), "today": _number(row.get("today")), "last_7d": _number(row.get("last_7d")), "has_revenue": False, } if app_key == "chromakey": if not self.has_table("chromakey_purchases"): return {"supported": False} row = await self._fetch_one( """ SELECT COUNT(*) AS total, SUM(CASE WHEN is_active = 1 AND (expiry_time IS NULL OR expiry_time > NOW()) THEN 1 ELSE 0 END) AS paid, SUM(CASE WHEN purchase_time >= CURDATE() THEN 1 ELSE 0 END) AS today, SUM(CASE WHEN purchase_time >= CURDATE() - INTERVAL 6 DAY THEN 1 ELSE 0 END) AS last_7d FROM chromakey_purchases """ ) return { "supported": True, "total": _number(row.get("total")), "paid": _number(row.get("paid")), "today": _number(row.get("today")), "last_7d": _number(row.get("last_7d")), "has_revenue": False, } config = PURCHASES.get(app_key) if not config or not self.has_table(config.table): return {"supported": False} status_placeholders = ", ".join(["%s"] * len(config.paid_statuses)) params = tuple(config.paid_statuses) * 3 row = await self._fetch_one( f""" SELECT COUNT(*) AS total, SUM(CASE WHEN status IN ({status_placeholders}) THEN 1 ELSE 0 END) AS paid, SUM(CASE WHEN status IN ({status_placeholders}) AND `{config.date_column}` >= CURDATE() THEN 1 ELSE 0 END) AS today, SUM(CASE WHEN status IN ({status_placeholders}) AND `{config.date_column}` >= CURDATE() - INTERVAL 6 DAY THEN 1 ELSE 0 END) AS last_7d FROM `{config.table}` """, params, ) revenue_rows = await self._fetch_all( f""" SELECT COALESCE(NULLIF(currency, ''), 'KRW') AS currency, COALESCE(SUM(price), 0) AS amount FROM `{config.table}` WHERE status IN ({status_placeholders}) GROUP BY COALESCE(NULLIF(currency, ''), 'KRW') ORDER BY currency """, config.paid_statuses, ) result = { "supported": True, "total": _number(row.get("total")), "paid": _number(row.get("paid")), "today": _number(row.get("today")), "last_7d": _number(row.get("last_7d")), "has_revenue": True, "revenue_by_currency": revenue_rows, } if app_key == "babynote" and self.has_table("babynote_kidsnote_payment_orders"): extra = await self._fetch_one( """ SELECT COUNT(*) AS paid_orders, COALESCE(SUM(total_price), 0) AS revenue, SUM(CASE WHEN approved_at >= CURDATE() THEN 1 ELSE 0 END) AS today, SUM(CASE WHEN approved_at >= CURDATE() - INTERVAL 6 DAY THEN 1 ELSE 0 END) AS last_7d FROM babynote_kidsnote_payment_orders WHERE status = 'paid' """ ) paid_orders = _number(extra.get("paid_orders")) result["total"] += paid_orders result["paid"] += paid_orders result["today"] += _number(extra.get("today")) result["last_7d"] += _number(extra.get("last_7d")) result["extra"] = [ {"label": "포토북 결제", "value": paid_orders} ] if _number(extra.get("revenue")): krw_row = next( ( item for item in result["revenue_by_currency"] if str(item.get("currency") or "").upper() == "KRW" ), None, ) if krw_row: krw_row["amount"] = _number(krw_row.get("amount")) + _number( extra.get("revenue") ) else: result["revenue_by_currency"].append( {"currency": "KRW", "amount": _number(extra.get("revenue"))} ) return result async def stats(self, app_key: str, category: str) -> dict[str, Any]: if app_key not in APP_LABELS: raise ValueError("지원하지 않는 앱입니다.") if category == "users": return await self.user_stats(app_key) if category == "activity": return await self.core_stats(app_key) if category == "purchases": return await self.purchase_stats(app_key) if category == "inquiries": return await self.inquiry_stats(app_key) users, core, purchases, inquiries = await asyncio.gather( self.user_stats(app_key), self.core_stats(app_key), self.purchase_stats(app_key), self.inquiry_stats(app_key), ) return { "users": users, "metrics": core.get("metrics", []), "purchases": purchases, "inquiries": inquiries, } async def fetch_events(self, start_at: datetime, end_at: datetime) -> list[dict[str, Any]]: events: list[dict[str, Any]] = [] for source in EVENT_SOURCES: if not self.has_table(source["table"]): continue rows = await self._fetch_all(source["sql"], (start_at, end_at)) for row in rows: row["source"] = source["source"] row["event_type"] = source["event_type"] row["event_key"] = ( f"{source['source']}:{row.get('event_id')}:{row.get('status') or 'new'}" ) events.append(row) events.sort(key=lambda item: item.get("event_time") or datetime.min) return events app_master_service = AppMasterService()