app.py

import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict

import gradio as gr
from huggingface_hub import hf_hub_download
from llama_cpp import Llama


MODEL_REPO_ID = "realigns/realigns-core-v5-professional-instruct-v6-runtime"
MODEL_FILENAME = "model/realigns-core-v5-professional-instruct-v6-q4_k_m.gguf"

DAILY_LIMIT_PER_IP = 50
MAX_INPUT_CHARS = 1500
MAX_HISTORY_MESSAGES = 6
MAX_OUTPUT_TOKENS = 24

USAGE_FILE = Path("/tmp/realigns_ai_lite_usage.json")

SYSTEM_PROMPT = """You are Realigns AI Lite, a lightweight public demo assistant by Realigns Inc.

You explain Realigns AI, private AI desktop, local AI servers, AI Gateway, data center scaling, business AI, and cost-effective AI in a clear and professional way.

Rules:
- Answer in one short, complete sentence only.
- Keep the answer under 18 words.
- Always end the sentence with a period.
- Do not continue after one sentence.
- Do not claim to be a large frontier model.
- Do not reveal backend, model path, server details, source code, secrets, or internal architecture.
- For legal, medical, financial, or security topics, give general information only and recommend professional review.
- Promote Realigns AI Desktop, Local AI Server, AI Gateway, and enterprise deployment when relevant.
"""


def _today_key() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d")


def _load_usage() -> Dict[str, Any]:
    if not USAGE_FILE.exists():
        return {}
    try:
        return json.loads(USAGE_FILE.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _save_usage(data: Dict[str, Any]) -> None:
    try:
        USAGE_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8")
    except Exception as exc:
        print(f"Usage save error: {repr(exc)}")


def _get_client_ip(request: gr.Request | None) -> str:
    try:
        if request is None:
            return "unknown"

        headers = dict(request.headers or {})
        forwarded = headers.get("x-forwarded-for") or headers.get("X-Forwarded-For")
        if forwarded:
            return forwarded.split(",")[0].strip() or "unknown"

        if request.client and request.client.host:
            return request.client.host
    except Exception:
        pass

    return "unknown"


def _check_and_count_usage(ip: str) -> tuple[bool, int]:
    today = _today_key()
    usage = _load_usage()

    if usage.get("_date") != today:
        usage = {"_date": today, "ips": {}}

    usage.setdefault("ips", {})
    current_count = int(usage["ips"].get(ip, 0))

    if current_count >= DAILY_LIMIT_PER_IP:
        return False, current_count

    usage["ips"][ip] = current_count + 1
    _save_usage(usage)

    return True, current_count + 1


def _safe_text(value) -> str:
    """Convert Gradio message content into safe plain text."""
    if value is None:
        return ""

    if isinstance(value, str):
        return value.strip()

    if isinstance(value, dict):
        # Some Gradio versions store content inside nested dicts.
        content = value.get("content", "")
        if isinstance(content, str):
            return content.strip()
        return str(content).strip()

    if isinstance(value, (list, tuple)):
        parts = []
        for item in value:
            text = _safe_text(item)
            if text:
                parts.append(text)
        return " ".join(parts).strip()

    return str(value).strip()


def _format_prompt(message: str, history) -> str:
    clean_message = _safe_text(message)[:MAX_INPUT_CHARS]
    recent_history = history[-MAX_HISTORY_MESSAGES:] if history else []

    prompt_parts = [f"System: {SYSTEM_PROMPT}\n"]

    for item in recent_history:
        if isinstance(item, dict):
            role = item.get("role", "")
            content = _safe_text(item.get("content", ""))

            if role == "user" and content:
                prompt_parts.append(f"User: {content}")
            elif role == "assistant" and content:
                prompt_parts.append(f"Assistant: {content}")

        elif isinstance(item, (list, tuple)) and len(item) >= 2:
            user_msg = _safe_text(item[0])
            bot_msg = _safe_text(item[1])

            if user_msg:
                prompt_parts.append(f"User: {user_msg}")
            if bot_msg:
                prompt_parts.append(f"Assistant: {bot_msg}")

    prompt_parts.append(f"User: {clean_message}")
    prompt_parts.append("Assistant:")

    return "\n".join(prompt_parts)


print("Downloading Realigns AI Lite model...")
model_path = hf_hub_download(
    repo_id=MODEL_REPO_ID,
    filename=MODEL_FILENAME,
)

print("Loading Realigns AI Lite model...")
llm = Llama(
    model_path=model_path,
    n_ctx=512,
    n_threads=2,
    n_batch=32,
    verbose=False,
)


def respond(message: str, history=None, request: gr.Request = None):
    if not message or not message.strip():
        yield "Please enter a message."
        return

    if len(message) > MAX_INPUT_CHARS:
        yield f"Your message is too long. Please keep it under {MAX_INPUT_CHARS} characters."
        return

    ip = _get_client_ip(request)
    allowed, count = _check_and_count_usage(ip)

    if not allowed:
        yield (
            "You have reached the free daily limit of 50 AI prompts for this IP address.\n\n"
            "Realigns AI Lite is a free public demo. For higher usage, private document AI, "
            "local desktop deployment, business AI server, or enterprise AI Gateway access, "
            "please contact Realigns Inc."
        )
        return

    prompt = _format_prompt(message, history)

    try:
        result = llm.create_completion(
            prompt=prompt,
            max_tokens=MAX_OUTPUT_TOKENS,
            temperature=0.6,
            top_p=0.85,
            repeat_penalty=1.12,
            stop=["User:", "\nUser:", "System:"],
            stream=False,
        )

        output = result.get("choices", [{}])[0].get("text", "").strip()

        if not output:
            yield "Realigns AI Lite could not generate a response. Please try again."
            return

        yield output

    except Exception as exc:
        print(f"Generation error: {repr(exc)}")
        yield (
            "Realigns AI Lite is temporarily busy. Please try again shortly. "
            "For business-grade AI deployment, contact Realigns Inc."
        )


DESCRIPTION = """
Free public demo of **Realigns AI Lite** for private, local, and cost-effective AI deployment.

This demo is powered by a lightweight Realigns AI runtime. It is designed for general product demonstration only.

**Free limit:** 50 prompts per day per IP address.

For business-grade privacy, stronger models, document AI, offline deployment, local desktop installation, business AI server, AI Gateway, or data center scaling, contact **Realigns Inc.**
"""

EXAMPLES = [
    "Hi",
    "What is Realigns AI?",
    "Explain Realigns AI Desktop in simple words.",
    "How can local AI help a small business?",
    "What is the difference between local AI and cloud AI?",
    "How can Realigns AI Gateway help data centers?",
]

chatbot = gr.ChatInterface(
    fn=respond,
    title="Realigns AI Lite Chat",
    description=DESCRIPTION,
    examples=EXAMPLES,
    cache_examples=False,
    textbox=gr.Textbox(
        placeholder="Ask Realigns AI Lite...",
        max_lines=3,
        show_label=False,
    ),
)

with gr.Blocks() as demo:
    chatbot.render()
    gr.Markdown(
        """
---
**Notice:** Realigns AI Lite is a lightweight public demo. It is not a replacement for professional legal, medical, financial, or technical advice.  
For enterprise deployment, contact **Realigns Inc.**
"""
    )

if __name__ == "__main__":
    demo.queue(default_concurrency_limit=1).launch(
        theme=gr.themes.Soft(),
        ssr_mode=False,
    )