upload: bridge_models.py

bridge_models.py

@@ -0,0 +1,84 @@
+"""
+Pydantic models for Level Bridge Chat API.
+Strict JSON Schema -- MCP-wrappable without modification.
+"""
+
+from __future__ import annotations
+from typing import Optional
+from pydantic import BaseModel, Field, field_validator
+import base64
+
+
+class Metrics(BaseModel):
+    cvr: Optional[float] = Field(None, description="CVR (%)")
+    ctr: Optional[float] = Field(None, description="CTR (%)")
+    cpa: Optional[float] = Field(None, description="CPA (円)")
+
+    def has_any(self) -> bool:
+        return any(v is not None for v in [self.cvr, self.ctr, self.cpa])
+
+
+class DashboardContext(BaseModel):
+    campaign_name: Optional[str] = None
+    industry: Optional[str] = None
+    metrics: Optional[Metrics] = None
+    image_base64: Optional[str] = None
+
+    @field_validator("image_base64")
+    @classmethod
+    def validate_image_size(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return v
+        import os
+        max_mb = float(os.environ.get("MAX_IMAGE_SIZE_MB", "5"))
+        size_bytes = len(v.encode("utf-8")) * 3 / 4  # base64 -> bytes approx
+        if size_bytes > max_mb * 1024 * 1024:
+            raise ValueError(f"image_base64 exceeds {max_mb}MB limit")
+        return v
+
+
+class BridgeRequest(BaseModel):
+    session_id: Optional[str] = None
+    message: str = Field(default="", description="User message (can be empty on Turn 1)")
+    dashboard_context: Optional[DashboardContext] = None
+
+
+class NeededInfo(BaseModel):
+    key: str
+    label: str
+    example: str
+
+
+class BestNow(BaseModel):
+    summary: str
+    actions: list[str]
+    confidence: str = Field(..., pattern="^(low|mid|high)$")
+    reasoning_basis: list[str]
+
+
+class NextLevelPreview(BaseModel):
+    current_level: str
+    next_level: Optional[str]
+    needed_info: list[NeededInfo]
+    what_will_be_possible: list[str]
+    expected_impact: str
+
+
+class BridgeResponse(BaseModel):
+    ok: bool = True
+    session_id: str
+    turn_number: int
+    inferred_level: str
+    level_confidence: str = Field(..., pattern="^(low|mid|high)$")
+    level_reason: str
+    best_now: BestNow
+    next_level_preview: NextLevelPreview
+    follow_up_question: Optional[str] = None
+
+
+class BridgeErrorResponse(BaseModel):
+    ok: bool = False
+    session_id: Optional[str] = None
+    error_code: str
+    message: str
+    fallback: Optional[dict] = None