"""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()