Simplify Evotor connect to manual token entry only

Resolves #2 — removes semi-automatic OAuth flow (Переподключить button,
/evotor/connect and /evotor/link routes) and makes manual token entry
the sole connect option. Adds step-by-step instructions with a direct
link to the app on Evotor marketplace (opens in new tab).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
mguschin
2026-03-10 13:01:12 +03:00
parent 577c5de200
commit 00b74b8aa9
2 changed files with 28 additions and 103 deletions

View File

@@ -1,7 +1,7 @@
import logging
import httpx
from datetime import datetime, timedelta
from datetime import datetime
from fastapi import APIRouter, Request, Depends, HTTPException
from fastapi.responses import RedirectResponse, JSONResponse
from fastapi.templating import Jinja2Templates
@@ -21,9 +21,6 @@ templates = Jinja2Templates(directory="web/templates")
EVOTOR_APP_URL = "https://market.evotor.ru/store/apps/{app_id}"
EVOTOR_STORES_URL = "https://api.evotor.ru/stores"
# Pending connections older than this are ignored during linking
PENDING_LINK_WINDOW_SECONDS = 300
@router.get("")
def evotor_page(
@@ -36,27 +33,16 @@ def evotor_page(
connection = db.query(EvotorConnection).filter(EvotorConnection.user_id == user.id).first()
error = request.query_params.get("error")
app_url = EVOTOR_APP_URL.format(app_id=settings.EVOTOR_APP_ID) if settings.EVOTOR_APP_ID else None
return templates.TemplateResponse("evotor.html", {
"request": request,
"user": user,
"connection": connection,
"error": error,
"app_url": app_url,
})
@router.get("/connect")
def evotor_connect(request: Request, user: User | None = Depends(get_current_user)):
if not user:
return RedirectResponse("/login", 303)
# Record when this user initiated a connect so we can link the incoming webhook
request.session["evotor_connect_user_id"] = user.id
request.session["evotor_connect_at"] = datetime.utcnow().isoformat()
url = EVOTOR_APP_URL.format(app_id=settings.EVOTOR_APP_ID)
return RedirectResponse(url, 302)
class EvotorTokenPayload(BaseModel):
userId: str
token: str
@@ -130,62 +116,6 @@ async def evotor_callback(
return JSONResponse({"status": "ok"})
@router.get("/link")
def evotor_link(
request: Request,
db: Session = Depends(get_db),
user: User | None = Depends(get_current_user),
):
"""
Called when the user returns to our app after authorizing on Evotor.
Links the most recently received unlinked token to this user.
"""
if not user:
return RedirectResponse("/login", 303)
connect_user_id = request.session.pop("evotor_connect_user_id", None)
connect_at_str = request.session.pop("evotor_connect_at", None)
if not connect_user_id or connect_user_id != user.id or not connect_at_str:
return RedirectResponse("/evotor?error=session_expired", 303)
try:
connect_at = datetime.fromisoformat(connect_at_str)
except ValueError:
return RedirectResponse("/evotor?error=session_expired", 303)
cutoff = connect_at - timedelta(seconds=10) # allow slight clock drift
now = datetime.utcnow()
if (now - connect_at).total_seconds() > PENDING_LINK_WINDOW_SECONDS:
return RedirectResponse("/evotor?error=link_timeout", 303)
# Find an unlinked connection received after the user clicked "Connect"
pending = (
db.query(EvotorConnection)
.filter(
EvotorConnection.user_id.is_(None),
EvotorConnection.connected_at >= cutoff,
)
.order_by(EvotorConnection.connected_at.desc())
.first()
)
if not pending:
return RedirectResponse("/evotor?error=token_not_received", 303)
# Detach any existing connection for this user
existing = db.query(EvotorConnection).filter(EvotorConnection.user_id == user.id).first()
if existing:
db.delete(existing)
db.flush()
pending.user_id = user.id
db.commit()
return RedirectResponse("/connections", 303)
@router.post("/token")
async def evotor_token_manual(
request: Request,