""" Ken's Tech Translator — Production Backend Secrets stay server-side. Prompt behavior is fully dynamic. Compatible with Azure OpenAI Responses API. Deploy target: HuggingFace Spaces (Docker) """ import os import logging import httpx from pathlib import Path from fastapi import FastAPI, Request from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse, JSONResponse # ── Logging ─────────────────────────────────────────────────────────────────── logging.basicConfig(level=logging.INFO) logger = logging.getLogger("ken-translator") BASE_DIR = Path(__file__).resolve().parent app = FastAPI(title="Ken's Tech Translator", version="2.0.0") # ══════════════════════════════════════════════════════════════════════════════ # SECRETS # ══════════════════════════════════════════════════════════════════════════════ AZURE_API_KEY = os.environ.get("AZURE_API_KEY", "") AZURE_ENDPOINT = os.environ.get( "AZURE_ENDPOINT", "https://kendrickfilbert-9824-resource.cognitiveservices.azure.com/openai/responses?api-version=2025-04-01-preview" ) MODEL_DEPLOYMENT = os.environ.get("MODEL_DEPLOYMENT", "gpt-5.3-chat-ken") # ══════════════════════════════════════════════════════════════════════════════ # PROMPT LIBRARY # ══════════════════════════════════════════════════════════════════════════════ BASE_SYSTEM_PROMPT = """ You are Ken's Tech Translator — a senior technical communication engine. Your job is to convert raw engineering content into clear, structured updates tailored precisely to the audience, tone, and features specified below. ## FEATURE CONFLICT RESOLUTION (MANDATORY) If Slack-Ready Format is enabled: - Override formatting ONLY (not content requirements) - Convert all required content into Slack format If multiple features conflict: Priority order: 1. Slack Format 2. Urgent Tone 3. Audience rules 4. Other features ## CORE RULES - Never hallucinate or invent data not present in the source text. - Never ignore the AUDIENCE, TONE, or FEATURES instructions below — they are mandatory. - Always include the REQUIRED sections defined for the selected audience. - Do not omit required sections under any condition. - All output must be in Markdown. - Be concise. Eliminate filler. Every sentence must serve the reader. """.strip() AUDIENCE_PROMPTS = { "executive": """ ## AUDIENCE: C-Suite Executive Write for a time-poor executive who cares ONLY about business outcomes. - Lead with impact, not process. - Quantify risk, cost, or opportunity wherever possible. - Use no more than 3–4 short sections total. - Avoid all technical jargon. If unavoidable, define it inline in one clause. - REQUIRED sections: Summary (2 sentences max), Business Context, Suggested Next Steps. - OMIT: Technical Details, Glossary (unless a term is business-critical). """, "pm": """ ## AUDIENCE: Project Manager / Tech Lead Write for someone managing execution who needs scope, timelines, and owners. - Focus on what needs to happen, by when, and who owns it. - Call out blockers, dependencies, and risks explicitly. - REQUIRED sections: Summary, Overview, Business Context (timeline + scope focus), Technical Details (architectural context), Suggested Next Steps (detailed What/Who/Priority). """, "sales": """ ## AUDIENCE: Sales / Client-Facing Write for someone presenting value to a customer or external stakeholder. - Frame everything as customer benefit, competitive advantage, or risk mitigation. - Avoid internal jargon and infrastructure details entirely. - Use confident, positive framing. - REQUIRED sections: Summary, Overview (benefits-first), Business Context (implications + timeline), Suggested Next Steps (client-action focused). """, "general": """ ## AUDIENCE: General (Mixed / Anyone) Write for a non-technical reader with no assumed background. - Use plain English and everyday analogies. - Define every technical term on first use. - Keep sentences short. Use bullet points for lists. - REQUIRED sections: All standard sections. Glossary is mandatory. """, } TONE_PROMPTS = { "professional": """ ## TONE: Professional Clear, neutral, business-standard language. Structured paragraphs. Formal but not stiff. """, "casual": """ ## TONE: Casual Friendly Write like a knowledgeable colleague explaining over coffee. - Use conversational language and contractions (e.g., "it's", "doesn't") - Avoid report-style phrasing - Keep it natural and human - Reduce repetition across sections """, "urgent": """ ## TONE: Urgent / Escalation This content requires immediate attention. Open with impact and risk. Use direct, action-forcing language. Bold the most critical points. Keep it tight — no preamble or pleasantries. """, "educational": """ ## TONE: Educational Write to teach, not just inform. Explain the "why" behind decisions. Use the Glossary section fully. Build from simple to complex. Include at least one analogy. """, } FEATURE_PROMPTS = { "layered": """ ## FEATURE: Layered Output (TL;DR → Deep) You MUST begin your ENTIRE response with a "⚡ TL;DR" block — 1–2 sentences, plain English, zero jargon. Place it before all other sections. """, "business_impact": """ ## FEATURE: Business Impact You MUST include an explicit "💰 Business Impact" subsection inside Business Context. Cover: estimated cost/risk/ROI if quantifiable, operational disruption, revenue implication. If numbers are absent from source, state qualitative risk level (High/Medium/Low) with reasoning. """, "analogies": """ ## FEATURE: Analogies & Examples You MUST include at least one concrete real-world analogy inside the Overview section. Format it exactly as: 💡 Analogy: [your analogy here] """, "action_items": """ ## FEATURE: Action Items The Suggested Next Steps section MUST use this exact Markdown table format: | Action | Owner | Priority | |--------|-------|----------| | [What to do] | [Suggested role] | Address Promptly / Upcoming Sprint / Backlog | Include a minimum of 2 rows and a maximum of 5 rows. """, "confidence": """ ## FEATURE: Confidence Notes You MUST add a final "🔍 Confidence Notes" section after all other sections. Rate confidence for each major claim: - ✅ High — clearly stated in source text - 🟡 Medium — reasonably inferred - 🔴 Low — assumed due to missing context Flag any ambiguous or missing information explicitly. """, "slack": """ ## FEATURE: Slack-Ready Format Reformat your ENTIRE output as a Slack message. This overrides all other formatting rules. Rules: - Maximum 300 words total. - Use *bold* for emphasis (Slack syntax), not Markdown headers. - Use emoji bullets (✅ 🔴 📌 ⚠️) instead of dashes or numbers. - Flat structure only — no nested headers or subsections. - End with exactly one line: 📌 *Action needed:* [single most important next step] """, } # ── Feature key normalization map (frontend label → internal key) ───────────── FEATURE_KEY_MAP = { "layered": "layered", "layered (tl;dr → deep)": "layered", "business_impact": "business_impact", "business impact": "business_impact", "analogies": "analogies", "analogies & examples": "analogies", "action_items": "action_items", "action items": "action_items", "confidence": "confidence", "confidence notes": "confidence", "slack": "slack", "slack-ready format": "slack", } # ── Audience key normalization map ──────────────────────────────────────────── AUDIENCE_KEY_MAP = { "executive": "executive", "c-suite": "executive", "csuite": "executive", "pm": "pm", "pm / lead": "pm", "lead": "pm", "sales": "sales", "sales / client": "sales", "client": "sales", "general": "general", } # ── Tone key normalization map ──────────────────────────────────────────────── TONE_KEY_MAP = { "professional": "professional", "casual": "casual", "casual friendly": "casual", "urgent": "urgent", "urgent / escalation":"urgent", "escalation": "urgent", "educational": "educational", } # Input limits (characters) — protects token budget and prevents abuse MAX_TEXT_LENGTH = 8_000 MAX_TEXT2_LENGTH = 8_000 # ══════════════════════════════════════════════════════════════════════════════ # PROMPT BUILDER # ══════════════════════════════════════════════════════════════════════════════ def build_system_prompt(audience: str, tone: str, features: list[str]) -> str: """ Assemble a deterministic system prompt from the base + audience + tone + feature blocks. Slack feature MUST be last (highest priority override). """ audience_key = AUDIENCE_KEY_MAP.get(audience.lower().strip(), "general") tone_key = TONE_KEY_MAP.get(tone.lower().strip(), "professional") # Normalize features safely normalized_features = [] for f in features: key = FEATURE_KEY_MAP.get(f.lower().strip()) if key and key not in normalized_features: normalized_features.append(key) # Enforce Slack LAST (highest priority) if "slack" in normalized_features: normalized_features = [ f for f in normalized_features if f != "slack" ] + ["slack"] parts = [ BASE_SYSTEM_PROMPT, AUDIENCE_PROMPTS.get(audience_key, AUDIENCE_PROMPTS["general"]), TONE_PROMPTS.get(tone_key, TONE_PROMPTS["professional"]), ] # Append features in correct order for feature_key in normalized_features: if feature_key in FEATURE_PROMPTS: parts.append(FEATURE_PROMPTS[feature_key]) return "\n\n".join(parts) def build_user_prompt( text: str, text2: str, mode: str, audience: str, tone: str, features: list[str], ) -> str: """ Build the user-turn prompt. Keeps source content clean and clearly delimited. """ feature_summary = ", ".join(features) if features else "None" header = ( f"[Configuration]\n" f"- Mode: {mode.capitalize()}\n" f"- Audience: {audience}\n" f"- Tone: {tone}\n" f"- Features: {feature_summary}\n\n" ) if mode == "comparison": body = ( f"--- SOURCE TEXT A ---\n{text}\n\n" f"--- SOURCE TEXT B ---\n{text2}\n\n" f"[Task] Provide a Comparison summary: summarize both texts, " f"highlight key differences, note resource implications, and suggest an approach." ) elif mode == "reverse": body = ( f"--- BUSINESS REQUEST ---\n{text}\n\n" f"[Task] Reverse-translate this business request into: " f"technical considerations, acceptance criteria, and questions for the engineering team." ) else: body = ( f"--- TECHNICAL SOURCE TEXT ---\n{text}\n\n" f"[Task] Translate the technical content above following all configuration rules." ) return header + body # ══════════════════════════════════════════════════════════════════════════════ # API ROUTES # ══════════════════════════════════════════════════════════════════════════════ @app.post("/api/translate") async def translate(request: Request): # ── 1. Parse body ────────────────────────────────────────────────────────── try: body = await request.json() except Exception: return JSONResponse({"error": "Invalid JSON body."}, status_code=400) if not body: return JSONResponse({"error": "Empty request body."}, status_code=400) text = (body.get("text") or "").strip() text2 = (body.get("text2") or "").strip() mode = (body.get("mode") or "standard").strip().lower() audience = (body.get("audience") or "general").strip().lower() tone = (body.get("tone") or "professional").strip().lower() features = [f for f in (body.get("features") or []) if isinstance(f, str)] # ── 2. Input validation ──────────────────────────────────────────────────── if not text: return JSONResponse({"error": "No text provided."}, status_code=400) if len(text) > MAX_TEXT_LENGTH: return JSONResponse( {"error": f"Input text exceeds {MAX_TEXT_LENGTH:,} character limit. Please shorten it."}, status_code=400, ) if mode not in ("standard", "comparison", "reverse"): return JSONResponse( {"error": f"Invalid mode '{mode}'. Must be standard, comparison, or reverse."}, status_code=400, ) if mode == "comparison": if not text2: return JSONResponse( {"error": "Comparison mode requires a second text (text2)."}, status_code=400, ) if len(text2) > MAX_TEXT2_LENGTH: return JSONResponse( {"error": f"Second text exceeds {MAX_TEXT2_LENGTH:,} character limit."}, status_code=400, ) # ── 3. Config check ──────────────────────────────────────────────────────── if not AZURE_API_KEY: logger.error("AZURE_API_KEY is not configured.") return JSONResponse( {"error": "Translation service is not configured. Contact the administrator."}, status_code=500, ) if not AZURE_ENDPOINT: logger.error("AZURE_ENDPOINT is not configured.") return JSONResponse( {"error": "Translation endpoint is not configured. Contact the administrator."}, status_code=500, ) # ── 4. Build prompts ─────────────────────────────────────────────────────── system_prompt = build_system_prompt(audience, tone, features) user_prompt = build_user_prompt(text, text2, mode, audience, tone, features) logger.info( "Translate request | mode=%s audience=%s tone=%s features=%s chars=%d", mode, audience, tone, features, len(text), ) # ── 5. Call Azure OpenAI ─────────────────────────────────────────────────── try: async with httpx.AsyncClient(timeout=90.0) as client: response = await client.post( AZURE_ENDPOINT, headers={ "Content-Type": "application/json", "api-key": AZURE_API_KEY, }, json={ "model": MODEL_DEPLOYMENT, "instructions": system_prompt, "input": user_prompt, "max_output_tokens": 16384, }, ) # Surface non-2xx HTTP errors clearly if response.status_code != 200: logger.warning("Azure returned HTTP %d: %s", response.status_code, response.text[:300]) return JSONResponse( {"error": f"Translation service returned an error (HTTP {response.status_code}). Try again."}, status_code=502, ) data = response.json() except httpx.TimeoutException: logger.warning("Azure request timed out.") return JSONResponse( {"error": "Request timed out. Try with shorter input or try again shortly."}, status_code=504, ) except httpx.RequestError as exc: logger.error("Network error contacting Azure: %s", exc) return JSONResponse( {"error": "Could not reach the translation service. Check connectivity."}, status_code=502, ) # ── 6. Parse Azure response ──────────────────────────────────────────────── # Guard against {"error": {...}} in a 200 response (Azure sometimes does this) error_field = data.get("error") if error_field: error_msg = ( error_field.get("message", "Unknown error from translation service.") if isinstance(error_field, dict) else str(error_field) ) logger.warning("Azure error in 200 response: %s", error_msg) return JSONResponse({"error": error_msg}, status_code=502) # Extract text from the Responses API output array output_text = _extract_output_text(data) if not output_text: logger.error("Empty output extracted from Azure response: %s", str(data)[:500]) return JSONResponse( {"error": "The AI returned an empty response. Please try again."}, status_code=502, ) return JSONResponse({"translation": output_text}) def _extract_output_text(data: dict) -> str: """ Safely extract assistant text from Azure OpenAI Responses API payload. Handles nested output → message → content → output_text structure. Returns empty string if nothing is found. """ output_text = "" output_list = data.get("output") or [] if not isinstance(output_list, list): return output_text for item in output_list: if not isinstance(item, dict): continue if item.get("type") != "message": continue content_list = item.get("content") or [] for content in content_list: if isinstance(content, dict) and content.get("type") == "output_text": output_text += content.get("text", "") return output_text.strip() # ══════════════════════════════════════════════════════════════════════════════ # HEALTH CHECK # ══════════════════════════════════════════════════════════════════════════════ @app.get("/api/health") async def health(): return { "status": "ok", "configured": bool(AZURE_API_KEY and AZURE_ENDPOINT), "model": MODEL_DEPLOYMENT, } # ══════════════════════════════════════════════════════════════════════════════ # SERVE FRONTEND # ══════════════════════════════════════════════════════════════════════════════ @app.get("/") async def root(): return FileResponse(BASE_DIR / "static" / "index.html") app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static")