1270 lines
44 KiB
Python
1270 lines
44 KiB
Python
"""
|
|
JARVIS API Backend
|
|
FastAPI application for JARVIS AI Assistant + Business Automation
|
|
"""
|
|
|
|
from fastapi import FastAPI, HTTPException, Depends, Header
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
import os
|
|
import asyncio
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
from urllib.parse import urlparse
|
|
import logging
|
|
import json
|
|
|
|
import psycopg2
|
|
import psycopg2.pool
|
|
from psycopg2.extras import RealDictCursor, Json
|
|
import redis.asyncio as aioredis
|
|
import aiohttp
|
|
from anthropic import Anthropic
|
|
import caldav
|
|
import imaplib
|
|
import smtplib
|
|
import ssl
|
|
from email import message_from_bytes
|
|
from email.header import decode_header
|
|
from email.message import EmailMessage
|
|
from email.utils import formatdate, make_msgid
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=os.getenv("LOG_LEVEL", "info").upper())
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ============ CONFIG ============
|
|
DATABASE_URL = os.getenv("DATABASE_URL")
|
|
REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379")
|
|
MILVUS_HOST = os.getenv("MILVUS_HOST", "milvus")
|
|
MILVUS_PORT = os.getenv("MILVUS_PORT", "19530")
|
|
OLLAMA_HOST = os.getenv("OLLAMA_HOST", "http://ollama:11434")
|
|
OLLAMA_EMBED_MODEL = os.getenv("OLLAMA_EMBED_MODEL", "nomic-embed-text")
|
|
N8N_URL = os.getenv("N8N_URL", "http://n8n:5678")
|
|
CLAUDE_API_KEY = os.getenv("CLAUDE_API_KEY")
|
|
CLAUDE_MODEL = os.getenv("CLAUDE_MODEL", "claude-sonnet-5")
|
|
API_KEY_ADMIN = os.getenv("API_KEY_ADMIN")
|
|
NEXTCLOUD_CALDAV_URL = os.getenv("NEXTCLOUD_CALDAV_URL")
|
|
NEXTCLOUD_USER = os.getenv("NEXTCLOUD_USER")
|
|
NEXTCLOUD_APP_PASSWORD = os.getenv("NEXTCLOUD_APP_PASSWORD")
|
|
EMAIL_IMAP_HOST = os.getenv("EMAIL_IMAP_HOST")
|
|
EMAIL_IMAP_PORT = int(os.getenv("EMAIL_IMAP_PORT", "143"))
|
|
EMAIL_SMTP_HOST = os.getenv("EMAIL_SMTP_HOST")
|
|
EMAIL_SMTP_PORT = int(os.getenv("EMAIL_SMTP_PORT", "465"))
|
|
EMAIL_USER = os.getenv("EMAIL_USER")
|
|
EMAIL_PASSWORD = os.getenv("EMAIL_PASSWORD")
|
|
CLAUDE_SYSTEM_PROMPT = os.getenv(
|
|
"CLAUDE_SYSTEM_PROMPT",
|
|
"Du bist JARVIS, ein KI-Assistent fuer Business-Automatisierung.",
|
|
)
|
|
|
|
# ============ INITIALIZATION ============
|
|
app = FastAPI(
|
|
title="JARVIS API",
|
|
description="AI Assistant + Business Automation Backend",
|
|
version="0.1.0"
|
|
)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["https://jarvis.mbo-tech-it.de"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
async def require_admin_key(x_admin_key: Optional[str] = Header(default=None)):
|
|
if not API_KEY_ADMIN or x_admin_key != API_KEY_ADMIN:
|
|
raise HTTPException(status_code=401, detail="Invalid or missing admin key")
|
|
|
|
|
|
pg_pool: Optional[psycopg2.pool.ThreadedConnectionPool] = None
|
|
claude_client: Optional[Anthropic] = None
|
|
DEFAULT_USER_ID: Optional[int] = None
|
|
|
|
|
|
@app.on_event("startup")
|
|
async def on_startup():
|
|
global pg_pool, claude_client, DEFAULT_USER_ID
|
|
|
|
if DATABASE_URL:
|
|
pg_pool = psycopg2.pool.ThreadedConnectionPool(1, 10, dsn=DATABASE_URL)
|
|
DEFAULT_USER_ID = await ensure_default_user()
|
|
else:
|
|
logger.warning("DATABASE_URL not set - persistence disabled")
|
|
|
|
if CLAUDE_API_KEY:
|
|
claude_client = Anthropic(api_key=CLAUDE_API_KEY)
|
|
else:
|
|
logger.warning("CLAUDE_API_KEY not set - chat endpoint will return 503")
|
|
|
|
|
|
@app.on_event("shutdown")
|
|
async def on_shutdown():
|
|
if pg_pool:
|
|
pg_pool.closeall()
|
|
|
|
|
|
# ============ TEXT PROCESSING ============
|
|
def chunk_text(text: str, chunk_size: int = 1000, overlap: int = 100) -> list:
|
|
if not text:
|
|
return []
|
|
if len(text) <= chunk_size:
|
|
return [text]
|
|
chunks = []
|
|
step = chunk_size - overlap
|
|
start = 0
|
|
while start < len(text):
|
|
end = start + chunk_size
|
|
chunks.append(text[start:end])
|
|
if end >= len(text):
|
|
break
|
|
start += step
|
|
return chunks
|
|
|
|
|
|
async def get_embedding(text: str) -> list:
|
|
timeout = aiohttp.ClientTimeout(total=30)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.post(
|
|
f"{OLLAMA_HOST}/api/embeddings",
|
|
json={"model": OLLAMA_EMBED_MODEL, "prompt": text},
|
|
) as resp:
|
|
if resp.status != 200:
|
|
body = await resp.text()
|
|
raise RuntimeError(f"Ollama embedding failed ({resp.status}): {body}")
|
|
data = await resp.json()
|
|
return data["embedding"]
|
|
|
|
|
|
# ============ DB HELPERS ============
|
|
async def db_query(query: str, params: tuple = None, fetch: Optional[str] = None):
|
|
"""Run a query against Postgres in a worker thread; fetch is None/'one'/'all'."""
|
|
if not pg_pool:
|
|
raise RuntimeError("Database not configured")
|
|
|
|
def _run():
|
|
conn = pg_pool.getconn()
|
|
try:
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
|
cur.execute(query, params)
|
|
result = None
|
|
if fetch == "one":
|
|
result = cur.fetchone()
|
|
elif fetch == "all":
|
|
result = cur.fetchall()
|
|
conn.commit()
|
|
return result
|
|
finally:
|
|
pg_pool.putconn(conn)
|
|
|
|
return await asyncio.to_thread(_run)
|
|
|
|
|
|
async def ensure_default_user() -> int:
|
|
"""No auth system yet (Phase 3) - all chat activity is attributed to one service user."""
|
|
row = await db_query("SELECT id FROM users ORDER BY id LIMIT 1", fetch="one")
|
|
if row:
|
|
return row["id"]
|
|
row = await db_query(
|
|
"""
|
|
INSERT INTO users (username, email, password_hash, role)
|
|
VALUES ('jarvis-service', 'jarvis-service@local', 'n/a', 'service')
|
|
RETURNING id
|
|
""",
|
|
fetch="one",
|
|
)
|
|
return row["id"]
|
|
|
|
|
|
async def create_conversation(user_id: int, title: Optional[str], context: Optional[dict]) -> int:
|
|
row = await db_query(
|
|
"INSERT INTO conversations (user_id, title, context) VALUES (%s, %s, %s) RETURNING id",
|
|
(user_id, title, Json(context) if context is not None else None),
|
|
fetch="one",
|
|
)
|
|
return row["id"]
|
|
|
|
|
|
async def get_conversation(conversation_id: int):
|
|
return await db_query(
|
|
"SELECT id, user_id, title, created_at FROM conversations WHERE id = %s",
|
|
(conversation_id,),
|
|
fetch="one",
|
|
)
|
|
|
|
|
|
async def get_messages(conversation_id: int):
|
|
return await db_query(
|
|
"""
|
|
SELECT role, content, tokens_used, created_at FROM messages
|
|
WHERE conversation_id = %s ORDER BY created_at ASC
|
|
""",
|
|
(conversation_id,),
|
|
fetch="all",
|
|
)
|
|
|
|
|
|
async def save_message(conversation_id: int, user_id: int, role: str, content: str, tokens_used: Optional[int]):
|
|
await db_query(
|
|
"""
|
|
INSERT INTO messages (conversation_id, user_id, role, content, tokens_used)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
""",
|
|
(conversation_id, user_id, role, content, tokens_used),
|
|
)
|
|
|
|
|
|
def _vector_literal(embedding: list) -> str:
|
|
return "[" + ",".join(repr(float(x)) for x in embedding) + "]"
|
|
|
|
|
|
async def insert_document(title: str, content: str, document_type: str) -> int:
|
|
row = await db_query(
|
|
"INSERT INTO documents (user_id, title, content, document_type) VALUES (%s, %s, %s, %s) RETURNING id",
|
|
(DEFAULT_USER_ID, title, content, document_type),
|
|
fetch="one",
|
|
)
|
|
return row["id"]
|
|
|
|
|
|
async def insert_document_chunks(document_id: int, chunks: list):
|
|
for index, (content, embedding) in enumerate(chunks):
|
|
await db_query(
|
|
"""
|
|
INSERT INTO document_chunks (document_id, chunk_index, content, embedding)
|
|
VALUES (%s, %s, %s, %s::vector)
|
|
""",
|
|
(document_id, index, content, _vector_literal(embedding)),
|
|
)
|
|
|
|
|
|
async def search_chunks(query_embedding: list, limit: int):
|
|
return await db_query(
|
|
"""
|
|
SELECT dc.content, dc.chunk_index, d.id AS document_id, d.title,
|
|
dc.embedding <=> %s::vector AS distance
|
|
FROM document_chunks dc
|
|
JOIN documents d ON d.id = dc.document_id
|
|
ORDER BY distance ASC
|
|
LIMIT %s
|
|
""",
|
|
(_vector_literal(query_embedding), limit),
|
|
fetch="all",
|
|
)
|
|
|
|
|
|
async def get_latest_weather():
|
|
return await db_query(
|
|
"""
|
|
SELECT location, temperature_c, condition_text, fetched_at
|
|
FROM weather_cache ORDER BY fetched_at DESC LIMIT 1
|
|
""",
|
|
fetch="one",
|
|
)
|
|
|
|
|
|
async def get_latest_emails(limit: int = 10):
|
|
return await db_query(
|
|
"SELECT sender, subject, received_at FROM email_cache ORDER BY received_at DESC LIMIT %s",
|
|
(limit,),
|
|
fetch="all",
|
|
)
|
|
|
|
|
|
async def insert_memory_fact(user_id: int, content: str) -> int:
|
|
row = await db_query(
|
|
"INSERT INTO memory_facts (user_id, content) VALUES (%s, %s) RETURNING id",
|
|
(user_id, content),
|
|
fetch="one",
|
|
)
|
|
return row["id"]
|
|
|
|
|
|
async def search_memory_facts(query: str) -> list:
|
|
return await db_query(
|
|
"SELECT id, content FROM memory_facts WHERE content ILIKE %s ORDER BY created_at ASC",
|
|
(f"%{query}%",),
|
|
fetch="all",
|
|
)
|
|
|
|
|
|
async def delete_memory_fact(fact_id: int):
|
|
await db_query("DELETE FROM memory_facts WHERE id = %s", (fact_id,))
|
|
|
|
|
|
async def get_all_memory_facts() -> list:
|
|
return await db_query(
|
|
"SELECT id, content FROM memory_facts ORDER BY created_at ASC",
|
|
fetch="all",
|
|
)
|
|
|
|
|
|
async def upsert_conversation_summary(conversation_id: int, summary: str, embedding: list):
|
|
await db_query(
|
|
"""
|
|
INSERT INTO conversation_summaries (conversation_id, summary, embedding)
|
|
VALUES (%s, %s, %s::vector)
|
|
ON CONFLICT (conversation_id) DO UPDATE
|
|
SET summary = EXCLUDED.summary, embedding = EXCLUDED.embedding, updated_at = CURRENT_TIMESTAMP
|
|
""",
|
|
(conversation_id, summary, _vector_literal(embedding)),
|
|
)
|
|
|
|
|
|
async def search_similar_conversation_summaries(query_embedding: list, exclude_conversation_id: int, limit: int = 3) -> list:
|
|
return await db_query(
|
|
"""
|
|
SELECT conversation_id, summary, embedding <=> %s::vector AS distance
|
|
FROM conversation_summaries
|
|
WHERE conversation_id != %s
|
|
ORDER BY distance ASC
|
|
LIMIT %s
|
|
""",
|
|
(_vector_literal(query_embedding), exclude_conversation_id, limit),
|
|
fetch="all",
|
|
)
|
|
|
|
|
|
async def build_memory_context(user_message: str, conversation_id: int) -> str:
|
|
sections = []
|
|
|
|
facts = await get_all_memory_facts()
|
|
if facts:
|
|
facts_lines = "\n".join(f"- {f['content']}" for f in facts)
|
|
sections.append(f"Bekannte Fakten ueber den Nutzer:\n{facts_lines}")
|
|
|
|
try:
|
|
query_embedding = await get_embedding(user_message)
|
|
summaries = await search_similar_conversation_summaries(query_embedding, conversation_id, limit=3)
|
|
except Exception as e:
|
|
logger.warning(f"Memory retrieval skipped, embedding/search failed: {e}")
|
|
summaries = []
|
|
|
|
if summaries:
|
|
summary_lines = "\n".join(f"- {s['summary']}" for s in summaries)
|
|
sections.append(f"Relevante fruehere Gespraeche:\n{summary_lines}")
|
|
|
|
return "\n\n".join(sections)
|
|
|
|
|
|
def _caldav_calendar():
|
|
dav_root = NEXTCLOUD_CALDAV_URL.split("/calendars/")[0] + "/"
|
|
client = caldav.DAVClient(url=dav_root, username=NEXTCLOUD_USER, password=NEXTCLOUD_APP_PASSWORD)
|
|
return client.calendar(url=NEXTCLOUD_CALDAV_URL)
|
|
|
|
|
|
def _list_upcoming_events_sync(days_ahead: int) -> list:
|
|
calendar = _caldav_calendar()
|
|
start = datetime.now()
|
|
end = start + timedelta(days=days_ahead)
|
|
results = calendar.search(start=start, end=end, event=True, expand=True)
|
|
events = []
|
|
for result in results:
|
|
comp = result.icalendar_component
|
|
dtstart = comp.get("dtstart").dt
|
|
dtend_prop = comp.get("dtend")
|
|
dtend = dtend_prop.dt if dtend_prop else dtstart
|
|
events.append(
|
|
{
|
|
"uid": str(comp.get("uid", "")),
|
|
"summary": str(comp.get("summary", "")),
|
|
"start": dtstart.isoformat(),
|
|
"end": dtend.isoformat(),
|
|
"description": str(comp.get("description", "")),
|
|
"all_day": not isinstance(dtstart, datetime),
|
|
}
|
|
)
|
|
events.sort(key=lambda e: e["start"])
|
|
return events
|
|
|
|
|
|
async def list_upcoming_events(days_ahead: int) -> list:
|
|
return await asyncio.to_thread(_list_upcoming_events_sync, days_ahead)
|
|
|
|
|
|
CALENDAR_TIMEZONE = ZoneInfo("Europe/Berlin")
|
|
|
|
|
|
def _as_calendar_local(dt: datetime) -> datetime:
|
|
"""Naive ISO datetimes (from Claude tool calls) mean local calendar time,
|
|
not UTC - without this, events land 1-2 hours off in the calendar."""
|
|
if dt.tzinfo is None:
|
|
return dt.replace(tzinfo=CALENDAR_TIMEZONE)
|
|
return dt
|
|
|
|
|
|
def _create_event_sync(summary: str, start: str, end: str, description: str = "") -> dict:
|
|
calendar = _caldav_calendar()
|
|
calendar.save_event(
|
|
dtstart=_as_calendar_local(datetime.fromisoformat(start)),
|
|
dtend=_as_calendar_local(datetime.fromisoformat(end)),
|
|
summary=summary,
|
|
description=description,
|
|
)
|
|
return {"summary": summary, "start": start, "end": end, "description": description}
|
|
|
|
|
|
async def create_event(summary: str, start: str, end: str, description: str = "") -> dict:
|
|
return await asyncio.to_thread(_create_event_sync, summary, start, end, description)
|
|
|
|
|
|
def _update_event_sync(
|
|
uid: str,
|
|
start: str = None,
|
|
end: str = None,
|
|
summary: str = None,
|
|
description: str = None,
|
|
) -> dict:
|
|
calendar = _caldav_calendar()
|
|
event = calendar.event_by_uid(uid)
|
|
comp = event.icalendar_component
|
|
if start is not None:
|
|
comp["dtstart"].dt = _as_calendar_local(datetime.fromisoformat(start))
|
|
if end is not None:
|
|
comp["dtend"].dt = _as_calendar_local(datetime.fromisoformat(end))
|
|
if summary is not None:
|
|
comp["summary"] = summary
|
|
if description is not None:
|
|
comp["description"] = description
|
|
event.save()
|
|
return {
|
|
"uid": uid,
|
|
"summary": str(comp.get("summary", "")),
|
|
"start": comp["dtstart"].dt.isoformat(),
|
|
"end": comp["dtend"].dt.isoformat(),
|
|
}
|
|
|
|
|
|
async def update_event(
|
|
uid: str,
|
|
start: str = None,
|
|
end: str = None,
|
|
summary: str = None,
|
|
description: str = None,
|
|
) -> dict:
|
|
return await asyncio.to_thread(_update_event_sync, uid, start, end, summary, description)
|
|
|
|
|
|
def _delete_event_sync(uid: str) -> dict:
|
|
calendar = _caldav_calendar()
|
|
event = calendar.event_by_uid(uid)
|
|
event.delete()
|
|
return {"deleted": True, "uid": uid}
|
|
|
|
|
|
async def delete_event(uid: str) -> dict:
|
|
return await asyncio.to_thread(_delete_event_sync, uid)
|
|
|
|
|
|
def _deck_base_url() -> str:
|
|
host = urlparse(NEXTCLOUD_CALDAV_URL).netloc
|
|
return f"https://{host}/index.php/apps/deck/api/v1.0"
|
|
|
|
|
|
async def _deck_request(method: str, path: str, json_body: dict = None):
|
|
"""Generic helper for the Nextcloud Deck REST API - all Deck business
|
|
logic goes through this single function so it is the only place tests
|
|
need to mock aiohttp directly."""
|
|
url = f"{_deck_base_url()}{path}"
|
|
auth = aiohttp.BasicAuth(NEXTCLOUD_USER, NEXTCLOUD_APP_PASSWORD)
|
|
headers = {"OCS-APIRequest": "true", "Content-Type": "application/json"}
|
|
timeout = aiohttp.ClientTimeout(total=30)
|
|
async with aiohttp.ClientSession(timeout=timeout, auth=auth, headers=headers) as session:
|
|
async with session.request(method, url, json=json_body) as resp:
|
|
if resp.status >= 400:
|
|
body = await resp.text()
|
|
raise RuntimeError(f"Deck API {method} {path} failed ({resp.status}): {body}")
|
|
if resp.status == 204:
|
|
return None
|
|
return await resp.json()
|
|
|
|
|
|
_deck_cache: dict = {}
|
|
|
|
_DECK_BOARD_TITLE = "Bestellungen"
|
|
_DECK_STACK_TITLES = ["Offen", "In Arbeit", "Erledigt"]
|
|
|
|
|
|
async def _ensure_deck_board() -> dict:
|
|
"""Finds (or lazily creates) the 'Bestellungen' board and its three
|
|
status stacks. Result is cached in _deck_cache after the first call -
|
|
board/stack ids don't change at runtime, and finding/creating them
|
|
costs several sequential Deck API calls."""
|
|
if "board_id" in _deck_cache:
|
|
return _deck_cache
|
|
|
|
boards = await _deck_request("GET", "/boards")
|
|
board = next((b for b in boards if b["title"] == _DECK_BOARD_TITLE), None)
|
|
if board is None:
|
|
board = await _deck_request("POST", "/boards", {"title": _DECK_BOARD_TITLE, "color": "0082C9"})
|
|
board_id = board["id"]
|
|
|
|
stacks = await _deck_request("GET", f"/boards/{board_id}/stacks")
|
|
existing_by_title = {s["title"]: s["id"] for s in stacks}
|
|
|
|
stack_ids = {}
|
|
for order, title in enumerate(_DECK_STACK_TITLES):
|
|
if title in existing_by_title:
|
|
stack_ids[title] = existing_by_title[title]
|
|
else:
|
|
created = await _deck_request("POST", f"/boards/{board_id}/stacks", {"title": title, "order": order})
|
|
stack_ids[title] = created["id"]
|
|
|
|
_deck_cache["board_id"] = board_id
|
|
_deck_cache["stacks"] = stack_ids
|
|
return _deck_cache
|
|
|
|
|
|
async def create_order(customer: str, description: str, due_date: str = None) -> dict:
|
|
board = await _ensure_deck_board()
|
|
stack_id = board["stacks"]["Offen"]
|
|
|
|
body = {"title": customer, "description": description, "type": "plain", "order": 999}
|
|
if due_date:
|
|
body["duedate"] = _as_calendar_local(datetime.fromisoformat(due_date)).isoformat()
|
|
|
|
card = await _deck_request("POST", f"/boards/{board['board_id']}/stacks/{stack_id}/cards", body)
|
|
return {
|
|
"id": card["id"],
|
|
"customer": customer,
|
|
"description": description,
|
|
"status": "Offen",
|
|
"due_date": due_date,
|
|
}
|
|
|
|
|
|
async def list_orders(status: str = None) -> list:
|
|
board = await _ensure_deck_board()
|
|
stacks = await _deck_request("GET", f"/boards/{board['board_id']}/stacks")
|
|
|
|
orders = []
|
|
for stack in stacks:
|
|
if status and stack["title"] != status:
|
|
continue
|
|
for card in stack.get("cards", []):
|
|
orders.append({
|
|
"id": card["id"],
|
|
"customer": card["title"],
|
|
"description": card.get("description", ""),
|
|
"status": stack["title"],
|
|
"due_date": card.get("duedate"),
|
|
})
|
|
return orders
|
|
|
|
|
|
def _decode_mime_words(value: str) -> str:
|
|
parts = decode_header(value)
|
|
decoded = []
|
|
for text, charset in parts:
|
|
if isinstance(text, bytes):
|
|
decoded.append(text.decode(charset or "utf-8", errors="replace"))
|
|
else:
|
|
decoded.append(text)
|
|
return "".join(decoded)
|
|
|
|
|
|
def _list_recent_emails_sync(limit: int = 10) -> list:
|
|
conn = imaplib.IMAP4(EMAIL_IMAP_HOST, EMAIL_IMAP_PORT)
|
|
conn.starttls(ssl.create_default_context())
|
|
conn.login(EMAIL_USER, EMAIL_PASSWORD)
|
|
conn.select("INBOX", readonly=True)
|
|
_, data = conn.search(None, "ALL")
|
|
ids = data[0].split()[-limit:]
|
|
|
|
emails = []
|
|
for msg_id in reversed(ids):
|
|
_, msg_data = conn.fetch(msg_id, "(BODY.PEEK[HEADER.FIELDS (FROM SUBJECT DATE)] FLAGS)")
|
|
flags_line = msg_data[0][0].decode("utf-8", errors="replace")
|
|
msg = message_from_bytes(msg_data[0][1])
|
|
emails.append(
|
|
{
|
|
"from": _decode_mime_words(msg.get("From", "")),
|
|
"subject": _decode_mime_words(msg.get("Subject", "")),
|
|
"date": msg.get("Date", ""),
|
|
"unread": "\\Seen" not in flags_line,
|
|
}
|
|
)
|
|
conn.logout()
|
|
return emails
|
|
|
|
|
|
async def list_recent_emails(limit: int = 10) -> list:
|
|
return await asyncio.to_thread(_list_recent_emails_sync, limit)
|
|
|
|
|
|
def _send_email_sync(to: str, subject: str, body: str) -> dict:
|
|
msg = EmailMessage()
|
|
msg["From"] = EMAIL_USER
|
|
msg["To"] = to
|
|
msg["Subject"] = subject
|
|
msg["Date"] = formatdate(localtime=True)
|
|
msg["Message-ID"] = make_msgid()
|
|
msg.set_content(body)
|
|
|
|
conn = smtplib.SMTP_SSL(EMAIL_SMTP_HOST, EMAIL_SMTP_PORT, context=ssl.create_default_context())
|
|
conn.login(EMAIL_USER, EMAIL_PASSWORD)
|
|
conn.send_message(msg)
|
|
conn.quit()
|
|
return {"to": to, "subject": subject, "sent": True}
|
|
|
|
|
|
async def send_email(to: str, subject: str, body: str) -> dict:
|
|
return await asyncio.to_thread(_send_email_sync, to, subject, body)
|
|
|
|
|
|
async def remember_fact(fact: str) -> dict:
|
|
fact_id = await insert_memory_fact(DEFAULT_USER_ID, fact)
|
|
return {"id": fact_id, "content": fact}
|
|
|
|
|
|
async def forget_fact(query: str) -> dict:
|
|
matches = await search_memory_facts(query)
|
|
if not matches:
|
|
return {"deleted": False, "matches": [], "message": "Kein passender Fakt gefunden."}
|
|
if len(matches) > 1:
|
|
return {"deleted": False, "matches": [{"id": m["id"], "content": m["content"]} for m in matches]}
|
|
await delete_memory_fact(matches[0]["id"])
|
|
return {"deleted": True, "content": matches[0]["content"]}
|
|
|
|
|
|
CALENDAR_TOOLS = [
|
|
{
|
|
"name": "list_calendar_events",
|
|
"description": "List upcoming events from the FFW-Onza-Alle calendar within the next N days.",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"days_ahead": {
|
|
"type": "integer",
|
|
"description": "How many days ahead to look, e.g. 7 for the next week",
|
|
}
|
|
},
|
|
"required": ["days_ahead"],
|
|
},
|
|
},
|
|
{
|
|
"name": "create_calendar_event",
|
|
"description": "Create a new event in the FFW-Onza-Alle calendar.",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"summary": {"type": "string", "description": "Event title"},
|
|
"start": {"type": "string", "description": "Start date/time in ISO 8601, e.g. 2026-09-20T10:00:00"},
|
|
"end": {"type": "string", "description": "End date/time in ISO 8601, e.g. 2026-09-20T11:00:00"},
|
|
"description": {"type": "string", "description": "Optional longer description"},
|
|
},
|
|
"required": ["summary", "start", "end"],
|
|
},
|
|
},
|
|
{
|
|
"name": "update_calendar_event",
|
|
"description": (
|
|
"Reschedule or rename an existing event in the FFW-Onza-Alle "
|
|
"calendar. Call list_calendar_events first if you don't already "
|
|
"know the event's uid. Only pass the fields that should change."
|
|
),
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"uid": {"type": "string", "description": "The event's uid, from list_calendar_events"},
|
|
"start": {"type": "string", "description": "New start date/time in ISO 8601"},
|
|
"end": {"type": "string", "description": "New end date/time in ISO 8601"},
|
|
"summary": {"type": "string", "description": "New event title"},
|
|
"description": {"type": "string", "description": "New description"},
|
|
},
|
|
"required": ["uid"],
|
|
},
|
|
},
|
|
{
|
|
"name": "delete_calendar_event",
|
|
"description": (
|
|
"Permanently delete an event from the FFW-Onza-Alle calendar. "
|
|
"Only call this after the user has explicitly confirmed the "
|
|
"deletion in the conversation."
|
|
),
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"uid": {"type": "string", "description": "The event's uid, from list_calendar_events"},
|
|
},
|
|
"required": ["uid"],
|
|
},
|
|
},
|
|
{
|
|
"name": "list_recent_emails",
|
|
"description": "List the most recent emails in the kontakt@mbo-tech-it.de inbox.",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"limit": {"type": "integer", "description": "How many recent emails to list, e.g. 10"}
|
|
},
|
|
"required": ["limit"],
|
|
},
|
|
},
|
|
{
|
|
"name": "send_email",
|
|
"description": (
|
|
"Send an email from kontakt@mbo-tech-it.de. Only call this after "
|
|
"the user has explicitly confirmed recipient, subject and body "
|
|
"in the conversation."
|
|
),
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"to": {"type": "string", "description": "Recipient email address"},
|
|
"subject": {"type": "string", "description": "Email subject"},
|
|
"body": {"type": "string", "description": "Plain text email body"},
|
|
},
|
|
"required": ["to", "subject", "body"],
|
|
},
|
|
},
|
|
]
|
|
|
|
MEMORY_TOOLS = [
|
|
{
|
|
"name": "remember_fact",
|
|
"description": "Store a fact the user explicitly asked to remember, so it is available in future chats.",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"fact": {"type": "string", "description": "The fact to remember, phrased as a standalone statement"},
|
|
},
|
|
"required": ["fact"],
|
|
},
|
|
},
|
|
{
|
|
"name": "forget_fact",
|
|
"description": (
|
|
"Search remembered facts matching a query and delete it if exactly "
|
|
"one matches. Only call after the user has explicitly confirmed "
|
|
"which fact to forget in the conversation."
|
|
),
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string", "description": "Text to search for among remembered facts"},
|
|
},
|
|
"required": ["query"],
|
|
},
|
|
},
|
|
]
|
|
|
|
CALENDAR_ASSISTANT_INSTRUCTIONS = (
|
|
"Du hast Zugriff auf den Kalender 'FFW-Onza-Alle' ueber die Tools "
|
|
"list_calendar_events, create_calendar_event, update_calendar_event und "
|
|
"delete_calendar_event. create_calendar_event und update_calendar_event "
|
|
"kannst du direkt aufrufen, sobald du die noetigen Angaben hast - keine "
|
|
"Rueckfrage noetig. Bevor du delete_calendar_event aufrufst, frage den "
|
|
"Nutzer aber immer explizit im Klartext nach Bestaetigung und rufe das "
|
|
"Tool erst auf, nachdem der Nutzer im naechsten Chat-Beitrag zugestimmt "
|
|
"hat. Fuer update_calendar_event und delete_calendar_event brauchst du "
|
|
"die uid des Termins - ruf dafuer zuerst list_calendar_events auf, falls "
|
|
"du sie noch nicht aus dem Gespraech kennst.\n\n"
|
|
"Du hast ausserdem Zugriff auf das Postfach kontakt@mbo-tech-it.de ueber "
|
|
"die Tools list_recent_emails und send_email. list_recent_emails kannst "
|
|
"du direkt aufrufen. Bevor du send_email aufrufst, zeige dem Nutzer "
|
|
"Empfaenger, Betreff und Text zur Kontrolle und rufe das Tool erst auf, "
|
|
"nachdem der Nutzer im naechsten Chat-Beitrag zugestimmt hat - "
|
|
"kontakt@mbo-tech-it.de ist die offizielle Firmenadresse, eine "
|
|
"versendete Mail laesst sich nicht zurueckholen."
|
|
)
|
|
|
|
MEMORY_ASSISTANT_INSTRUCTIONS = (
|
|
"Du hast ausserdem Zugriff auf ein chatuebergreifendes Gedaechtnis ueber "
|
|
"die Tools remember_fact und forget_fact. Wenn der Nutzer dich explizit "
|
|
"bittet, dir etwas zu merken (z.B. 'merke dir, dass...'), rufe "
|
|
"remember_fact direkt auf - keine Rueckfrage noetig. Wenn der Nutzer "
|
|
"dich bittet, einen gemerkten Fakt zu vergessen, frage zuerst explizit "
|
|
"im Klartext nach Bestaetigung, welcher Fakt gemeint ist, und rufe "
|
|
"forget_fact erst auf, nachdem der Nutzer zugestimmt hat. Gibt "
|
|
"forget_fact mehrere moegliche Treffer zurueck, liste sie im Chat auf "
|
|
"und frage nach, welcher gemeint ist, statt den falschen zu loeschen."
|
|
)
|
|
|
|
|
|
async def execute_tool(name: str, tool_input: dict) -> str:
|
|
if name == "list_calendar_events":
|
|
events = await list_upcoming_events(tool_input.get("days_ahead", 14))
|
|
return json.dumps(events)
|
|
if name == "create_calendar_event":
|
|
result = await create_event(
|
|
tool_input["summary"],
|
|
tool_input["start"],
|
|
tool_input["end"],
|
|
tool_input.get("description", ""),
|
|
)
|
|
return json.dumps(result)
|
|
if name == "update_calendar_event":
|
|
result = await update_event(
|
|
tool_input["uid"],
|
|
tool_input.get("start"),
|
|
tool_input.get("end"),
|
|
tool_input.get("summary"),
|
|
tool_input.get("description"),
|
|
)
|
|
return json.dumps(result)
|
|
if name == "delete_calendar_event":
|
|
result = await delete_event(tool_input["uid"])
|
|
return json.dumps(result)
|
|
if name == "list_recent_emails":
|
|
emails = await list_recent_emails(tool_input.get("limit", 10))
|
|
return json.dumps(emails)
|
|
if name == "send_email":
|
|
result = await send_email(tool_input["to"], tool_input["subject"], tool_input["body"])
|
|
return json.dumps(result)
|
|
if name == "remember_fact":
|
|
result = await remember_fact(tool_input["fact"])
|
|
return json.dumps(result)
|
|
if name == "forget_fact":
|
|
result = await forget_fact(tool_input["query"])
|
|
return json.dumps(result)
|
|
raise ValueError(f"Unknown tool: {name}")
|
|
|
|
|
|
MAX_TOOL_ROUNDS = 5
|
|
|
|
|
|
async def run_chat_completion(claude_messages: list, conversation_id: int):
|
|
latest_user_message = claude_messages[-1]["content"]
|
|
memory_context = await build_memory_context(latest_user_message, conversation_id)
|
|
|
|
system_prompt = f"{CLAUDE_SYSTEM_PROMPT}\n\n{CALENDAR_ASSISTANT_INSTRUCTIONS}\n\n{MEMORY_ASSISTANT_INSTRUCTIONS}"
|
|
if memory_context:
|
|
system_prompt = f"{system_prompt}\n\n{memory_context}"
|
|
|
|
messages = list(claude_messages)
|
|
total_input = 0
|
|
total_output = 0
|
|
|
|
for _ in range(MAX_TOOL_ROUNDS):
|
|
completion = await asyncio.to_thread(
|
|
claude_client.messages.create,
|
|
model=CLAUDE_MODEL,
|
|
max_tokens=1024,
|
|
system=system_prompt,
|
|
tools=CALENDAR_TOOLS + MEMORY_TOOLS,
|
|
messages=messages,
|
|
)
|
|
total_input += completion.usage.input_tokens
|
|
total_output += completion.usage.output_tokens
|
|
|
|
if completion.stop_reason != "tool_use":
|
|
response_text = "".join(b.text for b in completion.content if b.type == "text")
|
|
return response_text, total_output, total_input + total_output
|
|
|
|
tool_results = []
|
|
for block in completion.content:
|
|
if block.type != "tool_use":
|
|
continue
|
|
try:
|
|
result_text = await execute_tool(block.name, block.input)
|
|
tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": result_text})
|
|
except Exception as e:
|
|
tool_results.append(
|
|
{"type": "tool_result", "tool_use_id": block.id, "content": str(e), "is_error": True}
|
|
)
|
|
|
|
messages = messages + [
|
|
{"role": "assistant", "content": completion.content},
|
|
{"role": "user", "content": tool_results},
|
|
]
|
|
|
|
return (
|
|
"Entschuldigung, das hat zu viele Zwischenschritte gebraucht. Bitte formuliere die Anfrage neu.",
|
|
total_output,
|
|
total_input + total_output,
|
|
)
|
|
|
|
|
|
SUMMARY_SYSTEM_PROMPT = (
|
|
"Fasse das folgende Gespraech in 2-3 Saetzen auf Deutsch zusammen, damit "
|
|
"ein spaeterer Chat den Kontext wiedererkennt. Gib nur die Zusammenfassung "
|
|
"aus, ohne Einleitung."
|
|
)
|
|
|
|
|
|
async def update_conversation_summary(conversation_id: int):
|
|
try:
|
|
history = await get_messages(conversation_id)
|
|
claude_messages = [{"role": m["role"], "content": m["content"]} for m in history]
|
|
completion = await asyncio.to_thread(
|
|
claude_client.messages.create,
|
|
model=CLAUDE_MODEL,
|
|
max_tokens=200,
|
|
system=SUMMARY_SYSTEM_PROMPT,
|
|
messages=claude_messages,
|
|
)
|
|
summary_text = "".join(b.text for b in completion.content if b.type == "text")
|
|
embedding = await get_embedding(summary_text)
|
|
await upsert_conversation_summary(conversation_id, summary_text, embedding)
|
|
except Exception as e:
|
|
logger.error(f"Conversation summary update failed for conversation {conversation_id}: {e}")
|
|
|
|
|
|
_background_tasks: set = set()
|
|
|
|
|
|
def _spawn_background_task(coro):
|
|
task = asyncio.create_task(coro)
|
|
_background_tasks.add(task)
|
|
task.add_done_callback(_background_tasks.discard)
|
|
return task
|
|
|
|
|
|
# ============ MODELS ============
|
|
class ChatRequest(BaseModel):
|
|
conversation_id: Optional[int] = None
|
|
message: str
|
|
context: Optional[dict] = None
|
|
|
|
class ChatResponse(BaseModel):
|
|
conversation_id: int
|
|
response: str
|
|
tokens_used: int
|
|
timestamp: datetime
|
|
|
|
class TaskRequest(BaseModel):
|
|
title: str
|
|
description: str
|
|
task_type: str
|
|
priority: int = 0
|
|
|
|
class TaskResponse(BaseModel):
|
|
id: int
|
|
title: str
|
|
status: str
|
|
created_at: datetime
|
|
|
|
# ============ HEALTH CHECKS ============
|
|
async def check_postgres():
|
|
start = time.perf_counter()
|
|
try:
|
|
await asyncio.wait_for(db_query("SELECT 1", fetch="one"), timeout=3)
|
|
return {"status": "ok", "response_time_ms": round((time.perf_counter() - start) * 1000, 1)}
|
|
except Exception as e:
|
|
return {"status": "error", "response_time_ms": round((time.perf_counter() - start) * 1000, 1), "error": str(e)}
|
|
|
|
|
|
async def check_redis():
|
|
start = time.perf_counter()
|
|
try:
|
|
client = aioredis.from_url(REDIS_URL, socket_connect_timeout=3)
|
|
try:
|
|
await asyncio.wait_for(client.ping(), timeout=3)
|
|
finally:
|
|
await client.aclose()
|
|
return {"status": "ok", "response_time_ms": round((time.perf_counter() - start) * 1000, 1)}
|
|
except Exception as e:
|
|
return {"status": "error", "response_time_ms": round((time.perf_counter() - start) * 1000, 1), "error": str(e)}
|
|
|
|
|
|
async def check_milvus():
|
|
"""Plain TCP reachability check - avoids pulling in the full pymilvus client
|
|
just for a health probe (not used for vector search yet, see Phase 3)."""
|
|
start = time.perf_counter()
|
|
try:
|
|
reader, writer = await asyncio.wait_for(
|
|
asyncio.open_connection(MILVUS_HOST, int(MILVUS_PORT)), timeout=3
|
|
)
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
return {"status": "ok", "response_time_ms": round((time.perf_counter() - start) * 1000, 1)}
|
|
except Exception as e:
|
|
return {"status": "error", "response_time_ms": round((time.perf_counter() - start) * 1000, 1), "error": str(e)}
|
|
|
|
|
|
async def check_http(url: str):
|
|
start = time.perf_counter()
|
|
try:
|
|
timeout = aiohttp.ClientTimeout(total=3)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.get(url) as resp:
|
|
ok = resp.status < 500
|
|
return {"status": "ok" if ok else "error", "response_time_ms": round((time.perf_counter() - start) * 1000, 1)}
|
|
except Exception as e:
|
|
return {"status": "error", "response_time_ms": round((time.perf_counter() - start) * 1000, 1), "error": str(e)}
|
|
|
|
|
|
async def check_ollama():
|
|
return await check_http(f"{OLLAMA_HOST}/api/tags")
|
|
|
|
|
|
async def check_n8n():
|
|
return await check_http(f"{N8N_URL}/healthz")
|
|
|
|
|
|
# ============ HEALTH CHECK ============
|
|
@app.get("/health")
|
|
async def health_check():
|
|
postgres, redis_status = await asyncio.gather(check_postgres(), check_redis())
|
|
return {
|
|
"status": "healthy",
|
|
"timestamp": datetime.now(),
|
|
"services": {
|
|
"api": "ok",
|
|
"database": postgres["status"],
|
|
"cache": redis_status["status"],
|
|
}
|
|
}
|
|
|
|
# ============ CHAT ENDPOINTS ============
|
|
@app.post("/api/v1/chat", response_model=ChatResponse, dependencies=[Depends(require_admin_key)])
|
|
async def chat(request: ChatRequest):
|
|
if not claude_client:
|
|
raise HTTPException(status_code=503, detail="CLAUDE_API_KEY is not configured")
|
|
if not pg_pool:
|
|
raise HTTPException(status_code=503, detail="Database is not configured")
|
|
|
|
try:
|
|
logger.info(f"Chat request: {request.message[:50]}...")
|
|
|
|
if request.conversation_id is not None:
|
|
conversation = await get_conversation(request.conversation_id)
|
|
if not conversation:
|
|
raise HTTPException(status_code=404, detail="Conversation not found")
|
|
conversation_id = conversation["id"]
|
|
history = await get_messages(conversation_id)
|
|
else:
|
|
title = request.message[:100]
|
|
conversation_id = await create_conversation(DEFAULT_USER_ID, title, request.context)
|
|
history = []
|
|
|
|
await save_message(conversation_id, DEFAULT_USER_ID, "user", request.message, None)
|
|
|
|
claude_messages = [{"role": m["role"], "content": m["content"]} for m in history]
|
|
claude_messages.append({"role": "user", "content": request.message})
|
|
|
|
response_text, output_tokens, tokens_used = await run_chat_completion(claude_messages, conversation_id)
|
|
|
|
await save_message(conversation_id, DEFAULT_USER_ID, "assistant", response_text, output_tokens)
|
|
|
|
_spawn_background_task(update_conversation_summary(conversation_id))
|
|
|
|
return ChatResponse(
|
|
conversation_id=conversation_id,
|
|
response=response_text,
|
|
tokens_used=tokens_used,
|
|
timestamp=datetime.now()
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Chat error: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/api/v1/conversations/{conversation_id}", dependencies=[Depends(require_admin_key)])
|
|
async def get_conversation_history(conversation_id: int):
|
|
"""Get conversation history"""
|
|
if not pg_pool:
|
|
raise HTTPException(status_code=503, detail="Database is not configured")
|
|
|
|
conversation = await get_conversation(conversation_id)
|
|
if not conversation:
|
|
raise HTTPException(status_code=404, detail="Conversation not found")
|
|
|
|
messages = await get_messages(conversation_id)
|
|
return {
|
|
"conversation_id": conversation_id,
|
|
"messages": messages
|
|
}
|
|
|
|
# ============ TASK ENDPOINTS ============
|
|
@app.post("/api/v1/tasks", response_model=TaskResponse, dependencies=[Depends(require_admin_key)])
|
|
async def create_task(request: TaskRequest):
|
|
"""Create a new automation task"""
|
|
try:
|
|
logger.info(f"Creating task: {request.title}")
|
|
|
|
# TODO: Save to PostgreSQL
|
|
return TaskResponse(
|
|
id=1,
|
|
title=request.title,
|
|
status="pending",
|
|
created_at=datetime.now()
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Task creation error: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/api/v1/tasks", dependencies=[Depends(require_admin_key)])
|
|
async def list_tasks(status: Optional[str] = None, limit: int = 50):
|
|
"""List tasks"""
|
|
# TODO: Query PostgreSQL
|
|
return {
|
|
"tasks": [],
|
|
"total": 0
|
|
}
|
|
|
|
# ============ DOCUMENT/KNOWLEDGE BASE ============
|
|
@app.post("/api/v1/documents", dependencies=[Depends(require_admin_key)])
|
|
async def upload_document(title: str, content: str, document_type: str = "general"):
|
|
"""Upload document to knowledge base"""
|
|
if not pg_pool:
|
|
raise HTTPException(status_code=503, detail="Database is not configured")
|
|
if not content.strip():
|
|
raise HTTPException(status_code=422, detail="content must not be empty")
|
|
|
|
try:
|
|
logger.info(f"Uploading document: {title}")
|
|
document_id = await insert_document(title, content, document_type)
|
|
chunks = chunk_text(content)
|
|
embedded_chunks = [(c, await get_embedding(c)) for c in chunks]
|
|
await insert_document_chunks(document_id, embedded_chunks)
|
|
|
|
return {
|
|
"document_id": document_id,
|
|
"title": title,
|
|
"status": "indexed",
|
|
"chunk_count": len(chunks),
|
|
}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Document upload error: {str(e)}")
|
|
raise HTTPException(status_code=503, detail=f"Embedding failed: {str(e)}")
|
|
|
|
@app.get("/api/v1/documents", dependencies=[Depends(require_admin_key)])
|
|
async def search_documents(query: str, limit: int = 10):
|
|
"""Search knowledge base using vector similarity"""
|
|
if not pg_pool:
|
|
raise HTTPException(status_code=503, detail="Database is not configured")
|
|
if not query.strip():
|
|
raise HTTPException(status_code=422, detail="query must not be empty")
|
|
|
|
try:
|
|
query_embedding = await get_embedding(query)
|
|
rows = await search_chunks(query_embedding, limit)
|
|
return {
|
|
"query": query,
|
|
"results": [
|
|
{
|
|
"document_id": r["document_id"],
|
|
"title": r["title"],
|
|
"content": r["content"],
|
|
"chunk_index": r["chunk_index"],
|
|
"distance": float(r["distance"]),
|
|
}
|
|
for r in rows
|
|
],
|
|
"count": len(rows),
|
|
}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Document search error: {str(e)}")
|
|
raise HTTPException(status_code=503, detail=f"Search failed: {str(e)}")
|
|
|
|
# ============ AUTOMATION/WORKFLOW ============
|
|
@app.post("/api/v1/workflows/trigger", dependencies=[Depends(require_admin_key)])
|
|
async def trigger_workflow(workflow_id: str, data: dict):
|
|
"""Trigger n8n workflow"""
|
|
try:
|
|
logger.info(f"Triggering workflow: {workflow_id}")
|
|
|
|
# TODO: Call n8n API
|
|
return {
|
|
"workflow_id": workflow_id,
|
|
"execution_id": "exec_123",
|
|
"status": "triggered"
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Workflow trigger error: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
# ============ ADMIN ENDPOINTS ============
|
|
@app.get("/api/v1/admin/stats", dependencies=[Depends(require_admin_key)])
|
|
async def get_stats():
|
|
"""Get system statistics"""
|
|
if not pg_pool:
|
|
raise HTTPException(status_code=503, detail="Database is not configured")
|
|
|
|
conversations, tasks, documents = await asyncio.gather(
|
|
db_query("SELECT COUNT(*) AS n FROM conversations", fetch="one"),
|
|
db_query("SELECT COUNT(*) AS n FROM tasks", fetch="one"),
|
|
db_query("SELECT COUNT(*) AS n FROM documents", fetch="one"),
|
|
)
|
|
return {
|
|
"timestamp": datetime.now(),
|
|
"conversations": conversations["n"],
|
|
"tasks": tasks["n"],
|
|
"documents": documents["n"],
|
|
}
|
|
|
|
@app.get("/api/v1/admin/health/detailed", dependencies=[Depends(require_admin_key)])
|
|
async def detailed_health():
|
|
"""Detailed health check with all services"""
|
|
postgres, redis_status, milvus, ollama, n8n = await asyncio.gather(
|
|
check_postgres(), check_redis(), check_milvus(), check_ollama(), check_n8n()
|
|
)
|
|
components = {
|
|
"api": {"status": "ok", "response_time_ms": 0},
|
|
"postgres": postgres,
|
|
"redis": redis_status,
|
|
"milvus": milvus,
|
|
"ollama": ollama,
|
|
"n8n": n8n,
|
|
}
|
|
overall = "healthy" if all(c["status"] == "ok" for c in components.values()) else "degraded"
|
|
return {
|
|
"status": overall,
|
|
"components": components,
|
|
}
|
|
|
|
@app.get("/api/v1/weather", dependencies=[Depends(require_admin_key)])
|
|
async def get_weather():
|
|
"""Latest cached weather reading, populated by the n8n 'Wetter Crailsheim' workflow"""
|
|
if not pg_pool:
|
|
raise HTTPException(status_code=503, detail="Database is not configured")
|
|
|
|
row = await get_latest_weather()
|
|
if not row:
|
|
raise HTTPException(status_code=503, detail="Weather data not available yet")
|
|
|
|
return {
|
|
"location": row["location"],
|
|
"temperature_c": float(row["temperature_c"]),
|
|
"condition_text": row["condition_text"],
|
|
"fetched_at": row["fetched_at"].isoformat(),
|
|
}
|
|
|
|
@app.get("/api/v1/calendar/events", dependencies=[Depends(require_admin_key)])
|
|
async def get_calendar_events(days: int = 14):
|
|
"""Upcoming events from the FFW-Onza-Alle Nextcloud calendar"""
|
|
try:
|
|
events = await list_upcoming_events(days)
|
|
return {"events": events}
|
|
except Exception as e:
|
|
logger.error(f"Calendar fetch error: {str(e)}")
|
|
raise HTTPException(status_code=503, detail=f"Calendar unavailable: {str(e)}")
|
|
|
|
@app.get("/api/v1/emails", dependencies=[Depends(require_admin_key)])
|
|
async def get_cached_emails(limit: int = 10):
|
|
"""Recently cached emails, populated by the n8n 'Neue E-Mails Cache' workflow"""
|
|
if not pg_pool:
|
|
raise HTTPException(status_code=503, detail="Database is not configured")
|
|
|
|
rows = await get_latest_emails(limit)
|
|
return {
|
|
"emails": [
|
|
{
|
|
"sender": row["sender"],
|
|
"subject": row["subject"],
|
|
"received_at": row["received_at"].isoformat(),
|
|
}
|
|
for row in rows
|
|
]
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(
|
|
app,
|
|
host="0.0.0.0",
|
|
port=8000,
|
|
log_level=os.getenv("LOG_LEVEL", "info").lower()
|
|
)
|