Blame for utils.py

import io
import logging
import re
from datetime import datetime, timedelta, timezone

import streamlit as st

logger = logging.getLogger("talk_trace_ai")

DATETIME_FMT = "%Y-%m-%d %H:%M:%S"
DISPLAY_DATETIME_FMT = "%Y-%m-%d %p %I:%M:%S"
KST = timezone(timedelta(hours=9))


def get_kst_now():
    return datetime.now(tz=KST)


def get_kst_now_str():
    return get_kst_now().strftime(DATETIME_FMT)


def format_kst_datetime(value):
    if value is None:
        return "-"
    kst_tz = timezone(timedelta(hours=9))
    parsed_dt = None
    should_assume_utc = False
    if isinstance(value, datetime):
        parsed_dt = value
    else:
        raw = str(value).strip()
        if not raw:
            return "-"
        if "T" in raw and "+" not in raw and "Z" not in raw:
            should_assume_utc = True
        iso_candidate = raw.replace("Z", "+00:00")
        try:
            parsed_dt = datetime.fromisoformat(iso_candidate)
        except ValueError:
            for fmt in ("%Y-%m-%d %H:%M:%S.%f", DATETIME_FMT):
                try:
                    parsed_dt = datetime.strptime(raw, fmt)
                    break
                except ValueError:
                    continue
    if parsed_dt is None:
        return str(value)
    if parsed_dt.tzinfo is not None:
        parsed_dt = parsed_dt.astimezone(kst_tz)
    elif should_assume_utc:
        parsed_dt = parsed_dt.replace(tzinfo=timezone.utc).astimezone(kst_tz)
    return parsed_dt.strftime(DISPLAY_DATETIME_FMT)


def get_client_ip():
    try:
        headers = st.context.headers
    except Exception:
        return ""
    if not headers:
        return ""
    for key in ["x-forwarded-for", "x-real-ip", "cf-connecting-ip", "fly-client-ip",
                "X-Forwarded-For", "X-Real-Ip", "CF-Connecting-IP", "True-Client-Ip",
                "true-client-ip"]:
        raw_ip = headers.get(key)
        if raw_ip:
            return str(raw_ip).split(",")[0].strip()
    try:
        all_keys = list(headers.keys()) if headers else []
        logger.info("get_client_ip: IP 헤더 없음. 수신된 헤더 키: %s", all_keys)
    except Exception:
        pass
    return ""


def anonymize_ip(raw_ip: str) -> str | None:
    """IP를 익명화하여 반환합니다. 저장 불가 시 None 반환.

    IPv4: 첫 번째·마지막 옥텟 유지, 중간 0으로 대체 (예: 165.0.0.41)
    IPv6: 앞 4그룹·마지막 그룹 유지, 중간 :: 처리 (예: 2406:5900:117c:424b::4444)
          → 같은 Wi-Fi라도 기기별 구분 가능
    """
    ip = str(raw_ip or "").strip()
    if not ip:
        return None
    if ":" in ip:  # IPv6
        groups = [g for g in ip.split(":") if g]  # 빈 그룹 제거
        if len(groups) >= 5:
            return ":".join(groups[:4]) + "::" + groups[-1]
        if len(groups) >= 4:
            return ":".join(groups[:4]) + "::"
        return None
    parts = ip.split(".")  # IPv4
    if len(parts) == 4:
        return f"{parts[0]}.0.0.{parts[3]}"
    return None


def log_audit(event, room_name="", actor_name="", role="", **extra):
    logger.info(
        "AUDIT event=%s room=%s actor=%s role=%s extra=%s",
        event, room_name, actor_name, role, extra,
    )


def to_bool_flag(value):
    if isinstance(value, bool):
        return value
    if value is None:
        return False
    return str(value).strip().lower() in {"true", "t", "1", "yes", "y", "on"}


_FONT_CANDIDATES = [
    "/usr/share/fonts/truetype/nanum/NanumGothic.ttf",
    "/usr/share/fonts/truetype/fonts-japanese-gothic.ttf",
    "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
]
_FONT_BOLD_CANDIDATES = [
    "/usr/share/fonts/truetype/nanum/NanumGothicBold.ttf",
    "/usr/share/fonts/truetype/nanum/NanumGothic.ttf",
    "/usr/share/fonts/truetype/fonts-japanese-gothic.ttf",
]

_AI_SUBSECTION_LABELS = ("배움의 변화:", "성장한 점:", "한 줄 요약:")


def _get_pil_font(size: int):
    from PIL import ImageFont
    for path in _FONT_CANDIDATES:
        try:
            return ImageFont.truetype(path, size)
        except Exception:
            continue
    return ImageFont.load_default()


def _get_pil_font_bold(size: int):
    from PIL import ImageFont
    for path in _FONT_BOLD_CANDIDATES:
        try:
            return ImageFont.truetype(path, size)
        except Exception:
            continue
    return _get_pil_font(size)


_EMOJI_TEXT_SUBS = [
    # AI 피드백 카드 레이블
    ("✅", "[완료]"),
    ("🌱", "[성장]"),
    # 섹션 마커
    ("📌", ">>"),
    ("🔄", ">>"),
    ("🤖", "[AI]"),
    ("👨‍🏫", "[교사]"),
    # 기타 자주 쓰이는 이모지
    ("💡", "[아이디어]"),
    ("➕", "[보충]"),
    ("❓", "[질문]"),
    ("🔵", "[찬성]"),
    ("🔴", "[반대]"),
    ("👍", "[좋아요]"),
    ("🎉", ""),
    ("⚠️", "[주의]"),
]


def _strip_non_renderable(text: str) -> str:
    """이미지 렌더링 시 깨질 수 있는 이모지/심볼을 텍스트로 대체하거나 제거."""
    # 1단계: 알려진 이모지를 텍스트 레이블로 치환
    for emoji, replacement in _EMOJI_TEXT_SUBS:
        text = (text or "").replace(emoji, replacement)
    # 2단계: 나머지 렌더링 불가 문자 제거
    result = []
    for ch in (text or ""):
        code = ord(ch)
        # 4바이트 유니코드 (이모지 대부분)
        if code > 0xFFFF or 0xD800 <= code <= 0xDFFF:
            continue
        # BMP 내 이모지/기호 블록
        if 0x2190 <= code <= 0x27FF:  # 화살표, 기술기호, 딩뱃(✅ 등)
            continue
        if 0x2B00 <= code <= 0x2BFF:  # 기타 기호·화살표
            continue
        if 0xFE00 <= code <= 0xFEFF:  # 변형 선택자 등
            continue
        result.append(ch)
    return "".join(result)


def _wrap_to_width(draw, text: str, font, max_width: int) -> list:
    lines = []
    for para in _strip_non_renderable(text).split("\n"):
        if not para.strip():
            lines.append("")
            continue
        words = para.split()
        line: list = []
        for word in words:
            test = " ".join(line + [word])
            try:
                w = draw.textbbox((0, 0), test, font=font)[2]
            except Exception:
                w = len(test) * (font.size // 2 if hasattr(font, "size") else 9)
            if w <= max_width:
                line.append(word)
            else:
                if line:
                    lines.append(" ".join(line))
                    line = [word]
                else:
                    lines.append(word)
        if line:
            lines.append(" ".join(line))
    return lines or [""]


def create_analysis_image(
    student_name: str,
    topic: str,
    pre_opinion: str,
    post_opinion: str,
    ai_analysis: str,
    ai_feedback: str = "",
) -> bytes:
    """AI 배움 분석 결과를 PNG 이미지 bytes로 반환합니다."""
    from PIL import Image, ImageDraw

    W, PAD = 800, 40
    CONTENT_W = W - PAD * 2

    C_BG        = (255, 255, 255)
    C_HDR_BG    = (25,  80, 160)
    C_HDR_FG    = (255, 255, 255)
    C_HDR_SUB   = (170, 200, 240)
    C_ACCENT    = (25,  80, 160)
    C_TEXT      = (30,  30,  30)
    C_MUTED     = (110, 110, 110)
    C_DIVIDER   = (220, 220, 220)
    C_FTR_BG    = (245, 245, 248)

    f_title    = _get_pil_font(22)
    f_label    = _get_pil_font(15)
    f_sublabel = _get_pil_font_bold(16)
    f_body     = _get_pil_font(17)
    f_small    = _get_pil_font(13)

    LH_BODY    = 27   # body line height px
    LH_LABEL   = 22
    LH_SUBLBL  = 24
    SEC_GAP    = 16
    SUBSEC_GAP = 14   # extra gap between AI subsections
    HEADER_H   = 74
    FOOTER_H   = 38
    DIV_H      = 20

    def _ai_paragraphs(text):
        """AI 분석 텍스트를 (is_label, label_str, body_str) 튜플 리스트로 파싱."""
        result = []
        for para in _strip_non_renderable(text or "").split("\n"):
            para = para.strip()
            if not para:
                continue
            matched = next((lbl for lbl in _AI_SUBSECTION_LABELS if para.startswith(lbl)), None)
            if matched:
                result.append((True, matched, para[len(matched):].strip()))
            else:
                result.append((False, "", para))
        return result or [(False, "", "(분석 없음)")]

    # ── measure total height with dummy canvas ──────────────────────────
    dummy = Image.new("RGB", (W, 10))
    d = ImageDraw.Draw(dummy)

    def section_h(text):
        return LH_LABEL + 10 + len(_wrap_to_width(d, text, f_body, CONTENT_W)) * LH_BODY + SEC_GAP

    def ai_section_h(text):
        h = LH_LABEL + 10  # "AI 배움 분석" 헤더
        paras = _ai_paragraphs(text)
        for i, (is_lbl, lbl, body) in enumerate(paras):
            if is_lbl and i > 0:
                h += SUBSEC_GAP
            if is_lbl:
                h += LH_SUBLBL + 6
            h += len(_wrap_to_width(d, body if is_lbl else lbl + body, f_body, CONTENT_W)) * LH_BODY
        return h + SEC_GAP

    total_h = (
        HEADER_H + 18
        + LH_LABEL + 14      # info line
        + DIV_H
        + section_h(pre_opinion)
        + section_h(post_opinion)
        + DIV_H
        + ai_section_h(ai_analysis)
        + (DIV_H + section_h(ai_feedback) if ai_feedback else 0)
        + FOOTER_H
    )

    # ── render ──────────────────────────────────────────────────────────
    img  = Image.new("RGB", (W, total_h), C_BG)
    draw = ImageDraw.Draw(img)

    # Header
    draw.rectangle([0, 0, W, HEADER_H], fill=C_HDR_BG)
    draw.text((PAD, 14), "말자취 Talk-Trace AI", font=f_small, fill=C_HDR_SUB)
    draw.text((PAD, 34), "AI 배움 분석 결과", font=f_title, fill=C_HDR_FG)

    y = HEADER_H + 18

    # Student / topic info line
    info = f"학생: {_strip_non_renderable(student_name)}    |    주제: {_strip_non_renderable(topic)}"
    draw.text((PAD, y), info, font=f_label, fill=C_MUTED)
    y += LH_LABEL + 14

    def divider():
        nonlocal y
        draw.line([PAD, y, W - PAD, y], fill=C_DIVIDER, width=1)
        y += DIV_H

    def section(label, text):
        nonlocal y
        draw.text((PAD, y), label, font=f_label, fill=C_ACCENT)
        y += LH_LABEL + 10
        for line in _wrap_to_width(draw, text, f_body, CONTENT_W):
            draw.text((PAD, y), line, font=f_body, fill=C_TEXT)
            y += LH_BODY
        y += SEC_GAP

    def ai_section(text):
        nonlocal y
        draw.text((PAD, y), "AI 배움 분석", font=f_label, fill=C_ACCENT)
        y += LH_LABEL + 10
        for i, (is_lbl, lbl, body) in enumerate(_ai_paragraphs(text)):
            if is_lbl and i > 0:
                y += SUBSEC_GAP
            if is_lbl:
                # 소제목: 굵은 강조색
                draw.text((PAD, y), lbl, font=f_sublabel, fill=C_ACCENT)
                y += LH_SUBLBL + 6
                for line in _wrap_to_width(draw, body, f_body, CONTENT_W):
                    draw.text((PAD + 8, y), line, font=f_body, fill=C_TEXT)
                    y += LH_BODY
            else:
                full = (lbl + body).strip()
                for line in _wrap_to_width(draw, full, f_body, CONTENT_W):
                    draw.text((PAD, y), line, font=f_body, fill=C_TEXT)
                    y += LH_BODY
        y += SEC_GAP

    divider()
    section("토론 전 생각", pre_opinion or "(없음)")
    section("토론 후 생각", post_opinion or "(없음)")
    divider()
    ai_section(ai_analysis)
    if ai_feedback:
        divider()
        section("AI 피드백 카드 (잘한 점 / 발전할 점)", ai_feedback)

    # Footer
    draw.rectangle([0, total_h - FOOTER_H, W, total_h], fill=C_FTR_BG)
    now_str = datetime.now(tz=KST).strftime("%Y-%m-%d %H:%M")
    draw.text((PAD, total_h - FOOTER_H + 13), f"생성: {now_str}", font=f_small, fill=C_MUTED)

    buf = io.BytesIO()
    img.save(buf, format="PNG", optimize=True)
    return buf.getvalue()


def compact_ai_report_output(text):
    raw_lines = [str(line).strip() for line in str(text or "").splitlines()]
    cleaned = [line for line in raw_lines if line and not line.startswith("#")]
    if not cleaned:
        return ""
    normalized_text = " ".join(cleaned)
    normalized_text = re.sub(r"\s*(핵심요약 [123]:|베스트 학생:|선정 이유:)", r"\n\1", normalized_text)
    normalized_lines = [line.strip() for line in normalized_text.splitlines() if line.strip()]
    return "\n".join(normalized_lines[:5])