domovoy_bot/web/app.py

434 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
Web Admin Interface v4.2 REBORN - MEGA ADMIN PANEL
Полноценное управление ВСЕМ функционалом бота
"""
import logging
import json
import aiohttp
import re
from pathlib import Path
from fastapi import FastAPI, Request, Depends, HTTPException, status, Form, File, UploadFile
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, FileResponse
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, update, delete, desc, text
from database.models import User, Message, EmailExportLog, EmailAudit
from datetime import datetime, timedelta, timezone
import secrets
import config
# ===== ЧАСОВОЙ ПОЯС УЛЬЯНОВСК (UTC+4) =====
ULY_TZ = timezone(timedelta(hours=4))
def to_local(dt: datetime) -> datetime:
if dt is None: return None
# Если время без часового пояса (naive), считаем что оно в UTC (как в БД)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(ULY_TZ)
def fmt_local(dt: datetime, fmt: str = '%d.%m.%Y %H:%M') -> str:
local_dt = to_local(dt)
return local_dt.strftime(fmt) if local_dt else '-'
from database.db import AsyncSessionLocal
from database.models import (
User, Message, Poll, Ad, PaymentReminder, Schedule,
VerificationRequest, Service, Announcement, Event, ScheduledPost,
Broadcast, BroadcastRead, Digest, InitiativeGroup, EmailAudit, EmailExportLog
)
logger = logging.getLogger(__name__)
app = FastAPI(title="LKM37 MEGA Admin Panel", version="4.2.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.exception_handler(Exception)
async def debug_exception_handler(request: Request, exc: Exception):
logger.error(f"❌ WEB ERROR: {exc}", exc_info=True)
return JSONResponse(status_code=500, content={"message": f"Internal Server Error: {str(exc)}"})
templates = Jinja2Templates(directory="web/templates")
base_dir = Path(__file__).parent.parent
app.mount("/static", StaticFiles(directory="web/static"), name="static")
app.mount("/static/service_images", StaticFiles(directory=str(base_dir / "data" / "service_images")), name="service_images")
app.mount("/static/email_archive", StaticFiles(directory=str(base_dir / "services" / "email_auditor" / "exported_emails")), name="email_archive")
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=401, 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)))).scalar()
verified_count = (await session.execute(select(func.count(User.user_id)).where(User.verified == True))).scalar()
messages_count = (await session.execute(select(func.count(Message.id)))).scalar()
active_ads = (await session.execute(select(func.count(Ad.ad_id)).where(Ad.is_active == True))).scalar()
active_polls = (await session.execute(select(func.count(Poll.poll_id)).where(Poll.is_active == True))).scalar()
upcoming_events = (await session.execute(select(func.count(Event.id)).where(Event.is_active == True))).scalar()
pending_count = (await session.execute(select(func.count(VerificationRequest.id)).where(VerificationRequest.status == 'pending'))).scalar()
return templates.TemplateResponse(
request=request, name="dashboard.html",
context={
"username": username, "users_count": users_count, "verified_count": verified_count,
"messages_count": messages_count, "active_ads": active_ads, "active_polls": active_polls,
"upcoming_events": upcoming_events, "pending_count": pending_count
}
)
# ============================================================================
# ПОЛЬЗОВАТЕЛИ
# ============================================================================
@app.get("/users", response_class=HTMLResponse)
async def users_page(request: Request, search: str = "", filter: str = "", username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
stmt = select(User)
if filter == 'verified': stmt = stmt.where(User.verified == True)
elif filter == 'unverified': stmt = stmt.where(User.verified == False)
elif filter == 'ig':
ig_stmt = select(InitiativeGroup.user_id).where(InitiativeGroup.is_active == True)
ig_user_ids = [row[0] for row in (await session.execute(ig_stmt)).all()]
stmt = stmt.where(User.user_id.in_(ig_user_ids))
if search:
stmt = stmt.where((User.first_name.ilike(f"%{search}%")) | (User.username.ilike(f"%{search}%")) | (User.apartment.ilike(f"%{search}%")))
users = list((await session.execute(stmt.order_by(User.user_id.desc()).limit(100))).scalars().all())
total_count = (await session.execute(select(func.count(User.user_id)))).scalar()
verified_count = (await session.execute(select(func.count(User.user_id)).where(User.verified == True))).scalar()
ig_stmt = select(InitiativeGroup.user_id).where(InitiativeGroup.is_active == True)
ig_user_ids = set(row[0] for row in (await session.execute(ig_stmt)).all())
pending_stmt = select(VerificationRequest).where(VerificationRequest.status == 'pending')
pending_requests = list((await session.execute(pending_stmt)).scalars().all())
return templates.TemplateResponse(
request=request, name="users.html",
context={
"username": username, "users": users, "search": search, "current_filter": filter,
"total_count": total_count, "verified_count": verified_count, "unverified_count": total_count - verified_count,
"ig_count": len(ig_user_ids), "ig_user_ids": ig_user_ids, "pending_requests": [], "pending_count": len(pending_requests),
"active_unverified": 0
}
)
# ============================================================================
# РАССЫЛКИ
# ============================================================================
@app.get("/broadcast", response_class=HTMLResponse)
async def broadcast_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
broadcasts = list((await session.execute(select(Broadcast).order_by(Broadcast.created_at.desc()).limit(50))).scalars().all())
return templates.TemplateResponse(request=request, name="broadcast.html", context={"username": username, "broadcasts": broadcasts})
@app.post("/api/broadcast/create")
async def api_create_broadcast(
text: str = Form(...),
photo: UploadFile = File(None),
recipients: str = Form("all_and_chat"),
staircase: str = Form("all"),
username: str = Depends(get_current_admin)
):
from handlers.smart_broadcast import send_smart_broadcast
from bot_instance import get_bot
photo_file_id = None
if photo and photo.filename:
temp_path = Path("data") / photo.filename
with open(temp_path, "wb") as f: f.write(await photo.read())
async with aiohttp.ClientSession() as session:
url = f"https://api.telegram.org/bot{config.BOT_TOKEN}/sendPhoto"
data = aiohttp.FormData()
data.add_field('chat_id', str(config.ADMIN_USER_ID))
data.add_field('photo', open(temp_path, 'rb'))
async with session.post(url, data=data, proxy=config.get_proxy_url()) as resp:
res = await resp.json()
if res.get('ok'): photo_file_id = res['result']['photo'][-1]['file_id']
if temp_path.exists(): temp_path.unlink()
bot = get_bot()
success = await send_smart_broadcast(bot, text, recipients, photo_file_id, staircase)
return JSONResponse({"success": success})
# ============================================================================
# ФАЙЛЫ (МЕНЕДЖЕР)
# ============================================================================
@app.get("/files", response_class=HTMLResponse)
async def files_page(request: Request, username: str = Depends(get_current_admin)):
return templates.TemplateResponse(request=request, name="files.html", context={"username": username})
@app.get("/api/files/list")
async def api_list_files(username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
result = await session.execute(text("SELECT id, filename, file_id, file_type FROM assets ORDER BY created_at DESC"))
files = [{"id": r[0], "filename": r[1], "file_id": r[2], "file_type": r[3]} for r in result.all()]
return JSONResponse({"success": True, "files": files})
@app.post("/api/files/upload")
async def api_upload_file(file: UploadFile = File(...), username: str = Depends(get_current_admin)):
temp_path = Path("data") / file.filename
with open(temp_path, "wb") as f: f.write(await file.read())
file_ext = Path(file.filename).suffix.lower()
is_photo = file_ext in ['.jpg', '.jpeg', '.png', '.webp']
tg_method = "sendPhoto" if is_photo else "sendDocument"
try:
async with aiohttp.ClientSession() as session:
url = f"https://api.telegram.org/bot{config.BOT_TOKEN}/{tg_method}"
data = aiohttp.FormData()
data.add_field('chat_id', str(config.ADMIN_USER_ID))
data.add_field('photo' if is_photo else 'document', open(temp_path, 'rb'))
async with session.post(url, data=data, proxy=config.get_proxy_url()) as resp:
res = await resp.json()
if not res.get('ok'): raise Exception(res.get('description'))
fid = res['result']['photo'][-1]['file_id'] if is_photo else res['result']['document']['file_id']
async with AsyncSessionLocal() as sdb:
await sdb.execute(text("INSERT INTO assets (filename, file_id, file_type) VALUES (:fn, :fid, :ft)"),
{"fn": file.filename, "fid": fid, "ft": "photo" if is_photo else "document"})
await sdb.commit()
return JSONResponse({"success": True, "file_id": fid})
except Exception as e: return JSONResponse({"success": False, "error": str(e)}, status_code=500)
finally:
if temp_path.exists(): temp_path.unlink()
# ============================================================================
# ОСТАЛЬНЫЕ РОУТЫ (ЗАГЛУШКИ / БАЗОВЫЕ)
# ============================================================================
@app.get("/verification", response_class=HTMLResponse)
async def verification_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
pending = list((await session.execute(select(VerificationRequest).where(VerificationRequest.status == 'pending'))).scalars().all())
unverified = list((await session.execute(select(User).where(User.verified == False).limit(50))).scalars().all())
return templates.TemplateResponse(request=request, name="verification.html", context={"username": username, "pending_requests": pending, "unverified_users": unverified})
@app.get("/phones", response_class=HTMLResponse)
async def phones_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
services = list((await session.execute(select(Service).order_by(Service.category, Service.name))).scalars().all())
return templates.TemplateResponse(request=request, name="phones.html", context={"username": username, "services": services})
@app.get("/schedules", response_class=HTMLResponse)
async def schedules_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
schedules = list((await session.execute(select(Schedule).order_by(Schedule.start_time.desc()))).scalars().all())
return templates.TemplateResponse(request=request, name="schedules.html", context={"username": username, "schedules": schedules})
@app.get("/ads", response_class=HTMLResponse)
async def ads_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
ads = list((await session.execute(select(Ad).order_by(Ad.created_at.desc()).limit(100))).scalars().all())
return templates.TemplateResponse(request=request, name="ads.html", context={"username": username, "ads": ads})
@app.get("/polls", response_class=HTMLResponse)
async def polls_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
polls = list((await session.execute(select(Poll).order_by(Poll.created_at.desc()).limit(50))).scalars().all())
return templates.TemplateResponse(request=request, name="polls.html", context={"username": username, "polls": polls})
@app.get("/events", response_class=HTMLResponse)
async def events_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
events = list((await session.execute(select(Event).order_by(Event.event_date.desc()).limit(50))).scalars().all())
return templates.TemplateResponse(request=request, name="events.html", context={"username": username, "events": events})
@app.get("/digests", response_class=HTMLResponse)
async def digests_page(request: Request, username: str = Depends(get_current_admin)):
return templates.TemplateResponse(request=request, name="digests.html", context={"username": username, "pending_digests": [], "archive_digests": []})
@app.get("/export", response_class=HTMLResponse)
async def export_page(request: Request, username: str = Depends(get_current_admin)):
return templates.TemplateResponse(request=request, name="export.html", context={"username": username})
@app.get("/api/export/json")
async def api_export_json(username: str = Depends(get_current_admin)):
"""Экспорт данных в JSON с прямой отдачей файла"""
import json
from database.db import AsyncSessionLocal
export_dir = Path(__file__).parent.parent / "exports"
export_dir.mkdir(exist_ok=True)
# Время по Ульяновску для имени файла
now_uly = datetime.now(ULY_TZ)
timestamp = now_uly.strftime("%Y%m%d_%H%M%S")
filename = f"domovoy_export_{timestamp}.json"
filepath = export_dir / filename
try:
async with AsyncSessionLocal() as session:
# Получаем ВСЕХ пользователей
users_res = await session.execute(select(User))
users = users_res.scalars().all()
# Получаем ВСЕ сообщения (без лимита 1000)
messages_res = await session.execute(select(Message).order_by(Message.timestamp.asc()))
messages = messages_res.scalars().all()
data = {
"export_info": {
"date": fmt_local(now_uly),
"total_users": len(list(users)),
"total_messages": len(list(messages))
},
"users": [
{
"user_id": u.user_id, "username": u.username, "first_name": u.first_name,
"last_name": u.last_name, "apartment": u.apartment, "phone": u.phone,
"verified": u.verified, "rating": u.rating, "level": u.level
} for u in users
],
"messages": [
{
"id": m.id, "user_id": m.user_id, "text": m.text,
"date": fmt_local(m.timestamp), "topic": m.topic
} for m in messages
]
}
with open(filepath, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# Отдаем файл напрямую в браузер
return FileResponse(
path=filepath,
filename=filename,
media_type='application/json'
)
except Exception as e:
logger.error(f"❌ Export error: {e}", exc_info=True)
return JSONResponse(status_code=500, content={"success": False, "error": str(e)})
@app.get("/scheduled_posts", response_class=HTMLResponse)
async def scheduled_posts_page(request: Request, username: str = Depends(get_current_admin)):
return templates.TemplateResponse(request=request, name="scheduled_posts.html", context={"username": username})
# ============================================================================
# АУДИТ ПОЧТЫ
# ============================================================================
@app.get("/emails", response_class=HTMLResponse)
async def emails_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
# Получаем статистику по доменам
stmt = select(EmailAudit).order_by(EmailAudit.last_email_date.desc())
audit_data = list((await session.execute(stmt)).scalars().all())
# Получаем логи последнего экспорта
stmt_log = select(EmailExportLog).order_by(EmailExportLog.start_time.desc()).limit(1)
last_log = (await session.execute(stmt_log)).scalar_one_or_none()
# Получаем список последних 10 файлов (писем)
latest_emails = []
all_folder = base_dir / "services" / "email_auditor" / "exported_emails" / "all"
if all_folder.exists():
files = sorted(all_folder.glob("*.pdf"), key=lambda x: x.stat().st_mtime, reverse=True)[:10]
for f in files:
name = f.stem
# Парсим: YYYY-MM-DD_domain_subject
match = re.match(r'(\d{4}-\d{2}-\d{2})_([^_]+)_(.*)', name)
if match:
date_str, domain, subject = match.groups()
latest_emails.append({"date": date_str, "domain": domain, "subject": subject})
else:
# Вариант 2: domain_date_sender_subject
match = re.match(r'([^_]+)_(\d{4}-\d{2}-\d{2})_(.*)', name)
if match:
domain, date_str, rest = match.groups()
latest_emails.append({"date": date_str, "domain": domain, "subject": rest})
# Проверяем наличие мастер-файла
master_pdf_path = Path(__file__).parent.parent / "services" / "email_auditor" / "exported_emails" / "TOTAL_ARCHIVE_2025-2026.pdf"
has_master_pdf = master_pdf_path.exists()
last_update = None
if has_master_pdf:
mtime = master_pdf_path.stat().st_mtime
# Берем время как UTC
last_update = datetime.fromtimestamp(mtime, tz=timezone.utc)
return templates.TemplateResponse(
request=request, name="emails.html",
context={
"username": username, "audit_data": audit_data, "last_log": last_log,
"latest_emails": latest_emails,
"has_master_pdf": has_master_pdf, "last_update": last_update,
"fmt_local": fmt_local
}
)
@app.post("/api/emails/run")
async def api_run_email_audit(username: str = Depends(get_current_admin)):
"""Запуск аудита почты вручную (фоновым процессом)"""
import subprocess
import sys
import os
# Определяем пути
base_dir = Path(__file__).parent.parent
script_path = base_dir / "services" / "email_auditor" / "export_emails_v2.py"
cwd_path = base_dir / "services" / "email_auditor"
# Ищем питон в venv
venv_python = base_dir / "venv" / "bin" / "python3"
if not venv_python.exists():
# Попробуем просто python3 если venv не найден
venv_python = Path(sys.executable)
if not script_path.exists():
return JSONResponse(status_code=404, content={"message": f"Скрипт не найден: {script_path}"})
logger.info(f"🚀 Manual audit attempt. Python: {venv_python}, Script: {script_path}")
try:
# Устанавливаем PYTHONPATH
env = os.environ.copy()
env["PYTHONPATH"] = str(base_dir)
# Открываем лог-файл для записи вывода скрипта
audit_log_path = base_dir / "logs" / "manual_audit.log"
audit_log = open(audit_log_path, "a")
# Запускаем скрипт в фоне
process = subprocess.Popen(
[str(venv_python), str(script_path)],
cwd=str(cwd_path),
env=env,
stdout=audit_log,
stderr=audit_log,
start_new_session=True
)
return JSONResponse(content={"message": f"Процесс аудита успешно запущен (PID: {process.pid}). Результат придет в Telegram."})
except Exception as e:
logger.error(f"❌ Subprocess error: {e}", exc_info=True)
return JSONResponse(status_code=500, content={"message": f"Ошибка запуска процесса: {str(e)}"})
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()