mirror of
https://github.com/OHV-IT/collabrix.git
synced 2025-12-15 16:48:36 +01:00
671 lines
20 KiB
Python
671 lines
20 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlmodel import Session, select
|
|
from typing import List
|
|
from app.database import get_session
|
|
from app.models import User, Department, UserDepartmentLink, Channel, Snippet, SnippetDepartmentLink, Language, Translation, UserRole
|
|
from app.schemas import (
|
|
DepartmentCreate, DepartmentResponse,
|
|
ChannelCreate, ChannelResponse,
|
|
UserResponse,
|
|
LanguageCreate,
|
|
LanguageResponse,
|
|
TranslationGroupResponse,
|
|
TranslationUpdateRequest
|
|
)
|
|
from app.auth import get_current_user
|
|
from pydantic import BaseModel
|
|
from app.services.translations import (
|
|
ensure_default_languages,
|
|
ensure_translation_entries,
|
|
ensure_translations_for_language,
|
|
get_translation_blueprint,
|
|
update_translation_timestamp,
|
|
)
|
|
|
|
router = APIRouter(prefix="/admin", tags=["Admin"])
|
|
|
|
|
|
class UserDepartmentAssignment(BaseModel):
|
|
user_id: int
|
|
department_id: int
|
|
|
|
|
|
class UserAdminUpdate(BaseModel):
|
|
user_id: int
|
|
role: UserRole
|
|
|
|
|
|
class SnippetDepartmentAccess(BaseModel):
|
|
snippet_id: int
|
|
department_id: int
|
|
enabled: bool
|
|
|
|
|
|
|
|
def require_admin(current_user: User = Depends(get_current_user)) -> User:
|
|
"""Verify that the current user is an admin or superadmin"""
|
|
if current_user.role not in [UserRole.ADMIN, UserRole.SUPERADMIN]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin privileges required"
|
|
)
|
|
return current_user
|
|
|
|
|
|
def require_superadmin(current_user: User = Depends(get_current_user)) -> User:
|
|
"""Verify that the current user is a superadmin"""
|
|
if current_user.role != UserRole.SUPERADMIN:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Superadmin privileges required"
|
|
)
|
|
return current_user
|
|
|
|
|
|
# ========== User Management ==========
|
|
|
|
@router.get("/users", response_model=List[UserResponse])
|
|
def get_all_users(
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Get all users (Admin only)"""
|
|
statement = select(User)
|
|
users = session.exec(statement).all()
|
|
return users
|
|
|
|
|
|
@router.patch("/users/{user_id}/role")
|
|
def update_user_role(
|
|
user_id: int,
|
|
body: UserAdminUpdate,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Update a user's role (Superadmin only)"""
|
|
user = session.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not found"
|
|
)
|
|
|
|
# Prevent superadmin from demoting themselves
|
|
if admin.id == user_id and body.role != UserRole.SUPERADMIN:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Cannot change your own superadmin role"
|
|
)
|
|
|
|
user.role = body.role
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
|
|
return {"message": f"User {user.username} role updated to {body.role.value}", "role": body.role}
|
|
|
|
|
|
# ========== Department Management ==========
|
|
|
|
@router.post("/departments", response_model=DepartmentResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_department(
|
|
department_data: DepartmentCreate,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Create a new department (Admin only)"""
|
|
department = Department(**department_data.model_dump())
|
|
session.add(department)
|
|
session.commit()
|
|
session.refresh(department)
|
|
return department
|
|
|
|
|
|
@router.get("/departments", response_model=List[DepartmentResponse])
|
|
def get_all_departments(
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Get all departments (Admin only)"""
|
|
statement = select(Department)
|
|
departments = session.exec(statement).all()
|
|
return departments
|
|
|
|
|
|
@router.put("/departments/{department_id}", response_model=DepartmentResponse)
|
|
def update_department(
|
|
department_id: int,
|
|
department_data: DepartmentCreate,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Update a department (Admin only)"""
|
|
department = session.get(Department, department_id)
|
|
if not department:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Department not found"
|
|
)
|
|
|
|
# Update fields
|
|
department.name = department_data.name
|
|
department.description = department_data.description
|
|
|
|
session.add(department)
|
|
session.commit()
|
|
session.refresh(department)
|
|
return department
|
|
|
|
|
|
@router.patch("/departments/{department_id}/snippets")
|
|
def toggle_department_snippets(
|
|
department_id: int,
|
|
enabled: bool,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Enable or disable snippet access for entire department (master switch)"""
|
|
department = session.get(Department, department_id)
|
|
if not department:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Department not found"
|
|
)
|
|
|
|
department.snippets_enabled = enabled
|
|
session.add(department)
|
|
session.commit()
|
|
session.refresh(department)
|
|
|
|
return {
|
|
"message": f"Snippet access {'enabled' if enabled else 'disabled'} for department {department.name}",
|
|
"department_id": department_id,
|
|
"snippets_enabled": enabled
|
|
}
|
|
|
|
|
|
@router.delete("/departments/{department_id}")
|
|
def delete_department(
|
|
department_id: int,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Delete a department (Admin only)"""
|
|
department = session.get(Department, department_id)
|
|
if not department:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Department not found"
|
|
)
|
|
|
|
# Check if department has channels
|
|
channels_statement = select(Channel).where(Channel.department_id == department_id)
|
|
channels = session.exec(channels_statement).all()
|
|
if channels:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Cannot delete department. It has {len(channels)} channel(s). Please delete channels first."
|
|
)
|
|
|
|
# Delete user-department associations
|
|
links_statement = select(UserDepartmentLink).where(UserDepartmentLink.department_id == department_id)
|
|
links = session.exec(links_statement).all()
|
|
for link in links:
|
|
session.delete(link)
|
|
|
|
# Now delete the department
|
|
session.delete(department)
|
|
session.commit()
|
|
return {"message": f"Department '{department.name}' deleted successfully"}
|
|
|
|
|
|
# ========== User-Department Assignment ==========
|
|
|
|
@router.post("/departments/{department_id}/members")
|
|
def assign_user_to_department(
|
|
department_id: int,
|
|
user_id: int,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Assign a user to a department (Admin only)"""
|
|
# Check if department exists
|
|
department = session.get(Department, department_id)
|
|
if not department:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Department not found"
|
|
)
|
|
|
|
# Check if user exists
|
|
user = session.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not found"
|
|
)
|
|
|
|
# Check if already assigned
|
|
statement = select(UserDepartmentLink).where(
|
|
UserDepartmentLink.user_id == user_id,
|
|
UserDepartmentLink.department_id == department_id
|
|
)
|
|
existing = session.exec(statement).first()
|
|
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="User already assigned to this department"
|
|
)
|
|
|
|
# Create assignment
|
|
link = UserDepartmentLink(user_id=user_id, department_id=department_id)
|
|
session.add(link)
|
|
session.commit()
|
|
|
|
return {"message": f"User {user.username} assigned to {department.name}"}
|
|
|
|
|
|
@router.delete("/departments/{department_id}/members/{user_id}")
|
|
def remove_user_from_department(
|
|
department_id: int,
|
|
user_id: int,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Remove a user from a department (Admin only)"""
|
|
statement = select(UserDepartmentLink).where(
|
|
UserDepartmentLink.user_id == user_id,
|
|
UserDepartmentLink.department_id == department_id
|
|
)
|
|
link = session.exec(statement).first()
|
|
|
|
if not link:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not assigned to this department"
|
|
)
|
|
|
|
session.delete(link)
|
|
session.commit()
|
|
|
|
return {"message": "User removed from department"}
|
|
|
|
|
|
@router.get("/departments/{department_id}/members", response_model=List[UserResponse])
|
|
def get_department_members(
|
|
department_id: int,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Get all members of a department (Admin only)"""
|
|
department = session.get(Department, department_id)
|
|
if not department:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Department not found"
|
|
)
|
|
|
|
return department.users
|
|
|
|
|
|
# ========== Channel Management ==========
|
|
|
|
@router.get("/channels", response_model=List[ChannelResponse])
|
|
def get_all_channels(
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_admin)
|
|
):
|
|
"""Get all channels (Admin only)"""
|
|
statement = select(Channel)
|
|
channels = session.exec(statement).all()
|
|
return channels
|
|
|
|
|
|
@router.post("/channels", response_model=ChannelResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_channel(
|
|
channel_data: ChannelCreate,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_admin)
|
|
):
|
|
"""Create a new channel (Admin only)"""
|
|
# Verify department exists
|
|
department = session.get(Department, channel_data.department_id)
|
|
if not department:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Department not found"
|
|
)
|
|
|
|
channel = Channel(**channel_data.model_dump())
|
|
session.add(channel)
|
|
session.commit()
|
|
session.refresh(channel)
|
|
return channel
|
|
|
|
|
|
@router.put("/channels/{channel_id}", response_model=ChannelResponse)
|
|
def update_channel(
|
|
channel_id: int,
|
|
channel_data: ChannelCreate,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_admin)
|
|
):
|
|
"""Update a channel (Admin only)"""
|
|
channel = session.get(Channel, channel_id)
|
|
if not channel:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Channel not found"
|
|
)
|
|
|
|
# Verify department exists
|
|
department = session.get(Department, channel_data.department_id)
|
|
if not department:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Department not found"
|
|
)
|
|
|
|
channel.name = channel_data.name
|
|
channel.description = channel_data.description
|
|
channel.department_id = channel_data.department_id
|
|
session.add(channel)
|
|
session.commit()
|
|
session.refresh(channel)
|
|
return channel
|
|
|
|
|
|
@router.delete("/channels/{channel_id}")
|
|
def delete_channel(
|
|
channel_id: int,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_admin)
|
|
):
|
|
"""Delete a channel (Admin only)"""
|
|
channel = session.get(Channel, channel_id)
|
|
if not channel:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Channel not found"
|
|
)
|
|
|
|
session.delete(channel)
|
|
session.commit()
|
|
return {"message": f"Channel '{channel.name}' deleted"}
|
|
|
|
|
|
# ========== Language & Translation Management ==========
|
|
|
|
|
|
@router.get("/languages", response_model=List[LanguageResponse])
|
|
def get_languages(
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""List all available UI languages."""
|
|
ensure_default_languages(session)
|
|
statement = select(Language).order_by(Language.name)
|
|
return session.exec(statement).all()
|
|
|
|
|
|
@router.post("/languages", response_model=LanguageResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_language(
|
|
language_data: LanguageCreate,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Create a new UI language."""
|
|
code = language_data.code.strip().lower()
|
|
name = language_data.name.strip()
|
|
|
|
if not code or not name:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Code und Name sind erforderlich"
|
|
)
|
|
|
|
if " " in code:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Der Sprachcode darf keine Leerzeichen enthalten"
|
|
)
|
|
|
|
ensure_default_languages(session)
|
|
|
|
existing_code = session.exec(
|
|
select(Language).where(Language.code == code)
|
|
).first()
|
|
if existing_code:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Es existiert bereits eine Sprache mit diesem Code"
|
|
)
|
|
|
|
language = Language(code=code, name=name)
|
|
session.add(language)
|
|
session.commit()
|
|
session.refresh(language)
|
|
|
|
ensure_translation_entries(session)
|
|
ensure_translations_for_language(session, language)
|
|
|
|
return language
|
|
|
|
|
|
@router.delete("/languages/{language_id}")
|
|
def delete_language(
|
|
language_id: int,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Remove a UI language."""
|
|
language = session.get(Language, language_id)
|
|
if not language:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Sprache nicht gefunden"
|
|
)
|
|
|
|
if language.is_default:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Die Standardsprache kann nicht gelöscht werden"
|
|
)
|
|
|
|
translations = session.exec(
|
|
select(Translation).where(Translation.language_id == language.id)
|
|
).all()
|
|
for translation in translations:
|
|
session.delete(translation)
|
|
|
|
session.delete(language)
|
|
session.commit()
|
|
return {"message": "Sprache gelöscht"}
|
|
|
|
|
|
@router.get("/translations", response_model=List[TranslationGroupResponse])
|
|
def get_translations(
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Retrieve translation values grouped by attribute."""
|
|
ensure_default_languages(session)
|
|
ensure_translation_entries(session)
|
|
|
|
languages = session.exec(select(Language).order_by(Language.name)).all()
|
|
translations = session.exec(select(Translation)).all()
|
|
|
|
translations_by_key = {}
|
|
for translation in translations:
|
|
translations_by_key.setdefault(translation.key, []).append(translation)
|
|
|
|
blueprint = get_translation_blueprint()
|
|
response: List[TranslationGroupResponse] = []
|
|
created_entries = False
|
|
|
|
for blueprint_entry in blueprint:
|
|
key = blueprint_entry["key"]
|
|
entries = []
|
|
existing = {t.language_id: t for t in translations_by_key.get(key, [])}
|
|
|
|
for language in languages:
|
|
translation = existing.get(language.id)
|
|
if not translation:
|
|
translation = Translation(key=key, language_id=language.id, value="")
|
|
session.add(translation)
|
|
session.flush()
|
|
existing[language.id] = translation
|
|
created_entries = True
|
|
|
|
entries.append(
|
|
{
|
|
"translation_id": translation.id,
|
|
"language_id": language.id,
|
|
"language_code": language.code,
|
|
"language_name": language.name,
|
|
"value": translation.value,
|
|
}
|
|
)
|
|
|
|
response.append(
|
|
TranslationGroupResponse(
|
|
key=key,
|
|
label=blueprint_entry.get("label", key),
|
|
description=blueprint_entry.get("description"),
|
|
entries=entries,
|
|
)
|
|
)
|
|
|
|
if created_entries:
|
|
session.commit()
|
|
|
|
return response
|
|
|
|
|
|
@router.put("/translations")
|
|
def update_translation(
|
|
payload: TranslationUpdateRequest,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Update a single translation entry."""
|
|
translation = session.get(Translation, payload.translation_id)
|
|
if not translation:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Übersetzung nicht gefunden"
|
|
)
|
|
|
|
translation.value = payload.value
|
|
update_translation_timestamp(translation)
|
|
session.add(translation)
|
|
session.commit()
|
|
session.refresh(translation)
|
|
|
|
return {
|
|
"translation_id": translation.id,
|
|
"value": translation.value,
|
|
"updated_at": translation.updated_at,
|
|
}
|
|
|
|
|
|
# ========== Snippet Department Access Management ==========
|
|
|
|
@router.get("/snippets/{snippet_id}/departments")
|
|
def get_snippet_departments(
|
|
snippet_id: int,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Get all departments and their access status for a snippet.
|
|
By default, snippets are disabled for all departments.
|
|
Admins must explicitly enable access via department editing."""
|
|
snippet = session.get(Snippet, snippet_id)
|
|
if not snippet:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Snippet not found"
|
|
)
|
|
|
|
# Get all departments
|
|
all_departments = session.exec(select(Department)).all()
|
|
|
|
# Get snippet-department links (only enabled ones exist in DB by default)
|
|
links_statement = select(SnippetDepartmentLink).where(
|
|
SnippetDepartmentLink.snippet_id == snippet_id
|
|
)
|
|
links = {link.department_id: link.enabled for link in session.exec(links_statement).all()}
|
|
|
|
result = []
|
|
for dept in all_departments:
|
|
result.append({
|
|
"department_id": dept.id,
|
|
"department_name": dept.name,
|
|
"enabled": links.get(dept.id, False) # Default: False (disabled)
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
@router.post("/snippets/departments/toggle")
|
|
def toggle_snippet_department_access(
|
|
access_data: SnippetDepartmentAccess,
|
|
session: Session = Depends(get_session),
|
|
admin: User = Depends(require_superadmin)
|
|
):
|
|
"""Enable or disable a snippet for a specific department.
|
|
By default, all snippets are disabled for all departments.
|
|
This endpoint allows admins to explicitly grant or revoke access."""
|
|
# Verify snippet exists
|
|
snippet = session.get(Snippet, access_data.snippet_id)
|
|
if not snippet:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Snippet not found"
|
|
)
|
|
|
|
# Verify department exists
|
|
department = session.get(Department, access_data.department_id)
|
|
if not department:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Department not found"
|
|
)
|
|
|
|
# Check if link exists
|
|
statement = select(SnippetDepartmentLink).where(
|
|
SnippetDepartmentLink.snippet_id == access_data.snippet_id,
|
|
SnippetDepartmentLink.department_id == access_data.department_id
|
|
)
|
|
link = session.exec(statement).first()
|
|
|
|
if link:
|
|
if access_data.enabled:
|
|
# Update to enabled
|
|
link.enabled = access_data.enabled
|
|
session.add(link)
|
|
else:
|
|
# If disabling, remove the link entirely (cleaner approach)
|
|
session.delete(link)
|
|
else:
|
|
# Only create new link if enabling
|
|
if access_data.enabled:
|
|
new_link = SnippetDepartmentLink(
|
|
snippet_id=access_data.snippet_id,
|
|
department_id=access_data.department_id,
|
|
enabled=True
|
|
)
|
|
session.add(new_link)
|
|
# If disabling and no link exists, nothing to do (already disabled by default)
|
|
|
|
session.commit()
|
|
|
|
return {
|
|
"message": f"Snippet access {'enabled' if access_data.enabled else 'disabled'} for department",
|
|
"snippet_id": access_data.snippet_id,
|
|
"department_id": access_data.department_id,
|
|
"enabled": access_data.enabled
|
|
}
|
|
|