mirror of
https://github.com/Alexander-D-Karpov/akarpov
synced 2026-01-11 17:30:46 +03:00
72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
import os
|
|
from typing import Annotated
|
|
|
|
import django
|
|
from fastapi import FastAPI, Depends, Header, Cookie
|
|
from fastapi.responses import RedirectResponse, HTMLResponse
|
|
from sqlalchemy.orm import Session
|
|
from starlette.requests import Request
|
|
|
|
from redirect.db.curd import get_link_by_slug, LinkNotFoundException
|
|
from redirect.db.dependency import get_db
|
|
from redirect.settings import settings
|
|
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.local")
|
|
django.setup()
|
|
app = FastAPI()
|
|
|
|
|
|
from akarpov.tools.shortener.tasks import save_view_meta # noqa: This has to be imported strictly AFTER django setup
|
|
|
|
|
|
@app.exception_handler(LinkNotFoundException)
|
|
async def unicorn_exception_handler(request: Request, exc: LinkNotFoundException):
|
|
return HTMLResponse(
|
|
status_code=404,
|
|
# language=HTML
|
|
content="""
|
|
<html lang="en">
|
|
<head>
|
|
<title>Unknown Link</title>
|
|
<style>
|
|
h1 {
|
|
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", ui-system, sans-serif;
|
|
font-weight: bold;
|
|
font-size: medium;
|
|
text-align: center;
|
|
}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>Such link doesn't exist or has been revoked</h1>
|
|
</body>
|
|
</html>
|
|
""",
|
|
)
|
|
|
|
|
|
@app.get("/{slug}")
|
|
def redirect(
|
|
slug: str,
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
user_agent: Annotated[str | None, Header()] = None,
|
|
sessionid: Annotated[str | None, Cookie()] = None
|
|
) -> RedirectResponse:
|
|
"""Main route that redirects to a page based on the slug."""
|
|
if '+' in slug:
|
|
return RedirectResponse(url=f'/tools/shortener/p/{slug.replace("+", "")}')
|
|
|
|
link_id, link_target = get_link_by_slug(db, slug)
|
|
|
|
save_view_meta.apply_async(
|
|
kwargs={
|
|
"pk": link_id,
|
|
"ip": request.client.host,
|
|
"user_agent": user_agent,
|
|
"token": sessionid,
|
|
},
|
|
)
|
|
|
|
return RedirectResponse(url=(settings.relative_base + link_target) if link_target.startswith('/') else link_target)
|