ba4676b0af
✅ Регистрация и верификация жильцов ✅ Рейтинг активности (уровни) ✅ Детект шпионов (Score 0-100) ✅ Анти-мат система (3 предупреждения → бан) ✅ Админ-панель с рассылкой ✅ Учёт квартир (несколько жильцов) ✅ Телефоны экстренных служб и мастеров ✅ Еженедельный экспорт JSON ✅ Прокси (socks5) для обхода РКН Бекап: backups/versions/v1.0_mvp_2026-02-26/ Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
277 lines
9.5 KiB
Python
277 lines
9.5 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, "admin")
|
||
correct_password = secrets.compare_digest(credentials.password, str(config.ADMIN_USER_ID))
|
||
|
||
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
|
||
}
|
||
})
|
||
|
||
|
||
# =============================================================================
|
||
# ЗАПУСК
|
||
# =============================================================================
|
||
|
||
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()
|