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

9

Browse Source
This commit is contained in:
MoonTestUse1
2024-12-28 06:24:53 +06:00
parent bde484f1e0
commit 0c95c62f1d
31 changed files with 64 additions and 1417 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
backend/app/main.py
View File

@@ -5,6 +5,8 @@ from fastapi.openapi.utils import get_openapi
import logging
from logging.config import dictConfig
from .logging_config import logging_config
from .middleware import LoggingMiddleware
from .routers import auth
# Configure logging
dictConfig(logging_config)
@@ -18,6 +20,21 @@ app = FastAPI(
redoc_url=None
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Add logging middleware
app.add_middleware(LoggingMiddleware)
# Include routers
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
# Custom OpenAPI documentation
@app.get("/api/docs", include_in_schema=False)
async def custom_swagger_ui_html():
@@ -34,6 +51,4 @@ async def get_open_api_endpoint():
version="1.0.0",
description="API for managing support requests and employees",
routes=app.routes
)
# Existing middleware and routes...
)

4
backend/app/routers/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
"""API routes package"""
from . import auth
__all__ = ['auth']

42
backend/app/routers/auth.py Normal file
View File

@@ -0,0 +1,42 @@
"""Authentication routes"""
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from ..database import get_db
from ..crud import auth as auth_crud
from ..models.employee import EmployeeBase
from logging import getLogger
router = APIRouter()
logger = getLogger(__name__)
@router.post("/login")
async def login(credentials: dict, db: Session = Depends(get_db)):
"""Employee login endpoint"""
try:
employee = auth_crud.authenticate_employee(
db,
credentials.get("lastName"),
credentials.get("password")
)
if not employee:
raise HTTPException(status_code=401, detail="Неверные учетные данные")
return employee
except Exception as e:
logger.error(f"Login error: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Ошибка сервера")
@router.post("/admin")
async def admin_login(credentials: dict, db: Session = Depends(get_db)):
"""Admin login endpoint"""
try:
is_valid = auth_crud.authenticate_admin(
db,
credentials.get("username"),
credentials.get("password")
)
if not is_valid:
raise HTTPException(status_code=401, detail="Неверные учетные данные")
return {"status": "success"}
except Exception as e:
logger.error(f"Admin login error: {e}", exc_info=True)
raise HTTPException(status_code=500, detail="Ошибка сервера")

1
backend/app2/__init__.py
View File

@@ -1 +0,0 @@
# FastAPI application package

68
backend/app2/bot.py
View File

@@ -1,68 +0,0 @@
from aiogram import Bot, Dispatcher, types
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.filters import Command
from sqlalchemy.orm import Session
from ..database import get_db
from .. import models
# Создаем роутер для обработки callback'ов
from aiogram import Router
router = Router()
# Обработчик нажатия кнопки
@router.callback_query(lambda c: c.data.startswith('status_'))
async def process_status_button(callback_query: types.CallbackQuery):
try:
print(f"Hello world: {callback_query.data}")
# Получаем ID заявки и новый статус из callback_data
_, request_id, new_status = callback_query.data.split('_')
request_id = int(request_id)
# Получаем сессию базы данных
db = next(get_db())
# Обновляем статус в базе данных
request = db.query(models.Request).filter(models.Request.id == request_id).first()
if request:
request.status = new_status
db.commit()
# Обновляем сообщение в боте
await callback_query.message.edit_text(
f"Заявка №{request_id}\nСтатус: {new_status}",
reply_markup=get_updated_keyboard(request_id, new_status)
)
# Отправляем уведомление о успешном обновлении
await callback_query.answer("Статус успешно обновлен!")
else:
await callback_query.answer("Заявка не найдена!", show_alert=True)
except Exception as e:
print(f"Error in process_status_button: {e}")
await callback_query.answer("Произошла ошибка при обновлении статуса", show_alert=True)
def get_updated_keyboard(request_id: int, current_status: str) -> InlineKeyboardMarkup:
keyboard = InlineKeyboardMarkup(inline_keyboard=[])
if current_status != "in_progress":
keyboard.inline_keyboard.append([
InlineKeyboardButton(
text="В работе",
callback_data=f"status_{request_id}_in_progress"
)
])
if current_status != "completed":
keyboard.inline_keyboard.append([
InlineKeyboardButton(
text="Завершено",
callback_data=f"status_{request_id}_completed"
)
])
return keyboard
# В основном файле бота (где создается диспетчер)
dp = Dispatcher()
dp.include_router(router)

19
backend/app2/bot/__init__.py
View File

@@ -1,19 +0,0 @@
"""Bot initialization module"""
from aiogram import Bot, Dispatcher
from .config import settings
from .handlers import callbacks_router, start_router
# Initialize bot and dispatcher
bot = Bot(token=settings.bot_token)
dp = Dispatcher()
# Include routers only once during initialization
dp.include_router(callbacks_router)
dp.include_router(start_router)
async def start_bot():
"""Start the bot"""
try:
await dp.start_polling(bot, skip_updates=True)
except Exception as e:
print(f"Error starting bot: {e}")

69
backend/app2/bot/bot.py
View File

@@ -1,69 +0,0 @@
from aiogram import Bot, Dispatcher
from app.bot.config import settings
bot = Bot(token="7677506032:AAHduD5EePz3bE23DKlo35KoOp2_9lZuS34")
dp = Dispatcher()
from .handlers import start, status
async def start_bot():
"""Start the bot"""
try:
await dp.start_polling(bot, skip_updates=True)
finally:
await bot.session.close()
from aiogram import Bot, Dispatcher, types
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from sqlalchemy.orm import Session
from ..database import get_db
from .. import models
# Создаем обработчик нажатия кнопки
@dp.callback_query_handler(lambda c: c.data.startswith('status_'))
async def process_status_button(callback_query: types.CallbackQuery):
try:
# Получаем ID заявки и новый статус из callback_data
_, request_id, new_status = callback_query.data.split('_')
request_id = int(request_id)
# Получаем сессию базы данных
db = next(get_db())
# Обновляем статус в базе данных
request = db.query(models.Request).filter(models.Request.id == request_id).first()
if request:
request.status = new_status
db.commit()
# Обновляем сообщение в боте
await callback_query.message.edit_text(
f"Заявка №{request_id}\nСтатус: {new_status}",
reply_markup=get_updated_keyboard(request_id, new_status)
)
# Отправляем уведомление о успешном обновлении
await callback_query.answer("Статус успешно обновлен!")
else:
await callback_query.answer("Заявка не найдена!", show_alert=True)
except Exception as e:
print(f"Error in process_status_button: {e}")
await callback_query.answer("Произошла ошибка при обновлении статуса", show_alert=True)
def get_updated_keyboard(request_id: int, current_status: str) -> InlineKeyboardMarkup:
keyboard = InlineKeyboardMarkup()
if current_status != "in_progress":
keyboard.add(InlineKeyboardButton(
"В работе",
callback_data=f"status_{request_id}_in_progress"
))
if current_status != "completed":
keyboard.add(InlineKeyboardButton(
"Завершено",
callback_data=f"status_{request_id}_completed"
))
return keyboard

26
backend/app2/bot/config.py
View File

@@ -1,26 +0,0 @@
"""
Configuration module for the Telegram bot.
Contains all necessary settings and constants.
"""
from pydantic_settings import BaseSettings
from pydantic import Field
class Settings(BaseSettings):
"""Bot configuration settings"""
bot_token: str = Field("7677506032:AAHduD5EePz3bE23DKlo35KoOp2_9lZuS34", env="TELEGRAM_BOT_TOKEN")
chat_id: str = Field("5057752127", env="TELEGRAM_CHAT_ID")
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
# Create settings instance
settings = Settings()
# Request status constants
class RequestStatus:
"""Constants for request statuses"""
NEW = "new"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
CANCELLED = "cancelled"

73
backend/app2/bot/constants.py
View File

@@ -1,73 +0,0 @@
from enum import Enum
class RequestStatus(str, Enum):
NEW = "new"
IN_PROGRESS = "in_progress"
RESOLVED = "resolved"
CLOSED = "closed"
class RequestPriority(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class Department(str, Enum):
AHO = "aho"
GKH = "gkh"
GENERAL = "general"
class RequestType(str, Enum):
HARDWARE = "hardware"
SOFTWARE = "software"
NETWORK = "network"
ACCESS = "access"
OTHER = "other"
STATUS_LABELS = {
RequestStatus.NEW: "Новая",
RequestStatus.IN_PROGRESS: "В работе",
RequestStatus.RESOLVED: "Решена",
RequestStatus.CLOSED: "Закрыта",
}
PRIORITY_LABELS = {
RequestPriority.LOW: "Низкий",
RequestPriority.MEDIUM: "Средний",
RequestPriority.HIGH: "Высокий",
RequestPriority.CRITICAL: "Критический",
}
PRIORITY_EMOJI = {
RequestPriority.LOW: "🟢",
RequestPriority.MEDIUM: "🟡",
RequestPriority.HIGH: "🟠",
RequestPriority.CRITICAL: "🔴",
}
DEPARTMENT_LABELS = {
Department.AHO: "Административно-хозяйственный отдел",
Department.GKH: "Жилищно-коммунальное хозяйство",
Department.GENERAL: "Общий отдел",
}
REQUEST_TYPE_LABELS = {
RequestType.HARDWARE: "Проблемы с оборудованием",
RequestType.SOFTWARE: "Проблемы с ПО",
RequestType.NETWORK: "Проблемы с сетью",
RequestType.ACCESS: "Доступ к системам",
RequestType.OTHER: "Другое",
}
REQUEST_TYPE_EMOJI = {
RequestType.HARDWARE: "🖥️",
RequestType.SOFTWARE: "💿",
RequestType.NETWORK: "🌐",
RequestType.ACCESS: "🔑",
RequestType.OTHER: "📝",
}

6
backend/app2/bot/handlers/__init__.py
View File

@@ -1,6 +0,0 @@
"""Handlers initialization"""
from .callbacks import router as callbacks_router
from .start import router as start_router
from .callbacks import get_updated_keyboard
__all__ = ['callbacks_router', 'start_router', 'get_updated_keyboard']

70
backend/app2/bot/handlers/callbacks.py
View File

@@ -1,70 +0,0 @@
"""Handlers for callback queries"""
from aiogram import Router, types
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from sqlalchemy.orm import Session
from ...database import get_db
from ... import models
from ..config import RequestStatus
router = Router()
@router.callback_query(lambda c: c.data and c.data.startswith('status_'))
async def process_status_button(callback_query: types.CallbackQuery):
"""
Handle status update button clicks.
Updates request status in database and updates message in Telegram.
"""
try:
# Parse callback data
_, request_id, new_status = callback_query.data.split('_')
request_id = int(request_id)
# Get database session
db = next(get_db())
# Update request status
request = db.query(models.Request).filter(models.Request.id == request_id).first()
if request:
request.status = new_status
db.commit()
# Update message in Telegram
await callback_query.message.edit_text(
f"Заявка №{request_id}\n"
f"Статус: {new_status}\n"
f"Описание: {request.description}",
reply_markup=get_updated_keyboard(request_id, new_status)
)
await callback_query.answer("Статус успешно обновлен!")
else:
await callback_query.answer("Заявка не найдена!", show_alert=True)
except Exception as e:
print(f"Error in process_status_button: {e}")
await callback_query.answer(
"Произошла ошибка при обновлении статуса",
show_alert=True
)
def get_updated_keyboard(request_id: int, current_status: str) -> InlineKeyboardMarkup:
"""Create keyboard with status update buttons"""
keyboard = InlineKeyboardMarkup(inline_keyboard=[])
if current_status != RequestStatus.IN_PROGRESS:
keyboard.inline_keyboard.append([
InlineKeyboardButton(
text="В работе",
callback_data=f"status_{request_id}_{RequestStatus.IN_PROGRESS}"
)
])
if current_status != RequestStatus.COMPLETED:
keyboard.inline_keyboard.append([
InlineKeyboardButton(
text="Завершено",
callback_data=f"status_{request_id}_{RequestStatus.COMPLETED}"
)
])
return keyboard

11
backend/app2/bot/handlers/start.py
View File

@@ -1,11 +0,0 @@
"""Handler for start command and other basic commands"""
from aiogram import Router, types
from aiogram.filters import Command
from ..config import settings
router = Router()
@router.message(Command("start"))
async def cmd_start(message: types.Message):
"""Handle /start command"""
await message.answer("Бот для обработки заявок запущен!")

61
backend/app2/bot/handlers/status.py
View File

@@ -1,61 +0,0 @@
from aiogram import types, F
from logging import getLogger
from sqlalchemy.orm import Session
from ...database import get_db
from ...crud import requests
from ..bot import dp
from ..keyboards import create_status_keyboard
from ..messages import format_request_message
from ..constants import STATUS_LABELS
logger = getLogger(__name__)
@dp.callback_query(F.data.startswith("status_"))
async def process_status_update(callback: types.CallbackQuery):
try:
parts = callback.data.split("_")
logger.info(f"Received callback data: {callback.data}")
if len(parts) < 3:
logger.error(f"Invalid callback data format: {parts}")
await callback.answer("Неверный формат данных", show_alert=True)
return
request_id = int(parts[1])
new_status = "_".join(parts[2:]) if len(parts) > 3 else parts[2]
logger.info(
f"Processing status update: request_id={request_id}, new_status={new_status}"
)
db = next(get_db())
try:
updated_request = requests.update_request_status(db, request_id, new_status)
if not updated_request:
logger.warning(f"Request not found: {request_id}")
await callback.answer("Заявка не найдена", show_alert=True)
return
new_message = format_request_message(updated_request)
new_keyboard = create_status_keyboard(request_id, new_status)
await callback.message.edit_text(
text=new_message, parse_mode="HTML", reply_markup=new_keyboard
)
await callback.answer(f"Статус обновлен: {STATUS_LABELS[new_status]}")
logger.info(
f"Successfully updated request {request_id} to status {new_status}"
)
except ValueError as e:
logger.error(f"Value error while updating status: {e}")
await callback.answer(str(e), show_alert=True)
finally:
db.close()
except Exception as e:
logger.error(f"Error processing callback: {e}", exc_info=True)
await callback.answer(
"Произошла ошибка при обновлении статуса", show_alert=True
)

33
backend/app2/bot/keyboards.py
View File

@@ -1,33 +0,0 @@
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from logging import getLogger
from .constants import STATUS_LABELS
logger = getLogger(__name__)
def create_status_keyboard(
request_id: int, current_status: str
) -> InlineKeyboardMarkup:
status_transitions = {
"new": ["in_progress"],
"in_progress": ["resolved"],
"resolved": ["closed"],
"closed": [],
}
buttons = []
available_statuses = status_transitions.get(current_status, [])
for status in available_statuses:
callback_data = f"status_{request_id}_{status}"
logger.debug(f"Creating button with callback_data: {callback_data}")
buttons.append(
[
InlineKeyboardButton(
text=STATUS_LABELS[status], callback_data=callback_data
)
]
)
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons)
logger.debug(f"Created keyboard: {keyboard}")
return keyboard

37
backend/app2/bot/messages.py
View File

@@ -1,37 +0,0 @@
from datetime import datetime
from .constants import (
STATUS_LABELS,
PRIORITY_LABELS,
PRIORITY_EMOJI,
DEPARTMENT_LABELS,
REQUEST_TYPE_LABELS,
REQUEST_TYPE_EMOJI,
)
def format_request_message(request_data: dict) -> str:
created_at = datetime.fromisoformat(request_data["created_at"]).strftime(
"%d.%m.%Y %H:%M"
)
# Get translated values
department = DEPARTMENT_LABELS.get(
request_data["department"], request_data["department"]
)
request_type = REQUEST_TYPE_LABELS.get(
request_data["request_type"], request_data["request_type"]
)
priority = PRIORITY_LABELS.get(request_data["priority"], request_data["priority"])
status = STATUS_LABELS.get(request_data.get("status", "new"), "Неизвестно")
return (
f"📋 <b>Заявка #{request_data['id']}</b>\n\n"
f"👤 <b>Сотрудник:</b> {request_data['employee_last_name']} {request_data['employee_first_name']}\n"
f"🏢 <b>Отдел:</b> {department}\n"
f"🚪 <b>Кабинет:</b> {request_data['office']}\n"
f"{REQUEST_TYPE_EMOJI.get(request_data['request_type'], '📝')} <b>Тип заявки:</b> {request_type}\n"
f"{PRIORITY_EMOJI.get(request_data['priority'], '⚪')} <b>Приоритет:</b> {priority}\n\n"
f"📝 <b>Описание:</b>\n<blockquote>{request_data['description']}</blockquote>\n\n"
f"🕒 <b>Создана:</b> {created_at}\n"
f"📊 <b>Статус:</b> {status}"
)

23
backend/app2/bot/notifications.py
View File

@@ -1,23 +0,0 @@
"""Notifications module for the Telegram bot"""
from .config import settings
from . import bot
from .handlers import get_updated_keyboard
async def send_notification(request_data: dict):
"""Send notification about new request to Telegram chat"""
message_text = (
f"Новая заявка №{request_data['id']}\n"
f"Отдел: {request_data['department']}\n"
f"Тип: {request_data['request_type']}\n"
f"Приоритет: {request_data['priority']}\n"
f"Описание: {request_data['description']}"
)
try:
await bot.send_message(
chat_id=settings.chat_id,
text=message_text,
reply_markup=get_updated_keyboard(request_data['id'], "new")
)
except Exception as e:
print(f"Error sending notification: {e}")

26
backend/app2/crud/auth.py
View File

@@ -1,26 +0,0 @@
from sqlalchemy.orm import Session
from ..schemas import tables
from ..utils.auth import verify_password
def authenticate_employee(db: Session, last_name: str, password: str):
employee = db.query(tables.Employee).filter(tables.Employee.last_name == last_name).first()
if not employee:
return None
if not verify_password(password, employee.password):
return None
return {
"id": employee.id,
"firstName": employee.first_name,
"lastName": employee.last_name,
"department": employee.department,
"office": employee.office,
"createdAt": employee.created_at
}
def authenticate_admin(db: Session, username: str, password: str):
# Здесь можно добавить логику для админа, пока используем хардкод
if username == "admin" and password == "admin66":
return True
return False

45
backend/app2/crud/employees.py
View File

@@ -1,45 +0,0 @@
from sqlalchemy.orm import Session
from ..models import employee as models
from ..schemas import tables
from ..utils.auth import get_password_hash
def get_employee(db: Session, employee_id: int):
return db.query(tables.Employee).filter(tables.Employee.id == employee_id).first()
def get_employee_by_lastname(db: Session, last_name: str):
return (
db.query(tables.Employee).filter(tables.Employee.last_name == last_name).first()
)
def get_employees(db: Session, skip: int = 0, limit: int = 100):
return db.query(tables.Employee).offset(skip).limit(limit).all()
def create_employee(db: Session, employee: models.EmployeeCreate):
hashed_password = get_password_hash(employee.password)
db_employee = tables.Employee(
first_name=employee.first_name,
last_name=employee.last_name,
department=employee.department,
office=employee.office,
password=hashed_password,
)
db.add(db_employee)
db.commit()
db.refresh(db_employee)
return db_employee
def update_employee(db: Session, employee_id: int, data: dict):
db_employee = get_employee(db, employee_id)
if db_employee:
for key, value in data.items():
if key == "password":
value = get_password_hash(value)
setattr(db_employee, key, value)
db.commit()
db.refresh(db_employee)
return db_employee

207
backend/app2/crud/requests.py
View File

@@ -1,207 +0,0 @@
from sqlalchemy.orm import Session
from ..models import request as models
from ..schemas import tables
from sqlalchemy import select
from sqlalchemy.orm import Session
from ..schemas import tables
from datetime import datetime
def create_request(db: Session, request: models.RequestCreate):
db_request = tables.Request(
employee_id=request.employee_id,
department=request.department,
request_type=request.request_type,
priority=request.priority,
description=request.description,
status="new",
)
db.add(db_request)
db.commit()
db.refresh(db_request)
return db_request
def get_requests(db: Session, skip: int = 0, limit: int = 100):
requests = (
db.query(
tables.Request,
tables.Employee.last_name.label("employee_last_name"),
tables.Employee.first_name.label("employee_first_name"),
)
.join(tables.Employee)
.offset(skip)
.limit(limit)
.all()
)
return [
{
"id": req[0].id,
"employee_id": req[0].employee_id,
"department": req[0].department,
"request_type": req[0].request_type,
"priority": req[0].priority,
"status": req[0].status,
"description": req[0].description,
"created_at": req[0].created_at,
"employee_last_name": req[1],
"employee_first_name": req[2],
}
for req in requests
]
def get_requests_by_employee_lastname(db: Session, last_name: str):
requests = (
db.query(
tables.Request,
tables.Employee.last_name.label("employee_last_name"),
tables.Employee.first_name.label("employee_first_name"),
)
.join(tables.Employee)
.filter(tables.Employee.last_name.ilike(f"%{last_name}%"))
.all()
)
return [
{
"id": req[0].id,
"employee_id": req[0].employee_id,
"department": req[0].department,
"request_type": req[0].request_type,
"priority": req[0].priority,
"status": req[0].status,
"description": req[0].description,
"created_at": req[0].created_at,
"employee_last_name": req[1],
"employee_first_name": req[2],
}
for req in requests
]
def update_request_status(
db: Session, request_id: int, new_status: models.RequestStatus
):
try:
db_request = (
db.query(tables.Request).filter(tables.Request.id == request_id).first()
)
if not db_request:
return None
# Define valid status transitions
valid_transitions = {
models.RequestStatus.NEW: [models.RequestStatus.IN_PROGRESS],
models.RequestStatus.IN_PROGRESS: [models.RequestStatus.RESOLVED],
models.RequestStatus.RESOLVED: [models.RequestStatus.CLOSED],
models.RequestStatus.CLOSED: [],
}
current_status = models.RequestStatus(db_request.status)
if new_status not in valid_transitions[current_status]:
raise ValueError(
f"Invalid status transition from {current_status} to {new_status}"
)
db_request.status = new_status
db.commit()
db.refresh(db_request)
return db_request
except Exception as e:
db.rollback()
raise e
def get_request_details(db: Session, request_id: int):
"""Get detailed request information including employee details"""
request = (
db.query(tables.Request)
.join(tables.Employee)
.filter(tables.Request.id == request_id)
.first()
)
if not request:
return None
return {
"id": request.id,
"employee_last_name": request.employee.last_name,
"employee_first_name": request.employee.first_name,
"department": request.department,
"office": request.employee.office,
"request_type": request.request_type,
"priority": request.priority,
"description": request.description,
"status": request.status,
"created_at": request.created_at.isoformat(),
}
from sqlalchemy.orm import Session
from ..schemas import tables
from datetime import datetime
def get_request_details(db: Session, request_id: int):
"""Get detailed request information including employee details"""
request = (
db.query(tables.Request)
.join(tables.Employee)
.filter(tables.Request.id == request_id)
.first()
)
if not request:
return None
return {
"id": request.id,
"employee_last_name": request.employee.last_name,
"employee_first_name": request.employee.first_name,
"department": request.department,
"office": request.employee.office,
"request_type": request.request_type,
"priority": request.priority,
"description": request.description,
"status": request.status,
"created_at": request.created_at.isoformat(),
}
def update_request_status(db: Session, request_id: int, new_status: str):
"""Update request status with validation"""
try:
# Define valid status transitions
valid_transitions = {
"new": ["in_progress"],
"in_progress": ["resolved"],
"resolved": ["closed"],
"closed": [],
}
db_request = (
db.query(tables.Request).filter(tables.Request.id == request_id).first()
)
if not db_request:
return None
current_status = db_request.status
if new_status not in valid_transitions.get(current_status, []):
raise ValueError(
f"Invalid status transition from {current_status} to {new_status}"
)
db_request.status = new_status
db.commit()
db.refresh(db_request)
# Get full request details after update
return get_request_details(db, request_id)
except Exception as e:
db.rollback()
raise e

89
backend/app2/crud/statistics.py
View File

@@ -1,89 +0,0 @@
from sqlalchemy import func, text
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
from ..schemas import tables
from ..models.request import RequestStatus
def get_statistics(db: Session, period: str = "week"):
# Calculate date range based on period
now = datetime.now()
if period == "day":
start_date = now - timedelta(days=1)
elif period == "week":
start_date = now - timedelta(weeks=1)
elif period == "month":
start_date = now - timedelta(days=30)
else: # all time
start_date = datetime.min
# Total requests
total_requests = db.query(func.count(tables.Request.id)).scalar() or 0
# Resolved requests in period
resolved_requests = (
db.query(func.count(tables.Request.id))
.filter(tables.Request.status == RequestStatus.RESOLVED)
.filter(tables.Request.created_at >= start_date)
.scalar()
or 0
)
# Average resolution time (in hours)
avg_resolution = (
db.query(
func.avg(func.julianday("now") - func.julianday(tables.Request.created_at))
* 24
)
.filter(
tables.Request.status == RequestStatus.RESOLVED,
tables.Request.created_at >= start_date,
)
.scalar()
)
avg_resolution_time = f"{int(avg_resolution or 0)}ч" if avg_resolution else "0ч"
# Request volume over time
volume_data = (
db.query(
func.date(tables.Request.created_at).label("date"),
func.count(tables.Request.id).label("count"),
)
.filter(tables.Request.created_at >= start_date)
.group_by(text("date"))
.all()
)
# Request types distribution
type_distribution = (
db.query(tables.Request.request_type, func.count(tables.Request.id))
.group_by(tables.Request.request_type)
.all()
)
# Status distribution
status_distribution = (
db.query(tables.Request.status, func.count(tables.Request.id))
.group_by(tables.Request.status)
.all()
)
# Ensure all statuses are represented
all_statuses = {status.value: 0 for status in RequestStatus}
for status, count in status_distribution:
all_statuses[status] = count
status_data = [(status, count) for status, count in all_statuses.items()]
return {
"totalRequests": total_requests,
"resolvedRequests": resolved_requests,
"averageResolutionTime": avg_resolution_time,
"volumeLabels": [str(d[0]) for d in volume_data],
"volumeData": [d[1] for d in volume_data],
"typeLabels": [t[0] for t in type_distribution],
"typeData": [t[1] for t in type_distribution],
"statusLabels": [s[0] for s in status_data],
"statusData": [s[1] for s in status_data],
}

20
backend/app2/database.py
View File

@@ -1,20 +0,0 @@
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

56
backend/app2/logging_config.py
View File

@@ -1,56 +0,0 @@
"""Logging configuration for the application"""
logging_config = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"default": {
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
"datefmt": "%Y-%m-%d %H:%M:%S"
},
"access": {
"format": "%(asctime)s - %(name)s - %(levelname)s - %(client_addr)s - %(request_line)s - %(status_code)s",
"datefmt": "%Y-%m-%d %H:%M:%S"
}
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": "INFO",
"formatter": "default",
"stream": "ext://sys.stdout"
},
"file": {
"class": "logging.handlers.RotatingFileHandler",
"level": "INFO",
"formatter": "default",
"filename": "logs/app.log",
"maxBytes": 10485760, # 10MB
"backupCount": 5
},
"access_file": {
"class": "logging.handlers.RotatingFileHandler",
"level": "INFO",
"formatter": "access",
"filename": "logs/access.log",
"maxBytes": 10485760, # 10MB
"backupCount": 5
}
},
"loggers": {
"": { # Root logger
"handlers": ["console", "file"],
"level": "INFO"
},
"app": { # Application logger
"handlers": ["console", "file"],
"level": "INFO",
"propagate": False
},
"app.access": { # Access logger
"handlers": ["access_file"],
"level": "INFO",
"propagate": False
}
}
}

181
backend/app2/main.py
View File

@@ -1,181 +0,0 @@
from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from typing import List
from .models import employee as employee_models
from .models import request as request_models
from .schemas import tables
from .crud import employees, requests, auth, statistics
from .database import engine, get_db
from .models.request import StatusUpdate
from .bot.notifications import send_notification
from .bot import start_bot
import threading
import asyncio
tables.Base.metadata.create_all(bind=engine)
app = FastAPI()
def run_bot():
asyncio.run(start_bot())
bot_thread = threading.Thread(target=run_bot, daemon=True)
bot_thread.start()
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Auth endpoints
@app.post("/api/test/create-user")
def create_test_user(db: Session = Depends(get_db)):
test_user = employee_models.EmployeeCreate(
first_name="Test",
last_name="User",
department="general",
office="101",
password="test123"
)
return employees.create_employee(db=db, employee=test_user)
@app.post("/api/auth/login")
def login(credentials: dict, db: Session = Depends(get_db)):
print(f"Login attempt for: {credentials['lastName']}") # Добавьте для отладки
employee = auth.authenticate_employee(db, credentials["lastName"], credentials["password"])
if not employee:
raise HTTPException(
status_code=401,
detail="Неверная фамилия или пароль"
)
return employee
@app.post("/api/auth/admin")
def admin_login(credentials: dict, db: Session = Depends(get_db)):
if not auth.authenticate_admin(
db, credentials["username"], credentials["password"]
):
raise HTTPException(status_code=401, detail="Неверные учетные данные")
return {"success": True}
# Employee endpoints
@app.post("/api/employees/", response_model=employee_models.Employee)
def create_employee(
employee: employee_models.EmployeeCreate, db: Session = Depends(get_db)
):
db_employee = employees.get_employee_by_lastname(db, employee.last_name)
if db_employee:
raise HTTPException(status_code=400, detail="Last name already registered")
return employees.create_employee(db=db, employee=employee)
@app.get("/api/employees/", response_model=List[employee_models.Employee])
def read_employees(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
return employees.get_employees(db, skip=skip, limit=limit)
@app.patch("/api/employees/{employee_id}")
def update_employee(employee_id: int, data: dict, db: Session = Depends(get_db)):
return employees.update_employee(db, employee_id, data)
# Request endpoints
@app.post("/api/requests/")
async def create_request(
request: request_models.RequestCreate, db: Session = Depends(get_db)
):
# Create request in database
new_request = requests.create_request(db=db, request=request)
# Get employee details for the notification
employee = employees.get_employee(db, new_request.employee_id)
# Prepare notification data2
notification_data = {
"id": new_request.id,
"employee_last_name": employee.last_name,
"employee_first_name": employee.first_name,
"department": new_request.department,
"office": employee.office,
"request_type": new_request.request_type,
"priority": new_request.priority,
"description": new_request.description,
"created_at": new_request.created_at.isoformat(),
}
# Send notification to Telegram (non-blocking)
try:
await send_notification(notification_data)
except Exception as e:
print(f"Failed to send Telegram notification: {e}")
return new_request
@app.patch("/api/requests/{request_id}/status")
def update_request_status(
request_id: int,
status_update: request_models.StatusUpdate,
db: Session = Depends(get_db),
):
try:
request = requests.update_request_status(db, request_id, status_update.status)
if request is None:
raise HTTPException(status_code=404, detail="Request not found")
return request
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
@app.post("/api/requests/", response_model=request_models.Request)
def create_request(request_data: dict, db: Session = Depends(get_db)):
return requests.create_request(db=db, request_data=request_data)
@app.get("/api/requests/", response_model=List[request_models.RequestWithEmployee])
def read_requests(
skip: int = 0,
limit: int = 100,
last_name: str = None,
db: Session = Depends(get_db),
):
if last_name:
return requests.get_requests_by_employee_lastname(db, last_name)
return requests.get_requests(db, skip=skip, limit=limit)
@app.patch("/api/requests/{request_id}/status")
def update_request_status(request_id: int, status: str, db: Session = Depends(get_db)):
request = requests.update_request_status(db, request_id, status)
if request is None:
raise HTTPException(status_code=404, detail="Request not found")
return request
@app.patch("/api/requests/{request_id}/status")
def update_request_status(
request_id: int, status_update: StatusUpdate, db: Session = Depends(get_db)
):
try:
request = requests.update_request_status(db, request_id, status_update.status)
if request is None:
raise HTTPException(status_code=404, detail="Request not found")
return request
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
raise HTTPException(status_code=500, detail="Internal server error")
@app.get("/api/statistics")
def get_statistics(period: str = "week", db: Session = Depends(get_db)):
return statistics.get_statistics(db, period)

3
backend/app2/middleware/__init__.py
View File

@@ -1,3 +0,0 @@
from .logging import LoggingMiddleware
__all__ = ['LoggingMiddleware']

39
backend/app2/middleware/logging.py
View File

@@ -1,39 +0,0 @@
"""Logging middleware for request/response tracking"""
import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
import logging
logger = logging.getLogger("app.access")
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start_time = time.time()
# Log request
logger.info(
"Request started 1",
extra={
"client_addr": request.client.host,
"request_line": f"{request.method} {request.url.path}",
"status_code": "PENDING"
}
)
response = await call_next(request)
# Calculate processing time
process_time = time.time() - start_time
# Log response
logger.info(
"Request completed",
extra={
"client_addr": request.client.host,
"request_line": f"{request.method} {request.url.path}",
"status_code": response.status_code,
"process_time": f"{process_time:.2f}s"
}
)
return response

21
backend/app2/models/employee.py
View File

@@ -1,21 +0,0 @@
from pydantic import BaseModel
from datetime import datetime
class EmployeeBase(BaseModel):
last_name: str
first_name: str
department: str
office: str
class EmployeeCreate(EmployeeBase):
password: str
class Employee(EmployeeBase):
id: int
created_at: datetime
class Config:
from_attributes = True

50
backend/app2/models/request.py
View File

@@ -1,50 +0,0 @@
from pydantic import BaseModel
from datetime import datetime
from enum import Enum
class RequestStatus(str, Enum):
NEW = "new"
IN_PROGRESS = "in_progress"
RESOLVED = "resolved"
CLOSED = "closed"
class RequestPriority(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class StatusUpdate(BaseModel):
status: RequestStatus
class RequestBase(BaseModel):
department: str
request_type: str
priority: RequestPriority
description: str
class RequestCreate(RequestBase):
employee_id: int
class Request(RequestBase):
id: int
status: RequestStatus
created_at: datetime
employee_id: int
class Config:
from_attributes = True
class RequestWithEmployee(Request):
employee_last_name: str
employee_first_name: str
class Config:
from_attributes = True

34
backend/app2/schemas/tables.py
View File

@@ -1,34 +0,0 @@
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Enum
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from ..database import Base
from ..models.request import RequestStatus, RequestPriority
class Employee(Base):
__tablename__ = "employees"
id = Column(Integer, primary_key=True, index=True)
first_name = Column(String, nullable=False)
last_name = Column(String, nullable=False)
department = Column(String, nullable=False)
office = Column(String, nullable=False)
password = Column(String, nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
requests = relationship("Request", back_populates="employee")
class Request(Base):
__tablename__ = "requests"
id = Column(Integer, primary_key=True, index=True)
employee_id = Column(Integer, ForeignKey("employees.id"))
department = Column(String, nullable=False)
request_type = Column(String, nullable=False)
priority = Column(Enum(RequestPriority), nullable=False)
status = Column(Enum(RequestStatus), default=RequestStatus.NEW)
description = Column(String)
created_at = Column(DateTime(timezone=True), server_default=func.now())
employee = relationship("Employee", back_populates="requests")

11
backend/app2/utils/auth.py
View File

@@ -1,11 +0,0 @@
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)

45
backend/app2/utils/constants.py
View File

@@ -1,45 +0,0 @@
STATUS_LABELS = {
'new': 'Новая',
'in_progress': 'В работе',
'resolved': 'Решена',
'closed': 'Закрыта'
}
# Priority translations and emoji
PRIORITY_LABELS = {
'low': 'Низкий',
'medium': 'Средний',
'high': 'Высокий',
'critical': 'Критический'
}
PRIORITY_EMOJI = {
'low': '🟢',
'medium': '🟡',
'high': '🟠',
'critical': '🔴'
}
# Department translations
DEPARTMENT_LABELS = {
'aho': 'Административно-хозяйственный отдел',
'gkh': 'Жилищно-коммунальное хозяйство',
'general': 'Общий отдел'
}
# Request type translations and emoji
REQUEST_TYPE_LABELS = {
'hardware': 'Проблемы с оборудованием',
'software': 'Проблемы с ПО',
'network': 'Проблемы с сетью',
'access': 'Доступ к системам',
'other': 'Другое'
}
REQUEST_TYPE_EMOJI = {
'hardware': '🖥️',
'software': '💿',
'network': '🌐',
'access': '🔑',
'other': '📝'
}

90
backend/app2/utils/telegram.py
View File

@@ -1,90 +0,0 @@
from aiogram import Bot
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
import asyncio
from datetime import datetime
from logging import getLogger
from .constants import (
STATUS_LABELS, PRIORITY_LABELS, PRIORITY_EMOJI,
DEPARTMENT_LABELS, REQUEST_TYPE_LABELS, REQUEST_TYPE_EMOJI
)
# Initialize logger
logger = getLogger(__name__)
# Initialize bot with token
bot = Bot(token="7677506032:AAHEqNUr1lIUfNVbLwaWIaPeKKShsCyz3eo")
# Chat ID for notifications
CHAT_ID = "5057752127"
def create_status_keyboard(request_id: int, current_status: str) -> InlineKeyboardMarkup:
"""Create inline keyboard with status buttons"""
status_transitions = {
'new': ['in_progress'],
'in_progress': ['resolved'],
'resolved': ['closed'],
'closed': []
}
buttons = []
available_statuses = status_transitions.get(current_status, [])
for status in available_statuses:
callback_data = f"status_{request_id}_{status}"
logger.debug(f"Creating button with callback_data: {callback_data}")
buttons.append([
InlineKeyboardButton(
text=STATUS_LABELS[status],
callback_data=callback_data
)
])
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons)
logger.debug(f"Created keyboard: {keyboard}")
return keyboard
def format_request_message(request_data: dict) -> str:
"""Format request data into a message"""
created_at = datetime.fromisoformat(request_data['created_at']).strftime('%d.%m.%Y %H:%M')
# Get translated values
department = DEPARTMENT_LABELS.get(request_data['department'], request_data['department'])
request_type = REQUEST_TYPE_LABELS.get(request_data['request_type'], request_data['request_type'])
priority = PRIORITY_LABELS.get(request_data['priority'], request_data['priority'])
status = STATUS_LABELS.get(request_data.get('status', 'new'), 'Неизвестно')
return (
f"📋 <b>Заявка #{request_data['id']}</b>\n\n"
f"👤 <b>Сотрудник:</b> {request_data['employee_last_name']} {request_data['employee_first_name']}\n"
f"🏢 <b>Отдел:</b> {department}\n"
f"🚪 <b>Кабинет:</b> {request_data['office']}\n"
f"{REQUEST_TYPE_EMOJI.get(request_data['request_type'], '📝')} <b>Тип заявки:</b> {request_type}\n"
f"{PRIORITY_EMOJI.get(request_data['priority'], '⚪')} <b>Приоритет:</b> {priority}\n\n"
f"📝 <b>Описание:</b>\n{request_data['description']}\n\n"
f"🕒 <b>Создана:</b> {created_at}\n"
f"📊 <b>Статус:</b> {status}"
)
async def send_request_notification(request_data: dict):
"""Send notification about request to Telegram"""
try:
message = format_request_message(request_data)
keyboard = create_status_keyboard(request_data['id'], request_data.get('status', 'new'))
await bot.send_message(
chat_id=CHAT_ID,
text=message,
parse_mode="HTML",
reply_markup=keyboard
)
except Exception as e:
logger.error(f"Error sending Telegram notification: {e}", exc_info=True)
raise
def send_notification(request_data: dict):
"""Wrapper to run async notification in sync context"""
try:
asyncio.run(send_request_notification(request_data))
except Exception as e:
logger.error(f"Failed to send notification: {e}", exc_info=True)
raise