296 lines
9.2 KiB
Python
296 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, Depends, Form, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from auth import check_admin_credentials, get_current_admin, require_admin_page
|
|
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)
|
|
return templates.TemplateResponse(
|
|
"admin/dashboard.html",
|
|
{
|
|
"request": request,
|
|
"brand": settings.brand_name,
|
|
"links": links,
|
|
"site_base": site_base,
|
|
"flash": request.query_params.get("flash"),
|
|
"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),
|
|
},
|
|
)
|