Remnants/AdministrationItDepartmens
1
0
Fork 0
mirror of https://gitlab.com/MoonTestUse1/AdministrationItDepartmens.git synced 2025-08-14 00:25:46 +02:00
Code Activity

Создание чата2

Browse Source
This commit is contained in:
MoonTestUse1
2025-01-05 05:50:09 +06:00
parent ba0b0ec72b
commit 043871d7c6
11 changed files with 107 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
backend/app/routers/admin.py
View File

@@ -1,40 +0,0 @@
"""Admin router"""
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from ..database import get_db
from ..crud import requests, statistics
from ..schemas.request import Request
from ..utils.auth import get_current_admin
router = APIRouter()
@router.get("/statistics")
def get_statistics(
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Get system statistics"""
return statistics.get_request_statistics(db)
@router.get("/requests", response_model=List[Request])
def get_all_requests(
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Get all requests"""
return requests.get_requests(db)
@router.get("/requests/{request_id}", response_model=Request)
async def get_request_by_id(
request_id: int,
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""
Получить заявку по ID (только для админа)
"""
request = requests.get_request(db, request_id)
if request is None:
raise HTTPException(status_code=404, detail="Request not found")
return request

59
backend/app/routers/auth.py
View File

@@ -1,59 +0,0 @@
"""Authentication router"""
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from typing import Optional
from ..database import get_db
from ..crud import employees
from ..schemas.auth import Token
from ..utils.auth import verify_password
from ..utils.jwt import create_and_save_token
router = APIRouter()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
@router.post("/login", response_model=Token)
async def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
# Проверяем учетные данные сотрудника
employee = employees.get_employee_by_last_name(db, form_data.username)
if not employee or not verify_password(form_data.password, employee.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
# Создаем и сохраняем токен
access_token = create_and_save_token(employee.id, db)
return {
"access_token": access_token,
"token_type": "bearer"
}
@router.post("/admin/login", response_model=Token)
async def admin_login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
# Проверяем учетные данные администратора
if form_data.username != "admin" or form_data.password != "admin123":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
# Для админа используем специальный ID
admin_id = -1
access_token = create_and_save_token(admin_id, db)
return {
"access_token": access_token,
"token_type": "bearer"
}

117
backend/app/routers/employees.py
View File

@@ -1,117 +0,0 @@
"""Employee router"""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
import logging
from ..database import get_db
from ..crud import employees
from ..schemas.employee import Employee, EmployeeCreate, EmployeeUpdate
from ..utils.auth import get_current_admin, get_password_hash
# Настройка логирования
logger = logging.getLogger(__name__)
router = APIRouter(tags=["employees"])
@router.post("", response_model=Employee, status_code=status.HTTP_201_CREATED)
async def create_employee(
employee: EmployeeCreate,
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Create new employee"""
try:
logger.info(f"Creating employee: {employee}")
hashed_password = get_password_hash(employee.password)
return employees.create_employee(db, employee, hashed_password)
except Exception as e:
logger.error(f"Error creating employee: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Error creating employee"
)
@router.get("", response_model=List[Employee])
async def get_employees(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Get all employees"""
try:
logger.info("Getting all employees")
return employees.get_employees(db, skip=skip, limit=limit)
except Exception as e:
logger.error(f"Error getting employees: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Error getting employees"
)
@router.get("/{employee_id}", response_model=Employee)
async def get_employee(
employee_id: int,
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Get employee by ID"""
try:
logger.info(f"Getting employee by ID: {employee_id}")
db_employee = employees.get_employee(db, employee_id)
if db_employee is None:
raise HTTPException(status_code=404, detail="Employee not found")
return db_employee
except HTTPException:
raise
except Exception as e:
logger.error(f"Error getting employee: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Error getting employee"
)
@router.put("/{employee_id}", response_model=Employee)
async def update_employee(
employee_id: int,
employee: EmployeeUpdate,
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Update employee data"""
try:
logger.info(f"Updating employee {employee_id}: {employee}")
db_employee = employees.update_employee(db, employee_id, employee)
if db_employee is None:
raise HTTPException(status_code=404, detail="Employee not found")
return db_employee
except HTTPException:
raise
except Exception as e:
logger.error(f"Error updating employee: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Error updating employee"
)
@router.delete("/{employee_id}", response_model=Employee)
async def delete_employee(
employee_id: int,
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Delete employee"""
try:
logger.info(f"Deleting employee: {employee_id}")
db_employee = employees.delete_employee(db, employee_id)
if db_employee is None:
raise HTTPException(status_code=404, detail="Employee not found")
return db_employee
except HTTPException:
raise
except Exception as e:
logger.error(f"Error deleting employee: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Error deleting employee"
)

71
backend/app/routers/requests.py
View File

@@ -1,71 +0,0 @@
"""Requests router"""
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import List, Optional
from ..database import get_db
from ..crud import requests
from ..schemas.request import Request, RequestCreate, RequestUpdate
from ..models.request import RequestStatus
from ..utils.auth import get_current_employee, get_current_admin
from ..utils.telegram import notify_new_request
router = APIRouter()
@router.post("/", response_model=Request)
async def create_request(
request: RequestCreate,
db: Session = Depends(get_db),
current_employee: dict = Depends(get_current_employee)
):
"""Create new request"""
db_request = requests.create_request(db, request, current_employee["id"])
# Отправляем уведомление в Telegram
await notify_new_request(db_request.id)
return db_request
@router.get("/my", response_model=List[Request])
def get_employee_requests(
db: Session = Depends(get_db),
current_employee: dict = Depends(get_current_employee)
):
"""Get current employee's requests"""
return requests.get_employee_requests(db, current_employee["id"])
@router.get("/admin", response_model=List[Request])
def get_all_requests(
status: Optional[RequestStatus] = Query(None),
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Get all requests (admin only)"""
return requests.get_requests(db, status=status, skip=skip, limit=limit)
@router.patch("/{request_id}/status", response_model=Request)
def update_request_status(
request_id: int,
request_update: RequestUpdate,
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Update request status (admin only)"""
db_request = requests.update_request_status(db, request_id, request_update.status)
if db_request is None:
raise HTTPException(status_code=404, detail="Request not found")
return db_request
@router.get("/statistics")
def get_request_statistics(
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Get request statistics (admin only)"""
stats = requests.get_statistics(db)
return {
"total": stats["total"],
"by_status": {
status: count
for status, count in stats["by_status"].items()
}
}

16
backend/app/routers/statistics.py
View File

@@ -1,16 +0,0 @@
"""Statistics router"""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from ..database import get_db
from ..crud import statistics
from ..utils.auth import get_current_admin
router = APIRouter()
@router.get("/")
def get_statistics(
db: Session = Depends(get_db),
_: dict = Depends(get_current_admin)
):
"""Get system statistics"""
return statistics.get_request_statistics(db)