347 lines
11 KiB
Python
347 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
from urllib.parse import quote, unquote
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
|
|
from fastapi.responses import HTMLResponse, RedirectResponse, Response
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from auth import check_admin_credentials, get_current_admin, require_admin_page
|
|
from backup import dumps_backup, export_backup, import_backup, loads_backup
|
|
from config import get_settings
|
|
from database import get_db
|
|
from urlutil import public_base_url
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"])
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
|
|
|
|
|
@router.get("/login", response_class=HTMLResponse)
|
|
async def login_page(request: Request):
|
|
if require_admin_page(request):
|
|
return RedirectResponse("/admin", status_code=303)
|
|
settings = get_settings()
|
|
return templates.TemplateResponse(
|
|
"admin/login.html",
|
|
{"request": request, "brand": settings.brand_name, "error": None},
|
|
)
|
|
|
|
|
|
@router.post("/login")
|
|
async def login_submit(
|
|
request: Request,
|
|
username: str = Form(...),
|
|
password: str = Form(...),
|
|
):
|
|
settings = get_settings()
|
|
if not check_admin_credentials(username, password):
|
|
return templates.TemplateResponse(
|
|
"admin/login.html",
|
|
{
|
|
"request": request,
|
|
"brand": settings.brand_name,
|
|
"error": "Неверный логин или пароль",
|
|
},
|
|
status_code=401,
|
|
)
|
|
request.session["admin"] = username
|
|
return RedirectResponse("/admin", status_code=303)
|
|
|
|
|
|
@router.post("/logout")
|
|
async def logout(request: Request):
|
|
request.session.clear()
|
|
return RedirectResponse("/admin/login", status_code=303)
|
|
|
|
|
|
@router.get("", response_class=HTMLResponse)
|
|
async def admin_dashboard(request: Request):
|
|
if not require_admin_page(request):
|
|
return RedirectResponse("/admin/login", status_code=303)
|
|
|
|
settings = get_settings()
|
|
tab = (request.query_params.get("type") or "extend").strip().lower()
|
|
if tab not in ("extend", "traffic", "all"):
|
|
tab = "extend"
|
|
|
|
try:
|
|
page = max(1, int(request.query_params.get("page") or 1))
|
|
except ValueError:
|
|
page = 1
|
|
per_page = 10
|
|
offset = (page - 1) * per_page
|
|
|
|
db = await get_db()
|
|
try:
|
|
if tab == "all":
|
|
count_sql = "SELECT COUNT(*) AS c FROM guest_links"
|
|
list_sql = """
|
|
SELECT l.*,
|
|
(SELECT COUNT(*) FROM guest_claims c WHERE c.link_id = l.id) AS claims
|
|
FROM guest_links l
|
|
ORDER BY l.id DESC
|
|
LIMIT ? OFFSET ?
|
|
"""
|
|
count_args: tuple = ()
|
|
list_args: tuple = (per_page, offset)
|
|
else:
|
|
count_sql = "SELECT COUNT(*) AS c FROM guest_links WHERE link_type = ?"
|
|
list_sql = """
|
|
SELECT l.*,
|
|
(SELECT COUNT(*) FROM guest_claims c WHERE c.link_id = l.id) AS claims
|
|
FROM guest_links l
|
|
WHERE l.link_type = ?
|
|
ORDER BY l.id DESC
|
|
LIMIT ? OFFSET ?
|
|
"""
|
|
count_args = (tab,)
|
|
list_args = (tab, per_page, offset)
|
|
|
|
cursor = await db.execute(count_sql, count_args)
|
|
total = int((await cursor.fetchone())["c"])
|
|
cursor = await db.execute(list_sql, list_args)
|
|
links = [dict(row) for row in await cursor.fetchall()]
|
|
|
|
cursor = await db.execute(
|
|
"SELECT link_type, COUNT(*) AS c FROM guest_links GROUP BY link_type"
|
|
)
|
|
counts = {"extend": 0, "traffic": 0, "all": 0}
|
|
for row in await cursor.fetchall():
|
|
counts[row["link_type"]] = int(row["c"])
|
|
counts["all"] += int(row["c"])
|
|
finally:
|
|
await db.close()
|
|
|
|
total_pages = max(1, (total + per_page - 1) // per_page)
|
|
if page > total_pages:
|
|
page = total_pages
|
|
|
|
site_base = public_base_url(request)
|
|
flash = request.query_params.get("flash")
|
|
flash_error = request.query_params.get("flash_error")
|
|
if flash_error:
|
|
flash_error = unquote(flash_error)
|
|
return templates.TemplateResponse(
|
|
"admin/dashboard.html",
|
|
{
|
|
"request": request,
|
|
"brand": settings.brand_name,
|
|
"links": links,
|
|
"site_base": site_base,
|
|
"flash": flash,
|
|
"flash_error": flash_error,
|
|
"tab": tab,
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"total": total,
|
|
"total_pages": total_pages,
|
|
"counts": counts,
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/links")
|
|
async def create_link(
|
|
request: Request,
|
|
link_type: str = Form(...),
|
|
title: str = Form(""),
|
|
value: int = Form(...),
|
|
max_uses: int = Form(1),
|
|
duration_days: int = Form(0),
|
|
plan_traffic_gb: int = Form(0),
|
|
expires_at: str = Form(""),
|
|
note: str = Form(""),
|
|
_admin: str = Depends(get_current_admin),
|
|
):
|
|
if link_type not in ("extend", "traffic"):
|
|
raise HTTPException(400, "Неверный тип ссылки")
|
|
if value < 1:
|
|
raise HTTPException(400, "Значение должно быть >= 1")
|
|
if max_uses < 1:
|
|
raise HTTPException(400, "max_uses должен быть >= 1")
|
|
|
|
if link_type == "traffic":
|
|
if duration_days < 1:
|
|
raise HTTPException(
|
|
400, "Для трафика укажите срок действия (дней) — трафик временный"
|
|
)
|
|
plan_traffic_gb = 0
|
|
else:
|
|
duration_days = 0
|
|
if plan_traffic_gb < 1:
|
|
raise HTTPException(
|
|
400,
|
|
"Для продления укажите трафик тарифа (ГБ) — лимит зафиксируется при активации",
|
|
)
|
|
|
|
token = secrets.token_urlsafe(12)
|
|
expires = expires_at.strip() or None
|
|
if not title.strip():
|
|
if link_type == "extend":
|
|
title = f"Продление +{value} дн. / {plan_traffic_gb} ГБ"
|
|
else:
|
|
title = f"Трафик +{value} ГБ на {duration_days} дн."
|
|
|
|
db = await get_db()
|
|
try:
|
|
await db.execute(
|
|
"""
|
|
INSERT INTO guest_links
|
|
(token, link_type, title, value, max_uses, expires_at, note,
|
|
created_at, duration_days, plan_traffic_gb)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
token,
|
|
link_type,
|
|
title.strip(),
|
|
value,
|
|
max_uses,
|
|
expires,
|
|
note.strip(),
|
|
_now_iso(),
|
|
duration_days,
|
|
plan_traffic_gb,
|
|
),
|
|
)
|
|
await db.commit()
|
|
finally:
|
|
await db.close()
|
|
|
|
return RedirectResponse(
|
|
f"/admin?type={link_type}&flash=created", status_code=303
|
|
)
|
|
|
|
|
|
@router.post("/links/{link_id}/toggle")
|
|
async def toggle_link(
|
|
request: Request,
|
|
link_id: int,
|
|
_admin: str = Depends(get_current_admin),
|
|
):
|
|
tab = (request.query_params.get("type") or "extend").strip()
|
|
page = request.query_params.get("page") or "1"
|
|
db = await get_db()
|
|
try:
|
|
await db.execute(
|
|
"UPDATE guest_links SET is_active = CASE WHEN is_active = 1 THEN 0 ELSE 1 END WHERE id = ?",
|
|
(link_id,),
|
|
)
|
|
await db.commit()
|
|
finally:
|
|
await db.close()
|
|
return RedirectResponse(
|
|
f"/admin?type={tab}&page={page}&flash=toggled", status_code=303
|
|
)
|
|
|
|
|
|
@router.post("/links/{link_id}/delete")
|
|
async def delete_link(
|
|
request: Request,
|
|
link_id: int,
|
|
_admin: str = Depends(get_current_admin),
|
|
):
|
|
tab = (request.query_params.get("type") or "extend").strip()
|
|
page = request.query_params.get("page") or "1"
|
|
db = await get_db()
|
|
try:
|
|
await db.execute("DELETE FROM guest_links WHERE id = ?", (link_id,))
|
|
await db.commit()
|
|
finally:
|
|
await db.close()
|
|
return RedirectResponse(
|
|
f"/admin?type={tab}&page={page}&flash=deleted", status_code=303
|
|
)
|
|
|
|
|
|
@router.get("/links/{link_id}", response_class=HTMLResponse)
|
|
async def link_detail(request: Request, link_id: int):
|
|
if not require_admin_page(request):
|
|
return RedirectResponse("/admin/login", status_code=303)
|
|
|
|
settings = get_settings()
|
|
db = await get_db()
|
|
try:
|
|
cursor = await db.execute("SELECT * FROM guest_links WHERE id = ?", (link_id,))
|
|
link = await cursor.fetchone()
|
|
if not link:
|
|
raise HTTPException(404, "Ссылка не найдена")
|
|
cursor = await db.execute(
|
|
"SELECT * FROM guest_claims WHERE link_id = ? ORDER BY id DESC",
|
|
(link_id,),
|
|
)
|
|
claims = [dict(row) for row in await cursor.fetchall()]
|
|
cursor = await db.execute(
|
|
"""
|
|
SELECT * FROM traffic_grants
|
|
WHERE link_id = ?
|
|
ORDER BY id DESC
|
|
""",
|
|
(link_id,),
|
|
)
|
|
grants = [dict(row) for row in await cursor.fetchall()]
|
|
finally:
|
|
await db.close()
|
|
|
|
return templates.TemplateResponse(
|
|
"admin/link_detail.html",
|
|
{
|
|
"request": request,
|
|
"brand": settings.brand_name,
|
|
"link": dict(link),
|
|
"claims": claims,
|
|
"grants": grants,
|
|
"site_base": public_base_url(request),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/backup/export")
|
|
async def backup_export(_admin: str = Depends(get_current_admin)):
|
|
payload = await export_backup()
|
|
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
|
filename = f"remna-backup-{stamp}.json"
|
|
return Response(
|
|
content=dumps_backup(payload),
|
|
media_type="application/json; charset=utf-8",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
@router.post("/backup/import")
|
|
async def backup_import(
|
|
file: UploadFile = File(...),
|
|
_admin: str = Depends(get_current_admin),
|
|
):
|
|
raw = await file.read()
|
|
if not raw:
|
|
return RedirectResponse(
|
|
"/admin?type=all&flash_error=" + quote("пустой файл"),
|
|
status_code=303,
|
|
)
|
|
if len(raw) > 20 * 1024 * 1024:
|
|
return RedirectResponse(
|
|
"/admin?type=all&flash_error=" + quote("файл слишком большой (макс 20 МБ)"),
|
|
status_code=303,
|
|
)
|
|
try:
|
|
data = loads_backup(raw)
|
|
await import_backup(data)
|
|
except ValueError as exc:
|
|
return RedirectResponse(
|
|
"/admin?type=all&flash_error=" + quote(str(exc)[:200]),
|
|
status_code=303,
|
|
)
|
|
except Exception as exc:
|
|
return RedirectResponse(
|
|
"/admin?type=all&flash_error=" + quote(str(exc)[:200]),
|
|
status_code=303,
|
|
)
|
|
return RedirectResponse("/admin?type=all&flash=imported", status_code=303)
|