from dataclasses import dataclass, field
import gzip
import json
from pathlib import Path
import re

import requests


OBF_DUMP_URL = "https://static.openbeautyfacts.org/data/openbeautyfacts-products.jsonl.gz"


@dataclass
class Product:
    barcode: str
    brand: str
    name: str
    ingredients_raw: str
    inci_list: list[str]                # 함량 순으로 정렬된 성분 리스트
    categories: list[str] = field(default_factory=list)
    country: str = ""
    image_url: str = ""


def download_obf_dump(dest: Path) -> Path:
    """Open Beauty Facts JSON dump 다운로드 (약 500MB gzip)."""
    dest.parent.mkdir(parents=True, exist_ok=True)
    if dest.exists() and dest.stat().st_size > 100_000_000:
        return dest
    print(f"다운로드 중... {OBF_DUMP_URL}")
    with requests.get(OBF_DUMP_URL, stream=True, timeout=600) as r:
        r.raise_for_status()
        with open(dest, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192 * 1024):
                f.write(chunk)
    return dest


def parse_inci(ingredients_raw: str) -> list[str]:
    """INCI 성분 raw 텍스트를 함량 순 리스트로 파싱.
    
    - 콤마·세미콜론·마침표 구분
    - 소괄호 안은 alias · CI 번호 (예: "AQUA (WATER)"의 "WATER")
    - 대괄호 안은 제거 (nano 표기 등)
    - 알레르기 유발 성분 별표 tag 유지
    """
    if not ingredients_raw:
        return []
    # 정규화 · 소·대괄호 처리
    parts = re.split(r"[,;·]", ingredients_raw)
    cleaned = []
    for p in parts:
        # 대괄호 제거 (예: "[nano]")
        p = re.sub(r"\[.*?\]", "", p)
        # 소괄호 안 대체명을 별도 tag로 유지 (여기서는 통째 제거하지 않음)
        p = re.sub(r"\(.*?\)", "", p)
        p = p.strip().upper()
        # 알레르기 별표 (**), 물결 (~) 등 특수 tag 유지 여부 결정
        if p and len(p) > 1 and len(p) < 100:
            cleaned.append(p)
    return cleaned


def iter_cosmetics(dump_path: Path, min_ingredients: int = 3):
    """OBF dump 스트리밍 · 최소 성분 수 필터."""
    with gzip.open(dump_path, "rt", encoding="utf-8") as f:
        for line in f:
            try:
                d = json.loads(line)
            except json.JSONDecodeError:
                continue
            if not d.get("ingredients_text"):
                continue
            inci = parse_inci(d.get("ingredients_text", ""))
            if len(inci) < min_ingredients:
                continue
            yield Product(
                barcode=d.get("code", ""),
                brand=(d.get("brands") or "").split(",")[0].strip(),
                name=d.get("product_name", ""),
                ingredients_raw=d.get("ingredients_text", ""),
                inci_list=inci,
                categories=d.get("categories_tags", []),
                country=d.get("countries", ""),
                image_url=d.get("image_url", ""),
            )

import csv


@dataclass
class CosingEntry:
    inci_name: str
    cas: str
    einecs: str
    function: str            # e.g. "Skin conditioning · Emollient"
    restriction: str          # EU 규제 (예: "Concentration limit 0.5%")
    allergen_flag: bool = False


def load_cosing_dict(cosing_csv: Path) -> dict[str, CosingEntry]:
    """CosIng CSV → {INCI_name(uppercased): CosingEntry} 사전."""
    result: dict[str, CosingEntry] = {}
    with open(cosing_csv, encoding="utf-8", errors="ignore") as f:
        reader = csv.DictReader(f)
        for row in reader:
            key = (row.get("INCI name") or "").upper().strip()
            if not key:
                continue
            result[key] = CosingEntry(
                inci_name=key,
                cas=row.get("CAS #", "").strip(),
                einecs=row.get("EINECS/ELINCS #", "").strip(),
                function=row.get("Function", "").strip(),
                restriction=row.get("Restriction", "").strip(),
                allergen_flag="allergen" in row.get("Restriction", "").lower(),
            )
    return result


def normalize_inci(raw_name: str, cosing_dict: dict[str, CosingEntry]) -> str | None:
    """CosIng 사전으로 정규화. 부분 매칭 fallback."""
    upper = raw_name.upper().strip()
    if upper in cosing_dict:
        return upper
    # 부분 매칭 (예: "AQUA (WATER)" → "AQUA")
    for key in cosing_dict:
        if key in upper or upper in key:
            return key
    # Fuzzy matching (Levenshtein distance) 대안
    return None

from rdkit import Chem, RDLogger
from rdkit.Chem import AllChem
import numpy as np

RDLogger.DisableLog("rdApp.*")


def cas_to_smiles(cas_number: str, cache: dict | None = None) -> str | None:
    """CAS → SMILES. PubChem REST API [8]. 캐시 활용 권장 (반복 방지)."""
    if not cas_number:
        return None
    if cache and cas_number in cache:
        return cache[cas_number]
    url = f"https://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/name/{cas_number}/property/CanonicalSMILES/JSON"
    try:
        resp = requests.get(url, timeout=15)
        if resp.status_code != 200:
            if cache is not None:
                cache[cas_number] = None
            return None
        data = resp.json()
        smiles = data["PropertyTable"]["Properties"][0]["CanonicalSMILES"]
        if cache is not None:
            cache[cas_number] = smiles
        return smiles
    except (requests.RequestException, KeyError):
        if cache is not None:
            cache[cas_number] = None
        return None


def compute_ecfp(smiles: str, radius: int = 2, n_bits: int = 2048) -> np.ndarray:
    """ECFP (Extended Connectivity Fingerprint) 계산. Morgan Fingerprint."""
    mol = Chem.MolFromSmiles(smiles)
    if mol is None:
        return np.zeros(n_bits, dtype=np.uint8)
    fp = AllChem.GetMorganFingerprintAsBitVect(mol, radius, nBits=n_bits)
    return np.array(fp, dtype=np.uint8)


def build_ingredient_embeddings(
    cosing_dict: dict[str, CosingEntry],
    cache_path: Path,
    sleep_between: float = 0.2,  # PubChem 예의 (초당 5 request 이하)
) -> dict[str, np.ndarray]:
    """INCI 성분 → ECFP 임베딩 사전. PubChem 캐시 활용."""
    import time
    if cache_path.exists():
        return np.load(cache_path, allow_pickle=True).item()
    
    embeddings: dict[str, np.ndarray] = {}
    smiles_cache: dict[str, str | None] = {}
    for i, (inci_name, info) in enumerate(cosing_dict.items()):
        if i % 100 == 0:
            print(f"[{i}/{len(cosing_dict)}] fingerprint 계산 중")
        smiles = cas_to_smiles(info.cas, cache=smiles_cache)
        if smiles:
            embeddings[inci_name] = compute_ecfp(smiles)
        time.sleep(sleep_between)  # PubChem rate limit 준수
    
    np.save(cache_path, embeddings, allow_pickle=True)
    return embeddings

from typing import Literal, NamedTuple


Severity = Literal["info", "warn", "avoid"]


class Rule(NamedTuple):
    kind: str                          # "conflict" · "allergen" · "pregnancy" · "user_avoid"
    ingredient_a: str
    ingredient_b: str | None            # None if 단일 성분 룰
    severity: Severity
    reason: str
    reference: str
    context: dict[str, str] = {}


CONFLICT_RULES: list[Rule] = [
    Rule(
        kind="conflict",
        ingredient_a="RETINOL",
        ingredient_b="GLYCOLIC ACID",
        severity="warn",
        reason="레티노이드 + AHA 동시 사용은 자극·홍반 유발 가능. 밤·아침 분리 사용 권장.",
        reference="AAD (American Academy of Dermatology) retinoid guidelines",
    ),
    Rule(
        kind="conflict",
        ingredient_a="RETINOL",
        ingredient_b="SALICYLIC ACID",
        severity="warn",
        reason="레티노이드 + BHA 동시 사용은 자극 · 각질 파괴 과잉 유발 가능.",
        reference="Journal of the American Academy of Dermatology",
    ),
    Rule(
        kind="conflict",
        ingredient_a="RETINOL",
        ingredient_b="ASCORBIC ACID",
        severity="warn",
        reason="레티놀 + 순수 비타민 C는 pH 요구가 상충 (retinol 5.5, vit C 3.5). 안정성·자극 이슈.",
        reference="Skin Therapy Letter, vitamin C stability review",
    ),
    Rule(
        kind="conflict",
        ingredient_a="BENZOYL PEROXIDE",
        ingredient_b="RETINOL",
        severity="avoid",
        reason="BPO가 retinol을 산화·분해 → retinol 효과 소실 + 피부 자극.",
        reference="Journal of Drugs in Dermatology",
    ),
    Rule(
        kind="conflict",
        ingredient_a="NIACINAMIDE",
        ingredient_b="ASCORBIC ACID",
        severity="info",
        reason="고농도 동시 사용 시 nicotinic acid 생성 이론. 최근 연구는 대체로 문제 없음 (논쟁적).",
        reference="Cosmetic Dermatology reviews · 2024 update",
    ),
]


PREGNANCY_AVOID: list[Rule] = [
    Rule(
        kind="pregnancy", ingredient_a=name, ingredient_b=None,
        severity="avoid",
        reason=f"{name}는 임신 중 회피 권고 (기형 발생 위험 근거 or 예방적 회피).",
        reference="ACOG (American College of Obstetricians and Gynecologists) guidelines",
    )
    for name in [
        "RETINOL", "RETINALDEHYDE", "RETINYL PALMITATE", "TRETINOIN",
        "ADAPALENE", "TAZAROTENE",  # retinoid 계열
        "SALICYLIC ACID",             # BHA 고농도
        "HYDROQUINONE",                # 미백
    ]
]


EU_ALLERGENS_26 = [
    "LIMONENE", "LINALOOL", "CITRAL", "GERANIOL", "EUGENOL", "COUMARIN",
    "CITRONELLOL", "FARNESOL", "BENZYL ALCOHOL", "BENZYL BENZOATE",
    "BENZYL SALICYLATE", "BENZYL CINNAMATE", "CINNAMAL", "CINNAMYL ALCOHOL",
    "AMYLCINNAMAL", "AMYLCINNAMYL ALCOHOL", "HEXYL CINNAMAL",
    "HYDROXYCITRONELLAL", "HYDROXYISOHEXYL 3-CYCLOHEXENE CARBOXALDEHYDE",
    "ISOEUGENOL", "METHYL 2-OCTYNOATE", "ANISYL ALCOHOL", "ALPHA-ISOMETHYL IONONE",
    "EVERNIA PRUNASTRI EXTRACT", "EVERNIA FURFURACEA EXTRACT",
    "3-P-CUMENYL-2-METHYLPROPIONALDEHYDE",
]


def check_all_rules(
    inci_list: list[str],
    user_profile: dict,
) -> list[Rule]:
    """제품 성분 vs 룰 세트 → 관련 룰 리턴."""
    triggered = []
    upper_set = {name.upper() for name in inci_list}
    
    # 1. 상충 조합
    for rule in CONFLICT_RULES:
        if rule.ingredient_a in upper_set and (rule.ingredient_b or "") in upper_set:
            triggered.append(rule)
    
    # 2. 임신 계획
    if user_profile.get("pregnancy_planning", False):
        for rule in PREGNANCY_AVOID:
            if rule.ingredient_a in upper_set:
                triggered.append(rule)
    
    # 3. 사용자 알레르기 · 회피 성분
    for avoid in user_profile.get("allergies", []) + user_profile.get("avoid_ingredients", []):
        if avoid.upper() in upper_set:
            triggered.append(Rule(
                kind="user_avoid",
                ingredient_a=avoid.upper(),
                ingredient_b=None,
                severity="avoid",
                reason=f"사용자 자율 입력: {avoid} 회피 · 알레르기.",
                reference="user profile",
            ))
    
    # 4. EU 26 알레르기 유발 물질 (경고 수준)
    for allergen in EU_ALLERGENS_26:
        if allergen in upper_set:
            triggered.append(Rule(
                kind="allergen",
                ingredient_a=allergen,
                ingredient_b=None,
                severity="info",
                reason=f"EU 26 알레르기 유발 물질 목록 포함 (반드시 알레르기 발생은 아님).",
                reference="EU Regulation (EC) 1223/2009 Annex III",
            ))
    
    return triggered

import torch
from transformers import SegformerImageProcessor, SegformerForSemanticSegmentation


class FacialLesionSegmenter:
    """Segformer 기반 안면 병변 분할 (개념 예시).
    
    실전은 skin condition 특화 fine-tuned 모델 권장 (예: ISIC dermatology 벤치 fine-tune).
    이 예시는 자연 이미지 학습 baseline. 실제 서비스에서는 필수 재학습.
    """

    MODEL_ID = "nvidia/segformer-b0-finetuned-ade-512-512"  # 예시, 실제는 skin-specific

    def __init__(self, device: str = "cuda"):
        self.device = device
        self.processor = SegformerImageProcessor.from_pretrained(self.MODEL_ID)
        self.model = SegformerForSemanticSegmentation.from_pretrained(self.MODEL_ID).to(device).eval()

    @torch.no_grad()
    def segment(self, image: np.ndarray) -> dict:
        """RGB 이미지 → 병변 마스크 dict."""
        inputs = self.processor(images=image, return_tensors="pt").to(self.device)
        outputs = self.model(**inputs)
        logits = outputs.logits
        upsampled = torch.nn.functional.interpolate(
            logits, size=image.shape[:2], mode="bilinear", align_corners=False
        )
        pred_mask = upsampled.argmax(dim=1).squeeze().cpu().numpy()
        return {
            "mask": pred_mask,
            "class_ratios": {
                int(cls): float((pred_mask == cls).mean())
                for cls in np.unique(pred_mask)
            },
        }


LESION_TO_INGREDIENT_HINTS = {
    "acne": ["SALICYLIC ACID", "BENZOYL PEROXIDE", "NIACINAMIDE", "ZINC PCA"],
    "erythema": ["CENTELLA ASIATICA EXTRACT", "MADECASSOSIDE", "PANTHENOL"],
    "dryness": ["HYALURONIC ACID", "GLYCERIN", "CERAMIDE NP", "SQUALANE"],
    "hyperpigmentation": ["NIACINAMIDE", "ALPHA-ARBUTIN", "TRANEXAMIC ACID", "AZELAIC ACID"],
}

import anthropic


RECOMMEND_PROMPT = """당신은 화장품 성분·규제 전문가 어시스턴트입니다.
사용자 프로필과 후보 제품을 검토해 상위 3개를 자연어로 추천합니다.

사용자 프로필:
- 피부 타입: {skin_type}
- 관심 성분: {preferred}
- 회피 성분 · 알레르기: {avoid}
- 임신 계획: {pregnancy}
- 사용 이력: {history}

후보 제품 (INCI 상위 5 성분):
{products_formatted}

감지된 룰 (상충 · 알레르기 · 임신 회피):
{rules_formatted}

원칙:
1. 특정 브랜드 옹호 · 폄하 배제. 성분 · 물성 · 룰 근거만.
2. 각 추천에 짧은 이유(3~5문장): 어느 상위 성분이 사용자 목적에 맞는지 · 룰 감지 시 어떻게 대처했는지.
3. 반드시 하단에 "본 정보는 참고용이며 의료 조언이 아닙니다. 피부 문제는 피부과 전문의 상담을 권합니다." 포함.
4. 임신 계획이 있는 사용자에게 retinoid 관련 제품이 후보에 있으면 반드시 회피 권고 후 대체(bakuchiol 등) 제안.

응답 형식: Markdown.
"""


def format_products(products: list[dict]) -> str:
    lines = []
    for i, p in enumerate(products, 1):
        top5 = ", ".join(p["inci_list"][:5])
        lines.append(f"{i}. {p['brand']} / {p['name']}\n   상위 5 성분: {top5}")
    return "\n".join(lines)


def format_rules(rules: list[Rule]) -> str:
    if not rules:
        return "감지된 룰 없음."
    lines = []
    for r in rules:
        if r.ingredient_b:
            pair = f"{r.ingredient_a} × {r.ingredient_b}"
        else:
            pair = r.ingredient_a
        lines.append(f"- [{r.severity}] {pair}: {r.reason} (출처: {r.reference})")
    return "\n".join(lines)


def generate_recommendation(
    user_profile: dict,
    top_products: list[dict],
    all_rules: list[Rule],
    model: str = "claude-sonnet-4-5",
) -> str:
    """Claude가 최종 추천 자연어 요약."""
    client = anthropic.Anthropic()
    prompt = RECOMMEND_PROMPT.format(
        skin_type=user_profile.get("skin_type", ""),
        preferred=", ".join(user_profile.get("preferred", [])),
        avoid=", ".join(user_profile.get("avoid_ingredients", []) + user_profile.get("allergies", [])),
        pregnancy="예" if user_profile.get("pregnancy_planning", False) else "아니오",
        history="; ".join(user_profile.get("history", [])),
        products_formatted=format_products(top_products),
        rules_formatted=format_rules(all_rules),
    )
    resp = client.messages.create(
        model=model,
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}],
    )
    return resp.content[0].text

def full_advisor_pipeline(
    user_profile: dict,
    face_image: np.ndarray | None,
    product_db: list[Product],
    ingredient_embeddings: dict[str, np.ndarray],
    top_k: int = 10,
) -> dict:
    """사용자 프로필 + 안면 이미지 → 추천 제품 + 자연어 설명 + 룰 리포트."""
    # 1. 사용자 선호 성분 임베딩 평균
    preferred = [p.upper() for p in user_profile.get("preferred", [])]
    pref_embs = [ingredient_embeddings[p] for p in preferred if p in ingredient_embeddings]
    user_emb = np.mean(pref_embs, axis=0) if pref_embs else None
    
    # 2. 회피 성분 · 알레르기 · 임신 회피 성분 통합
    hard_avoid = {a.upper() for a in user_profile.get("avoid_ingredients", []) + user_profile.get("allergies", [])}
    if user_profile.get("pregnancy_planning", False):
        for rule in PREGNANCY_AVOID:
            hard_avoid.add(rule.ingredient_a)
    
    # 3. 후보 스코어링
    scored = []
    for prod in product_db:
        top5 = {i.upper() for i in prod.inci_list[:10]}
        # Hard avoid 있으면 즉시 배제
        if hard_avoid & top5:
            continue
        top5_embs = [
            ingredient_embeddings[i] for i in prod.inci_list[:5]
            if i in ingredient_embeddings
        ]
        if not top5_embs:
            continue
        prod_emb = np.mean(top5_embs, axis=0)
        if user_emb is not None:
            similarity = float(np.dot(prod_emb, user_emb) / (
                np.linalg.norm(prod_emb) * np.linalg.norm(user_emb) + 1e-8
            ))
        else:
            similarity = 0.5
        scored.append({
            "brand": prod.brand, "name": prod.name,
            "inci_list": prod.inci_list, "score": similarity,
            "categories": prod.categories, "image_url": prod.image_url,
        })
    
    # 4. 상위 K개 정렬
    scored.sort(key=lambda x: -x["score"])
    top_products = scored[:top_k]
    
    # 5. 룰 검사
    all_rules = []
    for prod in top_products:
        all_rules.extend(check_all_rules(prod["inci_list"], user_profile))
    
    # 6. 안면 병변 분석 (있으면)
    lesion_info = None
    if face_image is not None:
        segmenter = FacialLesionSegmenter()
        lesion_info = segmenter.segment(face_image)
    
    # 7. Claude 자연어 추천
    recommendation = generate_recommendation(user_profile, top_products, all_rules)
    
    return {
        "recommended_products": top_products,
        "triggered_rules": [r._asdict() for r in all_rules],
        "lesion_analysis": lesion_info,
        "recommendation_text": recommendation,
        "disclaimer": "본 정보는 참고용이며 의료 조언이 아닙니다. 피부 문제는 피부과 전문의 상담을 권합니다.",
    }


# 실행 예시 (페르소나 시나리오)
# user = {
#     "skin_type": "oily · acne-prone",
#     "preferred": ["NIACINAMIDE", "BAKUCHIOL", "ZINC PCA"],
#     "avoid_ingredients": ["ALCOHOL DENAT"],
#     "allergies": ["SALICYLIC ACID"],
#     "pregnancy_planning": True,
#     "history": ["니아신아마이드 3개월 효과 있음", "BHA 1개월 자극"],
# }
# result = full_advisor_pipeline(user, face_image=None, product_db=[...], ingredient_embeddings={...})
# print(result["recommendation_text"])