227 lines
9.0 KiB
Python
227 lines
9.0 KiB
Python
"""Loopback-only web UI, served by the same Python installation as Copienator."""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
from io import BytesIO
|
||
from pathlib import Path
|
||
import secrets
|
||
import socket
|
||
import threading
|
||
import time
|
||
|
||
from fastapi import FastAPI, Request, HTTPException
|
||
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response
|
||
from fastapi.staticfiles import StaticFiles
|
||
from pydantic import BaseModel, StrictStr, Field
|
||
import uvicorn
|
||
|
||
from copienator import CliError
|
||
from copienator.annotation_modifications import baseline, body_image, edited_bodies, read_instructions, validate_instructions
|
||
from copienator.evaluation_lock import EvaluationLock
|
||
from copienator.finalisation import Finalisation
|
||
from copienator.return_identity import original_copy_path
|
||
|
||
|
||
class Change(BaseModel):
|
||
revision: StrictStr
|
||
kind: StrictStr
|
||
value: StrictStr | list[StrictStr]
|
||
copy_id: StrictStr = Field(default='', alias='copy')
|
||
label: StrictStr = ''
|
||
|
||
|
||
class Regenerate(BaseModel):
|
||
revision: str
|
||
apply_draft: bool = False
|
||
incremental: bool = True
|
||
|
||
|
||
class Preview(BaseModel):
|
||
copy_id: str = Field(alias="copy")
|
||
text: str
|
||
source: str
|
||
target: str
|
||
|
||
|
||
def create_app(service: Finalisation, browser_token: str, stop=lambda: None):
|
||
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
|
||
|
||
cookie_name = 'copienator_' + browser_token[:12]
|
||
|
||
@app.middleware('http')
|
||
async def local_session(request: Request, call_next):
|
||
host = request.headers.get('host', '').split(':')[0]
|
||
if host not in ('127.0.0.1', 'localhost', 'testserver'):
|
||
return Response(status_code=403)
|
||
origin = request.headers.get('origin')
|
||
if origin and origin != str(request.base_url).rstrip('/'):
|
||
return Response(status_code=403)
|
||
if request.url.path == '/' and secrets.compare_digest(request.query_params.get('token', ''), browser_token):
|
||
response = RedirectResponse('/', status_code=303)
|
||
response.set_cookie(cookie_name, browser_token, httponly=True, samesite='strict')
|
||
return response
|
||
if not secrets.compare_digest(request.cookies.get(cookie_name, ''), browser_token):
|
||
return Response('Ouvrez le lien fourni par Copienator.', status_code=403)
|
||
response = await call_next(request)
|
||
response.headers['Cache-Control'] = 'no-store'
|
||
response.headers['X-Content-Type-Options'] = 'nosniff'
|
||
response.headers['Referrer-Policy'] = 'no-referrer'
|
||
response.headers['Content-Security-Policy'] = "default-src 'self'; img-src 'self' data: blob:; style-src 'self' 'unsafe-inline'; frame-ancestors 'none'"
|
||
return response
|
||
|
||
@app.exception_handler(ValueError)
|
||
@app.exception_handler(CliError)
|
||
@app.exception_handler(OSError)
|
||
async def invalid(request, exc):
|
||
return JSONResponse({'detail': str(exc)}, status_code=409)
|
||
|
||
@app.get('/api/state')
|
||
def state():
|
||
return service.state()
|
||
|
||
@app.get('/api/job')
|
||
def job():
|
||
with service.mutex:
|
||
return dict(service.job)
|
||
|
||
@app.post('/api/change')
|
||
def change(data: Change):
|
||
if data.kind != 'general' and not isinstance(data.value, str):
|
||
raise ValueError('Texte attendu.')
|
||
return service.save(data.model_dump(by_alias=True))
|
||
|
||
@app.post('/api/regenerate')
|
||
def regenerate(data: Regenerate):
|
||
return service.start_job(data.revision, data.apply_draft, data.incremental)
|
||
|
||
@app.post('/api/stop')
|
||
def shutdown():
|
||
with service.mutex:
|
||
if service.job['running']:
|
||
raise ValueError('Attendez la fin de la régénération avant de fermer le serveur.')
|
||
stop()
|
||
return {'stopped': True}
|
||
|
||
@app.get('/api/image')
|
||
def image(copy: str, label: str, body: bool = False):
|
||
directory = service.directory(copy)
|
||
service.label(label)
|
||
if body:
|
||
result = body_image(directory, label)
|
||
output = BytesIO()
|
||
result.save(output, format='PNG')
|
||
return Response(output.getvalue(), media_type='image/png')
|
||
path = directory / f'{label}.jpg'
|
||
if not path.is_file():
|
||
raise HTTPException(404, 'Aucune image pour cette réponse.')
|
||
return FileResponse(path)
|
||
|
||
@app.get('/api/document')
|
||
def document(copy: str, kind: str, label: str = ''):
|
||
directory = service.directory(copy)
|
||
if kind == 'original':
|
||
path = original_copy_path(service.root, directory.name[5:])
|
||
elif kind == 'output':
|
||
path = directory / 'Concat_F.pdf'
|
||
elif kind == 'context':
|
||
from copienator import utils
|
||
from pypdf import PdfWriter
|
||
paths = utils.pdf_images_of_contexts(service.root, service.label(label), service.labels)
|
||
if not paths:
|
||
raise HTTPException(404, 'Aucun contexte pour ce label.')
|
||
writer = PdfWriter()
|
||
for item in paths:
|
||
if not Path(item).resolve().is_relative_to(service.root):
|
||
raise HTTPException(403)
|
||
writer.append(item)
|
||
buffer = BytesIO()
|
||
writer.write(buffer)
|
||
return Response(buffer.getvalue(), media_type='application/pdf')
|
||
elif kind in ('question', 'solution'):
|
||
path = service.root / ('Text2' if kind == 'question' else 'Sol2') / f'{service.label(label)}.pdf'
|
||
else:
|
||
raise ValueError('Document inconnu.')
|
||
if path is None or not path.is_file():
|
||
raise HTTPException(404, 'Ce document est absent.')
|
||
if not path.resolve().is_relative_to(service.root):
|
||
raise HTTPException(403)
|
||
return FileResponse(path, media_type='application/pdf')
|
||
|
||
@app.post('/api/preview')
|
||
def preview(data: Preview):
|
||
with service.mutex:
|
||
if service.job['running']:
|
||
raise ValueError('Régénération en cours.')
|
||
directory = service.directory(data.copy_id)
|
||
service.label(data.source)
|
||
service.label(data.target)
|
||
instructions = read_instructions(data.text, service.mode)
|
||
validate_instructions(service.root, service.mode, instructions)
|
||
_, bodies, _ = edited_bodies(directory, baseline(directory),
|
||
[i for i in instructions if i.copy_id == directory.name[5:]])
|
||
result = {}
|
||
for label in (data.source, data.target):
|
||
image = bodies.get(label)
|
||
if image:
|
||
buffer = BytesIO()
|
||
image.save(buffer, format='PNG')
|
||
result[label] = 'data:image/png;base64,' + base64.b64encode(buffer.getvalue()).decode()
|
||
else:
|
||
result[label] = None
|
||
return result
|
||
|
||
app.mount('/', StaticFiles(directory=Path(__file__).parent / 'web_static', html=True), name='web')
|
||
return app
|
||
|
||
|
||
class WebSession:
|
||
def __init__(self, root: Path, mode: str, *, repository: Path | None = None):
|
||
self.lease = EvaluationLock(root)
|
||
self.lease.__enter__()
|
||
try:
|
||
self.service = Finalisation(root, mode, self.lease.token, self.lease.stream.fileno(), repository=repository)
|
||
if not self.service.copies():
|
||
raise ValueError('Générez les copies Anot, ou lisez les annotations Bnot/BGnot, avant d’ouvrir la finalisation web.')
|
||
self.token = secrets.token_urlsafe(32)
|
||
self.socket = socket.socket()
|
||
self.socket.bind(('127.0.0.1', 0))
|
||
self.port = self.socket.getsockname()[1]
|
||
self.app = create_app(self.service, self.token, self.request_stop)
|
||
self.server = uvicorn.Server(uvicorn.Config(self.app, log_level='warning', access_log=False))
|
||
self.thread = threading.Thread(target=self._run, daemon=True)
|
||
self.thread.start()
|
||
deadline = time.monotonic() + 5
|
||
while not self.server.started and self.thread.is_alive() and time.monotonic() < deadline:
|
||
time.sleep(.02)
|
||
if not self.server.started:
|
||
self.server.should_exit = True
|
||
raise RuntimeError('Le serveur web n’a pas démarré.')
|
||
except Exception:
|
||
self.lease.__exit__()
|
||
raise
|
||
|
||
@property
|
||
def url(self):
|
||
return f'http://127.0.0.1:{self.port}/?token={self.token}'
|
||
|
||
def _run(self):
|
||
try:
|
||
self.server.run(sockets=[self.socket])
|
||
finally:
|
||
# Never release the GUI/CLI lease while a generation worker is active.
|
||
while self.service.job['running']:
|
||
time.sleep(.1)
|
||
self.socket.close()
|
||
self.lease.__exit__()
|
||
|
||
def request_stop(self):
|
||
with self.service.mutex:
|
||
if self.service.job['running']:
|
||
raise ValueError('Attendez la fin de la régénération.')
|
||
self.service.closing = True
|
||
self.server.should_exit = True
|
||
|
||
@property
|
||
def alive(self):
|
||
return self.thread.is_alive()
|