"""HTTP clients for search API, reranker, and DashScope chat (relevance labeling)."""
  
  from __future__ import annotations
  

  import io
  import json

  import logging
  import threading

  import time
  import uuid

  from typing import Any, Dict, List, Optional, Sequence, Tuple
  
  import requests
  

  from .constants import VALID_LABELS

  from .logging_setup import setup_eval_logging
  from .prompts import classify_prompt, intent_analysis_prompt

  from .utils import build_label_doc_line, extract_json_blob, safe_json_dumps
  

  _VERBOSE_LOGGER_LOCK = threading.Lock()
  _eval_llm_verbose_logger_singleton: logging.Logger | None = None
  _eval_llm_verbose_path_logged = False

  _TRANSIENT_HTTP_STATUS_CODES = frozenset({408, 425, 429, 500, 502, 503, 504})
  _client_log = logging.getLogger("search_eval.clients")

  
  
  def _get_eval_llm_verbose_logger() -> logging.Logger:

      """File logger for full LLM prompts/responses under ``search_evaluation.eval_log_dir/verbose/``."""
      from config.loader import get_app_config
  
      se = get_app_config().search_evaluation
      setup_eval_logging(se.eval_log_dir)

      global _eval_llm_verbose_logger_singleton, _eval_llm_verbose_path_logged
      with _VERBOSE_LOGGER_LOCK:
          if _eval_llm_verbose_logger_singleton is not None:
              return _eval_llm_verbose_logger_singleton

          log_path = se.eval_log_dir / "verbose" / "eval_verbose.log"

          log_path.parent.mkdir(parents=True, exist_ok=True)
          lg = logging.getLogger("search_eval.verbose_llm")
          lg.setLevel(logging.INFO)
          if not lg.handlers:
              handler = logging.FileHandler(log_path, encoding="utf-8")
              handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
              lg.addHandler(handler)
              lg.propagate = False
          _eval_llm_verbose_logger_singleton = lg
          if not _eval_llm_verbose_path_logged:
              _eval_llm_verbose_path_logged = True
              logging.getLogger("search_eval").info(
                  "LLM verbose I/O log (full prompt + response): %s",
                  log_path.resolve(),
              )
          return lg
  
  
  def _log_eval_llm_verbose(
      *,
      phase: str,
      model: str,
      prompt: str,
      assistant_text: str,
      raw_response: str,
  ) -> None:
      log = _get_eval_llm_verbose_logger()
      sep = "=" * 80
      log.info("\n%s", sep)
      log.info("phase=%s model=%s", phase, model)
      log.info("%s\nFULL PROMPT (user message)\n%s", sep, prompt)
      log.info("%s\nASSISTANT CONTENT (parsed)\n%s", sep, assistant_text)
      log.info("%s\nRAW RESPONSE (JSON string)\n%s", sep, raw_response)
      log.info("%s\n", sep)
  

  def _canonicalize_judge_label(raw: str) -> str | None:
      s = str(raw or "").strip().strip('"').strip("'")
      if s in VALID_LABELS:
          return s
      low = s.lower()
      for v in VALID_LABELS:
          if v.lower() == low:
              return v
      return None
  
  

  def _describe_request_exception(exc: requests.exceptions.RequestException) -> str:
      if isinstance(exc, requests.exceptions.HTTPError):
          response = getattr(exc, "response", None)
          if response is None:
              return str(exc)
          body = str(getattr(response, "text", "") or "").strip()
          if len(body) > 600:
              body = body[:600].rstrip() + "...[truncated]"
          return f"status={response.status_code} body={body or '<empty>'}"
      return str(exc)
  
  

  class SearchServiceClient:
      def __init__(self, base_url: str, tenant_id: str):
          self.base_url = base_url.rstrip("/")
          self.tenant_id = str(tenant_id)
          self.session = requests.Session()

          # Batch eval depends on live backend responses; tolerate brief restarts.
          self.retry_attempts = 45
          self.retry_delay_sec = 2.0
  
      @staticmethod
      def _is_transient_request_error(exc: requests.exceptions.RequestException) -> bool:
          if isinstance(exc, (requests.exceptions.ConnectionError, requests.exceptions.Timeout)):
              return True
          if isinstance(exc, requests.exceptions.HTTPError):
              response = getattr(exc, "response", None)
              if response is None:
                  return True
              return int(response.status_code) in _TRANSIENT_HTTP_STATUS_CODES
          return False
  
      def _request_json(
          self,
          method: str,
          path: str,
          *,
          timeout: float,
          headers: Optional[Dict[str, str]] = None,
          json_payload: Optional[Dict[str, Any]] = None,
      ) -> Dict[str, Any]:
          last_exc: requests.exceptions.RequestException | None = None
          url = f"{self.base_url}{path}"
          for attempt in range(1, self.retry_attempts + 1):
              try:
                  response = self.session.request(
                      method=method,
                      url=url,
                      headers=headers,
                      json=json_payload,
                      timeout=timeout,
                  )
                  response.raise_for_status()
                  return response.json()
              except requests.exceptions.RequestException as exc:
                  last_exc = exc
                  if not self._is_transient_request_error(exc) or attempt >= self.retry_attempts:
                      raise
                  _client_log.warning(
                      "Transient search-eval request failure, retrying (%s/%s): %s %s error=%s",
                      attempt,
                      self.retry_attempts,
                      method.upper(),
                      url,
                      exc,
                  )
                  time.sleep(self.retry_delay_sec)
          if last_exc is not None:
              raise last_exc
          raise RuntimeError(f"unexpected request retry state for {method.upper()} {url}")
  
      def get_json(self, path: str, *, timeout: float = 20) -> Dict[str, Any]:
          return self._request_json("GET", path, timeout=timeout)

      def search(self, query: str, size: int, from_: int = 0, language: str = "en", *, debug: bool = False) -> Dict[str, Any]:

          request_id = uuid.uuid4().hex[:8]

          payload: Dict[str, Any] = {
              "query": query,
              "size": size,
              "from": from_,
              "language": language,
          }
          if debug:
              payload["debug"] = True

          response = self._request_json(

              "POST",
              "/search/",

              timeout=120,

              headers={
                  "Content-Type": "application/json",
                  "X-Tenant-ID": self.tenant_id,
                  "X-Request-ID": request_id,
              },

              json_payload=payload,

          )

          response["_eval_request_id"] = request_id
          return response

  
  
  class RerankServiceClient:
      def __init__(self, service_url: str):
          self.service_url = service_url.rstrip("/")
          self.session = requests.Session()
  
      def rerank(self, query: str, docs: Sequence[str], normalize: bool = False, top_n: Optional[int] = None) -> Tuple[List[float], Dict[str, Any]]:
          payload: Dict[str, Any] = {
              "query": query,
              "docs": list(docs),
              "normalize": normalize,
          }
          if top_n is not None:
              payload["top_n"] = int(top_n)
          response = self.session.post(self.service_url, json=payload, timeout=180)
          response.raise_for_status()
          data = response.json()
          return list(data.get("scores") or []), dict(data.get("meta") or {})
  
  
  class DashScopeLabelClient:

      """DashScope OpenAI-compatible chat: synchronous or Batch File API (JSONL job).
  
      Batch flow: https://help.aliyun.com/zh/model-studio/batch-interfaces-compatible-with-openai/

  
      Some regional endpoints (e.g. ``dashscope-us`` compatible-mode) do not implement ``/batches``;
      on HTTP 404 from batch calls we fall back to synchronous ``/chat/completions`` and stop using batch
      for subsequent requests on this client.

      """
  
      def __init__(
          self,
          model: str,
          base_url: str,
          api_key: str,
          batch_size: int = 40,
          *,
          batch_completion_window: str = "24h",
          batch_poll_interval_sec: float = 10.0,
          enable_thinking: bool = True,

          use_batch: bool = False,

      ):

          self.model = model
          self.base_url = base_url.rstrip("/")
          self.api_key = api_key
          self.batch_size = int(batch_size)

          self.batch_completion_window = str(batch_completion_window)
          self.batch_poll_interval_sec = float(batch_poll_interval_sec)
          self.enable_thinking = bool(enable_thinking)
          self.use_batch = bool(use_batch)

          self.session = requests.Session()

          self.retry_attempts = 4
          self.retry_delay_sec = 3.0

      def _auth_headers(self) -> Dict[str, str]:
          return {"Authorization": f"Bearer {self.api_key}"}
  
      def _completion_body(self, prompt: str) -> Dict[str, Any]:
          body: Dict[str, Any] = {
              "model": self.model,
              "messages": [{"role": "user", "content": prompt}],
              "temperature": 0,
              "top_p": 0.1,
              "enable_thinking": self.enable_thinking,
          }
          return body
  
      def _chat_sync(self, prompt: str) -> Tuple[str, str]:

          response = self.session.post(
              f"{self.base_url}/chat/completions",

              headers={**self._auth_headers(), "Content-Type": "application/json"},
              json=self._completion_body(prompt),

              timeout=180,
          )
          response.raise_for_status()
          data = response.json()
          content = str(((data.get("choices") or [{}])[0].get("message") or {}).get("content") or "").strip()
          return content, safe_json_dumps(data)
  

      def _chat_batch(self, prompt: str) -> Tuple[str, str]:
          """One chat completion via Batch File API (single-line JSONL job)."""
          custom_id = uuid.uuid4().hex
          body = self._completion_body(prompt)
          line_obj = {
              "custom_id": custom_id,
              "method": "POST",
              "url": "/v1/chat/completions",
              "body": body,
          }
          jsonl = json.dumps(line_obj, ensure_ascii=False, separators=(",", ":")) + "\n"
          auth = self._auth_headers()
  
          up = self.session.post(
              f"{self.base_url}/files",
              headers=auth,
              files={
                  "file": (
                      "eval_batch_input.jsonl",
                      io.BytesIO(jsonl.encode("utf-8")),
                      "application/octet-stream",
                  )
              },
              data={"purpose": "batch"},
              timeout=300,
          )
          up.raise_for_status()
          file_id = (up.json() or {}).get("id")
          if not file_id:
              raise RuntimeError(f"DashScope file upload returned no id: {up.text!r}")
  
          cr = self.session.post(
              f"{self.base_url}/batches",
              headers={**auth, "Content-Type": "application/json"},
              json={
                  "input_file_id": file_id,
                  "endpoint": "/v1/chat/completions",
                  "completion_window": self.batch_completion_window,
              },
              timeout=120,
          )
          cr.raise_for_status()
          batch_payload = cr.json() or {}
          batch_id = batch_payload.get("id")
          if not batch_id:
              raise RuntimeError(f"DashScope batches.create returned no id: {cr.text!r}")
  
          terminal = frozenset({"completed", "failed", "expired", "cancelled"})
          batch: Dict[str, Any] = dict(batch_payload)
          status = str(batch.get("status") or "")
          while status not in terminal:
              time.sleep(self.batch_poll_interval_sec)
              br = self.session.get(f"{self.base_url}/batches/{batch_id}", headers=auth, timeout=120)
              br.raise_for_status()
              batch = br.json() or {}
              status = str(batch.get("status") or "")
  
          if status != "completed":
              raise RuntimeError(
                  f"DashScope batch {batch_id} ended with status={status!r} errors={batch.get('errors')!r}"
              )
  
          out_id = batch.get("output_file_id")
          err_id = batch.get("error_file_id")
  
          row = self._find_batch_line_for_custom_id(out_id, custom_id, auth)
          if row is None:
              err_row = self._find_batch_line_for_custom_id(err_id, custom_id, auth)
              if err_row is not None:
                  raise RuntimeError(f"DashScope batch request failed: {err_row!r}")
              raise RuntimeError(f"DashScope batch output missing custom_id={custom_id!r}")
  
          resp = row.get("response") or {}
          sc = resp.get("status_code")
          if sc is not None and int(sc) != 200:
              raise RuntimeError(f"DashScope batch line error: {row!r}")
  
          data = resp.get("body") or {}
          content = str(((data.get("choices") or [{}])[0].get("message") or {}).get("content") or "").strip()
          return content, safe_json_dumps(row)
  

      def _chat(self, prompt: str, *, phase: str = "chat") -> Tuple[str, str]:

          last_exc: Exception | None = None
          for attempt in range(1, self.retry_attempts + 1):

              try:

                  if not self.use_batch:

                      content, raw = self._chat_sync(prompt)
                  else:

                      try:
                          content, raw = self._chat_batch(prompt)
                      except requests.exceptions.HTTPError as e:
                          resp = getattr(e, "response", None)
                          if resp is not None and resp.status_code == 404:
                              self.use_batch = False
                              content, raw = self._chat_sync(prompt)
                          else:
                              raise
                  break
              except Exception as exc:
                  last_exc = exc
                  is_request_error = isinstance(exc, requests.exceptions.RequestException)

                  is_transient = is_request_error and self._is_transient_request_error(exc)
                  if not is_transient or attempt >= self.retry_attempts:

                      raise

                  _client_log.warning(
                      "Transient DashScope error, retrying (%s/%s): phase=%s model=%s use_batch=%s error=%s",
                      attempt,
                      self.retry_attempts,
                      phase,
                      self.model,
                      self.use_batch,

                      _describe_request_exception(exc),

                  )
                  time.sleep(self.retry_delay_sec)
          else:
              if last_exc is not None:
                  raise last_exc
              raise RuntimeError(f"unexpected DashScope retry state for phase={phase}")

          _log_eval_llm_verbose(
              phase=phase,
              model=self.model,
              prompt=prompt,
              assistant_text=content,
              raw_response=raw,
          )
          return content, raw

      @staticmethod
      def _is_transient_request_error(exc: requests.exceptions.RequestException) -> bool:
          if isinstance(exc, (requests.exceptions.ConnectionError, requests.exceptions.Timeout)):
              return True
          if isinstance(exc, requests.exceptions.HTTPError):
              response = getattr(exc, "response", None)
              if response is None:
                  return True
              return int(response.status_code) in _TRANSIENT_HTTP_STATUS_CODES
          return False
  

      def _find_batch_line_for_custom_id(
          self,
          file_id: Optional[str],
          custom_id: str,
          auth: Dict[str, str],
      ) -> Optional[Dict[str, Any]]:
          if not file_id or str(file_id) in ("null", ""):
              return None
          r = self.session.get(f"{self.base_url}/files/{file_id}/content", headers=auth, timeout=300)
          r.raise_for_status()
          for raw in r.text.splitlines():
              raw = raw.strip()
              if not raw:
                  continue
              try:
                  obj = json.loads(raw)
              except json.JSONDecodeError:
                  continue
              if str(obj.get("custom_id")) == custom_id:
                  return obj
          return None
  

      def query_intent(self, query: str) -> Tuple[str, str]:
          prompt = intent_analysis_prompt(query)
          return self._chat(prompt, phase="query_intent")
  

      def classify_batch(

          self,
          query: str,
          docs: Sequence[Dict[str, Any]],

          *,
          query_intent_block: str = "",

      ) -> Tuple[List[str], str]:
          numbered_docs = [build_label_doc_line(idx + 1, doc) for idx, doc in enumerate(docs)]

          prompt = classify_prompt(query, numbered_docs, query_intent_block=query_intent_block)
          content, raw_response = self._chat(prompt, phase="relevance_classify")

          labels: List[str] = []

          for line in str(content or "").splitlines():

              canon = _canonicalize_judge_label(line)
              if canon is not None:
                  labels.append(canon)

          if len(labels) != len(docs):
              payload = extract_json_blob(content)
              if isinstance(payload, dict) and isinstance(payload.get("labels"), list):
                  labels = []
                  for item in payload["labels"][: len(docs)]:
                      if isinstance(item, dict):

                          raw_l = str(item.get("label") or "").strip()

                      else:

                          raw_l = str(item).strip()
                      canon = _canonicalize_judge_label(raw_l)
                      if canon is not None:
                          labels.append(canon)

          if len(labels) != len(docs) or any(label not in VALID_LABELS for label in labels):

              raise ValueError(f"unexpected classify output: {content!r}")

          return labels, raw_response