911 lines
46 KiB
Python
911 lines
46 KiB
Python
"""
|
||
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
|
||
|
||
@app.post("/api/emails/sync_ai")
|
||
async def api_sync_emails_to_ai(username: str = Depends(get_current_admin)):
|
||
sync_script = "/app/domovoy_drive_sync.py" if os.path.exists("/app/domovoy_drive_sync.py") else str(Path(config.MUTT_DIR).parent / "domovoy_drive_sync.py")
|
||
python_path = sys.executable
|
||
|
||
try:
|
||
# Запуск синхронизации
|
||
process = subprocess.run(
|
||
[python_path, sync_script],
|
||
capture_output=True, text=True, timeout=300
|
||
)
|
||
|
||
if process.returncode == 0:
|
||
# Уведомляем админа в ТГ
|
||
msg = "🤖 <b>AI Синхронизация Успешна!</b>\n\n📄 Архив почты обновлен на Google Drive.\n👉 Не забудь нажать <b>'Sync'</b> в NotebookLM!"
|
||
|
||
# Отправка через API Telegram (упрощенно)
|
||
async with aiohttp.ClientSession() as session:
|
||
url = f"https://api.telegram.org/bot{config.BOT_TOKEN}/sendMessage"
|
||
params = {
|
||
"chat_id": config.ADMIN_USER_ID,
|
||
"text": msg,
|
||
"parse_mode": "HTML"
|
||
}
|
||
# Используем прокси если нужно
|
||
proxy = f"http://{config.PROXY_HOST}:{config.PROXY_PORT}" if config.USE_PROXY else None
|
||
await session.get(url, params=params, proxy=proxy)
|
||
|
||
return {"success": True, "message": "Синхронизация завершена, уведомление отправлено."}
|
||
else:
|
||
return {"success": False, "error": process.stderr or process.stdout}
|
||
|
||
except Exception as e:
|
||
return JSONResponse(status_code=500, content={"success": False, "error": str(e)})
|
||
|
||
# ============================================================================
|
||
# ГЛАВНАЯ ПАНЕЛЬ (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"✅ <b>Ваша заявка на верификацию одобрена!</b>\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": "✅ <b>Верификация подтверждена!</b>\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"
|
||
last_update = datetime.fromtimestamp(master_pdf_path.stat().st_mtime, tz=timezone.utc) if master_pdf_path.exists() else None
|
||
|
||
# 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
|
||
|
||
pdf_size_mb = 0.0
|
||
if master_pdf_path.exists():
|
||
pdf_size_mb = round(master_pdf_path.stat().st_size / (1024 * 1024), 2)
|
||
|
||
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
|
||
subprocess.Popen([python_path, master_script], 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 = "/app/domovoy_drive_sync.py" if os.path.exists("/app/domovoy_drive_sync.py") else str(Path(config.MUTT_DIR).parent / "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
|
||
subprocess.Popen([python_path, sync_script, master_pdf], 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"
|
||
|
||
# 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)
|
||
|
||
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.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/infra/run_sudrf_parser")
|
||
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/infra/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)})
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
uvicorn.run(app, host="0.0.0.0", port=8000)
|