import os import json import re import requests import traceback from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt BUILD_VERSION = "semantic-v5-nli-resize-2026-05-15b" # ----------------------- # 1) 문장 분리 # ----------------------- SENT_SPLIT_RE = re.compile(r"[.!?]\s+|\n+") def split_sentences(text: str, limit: int = 10): t = (text or "").strip() if not t: return [] parts = [p.strip() for p in SENT_SPLIT_RE.split(t) if p.strip()] return parts[:limit] # ----------------------- # 2) (AI) 감성 분류 파이프라인 # ----------------------- _SENTIMENT = None def get_sentiment_pipeline(): global _SENTIMENT if _SENTIMENT is None: from transformers import pipeline _SENTIMENT = pipeline( "text-classification", model="daekeun-ml/koelectra-small-v3-nsmc", top_k=None ) return _SENTIMENT # ----------------------- # 3) (AI) NLI 모델 (문장쌍 추론 직접 수행 + 임베딩 크기 보정) # ----------------------- _NLI = None def get_nli(): global _NLI if _NLI is None: import torch import torch.nn.functional as F from transformers import AutoTokenizer, AutoModelForSequenceClassification model_id = "Huffon/klue-roberta-base-nli" # fast tokenizer에서 드물게 mismatch가 나는 케이스를 피하기 위해 slow tokenizer 시도 tok = AutoTokenizer.from_pretrained(model_id, use_fast=False) mdl = AutoModelForSequenceClassification.from_pretrained(model_id) mdl.eval() # tokenizer vocab과 model embedding 크기가 다르면 resize(IndexError 방지) try: vocab_size = len(tok) emb = mdl.get_input_embeddings() emb_size = int(emb.num_embeddings) if emb is not None else None if emb_size is not None and vocab_size != emb_size: mdl.resize_token_embeddings(vocab_size) except Exception: # resize 실패해도 일단 계속(에러는 semantic_crisis_confidence에서 잡힘) pass # entailment 라벨 인덱스 찾기(방어적으로) entail_id = None label2id = getattr(mdl.config, "label2id", None) or {} id2label = getattr(mdl.config, "id2label", None) or {} for k, v in label2id.items(): if str(k).upper() == "ENTAILMENT": entail_id = int(v) break if entail_id is None: for k, v in id2label.items(): if str(v).upper() == "ENTAILMENT": entail_id = int(k) break if entail_id is None: entail_id = 0 # fallback _NLI = (tok, mdl, entail_id, torch, F) return _NLI def nli_entail_prob(premise: str, hypothesis: str) -> float: tok, mdl, entail_id, torch, F = get_nli() inputs = tok( premise, hypothesis, return_tensors="pt", truncation=True, max_length=256 ) inputs.pop("token_type_ids", None) # <- 추가: RoBERTa에서 token_type_ids 때문에 터지는 문제 방지 with torch.no_grad(): logits = mdl(**inputs).logits probs = F.softmax(logits, dim=-1)[0] n = int(probs.shape[-1]) # entail_id 안전 범위 체크 if entail_id < 0 or entail_id >= n: entail_id = 0 return float(probs[entail_id].item()) # ----------------------- # 4) 직접 위기 신호(레벨 4) + 부정문 필터 # ----------------------- SUICIDE_RE = re.compile(r"(자살|자해|죽고\s?싶|죽고싶|목숨)", re.IGNORECASE) NEGATION_RE = re.compile(r"(아니|않|없|지\s?않)", re.IGNORECASE) def _is_negated_around(text: str, start: int, end: int, window: int = 10) -> bool: left = text[max(0, start - window):start] right = text[end:min(len(text), end + window)] return bool(NEGATION_RE.search(left) or NEGATION_RE.search(right)) def has_direct_selfharm_signal(text: str) -> bool: t = text or "" m = SUICIDE_RE.search(t) if not m: return False if _is_negated_around(t, m.start(), m.end()): return False return True def has_negated_selfharm_mention(text: str) -> bool: t = text or "" m = SUICIDE_RE.search(t) if not m: return False return _is_negated_around(t, m.start(), m.end()) # ----------------------- # 5) 규칙 기반 카테고리 + 가중치 # ----------------------- CATEGORIES = { "우울/절망": re.compile(r"(우울|절망|희망이\s?없|허무|공허)", re.IGNORECASE), "무기력/피로": re.compile(r"(무기력|기운이\s?없|피곤|지침)", re.IGNORECASE), "무쾌감": re.compile(r"(재미가\s?없|흥미가\s?없|즐겁지\s?않)", re.IGNORECASE), "자기비난": re.compile(r"(실패자|쓸모없|자책|혐오|죄책감)", re.IGNORECASE), "수면문제": re.compile(r"(불면|잠이\s?안|자주\s?깨|과수면|너무\s?잠)", re.IGNORECASE), "집중곤란": re.compile(r"(집중이\s?안|멍하|결정이\s?어렵)", re.IGNORECASE), } CATEGORY_WEIGHTS = { "우울/절망": 1.0, "무기력/피로": 0.2, "무쾌감": 0.9, "자기비난": 1.0, "수면문제": 0.5, "집중곤란": 0.5, } # ----------------------- # 6) 레벨/톤 메시지 # ----------------------- RISK_LEVELS = [ (0, "안정", "현재 기록에서 큰 위험 신호는 낮아 보입니다. 다만 변화가 있으면 계속 기록해 주세요."), (1, "주의", "부정 정서/스트레스 신호가 일부 보입니다. 수면·식사·활동 리듬을 점검하고 추적을 권장합니다."), (2, "관찰 필요", "반복되는 무기력/자기비난 등의 신호가 보입니다. 상담/진료를 고려해 보세요."), (3, "도움 권장", "위험 신호가 뚜렷합니다. 가능한 한 빠르게 전문가 상담/진료를 권장합니다."), (4, "긴급(즉시 도움)", "위기 신호가 감지되었습니다. 지금은 즉시 도움을 받는 것이 안전합니다."), ] # ----------------------- # 7) 카카오 장소 추천(키 없으면 비활성) # ----------------------- def kakao_places(lat, lng, radius=3000): key = os.getenv("KAKAO_REST_API_KEY") if not key or lat is None or lng is None: return [] url = "https://dapi.kakao.com/v2/local/search/keyword.json" headers = {"Authorization": f"KakaoAK {key}"} params = { "query": "정신건강의학과", "y": str(lat), "x": str(lng), "radius": str(radius), "sort": "distance", "size": 10, } r = requests.get(url, headers=headers, params=params, timeout=5) if r.status_code != 200: return [] return r.json().get("documents", []) # ----------------------- # 8) 감성 모델 부정 확률 p_neg # ----------------------- def negative_prob_from_text(pipe, text: str) -> float: t = (text or "").strip() if not t: return 0.5 def _p_neg(one: str) -> float: out = pipe(one) p_neg = None if isinstance(out, list) and len(out) > 0 and isinstance(out[0], list): for item in out[0]: label = str(item.get("label", "")).lower() score = float(item.get("score", 0.0)) if ("neg" in label) or (label in ["0", "label_0", "negative"]): p_neg = score elif isinstance(out, list) and len(out) > 0 and isinstance(out[0], dict): label = str(out[0].get("label", "")).lower() score = float(out[0].get("score", 0.0)) if ("neg" in label) or (label in ["0", "label_0", "negative"]): p_neg = score else: p_neg = 1.0 - score return 0.5 if p_neg is None else p_neg if len(t) <= 900: return _p_neg(t[:1000]) head = t[:700] tail = t[-700:] return (_p_neg(head) + _p_neg(tail)) / 2.0 # ----------------------- # 9) 의미 기반(간접) 위기 표현 → 레벨 3으로만 상향 # ----------------------- H_CRISE = "이 문장은 삶을 포기하고 싶다는 뜻이다." H_FATIGUE = "이 문장은 단순히 피곤하거나 스트레스가 있다는 뜻이다." H_NEGATE = "이 문장은 그런 뜻이 아니라고 부정한다." SEMANTIC_TRIGGER_RE = re.compile(r"(살기\s?싫|의미\s?없|그만\s?하고\s?싶|포기하고\s?싶)", re.IGNORECASE) def semantic_crisis_confidence(text: str) -> dict: t = (text or "").strip() if not t or not SEMANTIC_TRIGGER_RE.search(t): return {"score": 0.0, "sentence": None, "detail": None, "error": None} try: sents = split_sentences(t, limit=6) if not sents: return {"score": 0.0, "sentence": None, "detail": {"reason": "no_sentences"}, "error": None} best = None for s in sents: e_crisis = nli_entail_prob(s, H_CRISE) e_fatigue = nli_entail_prob(s, H_FATIGUE) e_negate = nli_entail_prob(s, H_NEGATE) raw = e_crisis - max(e_fatigue, e_negate) score = max(0.0, raw) cand = { "score": float(score), "sentence": s, "detail": { "e_crisis": float(e_crisis), "e_fatigue": float(e_fatigue), "e_negate": float(e_negate), "raw": float(raw), }, "error": None } if best is None or cand["score"] > best["score"]: best = cand return best except Exception as e: return {"score": 0.0, "sentence": None, "detail": None, "error": repr(e)} # ----------------------- # 10) 최종 위험도 # ----------------------- def risk_from_text(diary: str): t = (diary or "").strip() if has_direct_selfharm_signal(t): lvl, name, tone = RISK_LEVELS[4] return { "level": lvl, "name": name, "tone_message": tone, "override": "direct_selfharm", "signals": {"build_version": BUILD_VERSION} } pipe = get_sentiment_pipeline() p_neg = negative_prob_from_text(pipe, t) matched = [] weighted_sum = 0.0 weighted_max = 0.0 for cat, rx in CATEGORIES.items(): w = float(CATEGORY_WEIGHTS.get(cat, 0.5)) hits = len(rx.findall(t)) if hits > 0: matched.append(cat) weighted_sum += w * min(2, hits) weighted_max += w * 2.0 cat_norm = 0.0 if weighted_max == 0 else min(1.0, weighted_sum / weighted_max) # 감성 영향 낮추기 R = 0.2 * p_neg + 0.8 * cat_norm if R < 0.20: idx = 0 elif R < 0.40: idx = 1 elif R < 0.60: idx = 2 elif R < 0.80: idx = 3 else: idx = 4 # 부정된 자해 표현은 최소 레벨 1 if has_negated_selfharm_mention(t) and idx < 1: idx = 1 lvl, name, tone = RISK_LEVELS[idx] # 간접 위기(의미) → 레벨 3까지만 상향 sem = semantic_crisis_confidence(t) SEM_THRESHOLD = 0.20 if sem.get("score", 0.0) >= SEM_THRESHOLD and idx < 3: lvl, name, tone = RISK_LEVELS[3] return { "level": lvl, "name": name, "tone_message": tone, "override": "semantic_crisis_to_level3", "signals": { "build_version": BUILD_VERSION, "semantic_crisis": sem, "p_negative": p_neg, "matched_categories": matched, "category_norm": cat_norm, "R": R, } } return { "level": lvl, "name": name, "tone_message": tone, "signals": { "build_version": BUILD_VERSION, "semantic_crisis": sem, "p_negative": p_neg, "matched_categories": matched, "category_norm": cat_norm, "R": R, } } # ----------------------- # 11) API # ----------------------- @csrf_exempt def api_analyze(request): payload = json.loads(request.body.decode("utf-8")) diary = payload.get("diary", "") lat = payload.get("lat", None) lng = payload.get("lng", None) risk = risk_from_text(diary) kakao_key_present = bool(os.getenv("KAKAO_REST_API_KEY")) places = kakao_places(lat, lng) if (kakao_key_present and lat is not None and lng is not None) else [] safety = None if risk.get("override") == "direct_selfharm": safety = { "message": "위기 신호가 감지되었습니다. 즉시 도움을 받는 것이 안전합니다.", "hotline": "자살예방상담전화 109 (24시간)", "source": "https://www.korea.kr/multi/visualNewsView.do?newsId=148921874" } return JsonResponse({ "build_version": BUILD_VERSION, "risk": risk, "safety": safety, "places": places, "hospital_recommendation": { "enabled": kakao_key_present, "reason_disabled": None if kakao_key_present else "KAKAO_REST_API_KEY가 설정되지 않아 병원 추천 기능이 비활성화되어 있습니다." }, "disclaimer": "본 결과는 의학적 진단이 아니라 텍스트 기반 선별/안내입니다. 정확한 평가/진단은 전문가 상담이 필요합니다." })