jinojose/aipackage
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update installer and runtime setup flow

Browse Source
This commit is contained in:
Jino Jose committed 2026-07-09 10:30:10 +05:30
1 parent 8603fb2bc2
commit 7540da04a8
5 files changed
+385 -54
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

ansible/roles/cezen-backend/files/main.py
+262 -13
View File
@@ -10,7 +10,7 @@ Default admin on first run:
password: Cezen@2024! (forced change on first login)
"""
import os, sqlite3, subprocess, json, uuid, secrets, hashlib, shutil, tempfile, base64, zipfile, sys
import os, sqlite3, subprocess, json, uuid, secrets, hashlib, shutil, tempfile, base64, zipfile, sys, logging
from datetime import datetime, timedelta, timezone
from pathlib import Path
from contextlib import asynccontextmanager
@@ -705,7 +705,7 @@ def init_db():
utcnow())
)
db.commit()
print("[cezen] Default admin created — username: admin password: Cezen@2024!")
logger.info("Default admin account created — username: admin password: Cezen@2024! (change required on first login)")
# Seed default branding settings
defaults = {
@@ -803,6 +803,76 @@ app.add_middleware(
allow_headers=["*"],
)
# ── Structured logging ───────────────────────────────────────────────────────
# Enterprise basic: a real log file + levels, instead of the handful of bare
# print() calls this file previously relied on. Log location follows DATA_DIR
# so it lands in the same place as the rest of Nexus One AI's runtime state.
LOG_FILE = DATA_DIR / "cezen-api.log"
logger = logging.getLogger("cezen")
if not logger.handlers:
logger.setLevel(logging.INFO)
_fmt = logging.Formatter("%(asctime)s %(levelname)s [%(name)s] %(message)s")
try:
_file_handler = logging.FileHandler(str(LOG_FILE))
_file_handler.setFormatter(_fmt)
logger.addHandler(_file_handler)
except Exception:
pass # e.g. read-only filesystem — fall back to console only
_console_handler = logging.StreamHandler()
_console_handler.setFormatter(_fmt)
logger.addHandler(_console_handler)
# ── Request correlation ID ───────────────────────────────────────────────────
# Every request/response gets an X-Request-Id so a support engineer can tie a
# user-reported error to a specific log line, without needing to reproduce it.
@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
req_id = request.headers.get("X-Request-Id") or str(uuid.uuid4())
request.state.request_id = req_id
try:
response = await call_next(request)
except Exception:
# Let the global exception handler below produce the JSON body; just
# make sure this middleware doesn't swallow the request id in that path.
raise
response.headers["X-Request-Id"] = req_id
return response
# ── Global exception handler ─────────────────────────────────────────────────
# Any exception that isn't already an HTTPException (i.e. a genuine bug, not a
# handled validation/business-logic error) previously fell through to
# Starlette's plain-text "Internal Server Error" — breaking the JSON contract
# every other response follows, and giving the user/support engineer nothing
# to go on. This logs the real exception server-side and returns a safe,
# structured, correlatable error to the client instead.
@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
req_id = getattr(request.state, "request_id", None) or str(uuid.uuid4())
logger.error(
"Unhandled exception on %s %s [request_id=%s]: %s",
request.method, request.url.path, req_id, exc, exc_info=True,
)
return JSONResponse(
status_code=500,
content={
"detail": "An unexpected error occurred. This has been logged — if it persists, contact support with the request ID below.",
"code": "internal_error",
"request_id": req_id,
},
headers={"X-Request-Id": req_id},
)
# ── Friendly error helper ────────────────────────────────────────────────────
# Several endpoints previously returned raw Python/subprocess exception text
# (disk errno strings, Ollama CLI stderr, urllib exception reprs) straight to
# the client. This logs the real error server-side and raises a structured,
# actionable HTTPException instead — {code, message, remediation} rather than
# a bare string, so the portal can show "what happened / why / how to fix it".
def raise_api_error(status_code: int, code: str, message: str, remediation: str = "", *, log_context: str = "", log_exc=None):
if log_exc is not None:
logger.error("%s: %s", log_context or code, log_exc, exc_info=True)
raise HTTPException(status_code=status_code, detail={"code": code, "message": message, "remediation": remediation})
# ── Auth helpers ──────────────────────────────────────────────────────────────
def create_token(user_id: int, username: str, role: str) -> tuple[str, str]:
@@ -1492,7 +1562,10 @@ async def delete_model(model_name: str, request: Request, admin: dict = Depends(
db.commit()
db.close()
if "error" in result:
raise HTTPException(status_code=500, detail=result["error"])
raise_api_error(500, "model_delete_failed",
f"Could not delete model '{model_name}'.",
"Check that Ollama is running (systemctl status ollama) and that no chat session currently has this model loaded, then try again.",
log_context=f"model_delete({model_name})", log_exc=result["error"])
return {"ok": True}
# ── Offline model upload / load ───────────────────────────────────────────────
@@ -1536,9 +1609,21 @@ async def upload_model(
if not chunk:
break
await out.write(chunk)
except OSError as e:
dest.unlink(missing_ok=True)
if getattr(e, "errno", None) == 28: # ENOSPC
raise_api_error(500, "disk_full",
"The upload failed because the server has run out of disk space.",
f"Free up space in {UPLOAD_DIR} (or expand storage) and try the upload again.",
log_context="upload_model (disk full)", log_exc=e)
raise_api_error(500, "upload_failed", "The model upload failed while writing to disk.",
"Check that the server has enough free disk space and that the destination directory is writable, then retry.",
log_context="upload_model", log_exc=e)
except Exception as e:
dest.unlink(missing_ok=True)
raise HTTPException(status_code=500, detail=f"Upload failed: {e}")
raise_api_error(500, "upload_failed", "The model upload failed.",
"Check your network connection and try uploading again. If this keeps happening, contact support.",
log_context="upload_model", log_exc=e)
size_gb = round(dest.stat().st_size / 1e9, 2)
db = get_db()
@@ -1584,10 +1669,16 @@ async def load_model(body: dict, request: Request, admin: dict = Depends(admin_o
)
modelfile_path.unlink(missing_ok=True)
if result.returncode != 0:
raise HTTPException(status_code=500, detail=result.stderr or "ollama create failed")
except subprocess.TimeoutExpired:
raise_api_error(500, "model_load_failed",
f"Ollama could not register '{model_name}' as a model.",
"Confirm the uploaded .gguf file is not corrupted and that Ollama has enough free disk/memory, then try loading it again.",
log_context=f"load_model({model_name})", log_exc=result.stderr or "ollama create failed (no stderr)")
except subprocess.TimeoutExpired as e:
modelfile_path.unlink(missing_ok=True)
raise HTTPException(status_code=500, detail="ollama create timed out (>5 min)")
raise_api_error(500, "model_load_timeout",
f"Registering '{model_name}' with Ollama took too long and timed out (5 min limit).",
"This usually happens with very large model files or a slow/busy server. Try again during a quieter period, or check server load.",
log_context=f"load_model({model_name})", log_exc=e)
db = get_db()
audit(db, admin["sub"], admin["username"], "model_load",
@@ -2135,9 +2226,15 @@ def _chroma_req(method: str, path: str, **kwargs):
return json.loads(r.read().decode())
except urllib.error.HTTPError as e:
body = e.read().decode()
raise HTTPException(status_code=e.code, detail=f"ChromaDB: {body}")
raise_api_error(e.code, "chromadb_error",
"The knowledge base service (ChromaDB) rejected this request.",
"Check the ChromaDB service logs for details, then try again.",
log_context=f"_chroma_req({method} {path})", log_exc=body)
except Exception as e:
raise HTTPException(status_code=503, detail=f"ChromaDB unavailable: {e}")
raise_api_error(503, "chromadb_unavailable",
"The knowledge base service (ChromaDB) is unreachable.",
"Check that the ChromaDB service is running (systemctl status chromadb) and try again in a moment.",
log_context=f"_chroma_req({method} {path})", log_exc=e)
def _ollama_embed(texts: list[str], model: str = OLLAMA_EMBED) -> list[list[float]]:
"""Get embeddings from Ollama for a list of texts."""
@@ -2151,7 +2248,10 @@ def _ollama_embed(texts: list[str], model: str = OLLAMA_EMBED) -> list[list[floa
with urllib.request.urlopen(req, timeout=30) as r:
embeddings.append(json.loads(r.read().decode())["embedding"])
except Exception as e:
raise HTTPException(status_code=503, detail=f"Ollama embedding error: {e}")
raise_api_error(503, "embedding_failed",
"Could not generate embeddings — the Ollama embedding model is unreachable.",
f"Check that Ollama is running and that the '{model}' embedding model is pulled (ollama list), then retry.",
log_context=f"_ollama_embed(model={model})", log_exc=e)
return embeddings
# Collections
@@ -2514,7 +2614,10 @@ async def launch_job(body: dict, _licensed: dict = Depends(require_feature("fine
(utcnow(), job_id))
db.commit()
db.close()
raise HTTPException(status_code=500, detail=f"Failed to start training process: {e}")
raise_api_error(500, "training_launch_failed",
f"Could not start the training job for '{body['name']}'.",
"Check that the training script and base model are available on the server and that there's enough free GPU/CPU capacity, then try launching the job again.",
log_context=f"training_launch(job_id={job_id})", log_exc=e)
db.close()
_audit(None, admin["username"], "training_launch",
@@ -4437,7 +4540,7 @@ def _add_to_scheduler(job: dict):
misfire_grace_time=300,
)
except Exception as e:
print(f"[cezen] Failed to schedule job {job['id']}: {e}")
logger.warning("Failed to schedule job %s: %s", job['id'], e)
class ScheduledJobCreate(BaseModel):
@@ -5730,7 +5833,153 @@ async def process_meeting(file: UploadFile = File(...), meta: str = Form("{}"),
# ── Health ────────────────────────────────────────────────────────────────────
# Real subsystem checks for a unified Health Center (Phase 3 of the v1.0 GA
# polish pass). Each check returns {"status": "ok"|"warning"|"critical",
# "detail": "<human-readable, actionable message>"} — never a bare boolean —
# so the portal can show *why* something is degraded, not just that it is.
_LICENSE_STATUS_HEALTH = {
"valid": ("ok", "License is valid."),
"missing": ("warning", "No license installed — running in field-staging mode. Upload a signed license in Settings > License."),
"invalid_signature": ("critical", "The installed license failed signature verification. Re-upload a valid signed license file."),
"expired": ("critical", "The installed license has expired. Contact support@cezentech.com to renew."),
"not_yet_valid": ("warning", "The installed license is not yet in its valid date range."),
"machine_mismatch": ("critical", "The installed license is bound to different hardware. Contact support@cezentech.com."),
}
def _health_database() -> dict:
try:
conn = get_db()
conn.execute("SELECT 1").fetchone()
conn.close()
return {"status": "ok", "detail": "Database is reachable."}
except Exception as e:
return {"status": "critical", "detail": f"Cannot reach the database at {DB_PATH}: {e}"}
def _health_scheduler() -> dict:
try:
running = bool(_scheduler.running)
job_count = len(_scheduler.get_jobs())
if running:
return {"status": "ok", "detail": f"Scheduler is running ({job_count} job{'s' if job_count != 1 else ''} registered)."}
return {"status": "critical", "detail": "Scheduler process is not running — scheduled jobs will not fire. Restart the cezen-api service."}
except Exception as e:
return {"status": "critical", "detail": f"Could not read scheduler state: {e}"}
def _health_storage() -> dict:
try:
disk = psutil.disk_usage("/")
pct = disk.percent
free_gb = round(disk.free / 1e9, 1)
if pct >= 90:
return {"status": "critical", "detail": f"Disk is {pct:.0f}% full ({free_gb} GB free). Free up space or expand storage — installs and uploads may start failing."}
if pct >= 75:
return {"status": "warning", "detail": f"Disk is {pct:.0f}% full ({free_gb} GB free). Plan for cleanup or expansion soon."}
return {"status": "ok", "detail": f"Disk usage is {pct:.0f}% ({free_gb} GB free)."}
except Exception as e:
return {"status": "critical", "detail": f"Could not read disk usage: {e}"}
def _health_backup() -> dict:
try:
BACKUP_DIR.mkdir(parents=True, exist_ok=True)
files = sorted(BACKUP_DIR.glob("cezen-backup-*.zip"), key=lambda p: p.stat().st_mtime, reverse=True)
if not files:
return {"status": "warning", "detail": "No backups have been taken yet. Configure a backup schedule in Settings."}
age_hrs = (datetime.now(timezone.utc).timestamp() - files[0].stat().st_mtime) / 3600
when = f"{age_hrs:.0f} hour{'s' if age_hrs >= 2 else ''} ago" if age_hrs < 48 else f"{age_hrs/24:.0f} days ago"
if age_hrs > 7 * 24:
return {"status": "critical", "detail": f"Last backup was {when} — well overdue. Check the backup schedule."}
if age_hrs > 24:
return {"status": "warning", "detail": f"Last backup was {when}. Confirm the backup schedule is still running."}
return {"status": "ok", "detail": f"Last backup was {when}."}
except Exception as e:
return {"status": "critical", "detail": f"Could not check backup status: {e}"}
def _health_license() -> dict:
try:
entitlement = _entitlement_summary()
status_key = entitlement.get("license_status", "missing")
level, detail = _LICENSE_STATUS_HEALTH.get(status_key, ("warning", f"Unrecognized license status '{status_key}'."))
return {"status": level, "detail": detail}
except Exception as e:
return {"status": "warning", "detail": f"Could not evaluate license status: {e}"}
def _health_ssl() -> dict:
# Nexus One AI currently ships with HTTP only (no TLS termination configured
# in the nginx role) — report this honestly rather than claiming "ok".
return {"status": "warning", "detail": "No TLS/SSL certificate is configured — the portal is served over HTTP only. Consider placing this deployment behind TLS termination."}
def _health_portal() -> dict:
import socket
try:
s = socket.create_connection(("127.0.0.1", 80), timeout=1.5)
s.close()
return {"status": "ok", "detail": "Portal (nginx, port 80) is reachable."}
except OSError:
return {"status": "critical", "detail": "Portal (nginx, port 80) is not reachable. Check the nginx service."}
def _health_models() -> dict:
try:
data = ollama_get("/api/tags")
if data is None:
return {"status": "critical", "detail": "Ollama is not reachable at " + OLLAMA_URL + ". Check the ollama service."}
count = len(data.get("models", []))
if count == 0:
return {"status": "warning", "detail": "Ollama is reachable but no models are installed yet."}
return {"status": "ok", "detail": f"Ollama is reachable ({count} model{'s' if count != 1 else ''} installed)."}
except Exception as e:
return {"status": "critical", "detail": f"Could not reach Ollama: {e}"}
def _health_gpu_drivers_cuda() -> dict:
try:
out = subprocess.check_output(
["nvidia-smi", "--query-gpu=name,driver_version,temperature.gpu,utilization.gpu",
"--format=csv,noheader"], timeout=5, stderr=subprocess.STDOUT,
).decode().strip().split("\n")[0]
name, driver, temp, util = [p.strip() for p in out.split(",")]
gpu = {"status": "ok", "detail": f"{name} detected — driver {driver}, {temp}°C, {util} utilisation."}
except FileNotFoundError:
gpu = {"status": "critical", "detail": "nvidia-smi was not found. The NVIDIA driver may not be installed."}
except subprocess.TimeoutExpired:
gpu = {"status": "critical", "detail": "nvidia-smi timed out. The GPU driver may be unresponsive."}
except Exception as e:
gpu = {"status": "critical", "detail": f"GPU check failed: {e}"}
drivers = {"status": gpu["status"], "detail": gpu["detail"]}
try:
import re as _re
raw = subprocess.check_output(["nvidia-smi"], timeout=5, stderr=subprocess.STDOUT).decode()
m = _re.search(r"CUDA Version:\s*([\d.]+)", raw)
if m:
cuda = {"status": "ok", "detail": f"CUDA {m.group(1)} detected."}
else:
cuda = {"status": "warning", "detail": "nvidia-smi ran but no CUDA version could be parsed."}
except Exception as e:
cuda = {"status": "critical" if gpu["status"] == "critical" else "warning", "detail": f"Could not determine CUDA version: {e}"}
return {"gpu": gpu, "drivers": drivers, "cuda": cuda}
@app.get("/api/health")
async def health():
return {"status": "ok", "version": "1.0.0"}
gpu_group = _health_gpu_drivers_cuda()
checks = {
"api": {"status": "ok", "detail": "API process is responding."},
"database": _health_database(),
"scheduler": _health_scheduler(),
"storage": _health_storage(),
"backup": _health_backup(),
"license": _health_license(),
"ssl": _health_ssl(),
"portal": _health_portal(),
"models": _health_models(),
**gpu_group,
}
order = {"critical": 0, "warning": 1, "ok": 2}
overall = min((c["status"] for c in checks.values()), key=lambda s: order.get(s, 1))
return {
"status": overall,
"version": "1.0.0",
"generated_at": utcnow(),
"checks": checks,
}
ansible/roles/cezen-backend/files/rag_ingest.py
+25 -6
View File
@@ -20,7 +20,7 @@ Usage (called by main.py):
--ollama-url http://localhost:11434
"""
import argparse, json, os, sqlite3, sys, uuid
import argparse, json, logging, os, sqlite3, sys, uuid
from datetime import datetime, timezone
from pathlib import Path
@@ -38,6 +38,27 @@ CHUNK_SIZE = 512 # tokens/chars per chunk
CHUNK_OVERLAP = 64 # overlap between consecutive chunks
BATCH_SIZE = 16 # how many chunks to embed + upsert at once
# ── Logging ───────────────────────────────────────────────────────────────────
# Ingest runs as a detached subprocess of the API, so failures used to be
# surfaced to the portal as a raw Python traceback stuffed into error_msg.
# Now the full traceback goes to a log file (server-side only) and the DB gets
# a short, human-readable message instead.
_LOG_FILE = Path(args.db_path).resolve().parent / "cezen-ingest.log"
logger = logging.getLogger("cezen.ingest")
if not logger.handlers:
_handler = logging.FileHandler(_LOG_FILE)
_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
logger.addHandler(_handler)
logger.setLevel(logging.INFO)
def friendly_error(e: Exception, context: str) -> str:
"""Log the full exception + traceback server-side; return a short, actionable message for the UI."""
logger.error("doc_id=%s context=%s", args.doc_id, context, exc_info=True)
msg = str(e).strip() or e.__class__.__name__
if len(msg) > 200:
msg = msg[:200] + "..."
return f"{msg} (see server ingest log for full details, request context: {context})"
def utcnow():
return datetime.now(timezone.utc).isoformat()
@@ -256,8 +277,7 @@ def main():
processed += len(batch)
except Exception as e:
import traceback
set_status("failed", chunk_count=processed, error=f"{e}\n{traceback.format_exc()[:500]}")
set_status("failed", chunk_count=processed, error=friendly_error(e, "embed_and_upsert"))
sys.exit(1)
set_status("ready", chunk_count=total)
@@ -266,9 +286,8 @@ if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
set_status("failed", error="Interrupted")
set_status("failed", error="Ingest was interrupted before it could finish. Try uploading the document again.")
sys.exit(130)
except Exception as e:
import traceback
set_status("failed", error=f"{e}\n{traceback.format_exc()[:500]}")
set_status("failed", error=friendly_error(e, "main"))
sys.exit(1)