Escalapro+ 1.0

ai/__init__.py

@@ -0,0 +1,3 @@
+"""Helpers for interacting with the Flatlogic AI proxy from Django code."""
+
+from .local_ai_api import LocalAIApi, create_response, request, decode_json_from_response  # noqa: F401

ai/local_ai_api.py

@@ -0,0 +1,282 @@
+"""
+LocalAIApi — lightweight Python client for the Flatlogic AI proxy.
+
+Usage (inside the Django workspace):
+
+    from ai.local_ai_api import LocalAIApi
+
+    response = LocalAIApi.create_response({
+        "input": [
+            {"role": "system", "content": "You are a helpful assistant."},
+            {"role": "user", "content": "Summarise this text in two sentences."},
+        ],
+        "text": {"format": {"type": "json_object"}},
+    })
+
+    if response.get("success"):
+        data = LocalAIApi.decode_json_from_response(response)
+        # ...
+
+The helper automatically injects the project UUID header and falls back to
+reading executor/.env if environment variables are missing.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import ssl
+from typing import Any, Dict, Iterable, Optional
+from urllib import error as urlerror
+from urllib import request as urlrequest
+
+__all__ = [
+    "LocalAIApi",
+    "create_response",
+    "request",
+    "decode_json_from_response",
+]
+
+
+_CONFIG_CACHE: Optional[Dict[str, Any]] = None
+
+
+class LocalAIApi:
+    """Static helpers mirroring the PHP implementation."""
+
+    @staticmethod
+    def create_response(params: Dict[str, Any], options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        return create_response(params, options or {})
+
+    @staticmethod
+    def request(path: Optional[str] = None, payload: Optional[Dict[str, Any]] = None,
+                options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        return request(path, payload or {}, options or {})
+
+    @staticmethod
+    def decode_json_from_response(response: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        return decode_json_from_response(response)
+
+
+def create_response(params: Dict[str, Any], options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    """Signature compatible with the OpenAI Responses API."""
+    options = options or {}
+    payload = dict(params)
+
+    if not isinstance(payload.get("input"), list) or not payload["input"]:
+        return {
+            "success": False,
+            "error": "input_missing",
+            "message": 'Parameter "input" is required and must be a non-empty list.',
+        }
+
+    cfg = _config()
+    if not payload.get("model"):
+        payload["model"] = cfg["default_model"]
+
+    return request(options.get("path"), payload, options)
+
+
+def request(path: Optional[str], payload: Dict[str, Any], options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    """Perform a raw request to the AI proxy."""
+    cfg = _config()
+    options = options or {}
+
+    resolved_path = path or options.get("path") or cfg["responses_path"]
+    if not resolved_path:
+        return {
+            "success": False,
+            "error": "project_id_missing",
+            "message": "PROJECT_ID is not defined; cannot resolve AI proxy endpoint.",
+        }
+
+    project_uuid = cfg["project_uuid"]
+    if not project_uuid:
+        return {
+            "success": False,
+            "error": "project_uuid_missing",
+            "message": "PROJECT_UUID is not defined; aborting AI request.",
+        }
+
+    if "project_uuid" not in payload and project_uuid:
+        payload["project_uuid"] = project_uuid
+
+    url = _build_url(resolved_path, cfg["base_url"])
+    timeout = int(options.get("timeout", cfg["timeout"]))
+    verify_tls = options.get("verify_tls", cfg["verify_tls"])
+
+    headers: Dict[str, str] = {
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+        cfg["project_header"]: project_uuid,
+    }
+    extra_headers = options.get("headers")
+    if isinstance(extra_headers, Iterable):
+        for header in extra_headers:
+            if isinstance(header, str) and ":" in header:
+                name, value = header.split(":", 1)
+                headers[name.strip()] = value.strip()
+
+    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+    req = urlrequest.Request(url, data=body, method="POST")
+    for name, value in headers.items():
+        req.add_header(name, value)
+
+    context = None
+    if not verify_tls:
+        context = ssl.create_default_context()
+        context.check_hostname = False
+        context.verify_mode = ssl.CERT_NONE
+
+    try:
+        with urlrequest.urlopen(req, timeout=timeout, context=context) as resp:
+            status = resp.getcode()
+            response_body = resp.read().decode("utf-8", errors="replace")
+    except urlerror.HTTPError as exc:
+        status = exc.getcode()
+        response_body = exc.read().decode("utf-8", errors="replace")
+    except Exception as exc:  # pylint: disable=broad-except
+        return {
+            "success": False,
+            "error": "request_failed",
+            "message": str(exc),
+        }
+
+    decoded = None
+    if response_body:
+        try:
+            decoded = json.loads(response_body)
+        except json.JSONDecodeError:
+            decoded = None
+
+    if 200 <= status < 300:
+        return {
+            "success": True,
+            "status": status,
+            "data": decoded if decoded is not None else response_body,
+        }
+
+    error_message = "AI proxy request failed"
+    if isinstance(decoded, dict):
+        error_message = decoded.get("error") or decoded.get("message") or error_message
+    elif response_body:
+        error_message = response_body
+
+    return {
+        "success": False,
+        "status": status,
+        "error": error_message,
+        "response": decoded if decoded is not None else response_body,
+    }
+
+
+def decode_json_from_response(response: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    """Attempt to decode JSON emitted by the model (handles markdown fences)."""
+    text = _extract_text(response)
+    if text == "":
+        return None
+
+    try:
+        decoded = json.loads(text)
+        if isinstance(decoded, dict):
+            return decoded
+    except json.JSONDecodeError:
+        pass
+
+    stripped = text.strip()
+    if stripped.startswith("```json"):
+        stripped = stripped[7:]
+    if stripped.endswith("```"):
+        stripped = stripped[:-3]
+    stripped = stripped.strip()
+    if stripped and stripped != text:
+        try:
+            decoded = json.loads(stripped)
+            if isinstance(decoded, dict):
+                return decoded
+        except json.JSONDecodeError:
+            return None
+    return None
+
+
+def _extract_text(response: Dict[str, Any]) -> str:
+    payload = response.get("data") if response.get("success") else response.get("response")
+    if isinstance(payload, dict):
+        output = payload.get("output")
+        if isinstance(output, list):
+            combined = ""
+            for item in output:
+                content = item.get("content") if isinstance(item, dict) else None
+                if isinstance(content, list):
+                    for block in content:
+                        if isinstance(block, dict) and block.get("type") == "output_text" and block.get("text"):
+                            combined += str(block["text"])
+                if combined:
+                    return combined
+        choices = payload.get("choices")
+        if isinstance(choices, list) and choices:
+            message = choices[0].get("message")
+            if isinstance(message, dict) and message.get("content"):
+                return str(message["content"])
+    if isinstance(payload, str):
+        return payload
+    return ""
+
+
+def _config() -> Dict[str, Any]:
+    global _CONFIG_CACHE  # noqa: PLW0603
+    if _CONFIG_CACHE is not None:
+        return _CONFIG_CACHE
+
+    _ensure_env_loaded()
+
+    base_url = os.getenv("AI_PROXY_BASE_URL", "https://flatlogic.com")
+    project_id = os.getenv("PROJECT_ID") or None
+    responses_path = os.getenv("AI_RESPONSES_PATH")
+    if not responses_path and project_id:
+        responses_path = f"/projects/{project_id}/ai-request"
+
+    _CONFIG_CACHE = {
+        "base_url": base_url,
+        "responses_path": responses_path,
+        "project_id": project_id,
+        "project_uuid": os.getenv("PROJECT_UUID"),
+        "project_header": os.getenv("AI_PROJECT_HEADER", "project-uuid"),
+        "default_model": os.getenv("AI_DEFAULT_MODEL", "gpt-5"),
+        "timeout": int(os.getenv("AI_TIMEOUT", "30")),
+        "verify_tls": os.getenv("AI_VERIFY_TLS", "true").lower() not in {"0", "false", "no"},
+    }
+    return _CONFIG_CACHE
+
+
+def _build_url(path: str, base_url: str) -> str:
+    trimmed = path.strip()
+    if trimmed.startswith("http://") or trimmed.startswith("https://"):
+        return trimmed
+    if trimmed.startswith("/"):
+        return f"{base_url}{trimmed}"
+    return f"{base_url}/{trimmed}"
+
+
+def _ensure_env_loaded() -> None:
+    """Populate os.environ from executor/.env if variables are missing."""
+    if os.getenv("PROJECT_UUID") and os.getenv("PROJECT_ID"):
+        return
+
+    env_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".env"))
+    if not os.path.exists(env_path):
+        return
+
+    try:
+        with open(env_path, "r", encoding="utf-8") as handle:
+            for line in handle:
+                stripped = line.strip()
+                if not stripped or stripped.startswith("#") or "=" not in stripped:
+                    continue
+                key, value = stripped.split("=", 1)
+                key = key.strip()
+                value = value.strip().strip('\'"')
+                if key and not os.getenv(key):
+                    os.environ[key] = value
+    except OSError:
+        pass

core/templates/base.html

@@ -0,0 +1,11 @@
+<!DOCTYPE html>
+<html lang="en">
+  <head>
+    <meta charset="UTF-8">
+    <title>{% block title %}Knowledge Base{% endblock %}</title>
+    {% block head %}{% endblock %}
+  </head>
+  <body>
+    {% block content %}{% endblock %}
+  </body>
+</html>

core/templates/core/article_detail.html

@@ -0,0 +1,14 @@
+{% extends 'base.html' %}
+
+{% block title %}{{ article.title }}{% endblock %}
+
+{% block content %}
+<div class="container mt-5">
+    <h1>{{ article.title }}</h1>
+    <p class="text-muted">Published on {{ article.created_at|date:"F d, Y" }}</p>
+    <hr>
+    <div>
+        {{ article.content|safe }}
+    </div>
+</div>
+{% endblock %}