183 lines
6.4 KiB
Python
183 lines
6.4 KiB
Python
import os
|
|
import sys
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from fastapi import FastAPI, HTTPException, status
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from fastapi import Request
|
|
from fastapi.responses import RedirectResponse
|
|
from pydantic import BaseModel
|
|
|
|
from botdiscord.config import load_config, get_web_config, get_libretranslate_url, get_db_type
|
|
|
|
app = FastAPI(title="Panel de Configuración - Bots de Traducción")
|
|
|
|
SCRIPT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
templates = Jinja2Templates(directory=os.path.join(SCRIPT_DIR, "panel", "templates"))
|
|
|
|
class LoginForm(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
def get_config():
|
|
cfg = load_config()
|
|
return {
|
|
"discord": {"token": cfg.get("discord", {}).get("token", "")},
|
|
"telegram": {"token": cfg.get("telegram", {}).get("token", "")},
|
|
"libretranslate": {"url": cfg.get("libretranslate", {}).get("url", "")},
|
|
"web": cfg.get("web", {}),
|
|
"database": cfg.get("database", {}),
|
|
"languages": cfg.get("languages", {})
|
|
}
|
|
|
|
def verify_admin(username: str, password: str) -> bool:
|
|
web_config = get_web_config()
|
|
return (username == web_config.get("admin_username", "") and
|
|
password == web_config.get("admin_password", ""))
|
|
|
|
@app.get("/")
|
|
async def root(request: Request):
|
|
return templates.TemplateResponse("index.html", {"request": request})
|
|
|
|
@app.get("/login")
|
|
async def login_page(request: Request):
|
|
return templates.TemplateResponse("login.html", {"request": request})
|
|
|
|
@app.post("/login")
|
|
async def login(request: Request):
|
|
form = await request.form()
|
|
username = form.get("username", "")
|
|
password = form.get("password", "")
|
|
|
|
if verify_admin(username, password):
|
|
response = RedirectResponse(url="/dashboard", status_code=status.HTTP_303_SEE_OTHER)
|
|
response.set_cookie(key="auth", value="ok", httponly=True)
|
|
return response
|
|
|
|
return templates.TemplateResponse("login.html", {
|
|
"request": request,
|
|
"error": "Credenciales incorrectas"
|
|
})
|
|
|
|
@app.get("/dashboard")
|
|
async def dashboard(request: Request):
|
|
if request.cookies.get("auth") != "ok":
|
|
return RedirectResponse(url="/login")
|
|
|
|
config = get_config()
|
|
return templates.TemplateResponse("dashboard.html", {
|
|
"request": request,
|
|
"config": config
|
|
})
|
|
|
|
@app.get("/config")
|
|
async def config_page(request: Request):
|
|
if request.cookies.get("auth") != "ok":
|
|
return RedirectResponse(url="/login")
|
|
|
|
config = get_config()
|
|
return templates.TemplateResponse("config.html", {
|
|
"request": request,
|
|
"config": config
|
|
})
|
|
|
|
@app.get("/languages")
|
|
async def languages_page(request: Request):
|
|
if request.cookies.get("auth") != "ok":
|
|
return RedirectResponse(url="/login")
|
|
|
|
from botdiscord.database import get_available_languages, get_bot_languages
|
|
|
|
available = get_available_languages()
|
|
discord_langs = get_bot_languages("discord")
|
|
telegram_langs = get_bot_languages("telegram")
|
|
|
|
return templates.TemplateResponse("languages.html", {
|
|
"request": request,
|
|
"available_languages": available,
|
|
"discord_languages": discord_langs,
|
|
"telegram_languages": telegram_langs,
|
|
"libretranslate_url": get_libretranslate_url()
|
|
})
|
|
|
|
@app.post("/languages/sync")
|
|
async def sync_languages(request: Request):
|
|
if request.cookies.get("auth") != "ok":
|
|
raise HTTPException(status_code=401)
|
|
|
|
import aiohttp
|
|
url = get_libretranslate_url()
|
|
|
|
if not url:
|
|
return RedirectResponse(url="/languages?error=1", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
base_url = url.rstrip("/translate").rstrip("/translate/").rstrip("/")
|
|
if base_url.endswith("/translate"):
|
|
base_url = base_url[:-10]
|
|
base_url = base_url.rstrip("/")
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(f"{base_url}/languages", timeout=aiohttp.ClientTimeout(total=10)) as resp:
|
|
if resp.status == 200:
|
|
languages = await resp.json()
|
|
from botdiscord.database import get_available_languages, set_available_languages
|
|
|
|
existing = {lang["code"]: lang.get("flag", "") for lang in get_available_languages()}
|
|
|
|
for lang in languages:
|
|
lang["flag"] = existing.get(lang["code"], "")
|
|
|
|
set_available_languages(languages)
|
|
return RedirectResponse(url="/languages?synced=1", status_code=status.HTTP_303_SEE_OTHER)
|
|
else:
|
|
return RedirectResponse(url="/languages?error=1", status_code=status.HTTP_303_SEE_OTHER)
|
|
except Exception as e:
|
|
print(f"Error syncing languages: {e}")
|
|
return RedirectResponse(url="/languages?error=1", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
@app.post("/languages/bot")
|
|
async def update_bot_languages(request: Request):
|
|
if request.cookies.get("auth") != "ok":
|
|
raise HTTPException(status_code=401)
|
|
|
|
form = await request.form()
|
|
bot_type = form.get("bot_type", "discord")
|
|
lang_codes = form.getlist("lang_codes")
|
|
|
|
from botdiscord.database import set_bot_languages
|
|
set_bot_languages(bot_type, lang_codes)
|
|
|
|
return RedirectResponse(url="/languages?success=1", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
@app.post("/languages/flags")
|
|
async def update_language_flags(request: Request):
|
|
if request.cookies.get("auth") != "ok":
|
|
raise HTTPException(status_code=401)
|
|
|
|
from botdiscord.database import get_available_languages, set_available_languages
|
|
|
|
available = get_available_languages()
|
|
form = await request.form()
|
|
|
|
for lang in available:
|
|
flag_key = f"flag_{lang['code']}"
|
|
lang['flag'] = form.get(flag_key, "")
|
|
|
|
set_available_languages(available)
|
|
|
|
return RedirectResponse(url="/languages?success=1", status_code=status.HTTP_303_SEE_OTHER)
|
|
|
|
@app.get("/logout")
|
|
async def logout():
|
|
response = RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER)
|
|
response.delete_cookie("auth")
|
|
return response
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
web_config = get_web_config()
|
|
uvicorn.run(app, host=web_config.get("host", "0.0.0.0"), port=web_config.get("port", 8000))
|