fc7484889c
✅ WEB_ADMIN_LOGIN/PASSWORD в .env.example ✅ Конфиг web_admin в config.py ✅ API /api/command - выполнение команд ✅ API /api/commands - список команд ✅ Обновлён web_server.py с конфигом 📦 Бекап: backups/versions/v2.1.2_web_auth_2026-02-28/ Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
365 lines
13 KiB
Python
365 lines
13 KiB
Python
"""
|
||
Web Admin Interface - FastAPI приложение
|
||
Веб-интерфейс для администратора домового бота
|
||
"""
|
||
import logging
|
||
from fastapi import FastAPI, Request, Depends, HTTPException, status
|
||
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
||
from fastapi.staticfiles import StaticFiles
|
||
from fastapi.templating import Jinja2Templates
|
||
from fastapi.responses import HTMLResponse, JSONResponse
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, func
|
||
from datetime import datetime
|
||
import secrets
|
||
import config
|
||
|
||
from database.db import AsyncSessionLocal
|
||
from database.models import User, Message, Poll, Ad, PaymentReminder, Schedule
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Приложение
|
||
app = FastAPI(
|
||
title="Домовой Бот - Админ Панель",
|
||
description="Веб-интерфейс для управления домовым ботом",
|
||
version="1.11.0"
|
||
)
|
||
|
||
# Шаблоны
|
||
templates = Jinja2Templates(directory="web/templates")
|
||
|
||
# Статика
|
||
app.mount("/static", StaticFiles(directory="web/static"), name="static")
|
||
|
||
# Безопасность
|
||
security = HTTPBasic()
|
||
|
||
|
||
# =============================================================================
|
||
# АВТОРИЗАЦИЯ
|
||
# =============================================================================
|
||
|
||
def get_current_admin(credentials: HTTPBasicCredentials = Depends(security)):
|
||
"""Проверка админа"""
|
||
correct_username = secrets.compare_digest(credentials.username, config.WEB_ADMIN_LOGIN)
|
||
correct_password = secrets.compare_digest(credentials.password, config.WEB_ADMIN_PASSWORD)
|
||
|
||
if not (correct_username and correct_password):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Неверное имя пользователя или пароль",
|
||
headers={"WWW-Authenticate": "Basic"},
|
||
)
|
||
|
||
return credentials.username
|
||
|
||
|
||
# =============================================================================
|
||
# ГЛАВНАЯ СТРАНИЦА
|
||
# =============================================================================
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
async def dashboard(request: Request, username: str = Depends(get_current_admin)):
|
||
"""Главная страница - дашборд"""
|
||
async with AsyncSessionLocal() as session:
|
||
# Общая статистика
|
||
users_count = await session.execute(select(func.count(User.user_id)))
|
||
users_count = users_count.scalar()
|
||
|
||
messages_count = await session.execute(select(func.count(Message.id)))
|
||
messages_count = messages_count.scalar()
|
||
|
||
verified_count = await session.execute(
|
||
select(func.count(User.user_id)).where(User.verified == True)
|
||
)
|
||
verified_count = verified_count.scalar()
|
||
|
||
active_polls = await session.execute(
|
||
select(func.count(Poll.poll_id)).where(Poll.is_active == True)
|
||
)
|
||
active_polls = active_polls.scalar()
|
||
|
||
active_ads = await session.execute(
|
||
select(func.count(Ad.ad_id)).where(Ad.is_active == True)
|
||
)
|
||
active_ads = active_ads.scalar()
|
||
|
||
upcoming_schedules = await session.execute(
|
||
select(func.count(Schedule.schedule_id))
|
||
.where(Schedule.is_active == True)
|
||
.where(Schedule.start_time > datetime.utcnow())
|
||
)
|
||
upcoming_schedules = upcoming_schedules.scalar()
|
||
|
||
return templates.TemplateResponse("dashboard.html", {
|
||
"request": request,
|
||
"username": username,
|
||
"stats": {
|
||
"users": users_count,
|
||
"messages": messages_count,
|
||
"verified": verified_count,
|
||
"polls": active_polls,
|
||
"ads": active_ads,
|
||
"schedules": upcoming_schedules
|
||
}
|
||
})
|
||
|
||
|
||
# =============================================================================
|
||
# API - СТАТИСТИКА
|
||
# =============================================================================
|
||
|
||
@app.get("/api/stats")
|
||
async def api_stats(username: str = Depends(get_current_admin)):
|
||
"""API: Получить статистику"""
|
||
async with AsyncSessionLocal() as session:
|
||
# Топ пользователей по рейтингу
|
||
stmt = select(User).order_by(User.rating.desc()).limit(10)
|
||
result = await session.execute(stmt)
|
||
top_users = list(result.scalars().all())
|
||
|
||
# Активные опросы
|
||
stmt = select(Poll).where(Poll.is_active == True).limit(5)
|
||
result = await session.execute(stmt)
|
||
active_polls = list(result.scalars().all())
|
||
|
||
# Последние сообщения
|
||
stmt = select(Message).order_by(Message.timestamp.desc()).limit(10)
|
||
result = await session.execute(stmt)
|
||
recent_messages = list(result.scalars().all())
|
||
|
||
return {
|
||
"top_users": [
|
||
{
|
||
"id": u.user_id,
|
||
"name": u.full_name,
|
||
"apartment": u.apartment,
|
||
"rating": u.rating
|
||
} for u in top_users
|
||
],
|
||
"active_polls": [
|
||
{
|
||
"id": p.poll_id,
|
||
"question": p.question,
|
||
"votes": p.total_votes
|
||
} for p in active_polls
|
||
],
|
||
"recent_messages": [
|
||
{
|
||
"id": m.id,
|
||
"user_id": m.user_id,
|
||
"text": m.text[:100] if m.text else None,
|
||
"timestamp": m.timestamp.isoformat()
|
||
} for m in recent_messages
|
||
]
|
||
}
|
||
|
||
|
||
# =============================================================================
|
||
# ПОЛЬЗОВАТЕЛИ
|
||
# =============================================================================
|
||
|
||
@app.get("/users", response_class=HTMLResponse)
|
||
async def users_page(request: Request, search: str = "", username: str = Depends(get_current_admin)):
|
||
"""Страница пользователей"""
|
||
async with AsyncSessionLocal() as session:
|
||
stmt = select(User)
|
||
|
||
if search:
|
||
stmt = stmt.where(
|
||
(User.full_name.ilike(f"%{search}%")) |
|
||
(User.apartment.ilike(f"%{search}%")) |
|
||
(User.username.ilike(f"%{search}%"))
|
||
)
|
||
|
||
stmt = stmt.order_by(User.user_id.desc()).limit(100)
|
||
result = await session.execute(stmt)
|
||
users = list(result.scalars().all())
|
||
|
||
return templates.TemplateResponse("users.html", {
|
||
"request": request,
|
||
"username": username,
|
||
"users": users,
|
||
"search": search
|
||
})
|
||
|
||
|
||
@app.get("/api/users")
|
||
async def api_users(search: str = "", username: str = Depends(get_current_admin)):
|
||
"""API: Список пользователей"""
|
||
async with AsyncSessionLocal() as session:
|
||
stmt = select(User)
|
||
|
||
if search:
|
||
stmt = stmt.where(
|
||
(User.full_name.ilike(f"%{search}%")) |
|
||
(User.apartment.ilike(f"%{search}%"))
|
||
)
|
||
|
||
stmt = stmt.order_by(User.user_id.desc()).limit(100)
|
||
result = await session.execute(stmt)
|
||
users = list(result.scalars().all())
|
||
|
||
return {
|
||
"users": [
|
||
{
|
||
"id": u.user_id,
|
||
"name": u.full_name,
|
||
"username": u.username,
|
||
"apartment": u.apartment,
|
||
"verified": u.verified,
|
||
"rating": u.rating,
|
||
"messages": u.message_count
|
||
} for u in users
|
||
]
|
||
}
|
||
|
||
|
||
# =============================================================================
|
||
# ЭКСПОРТ
|
||
# =============================================================================
|
||
|
||
@app.get("/export", response_class=HTMLResponse)
|
||
async def export_page(request: Request, username: str = Depends(get_current_admin)):
|
||
"""Страница экспорта"""
|
||
return templates.TemplateResponse("export.html", {
|
||
"request": request,
|
||
"username": username
|
||
})
|
||
|
||
|
||
@app.get("/api/export/json")
|
||
async def api_export_json(username: str = Depends(get_current_admin)):
|
||
"""API: Экспорт в JSON"""
|
||
from services.analytics import ChatAnalytics
|
||
|
||
async with AsyncSessionLocal() as session:
|
||
analytics = ChatAnalytics(session)
|
||
filepath = await analytics.export_to_json()
|
||
|
||
return {
|
||
"success": True,
|
||
"filepath": filepath
|
||
}
|
||
|
||
|
||
# =============================================================================
|
||
# НАСТРОЙКИ
|
||
# =============================================================================
|
||
|
||
@app.get("/settings", response_class=HTMLResponse)
|
||
async def settings_page(request: Request, username: str = Depends(get_current_admin)):
|
||
"""Страница настроек"""
|
||
return templates.TemplateResponse("settings.html", {
|
||
"request": request,
|
||
"username": username,
|
||
"config": {
|
||
"bot_token": config.BOT_TOKEN[:20] + "...",
|
||
"admin_chat_id": config.ADMIN_CHAT_ID,
|
||
"admin_user_id": config.ADMIN_USER_ID,
|
||
"use_proxy": config.USE_PROXY,
|
||
"web_login": config.WEB_ADMIN_LOGIN,
|
||
"web_password": config.WEB_ADMIN_PASSWORD
|
||
}
|
||
})
|
||
|
||
|
||
# =============================================================================
|
||
# API - КОМАНДЫ БОТА
|
||
# =============================================================================
|
||
|
||
@app.post("/api/command")
|
||
async def api_command(
|
||
request: Request,
|
||
username: str = Depends(get_current_admin)
|
||
):
|
||
"""Выполнить команду бота"""
|
||
import json
|
||
from aiogram import Bot
|
||
|
||
data = await request.json()
|
||
command = data.get('command', '')
|
||
args = data.get('args', '')
|
||
|
||
if not command:
|
||
return JSONResponse({"success": False, "error": "Нет команды"})
|
||
|
||
try:
|
||
# Создаём бота
|
||
bot = Bot(token=config.BOT_TOKEN)
|
||
|
||
# Формируем полное сообщение команды
|
||
full_command = f"/{command}"
|
||
if args:
|
||
full_command += f" {args}"
|
||
|
||
# Отправляем команду в чат админа
|
||
result = await bot.send_message(
|
||
config.ADMIN_CHAT_ID,
|
||
full_command
|
||
)
|
||
|
||
await bot.session.close()
|
||
|
||
return JSONResponse({
|
||
"success": True,
|
||
"message": f"Команда /{command} отправлена",
|
||
"result_id": result.message_id
|
||
})
|
||
except Exception as e:
|
||
return JSONResponse({
|
||
"success": False,
|
||
"error": str(e)
|
||
}, status_code=500)
|
||
|
||
|
||
@app.get("/api/commands")
|
||
async def api_commands_list(username: str = Depends(get_current_admin)):
|
||
"""Список доступных команд"""
|
||
commands = {
|
||
"admin": [
|
||
{"cmd": "admin", "desc": "Админ-панель"},
|
||
{"cmd": "stats", "desc": "Статистика чата"},
|
||
{"cmd": "spies", "desc": "Подозрительные пользователи"},
|
||
{"cmd": "users", "desc": "Список жильцов"},
|
||
{"cmd": "verify", "desc": "Заявки на верификацию"},
|
||
{"cmd": "export", "desc": "Экспорт данных"},
|
||
{"cmd": "rating", "desc": "Рейтинг пользователей"},
|
||
],
|
||
"initiative_group": [
|
||
{"cmd": "ig_add", "desc": "Добавить в ИГ"},
|
||
{"cmd": "ig_remove", "desc": "Удалить из ИГ"},
|
||
{"cmd": "ig_list", "desc": "Список ИГ"},
|
||
{"cmd": "ig_stats", "desc": "Статистика ИГ"},
|
||
{"cmd": "ig_check", "desc": "Проверка участника"},
|
||
{"cmd": "ig_broadcast", "desc": "Рассылка для ИГ"},
|
||
],
|
||
"events": [
|
||
{"cmd": "event_add", "desc": "Добавить событие"},
|
||
{"cmd": "events", "desc": "Список событий"},
|
||
{"cmd": "participants", "desc": "Участники события"},
|
||
{"cmd": "event_cancel", "desc": "Отменить событие"},
|
||
],
|
||
"security": [
|
||
{"cmd": "unverify", "desc": "Отозвать верификацию"},
|
||
{"cmd": "unverified_list", "desc": "Список не верифицированных"},
|
||
{"cmd": "ban", "desc": "Забанить пользователя"},
|
||
{"cmd": "unban", "desc": "Разбанить"},
|
||
]
|
||
}
|
||
return JSONResponse({"commands": commands})
|
||
|
||
|
||
# =============================================================================
|
||
# ЗАПУСК
|
||
# =============================================================================
|
||
|
||
def run_web_server(host: str = "0.0.0.0", port: int = 8000):
|
||
"""Запустить веб-сервер"""
|
||
import uvicorn
|
||
uvicorn.run(app, host=host, port=port)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run_web_server()
|