""" 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()