777c80a0e1
- app.py: /api/models filtered to gemma+aya, /api/translate streams NDJSON from Ollama with stream:true, /healthz for k8s probes - static/index.html: dark-mode UI, three dropdowns (source/target/model), live token streaming with spinner+pulse+timer indicators - Dockerfile: two-stage uv build, slim Python 3.13 runtime - .gitea/workflows/build.yml: multi-arch (amd64+arm64) build, push to Scaleway registry on push to main - build.sh: manual fallback if Gitea Actions isn't available
92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
import os
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import FileResponse, StreamingResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
|
|
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://radagast.jpnadas.xyz:11434")
|
|
# Comma-separated substrings; a model name matches if it contains any of them.
|
|
# Set to empty string to show every installed model.
|
|
MODEL_FILTER = os.getenv("MODEL_FILTER", "gemma,aya").lower()
|
|
|
|
STATIC_DIR = Path(__file__).parent / "static"
|
|
|
|
app = FastAPI(title="finrod")
|
|
|
|
|
|
class TranslateRequest(BaseModel):
|
|
text: str
|
|
source: str
|
|
target: str
|
|
model: str
|
|
|
|
|
|
@app.get("/")
|
|
async def index() -> FileResponse:
|
|
return FileResponse(STATIC_DIR / "index.html")
|
|
|
|
|
|
@app.get("/healthz")
|
|
async def healthz() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/api/models")
|
|
async def list_models() -> list[str]:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
try:
|
|
r = await client.get(f"{OLLAMA_URL}/api/tags")
|
|
r.raise_for_status()
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(502, f"Ollama unreachable: {e}") from e
|
|
names = [m["name"] for m in r.json().get("models", [])]
|
|
filters = [f.strip() for f in MODEL_FILTER.split(",") if f.strip()]
|
|
if filters:
|
|
names = [n for n in names if any(f in n.lower() for f in filters)]
|
|
return sorted(names)
|
|
|
|
|
|
@app.post("/api/translate")
|
|
async def translate(req: TranslateRequest) -> StreamingResponse:
|
|
text = req.text.strip()
|
|
if not text:
|
|
raise HTTPException(400, "text is empty")
|
|
|
|
if req.source.lower() == "auto":
|
|
source_clause = "Detect the source language."
|
|
else:
|
|
source_clause = f"The source language is {req.source}."
|
|
|
|
prompt = (
|
|
f"You are a translation engine. {source_clause} "
|
|
f"Translate the text below into {req.target}. "
|
|
"Output ONLY the translation: no quotes, no explanations, no preamble, "
|
|
"no language labels. Preserve line breaks and formatting.\n\n"
|
|
f"Text:\n{text}"
|
|
)
|
|
|
|
async def relay():
|
|
async with httpx.AsyncClient(timeout=300) as client:
|
|
async with client.stream(
|
|
"POST",
|
|
f"{OLLAMA_URL}/api/generate",
|
|
json={
|
|
"model": req.model,
|
|
"prompt": prompt,
|
|
"stream": True,
|
|
"options": {"temperature": 0.2},
|
|
},
|
|
) as r:
|
|
r.raise_for_status()
|
|
async for line in r.aiter_lines():
|
|
if line:
|
|
yield line + "\n"
|
|
|
|
return StreamingResponse(relay(), media_type="application/x-ndjson")
|
|
|
|
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|