301 lines
9.5 KiB
Python
301 lines
9.5 KiB
Python
from fastapi import APIRouter, Depends, Request, Form, HTTPException
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
import httpx
|
|
from typing import Optional, List, Dict, Any
|
|
from app.deps import current_user
|
|
from app.services.config_loader import ConfigLoader
|
|
from app.config import get_settings
|
|
|
|
router = APIRouter()
|
|
settings = get_settings()
|
|
|
|
@router.get("/crumbforest/roles", response_class=HTMLResponse)
|
|
async def roles_dashboard(req: Request, user = Depends(current_user)):
|
|
"""
|
|
Show available roles based on user's group.
|
|
"""
|
|
# Determine user group (default to 'home' if not logged in, 'demo' if logged in but no group)
|
|
if not user:
|
|
user_group = 'home'
|
|
else:
|
|
user_group = user.get('user_group', 'demo')
|
|
|
|
# Load config
|
|
config = ConfigLoader.load_config()
|
|
all_roles = config.get('roles', {})
|
|
|
|
# Filter roles by group access
|
|
available_roles = {
|
|
rid: role for rid, role in all_roles.items()
|
|
if user_group in role.get('group_access', [])
|
|
}
|
|
|
|
# Get group config for template rendering
|
|
group_config = config.get('groups', {}).get(user_group, {})
|
|
|
|
return req.app.state.render(
|
|
req,
|
|
"crumbforest/roles_dashboard.html",
|
|
roles=available_roles,
|
|
user_group=user_group,
|
|
group_config=group_config
|
|
)
|
|
|
|
@router.get("/crumbforest/roles/{role_id}", response_class=HTMLResponse)
|
|
async def role_chat(req: Request, role_id: str, user = Depends(current_user)):
|
|
"""
|
|
Chat interface for a specific role.
|
|
"""
|
|
config = ConfigLoader.load_config()
|
|
role = config.get('roles', {}).get(role_id)
|
|
|
|
if not role:
|
|
raise HTTPException(404, "Role not found")
|
|
|
|
# Determine user group
|
|
if not user:
|
|
user_group = 'home'
|
|
else:
|
|
user_group = user.get('user_group', 'demo')
|
|
|
|
# Check access
|
|
if user_group not in role.get('group_access', []):
|
|
# Redirect to login if not logged in, else 403
|
|
if not user:
|
|
return RedirectResponse(f"/login?next={req.url.path}", status_code=302)
|
|
raise HTTPException(403, "Access denied")
|
|
|
|
group_config = config.get('groups', {}).get(user_group, {})
|
|
|
|
return req.app.state.render(
|
|
req,
|
|
"crumbforest/role_chat.html",
|
|
role=role,
|
|
group_config=group_config
|
|
)
|
|
|
|
@router.post("/crumbforest/roles/{role_id}/ask")
|
|
async def ask_role(
|
|
req: Request,
|
|
role_id: str,
|
|
question: str = Form(...),
|
|
user = Depends(current_user)
|
|
):
|
|
"""
|
|
Send question to role and get AI response.
|
|
"""
|
|
config = ConfigLoader.load_config()
|
|
role = config.get('roles', {}).get(role_id)
|
|
|
|
if not role:
|
|
raise HTTPException(404, "Role not found")
|
|
|
|
# Access check (simplified for API)
|
|
if not user:
|
|
raise HTTPException(401, "Login required")
|
|
|
|
user_group = user.get('user_group', 'demo')
|
|
if user_group not in role.get('group_access', []):
|
|
raise HTTPException(403, "Access denied")
|
|
|
|
# Get conversation history from session
|
|
history_key = f'role_history_{role_id}'
|
|
history = req.session.get(history_key, [])
|
|
|
|
# Build messages
|
|
messages = [
|
|
{"role": "system", "content": role['system_prompt']}
|
|
] + history + [
|
|
{"role": "user", "content": question}
|
|
]
|
|
|
|
# Call OpenRouter API
|
|
api_key = settings.openrouter_api_key
|
|
if not api_key:
|
|
return JSONResponse({
|
|
"answer": "⚠️ OpenRouter API Key missing in configuration.",
|
|
"error": "config_error"
|
|
}, status_code=500)
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
"https://openrouter.ai/api/v1/chat/completions",
|
|
headers={
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
"HTTP-Referer": "https://crumbforest.local", # Required by OpenRouter
|
|
"X-Title": "Crumbforest"
|
|
},
|
|
json={
|
|
"model": role['model'],
|
|
"temperature": role.get('temperature', 0.7),
|
|
"messages": messages
|
|
},
|
|
timeout=30.0
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
print(f"OpenRouter Error: {response.text}")
|
|
return JSONResponse({
|
|
"answer": "⚠️ Error communicating with AI provider.",
|
|
"details": response.text
|
|
}, status_code=502)
|
|
|
|
result = response.json()
|
|
|
|
answer = result['choices'][0]['message']['content']
|
|
|
|
# --- Code Execution Feature (Mock/Safe) ---
|
|
# If role has 'code_execution' feature and answer contains code block
|
|
if 'code_execution' in role.get('features', []) and '```' in answer:
|
|
# Simple mock execution for demonstration
|
|
# In production, this would need a secure sandbox (e.g. gVisor, Firecracker)
|
|
if "print(" in answer or "echo" in answer:
|
|
answer += "\n\n> [!NOTE]\n> **Code Execution Output** (Simulated):\n> ```\n> Hello from Crumbforest! 🌲\n> ```"
|
|
|
|
# Update history
|
|
history.append({"role": "user", "content": question})
|
|
history.append({"role": "assistant", "content": answer})
|
|
req.session[history_key] = history[-10:] # Keep last 10 exchanges
|
|
|
|
return JSONResponse({
|
|
"role": role_id,
|
|
"question": question,
|
|
"answer": answer,
|
|
"usage": result.get('usage', {})
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f"Exception in ask_role: {e}")
|
|
return JSONResponse({
|
|
"answer": "⚠️ Internal server error.",
|
|
"details": str(e)
|
|
}, status_code=500)
|
|
|
|
# --- Settings Endpoints ---
|
|
|
|
@router.get("/settings", response_class=HTMLResponse)
|
|
async def settings_page(req: Request, user = Depends(current_user)):
|
|
"""
|
|
User settings page.
|
|
"""
|
|
if not user:
|
|
return RedirectResponse(f"/login?next={req.url.path}", status_code=302)
|
|
|
|
config = ConfigLoader.load_config()
|
|
user_group = user.get('user_group', 'demo')
|
|
group_config = config.get('groups', {}).get(user_group, {})
|
|
theme_variants = config.get('theme_variants', {})
|
|
|
|
return req.app.state.render(
|
|
req,
|
|
"pages/settings.html",
|
|
group_config=group_config,
|
|
theme_variants=theme_variants
|
|
)
|
|
|
|
@router.post("/settings/theme")
|
|
async def update_theme(
|
|
req: Request,
|
|
theme: str = Form(...),
|
|
user = Depends(current_user)
|
|
):
|
|
"""
|
|
Update user theme preference.
|
|
"""
|
|
if not user:
|
|
raise HTTPException(401, "Login required")
|
|
|
|
# Verify theme exists
|
|
config = ConfigLoader.load_config()
|
|
if theme not in config.get('theme_variants', {}):
|
|
raise HTTPException(400, "Invalid theme")
|
|
|
|
# Update session
|
|
user['theme'] = theme
|
|
req.session['user'] = user
|
|
|
|
# Update DB
|
|
try:
|
|
from app.deps import get_db
|
|
from pymysql.cursors import DictCursor
|
|
with get_db().cursor(DictCursor) as cur:
|
|
cur.execute("UPDATE users SET theme=%s WHERE id=%s", (theme, user['id']))
|
|
except Exception as e:
|
|
print(f"Error updating theme in DB: {e}")
|
|
# Continue anyway, session is updated
|
|
|
|
return RedirectResponse("/settings", status_code=302)
|
|
|
|
@router.post("/settings/accessibility")
|
|
async def update_accessibility(
|
|
req: Request,
|
|
high_contrast: Optional[bool] = Form(None),
|
|
animation_reduced: Optional[bool] = Form(None),
|
|
font_size: str = Form("normal"),
|
|
user = Depends(current_user)
|
|
):
|
|
"""
|
|
Update accessibility settings.
|
|
"""
|
|
if not user:
|
|
raise HTTPException(401, "Login required")
|
|
|
|
import json
|
|
|
|
accessibility = {
|
|
"high_contrast": bool(high_contrast),
|
|
"animation_reduced": bool(animation_reduced),
|
|
"font_size": font_size
|
|
}
|
|
|
|
# Update session
|
|
user['accessibility'] = accessibility
|
|
|
|
# If high contrast is enabled, force that theme, otherwise revert to group default or keep current?
|
|
# For now, we just set the accessibility flags. The theme switcher handles the main theme.
|
|
# However, if high_contrast is ON, we might want to auto-switch to pico-high-contrast.
|
|
if accessibility['high_contrast']:
|
|
user['theme'] = 'pico-high-contrast'
|
|
elif user['theme'] == 'pico-high-contrast' and not accessibility['high_contrast']:
|
|
# Revert to default if turning off high contrast
|
|
user['theme'] = 'pico-default'
|
|
|
|
req.session['user'] = user
|
|
|
|
# Update DB
|
|
try:
|
|
from app.deps import get_db
|
|
from pymysql.cursors import DictCursor
|
|
with get_db().cursor(DictCursor) as cur:
|
|
cur.execute(
|
|
"UPDATE users SET accessibility=%s, theme=%s WHERE id=%s",
|
|
(json.dumps(accessibility), user['theme'], user['id'])
|
|
)
|
|
except Exception as e:
|
|
print(f"Error updating accessibility in DB: {e}")
|
|
|
|
return RedirectResponse("/settings", status_code=302)
|
|
|
|
@router.get("/crumbforest/roles/{role_id}/export")
|
|
async def export_history(req: Request, role_id: str, user = Depends(current_user)):
|
|
"""
|
|
Export chat history as JSON.
|
|
"""
|
|
if not user:
|
|
raise HTTPException(401, "Login required")
|
|
|
|
history_key = f'role_history_{role_id}'
|
|
history = req.session.get(history_key, [])
|
|
|
|
return JSONResponse({
|
|
"role": role_id,
|
|
"exported_at": str(import_datetime().now()),
|
|
"history": history
|
|
})
|
|
|
|
def import_datetime():
|
|
from datetime import datetime
|
|
return datetime
|