app.py - master - ring0/support-system

app.py
240 строк9 KB
ring0
Init project
22 июн 2026, 17:13
22 июн 2026, 17:139df53b7
import os

from dotenv import load_dotenv
from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse, FileResponse
from sqlalchemy.orm import Session

from db import engine, Base, get_db
from models import (
    User,
    Category,
    Priority,
    UserRegister,
    LoginRequest,
    TicketCreate,
    TicketUpdate,
    CommentCreate,
    CategoryCreate,
    PriorityCreate,
)
from manager import UserManager, TicketManager, CommentManager
from auth import create_access_token, get_current_user, require_roles

load_dotenv()

app = FastAPI(title="Система учёта заявок на техническую поддержку")

# Автоматическое создание таблиц при первом запуске приложения
Base.metadata.create_all(bind=engine)

user_manager = UserManager()
ticket_manager = TicketManager()
comment_manager = CommentManager()

STATIC_DIR = os.path.join(os.path.dirname(__file__), "static")


# ---------------------------------------------------------------------------
# Аутентификация и регистрация
# ---------------------------------------------------------------------------
@app.post("/api/auth/register")
async def register(data: UserRegister, db: Session = Depends(get_db)):
    """Регистрация нового пользователя"""
    result = user_manager.register(db, data)
    if "error" in result:
        return JSONResponse(content=result, status_code=400)
    return JSONResponse(content=result, status_code=201)


@app.post("/api/auth/login")
async def login(data: LoginRequest, db: Session = Depends(get_db)):
    """Аутентификация пользователя и выдача JWT-токена"""
    user = user_manager.authenticate(db, data.username, data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Неверный логин или пароль")
    token = create_access_token(user)
    return {"access_token": token, "token_type": "bearer", "user": user.to_dict()}


@app.get("/api/auth/me")
async def get_me(current_user: User = Depends(get_current_user)):
    """Информация о текущем пользователе"""
    return current_user.to_dict()


# ---------------------------------------------------------------------------
# Справочники: категории и приоритеты
# ---------------------------------------------------------------------------
@app.get("/api/categories")
async def get_categories(db: Session = Depends(get_db)):
    """Список категорий обращений"""
    return [c.to_dict() for c in db.query(Category).all()]


@app.get("/api/priorities")
async def get_priorities(db: Session = Depends(get_db)):
    """Список приоритетов (отсортированы по весу)"""
    items = db.query(Priority).order_by(Priority.weight.desc()).all()
    return [p.to_dict() for p in items]


@app.post("/api/categories")
async def create_category(
    data: CategoryCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_roles("admin")),
):
    """Добавление категории (только администратор)"""
    category = Category(name=data.name)
    db.add(category)
    db.commit()
    db.refresh(category)
    return JSONResponse(content=category.to_dict(), status_code=201)


@app.post("/api/priorities")
async def create_priority(
    data: PriorityCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_roles("admin")),
):
    """Добавление приоритета (только администратор)"""
    priority = Priority(name=data.name, weight=data.weight)
    db.add(priority)
    db.commit()
    db.refresh(priority)
    return JSONResponse(content=priority.to_dict(), status_code=201)


# ---------------------------------------------------------------------------
# Пользователи (для администратора)
# ---------------------------------------------------------------------------
@app.get("/api/users")
async def get_users(
    role: str = None,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_roles("admin")),
):
    """Список пользователей с опциональной фильтрацией по роли"""
    users = user_manager.get_users(db, role)
    return [u.to_dict() for u in users]


# ---------------------------------------------------------------------------
# Заявки
# ---------------------------------------------------------------------------
@app.get("/api/tickets")
async def get_tickets(
    status: str = None,
    category_id: int = None,
    priority_id: int = None,
    assignee_id: int = None,
    author_id: int = None,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Перечень заявок с фильтрацией; видимость зависит от роли пользователя"""
    filters = {
        "status": status,
        "category_id": category_id,
        "priority_id": priority_id,
        "assignee_id": assignee_id,
        "author_id": author_id,
    }
    tickets = ticket_manager.get_tickets(db, current_user, filters)
    return [t.to_dict() for t in tickets]


@app.post("/api/tickets")
async def create_ticket(
    data: TicketCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Создание новой заявки с проверкой дубликатов"""
    result = ticket_manager.create_ticket(db, data, current_user)
    if "error" in result:
        return JSONResponse(content=result, status_code=result.get("code", 400))
    return JSONResponse(content=result, status_code=201)


@app.get("/api/tickets/{ticket_id}")
async def get_ticket(
    ticket_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Получение одной заявки по идентификатору"""
    ticket = ticket_manager.get_ticket(db, ticket_id)
    if not ticket:
        raise HTTPException(status_code=404, detail="Заявка не найдена")
    return ticket.to_dict()


@app.put("/api/tickets/{ticket_id}")
async def update_ticket(
    ticket_id: int,
    data: TicketUpdate,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_roles("specialist", "admin")),
):
    """Обновление заявки (специалист — статус, администратор — любые поля)"""
    result = ticket_manager.update_ticket(db, ticket_id, data, current_user)
    if "error" in result:
        return JSONResponse(content=result, status_code=result.get("code", 400))
    return JSONResponse(content=result)


# ---------------------------------------------------------------------------
# Комментарии
# ---------------------------------------------------------------------------
@app.get("/api/tickets/{ticket_id}/comments")
async def get_comments(
    ticket_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Получение комментариев заявки"""
    comments = comment_manager.get_comments(db, ticket_id)
    return [c.to_dict() for c in comments]


@app.post("/api/tickets/{ticket_id}/comments")
async def add_comment(
    ticket_id: int,
    data: CommentCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Добавление комментария к заявке"""
    result = comment_manager.add_comment(db, ticket_id, data, current_user)
    if "error" in result:
        return JSONResponse(content=result, status_code=result.get("code", 400))
    return JSONResponse(content=result, status_code=201)


# ---------------------------------------------------------------------------
# Раздача статических страниц (SPA по ролям)
# ---------------------------------------------------------------------------
def _page(filename: str) -> HTMLResponse:
    with open(os.path.join(STATIC_DIR, filename), "r", encoding="utf-8") as f:
        return HTMLResponse(f.read())


@app.get("/", response_class=HTMLResponse)
async def user_interface():
    """Интерфейс заявителя"""
    return _page("user.html")


@app.get("/specialist", response_class=HTMLResponse)
async def specialist_interface():
    """Личный кабинет специалиста"""
    return _page("specialist.html")


@app.get("/admin", response_class=HTMLResponse)
async def admin_interface():
    """Административная панель"""
    return _page("admin.html")