| 1 | from __future__ import annotations |
| 2 | |
| 3 | from helpers.api import ApiHandler, Input, Output, Request, Response |
| 4 | from plugins._migrate_agents.helpers.migration import MAX_TOTAL_BYTES, Upload, parse_bundle, preview |
| 5 | |
| 6 | |
| 7 | def uploaded_files(request: Request) -> list[Upload]: |
| 8 | result: list[Upload] = [] |
| 9 | total = 0 |
| 10 | for item in request.files.getlist("files[]"): |
| 11 | data = item.stream.read(MAX_TOTAL_BYTES + 1) |
| 12 | total += len(data) |
| 13 | if total > MAX_TOTAL_BYTES: |
| 14 | raise ValueError("Upload exceeds the 256 MiB limit") |
| 15 | result.append(Upload(item.filename or "upload", data)) |
| 16 | if not result: |
| 17 | raise ValueError("Choose at least one export file or folder") |
| 18 | return result |
| 19 | |
| 20 | |
| 21 | class MigrationPreview(ApiHandler): |
| 22 | async def process(self, input: Input, request: Request) -> Output: |
| 23 | try: |
| 24 | source = str(request.form.get("source") or "").strip().lower() |
| 25 | return preview(parse_bundle(source, uploaded_files(request))) |
| 26 | except ValueError as exc: |
| 27 | return Response(str(exc), 400) |