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