"""Opik REST 에서 trace 를 시간 창으로 걷어온다 (읽기 전용)."""
import json
import logging
import urllib.parse
import urllib.request
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)

PAGE_SIZE = 200


def _fetch_paged(
    kind: str,
    base_url: str,
    workspace: str,
    project_name: str,
    since: str,
    until: str = "9999",
) -> List[Dict[str, Any]]:
    """`traces`/`spans` 공통 페이지 순회 — 기간 자르기 규칙은 하나뿐이다.

    start_time 이 [since, until) 인 행 전부 (최신순 페이지 순회).
    base_url 예: http://192.168.133.87:5173/api
    since/until: ISO 문자열 비교 (Opik start_time 은 ISO·UTC — 문자열
    비교가 시간 순서와 일치한다).

    ★**여기서 거르지 않는다.** 시험 기록 판별은 span 과 부모 trace 를
    **함께 봐야** 하므로(`select_call_rows`) 이 함수는 raw 로 준다.
    """
    traces: List[Dict[str, Any]] = []
    page = 1
    while True:
        q = urllib.parse.urlencode({
            "project_name": project_name,
            "size": PAGE_SIZE,
            "page": page,
            "sorting": json.dumps(
                [{"field": "start_time", "direction": "DESC"}]),
        })
        req = urllib.request.Request(
            f"{base_url}/v1/private/{kind}?{q}",
            headers={"Comet-Workspace": workspace})
        with urllib.request.urlopen(req, timeout=60) as r:
            d = json.loads(r.read())
        content = d.get("content", [])
        if not content:
            break
        stop = False
        for t in content:
            st = t.get("start_time", "")
            if st < since:
                stop = True
                break
            if st < until:
                traces.append(t)
        if stop or page * PAGE_SIZE >= int(d.get("total", 0)):
            break
        page += 1
    logger.info("%s %d건 수집 (since=%s)", kind, len(traces), since)
    return traces


def fetch_traces(
    base_url: str,
    workspace: str,
    project_name: str,
    since: str,
    until: str = "9999",
) -> List[Dict[str, Any]]:
    """v1 호환 — 기존 소비처가 이것을 부른다."""
    return _fetch_paged("traces", base_url, workspace, project_name,
                        since, until)


def _step_tag(row: Dict[str, Any]) -> Optional[str]:
    """`step:` 축 태그만 — 없으면 None. 이름으로 안 떨어진다."""
    for tag in (row.get("tags") or []):
        if isinstance(tag, str) and tag.startswith("step:"):
            return tag[len("step:"):]
    return None


def step_of(trace: Dict[str, Any]) -> str:
    """trace/span 의 스텝 이름.

    1순위 = `step:` 축 태그(v2). 2순위 = litellm 메타의 `trace_name` 마지막
    조각(v1 — 지난 자료를 계속 읽기 위해). 3순위 = 이름.
    """
    tag = _step_tag(trace)
    if tag:
        return tag
    md = trace.get("metadata") or {}
    tn = md.get("trace_name") if isinstance(md, dict) else None
    if isinstance(tn, str) and ">" in tn:
        return tn.split(">")[-1].strip()
    if isinstance(tn, str) and tn:
        return tn
    return str(trace.get("name") or "?")


def step_of_call(span: Dict[str, Any],
                 trace_index: Dict[str, Dict[str, Any]]) -> str:
    """호출(span) 하나의 스텝 이름.

    ① 자기 `step:` 태그 (v2)
    ② 부모 trace 의 스텝 (v1 — `{step}/{model}` 이름이나 `metadata.trace_name`)
    ③ 그래도 없으면 span 이름

    ★②를 ③보다 먼저 본다. `step_of` 를 먼저 부르면 span 이름으로 떨어져
    부모를 영영 안 본다 — litellm span 이름(`{model}_{obj}_{created}`)은
    **절대 비지 않기** 때문이다.
    """
    tag = _step_tag(span)
    if tag:
        return tag
    parent = trace_index.get(str(span.get("trace_id")))
    if parent is not None:
        ptag = _step_tag(parent)
        if ptag:
            return ptag
        pname = step_of(parent)
        if pname and pname != "?":
            return pname
    return step_of(span)


#: 시험이 쓰는 가짜 신원 — 실제 프로젝트 UUID 가 아니다.
#: 늘려야 할 일이 생기면 여기에 보탠다(코드가 아니라 목록이라 안전하다).
_TEST_PROJECT_IDS = frozenset({"p-i2i", "p", "proj-1", "p1", "test"})


def is_test_origin(row: Dict[str, Any]) -> bool:
    """시험이 남긴 기록인가 — 감사 셈에서 뺀다.

    2026-08-23 이전에는 돈 가드가 집 안 주소를 통과시켜 시험이 프로덕션
    프로젝트에 썼다. 지우지 않고 여기서 거른다 — 지우는 것은 되돌릴 수 없다.

    ★**이 그물은 완전하지 않다.** 두 갈래로 잡는다:

    1. 신원 없이 시간 0 — `record_provider_call` 을 scope 밖에서 부른 시험
    2. **가짜 신원** — `generation_context("p-i2i", …)` 안에서 부르면
       `project_id="p-i2i"` 가 실려 1번 그물을 **통과한다**

    그래도 못 잡는 것이 남는다(litellm 경로를 타는 시험 등). 그래서 리포트에
    **「걸러진 수」와 함께 「이 그물이 못 잡는 모양」**을 같이 적는다 —
    「0 건」이 「없다」로 읽히면 안 된다.
    """
    md = row.get("metadata") or {}
    if not isinstance(md, dict):
        return False
    pid = md.get("project_id")
    # ② 가짜 신원 — 실제 프로젝트 id 는 UUID 다
    if isinstance(pid, str) and pid in _TEST_PROJECT_IDS:
        return True
    # ① 신원 없이 시간 0 (이미지 기록 모양일 때만)
    if "step" not in md or "duration_ms" not in md:
        return False
    return md.get("duration_ms") == 0 and not pid


def test_filter_caveat() -> str:
    """리포트에 함께 실을 한 줄 — 이 그물의 한계를 밝힌다."""
    return ("★시험 기록 필터는 완전하지 않다 — 가짜 신원 목록에 없는 값이나 "
            "litellm 경로를 탄 시험 기록은 통과한다. 「걸러진 수」를 "
            "「오염 총량」으로 읽지 말 것.")


def select_call_rows(spans: List[Dict[str, Any]],
                     trace_index: Dict[str, Dict[str, Any]]
                     ) -> List[Dict[str, Any]]:
    """감사 대상 span 만 고른다 — **자기 또는 부모가 시험이면 뺀다**.

    span metadata 에는 `step` 이 없어서(`image_tracer` 가 trace 에만 넣는다)
    span 만 보면 시험 판별이 안 된다. 부모 trace 에는 있다. **둘을 함께 본다.**
    안 그러면 부모만 지워지고 자식 span 이 새 감사 SOT 에 그대로 들어가
    「시험 기록 0건」이 거짓 통과한다.

    ★부모를 못 찾은 span 은 **버리지 않는다** — 못 찾은 것과 시험인 것은 다르다.
    """
    out: List[Dict[str, Any]] = []
    for sp in spans:
        if is_test_origin(sp):
            continue
        parent = trace_index.get(str(sp.get("trace_id")))
        if parent is not None and is_test_origin(parent):
            continue
        out.append(sp)
    return out


def fetch_spans(
    base_url: str,
    workspace: str,
    project_name: str,
    since: str,
    until: str = "9999",
) -> List[Dict[str, Any]]:
    """감사 대상 = **호출 하나당 한 행**. 그것은 언제나 span 이다. (raw)

    ★span 과 trace 를 그냥 합치면 **지난 자료를 두 번 센다.**
    `ImageTracer.log` 는 호출 하나마다 trace 를 만들고 그 밑에 span 도 만든다.
    litellm 도 `trace_id` 가 없으면 trace 를 만들고 span 은 **항상** 만든다.
    즉 v1 호출 한 건이 **trace 1 + span 1** 로 남아 있다.

    두 배가 된 숫자로는 「팩이 얼마나 커졌나」를 못 묻는다 — 이 도구의 존재
    이유가 사라진다. 그래서 **세는 것은 언제나 span** 이고 trace 는
    묶음·이름을 얻는 데만 쓴다(`fetch_trace_index`).

    ★인자·기간 계약은 `fetch_traces` 와 **같게** 둔다. 다르면 소비처가
    조용히 전 기간을 센다.
    """
    return _fetch_paged("spans", base_url, workspace, project_name,
                        since, until)


def fetch_trace_index(
    base_url: str,
    workspace: str,
    project_name: str,
    since: str,
    until: str = "9999",
    *,
    pad_hours: int = 24,
) -> Dict[str, Dict[str, Any]]:
    """`trace_id` → trace. **세는 데는 안 쓴다** — 묶음·이름·시험 판별용.

    ★창을 `pad_hours` 만큼 **앞으로 넓힌다.** 창 경계 직전에 시작한 긴 trace 의
    창 안 span 이 부모를 잃으면, 시험 판별도 legacy 스텝 fallback 도 함께
    깨진다. 부모를 못 찾는 것이 여기서는 가장 나쁘다.
    """
    import datetime as _dt

    try:
        _s = _dt.datetime.fromisoformat(since.replace("Z", "+00:00"))
        wide = (_s - _dt.timedelta(hours=pad_hours)).isoformat()
    except ValueError:
        wide = since          # 형식이 다르면 그대로 — 좁히지는 않는다
    return {t["id"]: t for t in _fetch_paged(
        "traces", base_url, workspace, project_name, wide, until)
        if t.get("id")}


def messages_of(trace: Dict[str, Any]) -> Dict[str, str]:
    """input 에서 role 별 텍스트 결합 (system/user). 비텍스트 파트는
    JSON 직렬화 길이로만 세지 않고 문자열화해 포함한다."""
    out = {"system": "", "user": ""}
    inp = trace.get("input")
    if isinstance(inp, list):
        for m in inp:
            if not isinstance(m, dict):
                continue
            role = m.get("role")
            if role not in out:
                continue
            c = m.get("content")
            out[role] += c if isinstance(c, str) else json.dumps(
                c, ensure_ascii=False)
    elif isinstance(inp, dict):
        # record_provider_call 계열 — prompt 필드가 본문이다.
        p = inp.get("prompt")
        if isinstance(p, str):
            out["user"] = p
    return out