"""
Web Admin Interface v4.8 REBORN - MEGA ADMIN PANEL
Полноценное управление ВСЕМ функционалом бота:
- Dashboard, Users, Verification, Phones, Ads, Polls, Events, Schedules
- Smart Broadcasts (фото + док через Artifacts)
- Scheduled Posts (фото + док через Artifacts)
- Weekly Digests (v3.1)
- Chat Exporter (JSON + Gitea Push)
- Email Auditor (аудит переписки)
- Artifact Manager (хранилище медиа ID)
"""
import logging
import json
import aiohttp
import re
import os
import sys
import secrets
import subprocess
import uuid
from pathlib import Path
from datetime import datetime, timedelta, timezone
# ГАРАНТИРУЕМ, что корень проекта в пути поиска модулей
BASE_DIR = Path(__file__).parent.parent
if str(BASE_DIR) not in sys.path:
sys.path.insert(0, str(BASE_DIR))
from fastapi import FastAPI, Request, Depends, HTTPException, status, Form, File, UploadFile, Query
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
import config
from database.db import AsyncSessionLocal
from database.models import (
User, Message as DBMessage, Poll, Ad, PaymentReminder, Schedule,
VerificationRequest, Service, Announcement, Event, ScheduledPost,
Broadcast, BroadcastRead, Digest, InitiativeGroup, EmailAudit, EmailExportLog,
ProfileHistory, Achievement, Thank, SpyLog, Artifact, Speedtest
)
# ===== ЧАСОВОЙ ПОЯС УЛЬЯНОВСК (UTC+4) =====
ULY_TZ = timezone(timedelta(hours=4))
def to_local(dt) -> datetime:
if dt is None: return None
if isinstance(dt, str):
try:
dt = datetime.fromisoformat(dt.replace(' ', 'T'))
except: return None
if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(ULY_TZ)
def fmt_local(dt, fmt: str = '%d.%m.%Y %H:%M') -> str:
local_dt = to_local(dt)
return local_dt.strftime(fmt) if local_dt else '-'
logger = logging.getLogger(__name__)
logger.info(f"!!! [STARTUP] Unified Web App loaded. PID: {os.getpid()}")
app = FastAPI(title="LKM37 MEGA Admin Panel", version="4.8.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
templates = Jinja2Templates(directory=str(BASE_DIR / "web" / "templates"))
app.mount("/static", StaticFiles(directory=str(BASE_DIR / "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=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect login or password",
headers={"WWW-Authenticate": "Basic"},
)
return credentials.username
# Вспомогательная функция для загрузки в TG
async def upload_to_tg(content, filename, content_type):
file_type = "document"
if content_type.startswith("image/"): file_type = "photo"
proxy_url = f"http://{config.PROXY_HOST}:{config.PROXY_PORT}" if config.USE_PROXY else None
async with aiohttp.ClientSession() as session:
data = aiohttp.FormData()
data.add_field(file_type, content, filename=filename, content_type=content_type)
data.add_field('chat_id', str(config.ADMIN_USER_ID))
data.add_field('caption', f"📤 Загрузка через веб-панель: {filename}")
method = "sendPhoto" if file_type == "photo" else "sendDocument"
url = f"https://api.telegram.org/bot{config.BOT_TOKEN}/{method}"
async with session.post(url, data=data, proxy=proxy_url) as resp:
tg_res = await resp.json()
if not tg_res.get("ok"):
raise Exception(f"TG Error: {tg_res.get('description')}")
msg = tg_res["result"]
file_id = msg["photo"][-1]["file_id"] if file_type == "photo" else msg["document"]["file_id"]
return file_id, file_type
# ============================================================================
# ГЛАВНАЯ ПАНЕЛЬ (DASHBOARD)
# ============================================================================
@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(DBMessage.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 == 'active_unverified': stmt = stmt.where(User.verified == False, User.message_count >= 50)
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.last_name.ilike(f"%{search}%")) |
(User.username.ilike(f"%{search}%")) |
(User.apartment.ilike(f"%{search}%"))
)
stmt = stmt.order_by(User.user_id.desc()).limit(100)
users = list((await session.execute(stmt)).scalars().all())
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())
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()
return templates.TemplateResponse(request=request, name="users.html", context={
"username": username, "users": users, "search": search, "current_filter": filter,
"stats": {"total": total_count, "verified": verified_count, "unverified": total_count - verified_count, "ig": len(ig_user_ids)},
"ig_user_ids": ig_user_ids
})
@app.post("/api/user/{user_id}/update")
async def api_update_user(user_id: int, request: Request, username: str = Depends(get_current_admin)):
data = await request.json()
async with AsyncSessionLocal() as session:
update_data = {}
if 'apartment' in data: update_data['apartment'] = data['apartment'].upper()
if 'phone' in data: update_data['phone'] = data['phone']
if 'verified' in data:
update_data['verified'] = bool(data['verified'])
if data['verified']: update_data['verification_date'] = datetime.utcnow()
if update_data:
await session.execute(update(User).where(User.user_id == user_id).values(**update_data))
await session.commit()
return JSONResponse({"success": True})
# ============================================================================
# ВЕРИФИКАЦИЯ
# ============================================================================
@app.get("/verification", response_class=HTMLResponse)
async def verification_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
pending = (await session.execute(select(VerificationRequest).where(VerificationRequest.status == 'pending').order_by(VerificationRequest.created_at.desc()))).scalars().all()
unverified = (await session.execute(select(User).where(User.verified == False).order_by(User.join_date.desc()).limit(50))).scalars().all()
return templates.TemplateResponse(request=request, name="verification.html", context={"username": username, "pending_requests": pending, "unverified_users": unverified})
@app.post("/api/verification/{req_id}/approve")
async def api_approve_verification(req_id: int, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
req = (await session.execute(select(VerificationRequest).where(VerificationRequest.id == req_id))).scalar_one_or_none()
if req:
req.status = 'approved'
req.reviewed_at = datetime.utcnow()
await session.execute(update(User).where(User.user_id == req.user_id).values(verified=True, apartment=req.apartment, verification_date=datetime.utcnow()))
await session.commit()
# Уведомляем пользователя
try:
proxy_url = config.get_proxy_url()
async with aiohttp.ClientSession() as http_session:
await http_session.post(
f"https://api.telegram.org/bot{config.BOT_TOKEN}/sendMessage",
json={
"chat_id": req.user_id,
"text": f"✅ Ваша заявка на верификацию одобрена!\n\nКвартира: {req.apartment}\nТеперь вам доступны все функции бота.",
"parse_mode": "HTML"
},
proxy=proxy_url,
timeout=10
)
except Exception as e:
logger.error(f"Error notifying user {req.user_id}: {e}")
return JSONResponse({"success": True})
@app.post("/api/verification/{req_id}/reject")
async def api_reject_verification(req_id: int, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
req = (await session.execute(select(VerificationRequest).where(VerificationRequest.id == req_id))).scalar_one_or_none()
if req:
req.status = 'rejected'
req.reviewed_at = datetime.utcnow()
await session.commit()
return JSONResponse({"success": True})
@app.post("/api/user/{user_id}/verify")
async def api_verify_user(user_id: int, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
user = (await session.execute(select(User).where(User.user_id == user_id))).scalar_one_or_none()
if not user:
return JSONResponse({"success": False, "error": "Пользователь не найден"})
user.verified = True
user.verification_date = datetime.utcnow()
await session.commit()
# Уведомляем пользователя через бот
try:
proxy_url = config.get_proxy_url()
async with aiohttp.ClientSession() as http_session:
await http_session.post(
f"https://api.telegram.org/bot{config.BOT_TOKEN}/sendMessage",
json={
"chat_id": user_id,
"text": "✅ Верификация подтверждена!\n\nТеперь у вас есть полный доступ ко всем функциям бота LKM37.",
"parse_mode": "HTML"
},
proxy=proxy_url,
timeout=10
)
except Exception as e:
logger.error(f"Error notifying user {user_id}: {e}")
return JSONResponse({"success": True})
@app.post("/api/user/{user_id}/unverify")
async def api_unverify_user(user_id: int, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
user = (await session.execute(select(User).where(User.user_id == user_id))).scalar_one_or_none()
if not user:
return JSONResponse({"success": False, "error": "Пользователь не найден"})
user.verified = False
user.verification_date = None
await session.commit()
return JSONResponse({"success": True})
@app.post("/api/user/{user_id}/toggle_ai")
async def api_toggle_ai(user_id: int, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
user = (await session.execute(select(User).where(User.user_id == user_id))).scalar_one_or_none()
if not user:
return JSONResponse({"success": False, "error": "Пользователь не найден"})
user.has_ai_access = not user.has_ai_access
await session.commit()
return JSONResponse({"success": True, "has_ai_access": user.has_ai_access})
# ============================================================================
# ТЕЛЕФОНЫ / ОБЪЯВЛЕНИЯ / ОПРОСЫ / СОБЫТИЯ / ГРАФИКИ
# ============================================================================
@app.get("/phones", response_class=HTMLResponse)
async def phones_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
services = (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("/ads", response_class=HTMLResponse)
async def ads_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
ads = (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 = (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 = (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("/schedules", response_class=HTMLResponse)
async def schedules_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
schedules = (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})
# ============================================================================
# РАССЫЛКИ (SMART BROADCAST)
# ============================================================================
@app.get("/broadcast", response_class=HTMLResponse)
async def broadcast_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
broadcasts = (await session.execute(select(Broadcast).order_by(Broadcast.created_at.desc()).limit(50))).scalars().all()
artifacts = (await session.execute(select(Artifact).order_by(Artifact.filename))).scalars().all()
broadcasts_data = []
for b in broadcasts:
read_count = (await session.execute(select(func.count(BroadcastRead.id)).where(BroadcastRead.broadcast_id == b.id))).scalar() or 0
broadcasts_data.append({
'id': b.id, 'text': b.text[:100] + '...' if len(b.text) > 100 else b.text,
'sent_at': fmt_local(b.sent_at) if b.sent_at else '-',
'total_sent': b.total_sent, 'read_count': read_count,
'unread_count': b.total_sent - read_count,
'read_percent': round((read_count / b.total_sent * 100), 1) if b.total_sent > 0 else 0,
'broadcast_type': b.broadcast_type or 'regular',
'is_reminder_sent': b.is_reminder_sent
})
return templates.TemplateResponse(request=request, name="broadcast.html", context={"username": username, "broadcasts": broadcasts_data, "artifacts": artifacts})
@app.post("/api/broadcast/create")
async def api_create_broadcast(
text: str = Form(...),
photo: UploadFile = File(None),
document: UploadFile = File(None),
artifact_photo_id: str = Form(None),
artifact_document_id: str = Form(None),
recipients: str = Form("all_and_chat"),
staircase: str = Form("all"),
target_user_id: int = Form(None),
has_read_button: bool = Form(True),
track_reads: bool = Form(True),
dry_run: bool = Form(False),
username: str = Depends(get_current_admin)
):
try:
from database.models import User, Broadcast, InitiativeGroup
photo_file_id = artifact_photo_id
document_file_id = artifact_document_id
# Если загружена новая картинка
if photo and photo.filename and not photo_file_id:
content = await photo.read()
photo_file_id, _ = await upload_to_tg(content, photo.filename, photo.content_type)
# Если загружен новый документ
if document and document.filename and not document_file_id:
content = await document.read()
document_file_id, _ = await upload_to_tg(content, document.filename, document.content_type)
async with AsyncSessionLocal() as session:
stmt = select(User).where(User.verified == True, User.is_banned == False)
if recipients == 'single_user' and target_user_id:
stmt = stmt.where(User.user_id == target_user_id)
else:
if staircase != 'all':
stmt = stmt.where(User.staircase == int(staircase))
if recipients == 'ig_only':
ig_stmt = select(InitiativeGroup.user_id).where(InitiativeGroup.is_active == True)
ig_ids = [row[0] for row in (await session.execute(ig_stmt)).all()]
stmt = stmt.where(User.user_id.in_(ig_ids))
users = (await session.execute(stmt)).scalars().all()
if dry_run:
return JSONResponse({"success": True, "recipients_count": len(users)})
new_broadcast = Broadcast(
text=text, photo_file_id=photo_file_id,
sent_at=datetime.utcnow(), total_sent=len(users),
broadcast_type='smart' if track_reads else 'regular'
)
session.add(new_broadcast)
await session.commit()
# Отправка сообщений
from handlers.smart_broadcast import send_smart_broadcast
from bot_instance import get_bot
import asyncio
bot = get_bot()
user_ids = [u.user_id for u in users]
# Запускаем в фоне, чтобы не блокировать веб-интерфейс
asyncio.create_task(send_smart_broadcast(
bot=bot,
text=text,
user_ids=user_ids,
photo_file_id=photo_file_id,
document_file_id=document_file_id,
broadcast_id=new_broadcast.id
))
return JSONResponse({"success": True, "message": f"Рассылка создана. Получателей: {len(users)}", "broadcast_id": new_broadcast.id})
except Exception as e:
logger.error(f"Broadcast error: {e}")
return JSONResponse(status_code=500, content={"success": False, "error": str(e)})
@app.get("/api/broadcast/readers/{broadcast_id}")
async def api_broadcast_readers(broadcast_id: int, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
stmt = select(User.first_name, User.last_name, User.apartment, BroadcastRead.read_at).join(BroadcastRead).where(BroadcastRead.broadcast_id == broadcast_id)
results = (await session.execute(stmt)).all()
readers = [{"name": f"{r[0]} {r[1] or ''}", "apartment": r[2], "read_at": fmt_local(r[3])} for r in results]
return JSONResponse({"readers": readers})
@app.get("/api/broadcast/non_readers/{broadcast_id}")
async def api_broadcast_non_readers(broadcast_id: int, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
read_ids_stmt = select(BroadcastRead.user_id).where(BroadcastRead.broadcast_id == broadcast_id)
read_ids = [r[0] for r in (await session.execute(read_ids_stmt)).all()]
stmt = select(User.first_name, User.last_name, User.apartment).where(User.verified == True, User.user_id.not_in(read_ids))
results = (await session.execute(stmt)).all()
non_readers = [{"name": f"{r[0]} {r[1] or ''}", "apartment": r[2]} for r in results]
return JSONResponse({"non_readers": non_readers})
# ============================================================================
# ЗАПЛАНИРОВАННЫЕ ПОСТЫ
# ============================================================================
@app.get("/scheduled_posts", response_class=HTMLResponse)
async def scheduled_posts_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
artifacts = (await session.execute(select(Artifact).order_by(Artifact.filename))).scalars().all()
return templates.TemplateResponse(request=request, name="scheduled_posts.html", context={"username": username, "artifacts": artifacts})
@app.get("/api/scheduled_posts/list")
async def api_list_scheduled_posts(username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
posts = (await session.execute(select(ScheduledPost).order_by(ScheduledPost.scheduled_time.desc()))).scalars().all()
posts_data = []
for post in posts:
posts_data.append({
"id": post.id, "text": post.text, "topic_name": post.topic_name,
"topic_emoji": post.get_topic_emoji(), "recipients": post.recipients,
"scheduled_time": fmt_local(post.scheduled_time), "status": post.status,
"status_emoji": post.get_status_emoji(), "has_photo": bool(post.photo_file_id or post.document_file_id)
})
return JSONResponse({"success": True, "posts": posts_data})
@app.post("/api/scheduled_posts/create")
async def api_create_scheduled_post(
text: str = Form(...),
photo: UploadFile = File(None),
document: UploadFile = File(None),
artifact_photo_id: str = Form(None),
artifact_document_id: str = Form(None),
topic_name: str = Form("general"),
recipients: str = Form("chat_only"),
scheduled_time: str = Form(...),
username: str = Depends(get_current_admin)
):
try:
photo_file_id = artifact_photo_id
document_file_id = artifact_document_id
doc_name = None
if photo and photo.filename and not photo_file_id:
content = await photo.read()
photo_file_id, _ = await upload_to_tg(content, photo.filename, photo.content_type)
if document and document.filename and not document_file_id:
content = await document.read()
document_file_id, _ = await upload_to_tg(content, document.filename, document.content_type)
doc_name = document.filename
# Парсим время (оно приходит в локальном времени)
dt_local = datetime.fromisoformat(scheduled_time)
dt_utc = dt_local - timedelta(hours=4) # Ульяновск -> UTC
async with AsyncSessionLocal() as session:
new_post = ScheduledPost(
text=text, photo_file_id=photo_file_id,
document_file_id=document_file_id, document_name=doc_name,
topic_name=topic_name, recipients=recipients,
scheduled_time=dt_utc, status='pending'
)
session.add(new_post)
await session.commit()
return JSONResponse({"success": True, "message": f"Пост запланирован на {fmt_local(dt_utc)}"})
except Exception as e:
return JSONResponse(status_code=500, content={"success": False, "error": str(e)})
# ============================================================================
# ДАЙДЖЕСТЫ (V3.1)
# ============================================================================
@app.get("/digests", response_class=HTMLResponse)
async def digests_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
stmt_pending = select(Digest).where(Digest.status.in_(['draft', 'pending_approval'])).order_by(Digest.created_at.desc())
pending = (await session.execute(stmt_pending)).scalars().all()
stmt_archive = select(Digest).where(Digest.status.in_(['approved', 'sent', 'rejected'])).order_by(Digest.created_at.desc()).limit(20)
archive = (await session.execute(stmt_archive)).scalars().all()
return templates.TemplateResponse(request=request, name="digests.html", context={"username": username, "pending_digests": pending, "archive_digests": archive})
@app.post("/api/digest/generate")
async def api_generate_digest(username: str = Depends(get_current_admin)):
from services.digest_service import DigestService
try:
stats = await DigestService.collect_weekly_stats()
digest_text = DigestService.format_digest_text(stats)
now = datetime.now()
week_num = now.isocalendar()[1]
async with AsyncSessionLocal() as session:
new_digest = Digest(year=now.year, week_number=week_num, period_start=now - timedelta(days=7), period_end=now, digest_text=digest_text, status='draft')
session.add(new_digest)
await session.commit()
return JSONResponse({"success": True, "message": f"Дайджест за {week_num} неделю успешно сгенерирован (черновик)"})
except Exception as e:
return JSONResponse(status_code=500, content={"success": False, "error": str(e)})
# ============================================================================
# ФАЙЛЫ / АРТЕФАКТЫ (Artifact Manager)
# ============================================================================
@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:
artifacts = (await session.execute(select(Artifact).order_by(Artifact.created_at.desc()))).scalars().all()
files_data = []
for a in artifacts:
files_data.append({"id": a.id, "filename": a.filename, "file_type": a.file_type.upper(), "file_id": a.file_id, "created_at": fmt_local(a.created_at), "is_artifact": True})
export_path = BASE_DIR / "exports"
if export_path.exists():
for f in export_path.glob("*.json"):
files_data.append({"filename": f.name, "file_type": "EXPORT", "size": f"{round(f.stat().st_size / 1024, 1)} KB", "path": f"/exports/{f.name}", "is_artifact": False})
log_path = BASE_DIR / "logs"
if log_path.exists():
for f in log_path.glob("*.log"):
files_data.append({"filename": f.name, "file_type": "LOG", "size": f"{round(f.stat().st_size / 1024, 1)} KB", "path": f"/api/files/download/logs/{f.name}", "is_artifact": False})
return JSONResponse({"success": True, "files": files_data})
@app.post("/api/files/upload")
async def api_upload_artifact(file: UploadFile = File(...), username: str = Depends(get_current_admin)):
try:
content = await file.read()
file_id, file_type = await upload_to_tg(content, file.filename, file.content_type)
async with AsyncSessionLocal() as db_session:
new_artifact = Artifact(filename=file.filename, file_type=file_type, file_id=file_id)
db_session.add(new_artifact)
await db_session.commit()
return JSONResponse({"success": True, "file_id": file_id, "filename": file.filename})
except Exception as e:
return JSONResponse(status_code=500, content={"success": False, "error": str(e)})
@app.get("/api/files/download/logs/{filename}")
async def download_log(filename: str, username: str = Depends(get_current_admin)):
filepath = BASE_DIR / "logs" / filename
if not filepath.exists(): raise HTTPException(status_code=404, detail="File not found")
return FileResponse(filepath, filename=filename)
# ============================================================================
# ЭКСПОРТ (JSON + GITEA)
# ============================================================================
@app.get("/export", response_class=HTMLResponse)
async def export_page(request: Request, username: str = Depends(get_current_admin)):
from services.chat_exporter import ChatExporter
async with AsyncSessionLocal() as session:
exporter = ChatExporter(session)
stats = await exporter.get_stats()
return templates.TemplateResponse(request=request, name="export.html", context={"username": username, "export_stats": stats, "fmt_local": fmt_local})
@app.get("/api/export/json")
async def api_export_json(username: str = Depends(get_current_admin)):
from services.chat_exporter import ChatExporter
try:
async with AsyncSessionLocal() as session:
exporter = ChatExporter(session)
result = await exporter.run_full_export()
return JSONResponse(result)
except Exception as e:
return JSONResponse(status_code=500, content={"success": False, "error": str(e)})
@app.get("/exports/{filename}")
async def download_export(filename: str, username: str = Depends(get_current_admin)):
filepath = BASE_DIR / "exports" / filename
if not filepath.exists(): raise HTTPException(status_code=404, detail="File not found")
return FileResponse(filepath, filename=filename, media_type='application/json')
# ============================================================================
# ПОЧТА (EMAIL AUDITOR)
# ============================================================================
@app.get("/emails", response_class=HTMLResponse)
async def emails_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
audit_data = (await session.execute(select(EmailAudit).order_by(EmailAudit.last_email_date.desc()))).scalars().all()
last_log = (await session.execute(select(EmailExportLog).order_by(EmailExportLog.start_time.desc()).limit(1))).scalar_one_or_none()
latest_emails = []
all_folder = Path(config.MUTT_DIR) / "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:
# Try to parse date and subject
match = re.match(r'(\d{4}-\d{2}-\d{2})_([^_]+)_(.*)', f.stem)
if match:
latest_emails.append({"date": match.group(1), "domain": match.group(2), "subject": match.group(3)})
else:
latest_emails.append({"date": "-", "domain": "other", "subject": f.stem})
master_pdf_path = Path(config.MUTT_DIR) / "ALL_EMAILS_CONSOLIDATED.pdf"
parts = list(Path(config.MUTT_DIR).glob("ALL_EMAILS_CONSOLIDATED_part*.pdf"))
has_master_pdf = master_pdf_path.exists() or bool(parts)
pdf_size_mb = 0.0
last_update = None
if master_pdf_path.exists():
pdf_size_mb = round(master_pdf_path.stat().st_size / (1024 * 1024), 2)
last_update = datetime.fromtimestamp(master_pdf_path.stat().st_mtime, tz=timezone.utc)
elif parts:
pdf_size_mb = round(sum(p.stat().st_size for p in parts) / (1024 * 1024), 2)
last_update = datetime.fromtimestamp(max(p.stat().st_mtime for p in parts), tz=timezone.utc)
# Drive sync status
sync_time_file = Path(config.MUTT_DIR) / "last_ai_sync.txt"
last_sync_time = None
if sync_time_file.exists():
try:
time_str = sync_time_file.read_text().strip()
last_sync_time = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
except:
last_sync_time = datetime.fromtimestamp(sync_time_file.stat().st_mtime, tz=timezone.utc)
# Global stats calculation
total_inbound = sum(d.inbound_count for d in audit_data)
total_outbound = sum(d.outbound_count for d in audit_data)
total_emails = total_inbound + total_outbound
total_attachments = 0
export_log_path = Path(config.MUTT_DIR) / "export.log"
if export_log_path.exists():
try:
content = export_log_path.read_text(errors='ignore')
matches = re.findall(r"Has (\d+) attachment", content)
total_attachments = sum(int(m) for m in matches)
except Exception as e:
pass
last_email_date = None
if audit_data:
dates = [d.last_email_date for d in audit_data if d.last_email_date]
if dates:
last_email_date = max(dates)
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": master_pdf_path.exists(), "last_update": last_update,
"last_sync_time": last_sync_time, "fmt_local": fmt_local,
"total_inbound": total_inbound, "total_outbound": total_outbound,
"total_emails": total_emails, "total_attachments": total_attachments,
"pdf_size_mb": pdf_size_mb, "last_email_date": last_email_date
})
@app.post("/api/emails/run_full_sync")
async def api_emails_run_full_sync(username: str = Depends(get_current_admin)):
master_script = str(Path(config.MUTT_DIR) / "master_ai_sync.py")
python_path = sys.executable
if not os.path.exists(master_script):
return JSONResponse(status_code=404, content={"message": "Master sync script not found"})
try:
# Run master sync script in background with logging
log_path = Path(config.MUTT_DIR) / "master_sync.log"
log_file = open(log_path, "a")
subprocess.Popen([python_path, master_script], stdout=log_file, stderr=log_file, start_new_session=True)
return JSONResponse(content={"message": "Полная синхронизация (Сбор -> PDF -> G-Drive) запущена в фоне. Следите за уведомлениями в Telegram."})
except Exception as e:
return JSONResponse(status_code=500, content={"message": str(e)})
@app.post("/api/emails/sync_ai")
async def api_emails_sync_ai(username: str = Depends(get_current_admin)):
sync_script = "/home/matrixhasyou/domovoy_drive_sync.py" if os.path.exists("/home/matrixhasyou/domovoy_drive_sync.py") else "/app/domovoy_drive_sync.py"
python_path = sys.executable
master_pdf = str(Path(config.MUTT_DIR) / "TOTAL_ARCHIVE_2025-2026.pdf")
if not os.path.exists(sync_script):
return JSONResponse(status_code=404, content={"message": "Sync script not found"})
try:
# Run sync script in background with logging
log_path = Path(config.MUTT_DIR) / "master_sync.log"
log_file = open(log_path, "a")
subprocess.Popen([python_path, sync_script, master_pdf], stdout=log_file, stderr=log_file, start_new_session=True)
return JSONResponse(content={"message": "Синхронизация с Google Drive запущена в фоне. Это займет около минуты."})
except Exception as e:
return JSONResponse(status_code=500, content={"message": str(e)})
@app.get("/api/emails/pipeline_status")
async def api_emails_pipeline_status(username: str = Depends(get_current_admin)):
import time
status_file = Path(config.MUTT_DIR) / "pipeline_status.json"
master_pdf_path = Path(config.MUTT_DIR) / "ALL_EMAILS_CONSOLIDATED.pdf"
parts = list(Path(config.MUTT_DIR).glob("ALL_EMAILS_CONSOLIDATED_part*.pdf"))
# Calculate fallback values
pdf_size_mb = 0.0
if master_pdf_path.exists():
pdf_size_mb = round(master_pdf_path.stat().st_size / (1024 * 1024), 2)
elif parts:
pdf_size_mb = round(sum(p.stat().st_size for p in parts) / (1024 * 1024), 2)
total_emails = 0
total_inbound = 0
total_outbound = 0
async with AsyncSessionLocal() as session:
audit_data = (await session.execute(select(EmailAudit))).scalars().all()
total_inbound = sum(d.inbound_count for d in audit_data)
total_outbound = sum(d.outbound_count for d in audit_data)
total_emails = total_inbound + total_outbound
total_attachments = 0
export_log_path = Path(config.MUTT_DIR) / "export.log"
if export_log_path.exists():
try:
content = export_log_path.read_text(errors='ignore')
matches = re.findall(r"Has (\d+) attachment", content)
total_attachments = sum(int(m) for m in matches)
except Exception as e:
pass
sync_time_file = Path(config.MUTT_DIR) / "last_ai_sync.txt"
last_sync_time = "-"
if sync_time_file.exists():
try:
last_sync_time = sync_time_file.read_text().strip()
except:
pass
data = {
"status": "idle",
"step": "Неактивен",
"updated_at": "-",
"new_emails": 0,
"total_emails": total_emails,
"total_inbound": total_inbound,
"total_outbound": total_outbound,
"total_attachments": total_attachments,
"pdf_size_mb": pdf_size_mb,
"last_sync_time": last_sync_time
}
if status_file.exists():
try:
with open(status_file) as f:
file_data = json.load(f)
data.update(file_data)
mtime = status_file.stat().st_mtime
if data.get("status") == "running" and (time.time() - mtime > 600):
data["status"] = "stuck"
data["step"] = "Процесс завис или был остановлен"
except Exception as e:
data["status"] = "error"
data["message"] = str(e)
return JSONResponse(content=data)
@app.get("/infra", response_class=HTMLResponse)
async def get_infra_page(request: Request, username: str = Depends(get_current_admin)):
async with AsyncSessionLocal() as session:
stmt = select(Speedtest).order_by(Speedtest.timestamp.desc()).limit(10)
rows = (await session.execute(stmt)).scalars().all()
speedtests = []
for r in rows:
speedtests.append({
"timestamp": fmt_local(r.timestamp),
"download": r.download,
"upload": r.upload,
"ping": r.ping,
"server_name": r.server_name
})
return templates.TemplateResponse(request=request, name="infra.html", context={"username": username, "speedtests": speedtests})
@app.get("/sudrf")
async def sudrf_page(request: Request, username: str = Depends(get_current_admin)):
return templates.TemplateResponse(request=request, name="sudrf.html", context={"username": username})
@app.post("/api/infra/run_speedtest")
async def api_run_speedtest(username: str = Depends(get_current_admin)):
script_path = "/app/infra/speedtest_monitor.py" if os.path.exists("/app/infra/speedtest_monitor.py") else "/home/matrixhasyou/infra/speedtest_monitor.py"
try:
subprocess.Popen([sys.executable if os.path.exists("/app/infra/speedtest_monitor.py") else "python3", script_path], start_new_session=True)
return JSONResponse(content={"message": "Тест скорости запущен в фоне. Результаты появятся через минуту."})
except Exception as e:
return JSONResponse(status_code=500, content={"message": str(e)})
@app.post("/api/infra/run_gis_harvest")
async def api_run_gis_harvest(username: str = Depends(get_current_admin)):
try:
# 1. Запуск Жнеца в Docker
subprocess.run(["docker", "exec", "lkm37-gis-scraper", "pkill", "-f", "harvester_v5.py"])
subprocess.Popen(["docker", "exec", "-d", "lkm37-gis-scraper", "python3", "/app/services/gis_harvester/harvester_v5.py"])
# 2. Запуск PDF-конвейера
pipeline_script = str(Path(__file__).resolve().parent.parent / "services" / "gis_pdf_pipeline" / "pipeline_main.py")
venv_python = sys.executable
subprocess.Popen([venv_python, pipeline_script], start_new_session=True)
return JSONResponse(content={"message": "Процесс ГИС ЖКХ запущен (Жатва + PDF). Следите за Telegram."})
except Exception as e:
return JSONResponse(status_code=500, content={"message": str(e)})
@app.get("/api/infra/gis_status")
async def api_gis_status(username: str = Depends(get_current_admin)):
try:
# Проверка процесса
proc = subprocess.run(["docker", "exec", "lkm37-gis-scraper", "ps", "aux"], capture_output=True, text=True)
is_alive = "harvester_v5.py" in proc.stdout
# Чтение лога
log_proc = subprocess.run(["docker", "exec", "lkm37-gis-scraper", "tail", "-n", "30", "/app/services/gis_harvester/harvester.log"], capture_output=True, text=True)
return JSONResponse(content={
"alive": is_alive,
"log": log_proc.stdout or "Log is empty or file not found."
})
except Exception as e:
return JSONResponse(status_code=500, content={"message": str(e)})
@app.post("/api/sudrf/run")
async def api_run_sudrf_parser(username: str = Depends(get_current_admin)):
try:
subprocess.run(["docker", "exec", "lkm37-gis-scraper", "pkill", "-f", "sudrf_reaper.py"])
subprocess.Popen(["docker", "exec", "-d", "lkm37-gis-scraper", "python3", "/app/sudrf_reaper.py"])
return JSONResponse(content={"message": "Сессия ГАС Правосудие успешно запущена. Перейдите по ссылке VNC и авторизуйтесь."})
except Exception as e:
return JSONResponse(status_code=500, content={"message": str(e)})
@app.get("/api/sudrf/status")
async def api_sudrf_status(username: str = Depends(get_current_admin)):
try:
proc = subprocess.run(["docker", "exec", "lkm37-gis-scraper", "ps", "aux"], capture_output=True, text=True)
is_alive = "sudrf_reaper.py" in proc.stdout
log_proc = subprocess.run(["docker", "exec", "lkm37-gis-scraper", "tail", "-n", "30", "/app/logs/sudrf.log"], capture_output=True, text=True)
return JSONResponse(content={
"alive": is_alive,
"log": log_proc.stdout or "Log is empty or file not found."
})
except Exception as e:
return JSONResponse(status_code=500, content={"message": str(e)})
@app.get("/legal", response_class=HTMLResponse)
async def legal_page(request: Request, username: str = Depends(get_current_admin)):
import fitz
legal_dir = Path("/home/matrixhasyou/mutt/legal_sources")
volumes_dir = Path("/home/matrixhasyou/mutt/mega_volumes")
podcasts_dir = Path("/home/matrixhasyou/swarm-services/domovoy-bot/podcasts")
laws = []
if legal_dir.exists():
for f in sorted(legal_dir.glob("*.pdf")):
if f.is_file() and not f.name.startswith("."):
size_mb = f.stat().st_size / (1024 * 1024)
size_formatted = f"{size_mb:.2f} МБ" if size_mb >= 1 else f"{f.stat().st_size / 1024:.1f} КБ"
try:
doc = fitz.open(f)
pages = len(doc)
except Exception:
pages = 0
title = f.stem.replace("_", " ").replace("Полный", "").strip()
laws.append({
"filename": f.name,
"title": title,
"pages": pages,
"size_formatted": size_formatted
})
mega_volumes = []
if volumes_dir.exists():
for f in sorted(volumes_dir.glob("*.pdf")):
if f.is_file() and not f.name.startswith("."):
size_mb = f.stat().st_size / (1024 * 1024)
mega_volumes.append({
"name": f.name,
"size_formatted": f"{size_mb:.1f} МБ"
})
latest_podcast = None
if podcasts_dir.exists():
podcast_files = sorted(podcasts_dir.glob("*.*"), key=lambda x: x.stat().st_mtime, reverse=True)
if podcast_files:
p = podcast_files[0]
latest_podcast = {
"name": p.name,
"size": f"{p.stat().st_size / (1024 * 1024):.1f} МБ",
"date": datetime.fromtimestamp(p.stat().st_mtime).strftime("%d.%m.%Y %H:%M")
}
return templates.TemplateResponse(
request=request,
name="legal.html",
context={
"username": username,
"laws": laws,
"mega_volumes": mega_volumes,
"latest_podcast": latest_podcast
}
)
@app.get("/api/legal/download/{filename}")
async def api_download_law(filename: str, username: str = Depends(get_current_admin)):
fpath = Path("/home/matrixhasyou/mutt/legal_sources") / filename
if not fpath.exists() or not fpath.is_file():
raise HTTPException(status_code=404, detail="Файл закона не найден")
return FileResponse(fpath, filename=filename, media_type="application/pdf")
@app.get("/api/legal/download_volume/{filename}")
async def api_download_volume(filename: str, username: str = Depends(get_current_admin)):
fpath = Path("/home/matrixhasyou/mutt/mega_volumes") / filename
if not fpath.exists() or not fpath.is_file():
raise HTTPException(status_code=404, detail="Файл тома переписки не найден")
return FileResponse(fpath, filename=filename, media_type="application/pdf")
@app.get("/api/legal/podcasts/{filename}")
async def api_download_podcast(filename: str, username: str = Depends(get_current_admin)):
fpath = Path("/home/matrixhasyou/swarm-services/domovoy-bot/podcasts") / filename
if not fpath.exists() or not fpath.is_file():
raise HTTPException(status_code=404, detail="Подкаст не найден")
return FileResponse(fpath, filename=filename)
@app.post("/api/legal/ask_ai")
async def api_legal_ask_ai(request: Request, username: str = Depends(get_current_admin)):
try:
body = await request.json()
query = body.get("query", "").strip()
deep = body.get("deep", False)
if not query:
return JSONResponse(content={"success": False, "error": "Текст вопроса пуст"})
from services.ai_legal_engine import AILegalEngine
engine = AILegalEngine()
role = "lawyer" if deep else "resident"
res = await engine.query(prompt=query, role=role)
return JSONResponse(content=res)
except Exception as e:
return JSONResponse(content={"success": False, "error": str(e)})
@app.post("/api/legal/notebooklm_sync")
async def api_legal_notebooklm_sync(username: str = Depends(get_current_admin)):
try:
from services.notebooklm_podcast_generator import NotebookLMPodcastGenerator
generator = NotebookLMPodcastGenerator()
res = generator.sync_all_sources()
if res.get("success"):
return JSONResponse(content={"message": f"Синхронизация успешна! Загружено файлов: {len(res.get('uploaded_sources', []))}"})
return JSONResponse(content={"message": f"Статус: {res.get('error', 'Сессия требует настройки')}"})
except Exception as e:
return JSONResponse(status_code=500, content={"message": str(e)})
@app.post("/api/legal/generate_podcast")
async def api_legal_generate_podcast(username: str = Depends(get_current_admin)):
try:
from services.notebooklm_podcast_generator import NotebookLMPodcastGenerator
generator = NotebookLMPodcastGenerator()
if generator.is_authenticated():
res = generator.generate_audio_overview()
return JSONResponse(content={"message": "Генерация Audio Overview в NotebookLM запущена!"})
else:
# Fallback генерация сценария через Gemini 2.5 Pro
res = generator.generate_fallback_dialogue_script(
"Анализ судебных споров дома с УК и ГЖИ по отоплению, перерасчетам и отпискам прокуратуры."
)
return JSONResponse(content={"message": f"Сгенерирован сценарий подкаста через Gemini 2.5 Pro: {len(res.get('script', ''))} симв."})
except Exception as e:
return JSONResponse(status_code=500, content={"message": str(e)})
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)