retuve.app.routes.ui

  1import base64
  2import os
  3import re
  4import secrets
  5import shutil
  6import time
  7from datetime import datetime, timedelta
  8from hashlib import sha256
  9from tempfile import mkdtemp
 10
 11from cycler import V
 12from fastapi import (
 13    APIRouter,
 14    Depends,
 15    Header,
 16    HTTPException,
 17    Request,
 18    Response,
 19)
 20from fastapi.responses import HTMLResponse, JSONResponse
 21
 22from retuve.app.helpers import API_RESULTS_URL_ACCESS, web_templates
 23from retuve.app.utils import (
 24    API_TOKEN_STORE,
 25    TOKEN_STORE,
 26    generate_token,
 27    get_sorted_dicom_images,
 28    save_dicom_and_get_results,
 29    save_results,
 30    validate_api_token,
 31    validate_auth_token,
 32)
 33from retuve.keyphrases.config import Config
 34from retuve.trak.data import extract_files
 35
 36
 37def basic_auth_dependency(
 38    authorization: str = Header(None), response: Response = None
 39):
 40    if not authorization or not authorization.startswith("Basic "):
 41        raise HTTPException(
 42            status_code=401,
 43            detail="Unauthorized",
 44            headers={"WWW-Authenticate": "Basic"},
 45        )
 46    try:
 47        valid_username = Config.global_config.username
 48        valid_password = Config.global_config.password
 49
 50        auth_token = authorization.split(" ")[1]
 51        decoded_credentials = base64.b64decode(auth_token).decode("utf-8")
 52        username, password = decoded_credentials.split(":")
 53        if username != valid_username or password != valid_password:
 54            raise ValueError("Invalid credentials")
 55
 56        # Generate authentication token
 57        auth_token = generate_token()
 58        api_token = generate_token()
 59
 60        # Store tokens with expiration timestamps
 61        expiration = datetime.utcnow() + timedelta(hours=24)
 62        TOKEN_STORE[auth_token] = {
 63            "username": username,
 64            "expires": expiration,
 65        }
 66        API_TOKEN_STORE[api_token] = {
 67            "username": username,
 68            "expires": expiration,
 69        }
 70
 71        return {"auth_token": auth_token, "api_token": api_token}
 72
 73    except Exception:
 74        raise HTTPException(
 75            status_code=401,
 76            detail="Unauthorized",
 77            headers={"WWW-Authenticate": "Basic"},
 78        )
 79
 80
 81router = APIRouter()
 82
 83
 84@router.get("/", response_class=HTMLResponse)
 85async def web(
 86    request: Request,
 87    tokens: dict = Depends(basic_auth_dependency),
 88):
 89    """
 90    Open the main page of the Retuve Web Interface.
 91    """
 92    keyphrases = [keyphrase for keyphrase in Config.configs.keys()]
 93
 94    response = web_templates.TemplateResponse(
 95        f"index.html",
 96        {
 97            "request": request,
 98            "keyphrases": keyphrases,
 99        },
100    )
101
102    response.set_cookie(
103        "auth_token",
104        tokens["auth_token"],
105        expires=3600,
106        httponly=True,
107        secure=True,
108        samesite="Strict",
109    )
110    response.set_cookie(
111        "api_token",
112        tokens["api_token"],
113        expires=3600,
114        httponly=True,
115        secure=True,
116        samesite="Strict",
117    )
118
119    return response
120
121
122@router.get("/ui/live/", response_class=HTMLResponse)
123async def live_ui(request: Request):
124    """
125    Open the live page of the Retuve Web Interface.
126    """
127    auth_token = request.cookies.get("auth_token")
128    validate_auth_token(auth_token)
129
130    return web_templates.TemplateResponse(
131        f"live.html",
132        {
133            "request": request,
134            "keyphrase": Config.live_config.name,
135            "url": Config.live_config.api.url,
136        },
137    )
138
139
140@router.get("/ui/{keyphrase}", response_class=HTMLResponse)
141async def web_keyphrase(request: Request, keyphrase: str):
142    """
143    Open the page for a specific keyphrase.
144    """
145
146    try:
147        config = Config.get_config(keyphrase)
148    except ValueError:
149        raise HTTPException(
150            status_code=404, detail=f"Keyphrase {keyphrase} not found"
151        )
152    files_data = extract_files(config.api.db_path)
153
154    auth_token = request.cookies.get("auth_token")
155    validate_auth_token(auth_token)
156
157    return web_templates.TemplateResponse(
158        f"main.html",
159        {
160            "request": request,
161            "files": files_data,
162            "hip_mode": config.batch.hip_mode,
163            "keyphrase": keyphrase,
164            "url": config.api.url,
165        },
166    )
167
168
169@router.get("/ui/upload/", response_class=HTMLResponse)
170async def upload_form(request: Request):
171    """
172    Open the upload form for the Retuve Web Interface.
173    """
174
175    auth_token = request.cookies.get("auth_token")
176    validate_auth_token(auth_token)
177
178    keyphrases = [keyphrase for keyphrase in Config.configs.keys()]
179
180    return web_templates.TemplateResponse(
181        f"upload.html",
182        {
183            "request": request,
184            "keyphrases": keyphrases,
185        },
186    )
187
188
189@router.get("/ui/download/{keyphrase}")
190async def download_files(
191    request: Request, keyphrase: str, pattern: str = None
192):
193    """
194    Download files from the Retuve Web Interface.
195    """
196
197    auth_token = request.cookies.get("auth_token")
198    validate_auth_token(auth_token)
199
200    if not pattern:
201        raise HTTPException(status_code=400, detail="Pattern is required")
202
203    config = Config.get_config(keyphrase)
204    savedir = config.api.savedir
205
206    # Create a temporary directory to hold the folders
207    temp_dir = mkdtemp()
208
209    # Find all folders in savedir that match the pattern
210    # Copy each matching folder to the temporary directory
211    for folder in os.listdir(savedir):
212        if re.search(pattern, folder) and os.path.isdir(
213            os.path.join(savedir, folder)
214        ):
215            source_folder = os.path.join(savedir, folder)
216            destination_folder = os.path.join(temp_dir, folder)
217            shutil.copytree(source_folder, destination_folder)
218
219    # Create a zip file of the temporary directory
220    zip_file = os.path.join(savedir, f"{pattern}.zip")
221    shutil.make_archive(zip_file[:-4], "zip", temp_dir)
222
223    zip_file_url = (
224        f"{config.api.url}/{API_RESULTS_URL_ACCESS}/{keyphrase}/{pattern}.zip"
225    )
226
227    # Optionally, remove the temporary directory after creating the zip file
228    shutil.rmtree(temp_dir)
229
230    return web_templates.TemplateResponse(
231        "download.html",
232        {
233            "request": request,
234            "zip_file_url": zip_file_url,
235            "keyphrase": keyphrase,
236        },
237    )