"""FabriX(사내 LLM, OpenAI 호환 Serving) 접속 규칙 — 순수 함수만. Django 안 봄. ABAP_OPENCODE `apps_ito/aaf` 의 passthrough 중계에서 CodeAssist 에 필요한 규칙만 옮겼음: - 와이어는 OpenAI chat/completions 그대로. 인증은 Authorization 이 아니라 커스텀 헤더 3종. - 실제 모델은 body 의 model 이 아니라 `x-llm-model-id` 헤더가 고름. body model 은 고정 경로. - 고객사 게이트웨이가 토큰 형식(Bearer/날것)·클라이언트 헤더 이름을 서버마다 다르게 받아서, 401 이면 다른 조합으로 한 번씩 더 찔러보고 통과한 조합을 기억함(2026-09-16 ABAP 쪽 실측). """ from __future__ import annotations from dataclasses import dataclass, field, replace DEFAULT_BODY_MODEL = "/mnt/models" # body.model — 공식 예제 고정값. 모델 선택은 헤더가 함 TOKEN_HDR = "x-openapi-token" EMAIL_HDR = "x-generative-ai-user-email" MODEL_HDR = "x-llm-model-id" CLIENT_HDRS = ("x-generative-ai-client", "x-fabrix-client") # 고객사는 앞 것만 받음(뒤는 401) ENV_KEYS = ( "AAF_FABRIX_BASE_URL", # …/openapi/llm 까지. 뒤에 /chat/completions 붙임 "AAF_FABRIX_MODEL_ID", # 기본 모델 id (x-llm-model-id) "AAF_FABRIX_MODELS", # "605:Gemma4,339:GaussO Flash,581:GaussO Think" — 화면에서 고를 수 있는 목록 "AAF_FABRIX_CLIENT_KEY", "AAF_FABRIX_OPENAPI_TOKEN", "AAF_FABRIX_USER_EMAIL", "AAF_FABRIX_CLIENT_HEADER", # 기본 x-generative-ai-client "AAF_FABRIX_TOKEN_PREFIX", # "bearer" 면 토큰 앞에 "Bearer " 붙임. 기본은 날것 "AAF_FABRIX_MODEL", # body.model. 기본 /mnt/models "AAF_FABRIX_MAX_TOKENS", # 있으면 max_completion_tokens 로 실음 "AAF_RELAY_STREAM_USAGE", # "1" 이면 stream_options.include_usage — 제공자가 거부하면 끔 "AAF_GATEWAY_KEY", # OpenCode 가 Authorization: Bearer 로 보내는 키. 비우면 검사 안 함 ) def parse_models(spec: str) -> dict[str, str]: """"339:GaussO Flash,581:GaussO Think" → {"339": "GaussO Flash", "581": "GaussO Think"}.""" out: dict[str, str] = {} for item in (x.strip() for x in (spec or "").split(",")): if not item: continue mid, _, name = item.partition(":") if mid.strip(): out[mid.strip()] = name.strip() or mid.strip() return out def token_value(token: str, prefix: str = "") -> str: """x-openapi-token 값. 저장값에 Bearer 가 붙어 있어도 떼고, prefix=bearer 일 때만 다시 붙임.""" t = (token or "").strip() if t.lower().startswith("bearer "): t = t[7:].strip() return f"Bearer {t}" if t and prefix.strip().lower() == "bearer" else t @dataclass(frozen=True) class FabrixConfig: base_url: str = "" model_id: str = "" client_key: str = "" token: str = "" email: str = "" client_header: str = CLIENT_HDRS[0] token_prefix: str = "" body_model: str = DEFAULT_BODY_MODEL models: dict[str, str] = field(default_factory=dict) max_tokens: int | None = None stream_usage: bool = False gateway_key: str = "" connect_timeout_s: float = 10.0 read_timeout_s: float = 120.0 # 스트림 조각 사이 무수신 한계 total_timeout_s: float = 600.0 @classmethod def from_env(cls, env: dict[str, str]) -> "FabrixConfig": g = lambda k: (env.get(k) or "").strip() # noqa: E731 max_tokens = g("AAF_FABRIX_MAX_TOKENS") return cls( base_url=g("AAF_FABRIX_BASE_URL").rstrip("/"), model_id=g("AAF_FABRIX_MODEL_ID"), client_key=g("AAF_FABRIX_CLIENT_KEY"), token=g("AAF_FABRIX_OPENAPI_TOKEN"), email=g("AAF_FABRIX_USER_EMAIL"), client_header=g("AAF_FABRIX_CLIENT_HEADER") or CLIENT_HDRS[0], token_prefix=g("AAF_FABRIX_TOKEN_PREFIX"), body_model=g("AAF_FABRIX_MODEL") or DEFAULT_BODY_MODEL, models=parse_models(g("AAF_FABRIX_MODELS")), max_tokens=int(max_tokens) if max_tokens.isdigit() else None, stream_usage=g("AAF_RELAY_STREAM_USAGE") == "1", gateway_key=g("AAF_GATEWAY_KEY"), ) def missing(self) -> list[str]: """비어 있으면 안 되는 설정 이름. 비어 있는 채로 부르면 정체불명 401/404 라 미리 막음.""" need = { "AAF_FABRIX_BASE_URL": self.base_url, "AAF_FABRIX_MODEL_ID": self.model_id, "AAF_FABRIX_CLIENT_KEY": self.client_key, "AAF_FABRIX_OPENAPI_TOKEN": self.token, } return [k for k, v in need.items() if not v] @property def url(self) -> str: return self.base_url.rstrip("/") + "/chat/completions" def model_list(self) -> dict[str, str]: """화면에 보여줄 모델 목록. AAF_FABRIX_MODELS 없으면 기본 모델 하나.""" return dict(self.models) or {self.model_id: "FabriX"} def pick_model_id(self, requested: str) -> str: """OpenCode 가 보낸 model 이 허용 목록에 있으면 그것, 아니면 기본 모델.""" rm = (requested or "").strip() return rm if rm and rm in self.models else self.model_id def headers(self, model_id: str | None = None) -> dict[str, str]: return { "Content-Type": "application/json", MODEL_HDR: model_id or self.model_id, self.client_header: self.client_key, TOKEN_HDR: token_value(self.token, self.token_prefix), EMAIL_HDR: self.email, # 빈 값이라도 실음 — 공식 예제가 그럼 } def prepare(self, payload: dict) -> tuple[dict, dict[str, str]]: """OpenCode 가 보낸 OpenAI 요청 → FabriX 로 보낼 (body, headers).""" body = dict(payload) headers = self.headers(self.pick_model_id(str(payload.get("model") or ""))) body["model"] = self.body_model if body.get("stream") and self.stream_usage: so = dict(body.get("stream_options") or {}) so["include_usage"] = True body["stream_options"] = so if self.max_tokens: body.setdefault("max_completion_tokens", self.max_tokens) return body, headers def with_models(self, models: dict[str, str]) -> "FabrixConfig": return replace(self, models=models) Variant = tuple[str, str | None] # (토큰 접두 "bearer"|"raw", 클라이언트 헤더 이름) def auth_variants(headers: dict[str, str], last_ok: Variant | None = None) -> list[tuple[Variant, dict[str, str]]]: """설정된 헤더에서 시작해 (Bearer/날것) × (클라이언트 헤더 이름) 조합을 만듦. 순서: 마지막에 통과한 조합 → 설정된 조합 → 나머지. 토큰이 없으면 원본 하나만.""" tok = headers.get(TOKEN_HDR) if not tok: return [(("raw", None), headers)] raw = tok[7:].strip() if tok.lower().startswith("bearer ") else tok.strip() client_val = next((headers[h] for h in CLIENT_HDRS if h in headers), None) client_names: tuple[str | None, ...] = CLIENT_HDRS if client_val is not None else (None,) variants: dict[Variant, dict[str, str]] = {} for prefix in ("bearer", "raw"): for chdr in client_names: h = {k: v for k, v in headers.items() if k not in CLIENT_HDRS} h[TOKEN_HDR] = f"Bearer {raw}" if prefix == "bearer" else raw if chdr: h[chdr] = client_val or "" variants[(prefix, chdr)] = h current: Variant = ( "bearer" if tok.lower().startswith("bearer ") else "raw", next((h for h in CLIENT_HDRS if h in headers), None), ) order: list[Variant] = [current] if last_ok and last_ok in variants and last_ok != current: order.insert(0, last_ok) order += [k for k in variants if k not in order] return [(k, variants[k]) for k in order]