"""FlowDeck — unified import API (v5.6.0, Phase 0/1/2). Exposes the importer registry, a dry-run preview, a synchronous run and an optional background job with polling. Works with the existing workspace cookie (``flowdeck_workspace``) and session. """ from __future__ import annotations import json import logging from fastapi import APIRouter, HTTPException, Request from fastapi.responses import HTMLResponse, RedirectResponse, Response from app.auth.session import SessionManager from app.db import get_conn from app.services.automations import fire_event from app.services.importers import ( get_job, list_jobs, list_sources, parse_upload, preview_result, resolve_relations, run_import, start_import_job, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/import", tags=["import"]) page_router = APIRouter(tags=["import"]) MAX_UPLOAD_BYTES = 200 * 1024 * 1024 def _current_user(request: Request) -> dict: return SessionManager.decode_session(request.cookies.get("flowdeck_session", "")) or {} @page_router.get("/import", response_class=HTMLResponse) def import_page(request: Request): """Standalone import wizard (source picker, dry-run, mapping, progress).""" user = _current_user(request) if not user: return RedirectResponse("/auth/login?provider=local", status_code=302) from app.templating import ENV env = ENV return HTMLResponse(content=env.get_template("import.html").render(user=user)) def _workspace(request: Request) -> tuple[int | None, str]: """Resolve (workspace_id, login) from the workspace cookie + session.""" ws_id: int | None = None cookie = request.cookies.get("flowdeck_workspace", "") try: value = int(cookie) if value > 0: ws_id = value except (ValueError, TypeError): pass user = _current_user(request) login = user.get("login", "") if user else "" return ws_id, login async def _read_upload(request: Request) -> tuple[str, bytes, str | None]: form = await request.form() upload = form.get("file") if upload is None or not hasattr(upload, "filename"): raise HTTPException(400, "file field required") filename = (upload.filename or "import").replace("\\", "/").rsplit("/", 1)[-1] data = await upload.read() if len(data) > MAX_UPLOAD_BYTES: raise HTTPException(413, "File too large (max 200 MB)") source_id = form.get("source") or None return filename, data, source_id @router.get("/sources") def import_sources(request: Request): """List every available importer for the UI source picker.""" return {"sources": list_sources()} @router.post("/preview") async def import_preview(request: Request): """Dry-run: parse the upload and describe what would be created.""" filename, data, source_id = await _read_upload(request) imp, result = parse_upload(filename, data, source_id) if imp is None: raise HTTPException(400, "Format non reconnu — choisissez une source") out = preview_result(result) out["detected_source"] = imp.source_id out["source_label"] = imp.label return out @router.post("/run") async def import_run(request: Request): """Import an upload (synchronously, or as a background job when async=true).""" form = await request.form() upload = form.get("file") if upload is None or not hasattr(upload, "filename"): raise HTTPException(400, "file field required") filename = (upload.filename or "import").replace("\\", "/").rsplit("/", 1)[-1] data = await upload.read() if len(data) > MAX_UPLOAD_BYTES: raise HTTPException(413, "File too large (max 200 MB)") source_id = form.get("source") or None parent_id = _int_or_none(form.get("parent_id")) target = _int_or_none(form.get("target_collection_id")) dedup = str(form.get("dedup", "true")).lower() not in ("false", "0", "no") async_mode = str(form.get("async", "false")).lower() in ("true", "1", "yes") mapping = _parse_mapping(form.get("mapping")) mode = _parse_mode(form.get("mode")) ws_id, login = _workspace(request) if async_mode: job = start_import_job( filename=filename, data=data, source_id=source_id, workspace_id=ws_id, workspace_name=login, user_login=login, parent_page_id=parent_id, target_collection_id=target, dedup=dedup, mapping=mapping, mode=mode, ) return {"status": "queued", "job_id": job["id"]} imp, result = parse_upload(filename, data, source_id) if imp is None: raise HTTPException(400, "Format non reconnu — choisissez une source") report = run_import( result, workspace_id=ws_id, workspace_name=login, user_login=login, parent_page_id=parent_id, target_collection_id=target, dedup=dedup, mapping=mapping, mode=mode, ) report["detected_source"] = imp.source_id if imp.source_id == "notion": with get_conn() as conn: report["relations"] = resolve_relations(conn, ws_id) for page_id in report.get("page_ids", [])[:100]: try: await fire_event("page.created", {"page_id": page_id, "title": "", "workspace": login}) except Exception: # noqa: BLE001 - events are best-effort pass return report @router.post("/forge") async def import_forge(request: Request): """Import a forge repo's issues (+ labels/milestones) into collections.""" try: body = await request.json() except Exception: raise HTTPException(400, "Invalid JSON body") from None provider = str(body.get("provider") or "gitea").lower() owner = str(body.get("owner") or "").strip() repo = str(body.get("repo") or "").strip() if not owner or not repo: raise HTTPException(400, "owner and repo are required") state = str(body.get("state") or "all") include_labels = bool(body.get("include_labels", True)) include_milestones = bool(body.get("include_milestones", True)) ws_id, login = _workspace(request) if provider == "gitea": from app.services.gitea_client import get_user_gitea_client from app.services.importers.forge import GiteaForgeAdapter client = get_user_gitea_client(request) if client is None: raise HTTPException(400, "Gitea non connecté") adapter = GiteaForgeAdapter(client) elif provider == "github": from app.services.github_adapter import GitHubAdapter token = _user_oauth_token(request, "github") if not token: raise HTTPException(400, "GitHub non connecté") adapter = GitHubAdapter(token) else: raise HTTPException(400, "provider must be 'gitea' or 'github'") from app.services.importers.forge import fetch_forge_issues result = await fetch_forge_issues( adapter, owner, repo, provider=provider, state=state, include_labels=include_labels, include_milestones=include_milestones, ) report = run_import( result, workspace_id=ws_id, workspace_name=login, user_login=login, ) report["detected_source"] = f"forge:{provider}" return report @router.post("/forge-repo") async def import_forge_repo(request: Request): """Import a forge repo's text files as pages (folder hierarchy preserved).""" try: body = await request.json() except Exception: raise HTTPException(400, "Invalid JSON body") from None provider = str(body.get("provider") or "gitea").lower() owner = str(body.get("owner") or "").strip() repo = str(body.get("repo") or "").strip() if not owner or not repo: raise HTTPException(400, "owner and repo are required") path = str(body.get("path") or "") max_files = min(int(body.get("max_files") or 200), 1000) ws_id, login = _workspace(request) adapter = _forge_adapter(request, provider) from app.services.importers.forge_repo import fetch_forge_repo result = await fetch_forge_repo( adapter, owner, repo, provider=provider, path=path, max_files=max_files, ) report = run_import(result, workspace_id=ws_id, workspace_name=login, user_login=login) report["detected_source"] = f"forge-repo:{provider}" return report @router.post("/url") async def import_url(request: Request): """Web clipper: fetch a URL and create a page (bookmark card + content).""" try: body = await request.json() except Exception: raise HTTPException(400, "Invalid JSON body") from None url = str(body.get("url") or "").strip() if not url: raise HTTPException(400, "url is required") ws_id, login = _workspace(request) from app.services.importers.url_fetch import fetch_url_result try: result = await fetch_url_result(url) except ValueError as exc: raise HTTPException(400, str(exc)) from None if not result.pages: raise HTTPException(422, "; ".join(result.warnings) or "Page introuvable") report = run_import(result, workspace_id=ws_id, workspace_name=login, user_login=login) report["detected_source"] = "url" return report @router.post("/run-batch") async def import_run_batch(request: Request): """Import several uploaded files sequentially, returning one report each.""" form = await request.form() uploads = form.getlist("file") if not uploads: raise HTTPException(400, "file field required") source_id = form.get("source") or None parent_id = _int_or_none(form.get("parent_id")) target = _int_or_none(form.get("target_collection_id")) dedup = str(form.get("dedup", "true")).lower() not in ("false", "0", "no") mode = _parse_mode(form.get("mode")) mapping = _parse_mapping(form.get("mapping")) ws_id, login = _workspace(request) results: list[dict] = [] summary = {"files": 0, "pages_created": 0, "rows_created": 0, "errors": 0} for upload in uploads: filename = (getattr(upload, "filename", "") or "import").replace("\\", "/").rsplit("/", 1)[-1] data = await upload.read() if len(data) > MAX_UPLOAD_BYTES: results.append({"filename": filename, "report": {"status": "error", "errors": [{"title": filename, "error": "File too large"}]}}) summary["errors"] += 1 continue imp, result = parse_upload(filename, data, source_id) if imp is None: results.append({"filename": filename, "report": {"status": "error", "errors": [{"title": filename, "error": "Format non reconnu"}]}}) summary["errors"] += 1 continue report = run_import( result, workspace_id=ws_id, workspace_name=login, user_login=login, parent_page_id=parent_id, target_collection_id=target, dedup=dedup, mapping=mapping, mode=mode, ) report["detected_source"] = imp.source_id results.append({"filename": filename, "report": report}) summary["files"] += 1 summary["pages_created"] += report.get("pages_created", 0) summary["rows_created"] += report.get("rows_created", 0) summary["errors"] += len(report.get("errors", [])) return {"status": "ok", "summary": summary, "results": results} @router.post("/relations/resolve") def import_resolve_relations(request: Request): """Convert text columns referencing another collection into relation props.""" ws_id, _ = _workspace(request) with get_conn() as conn: return resolve_relations(conn, ws_id) @router.get("/jobs") def import_jobs(request: Request): return {"jobs": list_jobs()} @router.get("/jobs/{job_id}") def import_job(job_id: str): job = get_job(job_id) if not job: raise HTTPException(404, "Job not found") return job @router.get("/jobs/{job_id}/report") def import_job_report(job_id: str): """Download a job's import report as JSON.""" job = get_job(job_id) if not job: raise HTTPException(404, "Job not found") payload = json.dumps(job.get("report") or {}, ensure_ascii=False, indent=2) return Response( content=payload, media_type="application/json", headers={"Content-Disposition": f'attachment; filename="import-{job_id}.json"'}, ) def _forge_adapter(request: Request, provider: str): if provider == "gitea": from app.services.gitea_client import get_user_gitea_client from app.services.importers.forge import GiteaForgeAdapter client = get_user_gitea_client(request) if client is None: raise HTTPException(400, "Gitea non connecté") return GiteaForgeAdapter(client) if provider == "github": from app.services.github_adapter import GitHubAdapter token = _user_oauth_token(request, "github") if not token: raise HTTPException(400, "GitHub non connecté") return GitHubAdapter(token) raise HTTPException(400, "provider must be 'gitea' or 'github'") def _int_or_none(value) -> int | None: try: ivalue = int(value) return ivalue if ivalue > 0 else None except (ValueError, TypeError): return None def _parse_mapping(value) -> dict[str, str] | None: if not value: return None try: parsed = json.loads(value) except (ValueError, TypeError): return None if isinstance(parsed, dict): return {str(k): str(v) for k, v in parsed.items() if v} return None def _parse_mode(value) -> str | None: mode = str(value or "").strip().lower() return mode if mode in ("skip", "update", "duplicate") else None def _user_oauth_token(request: Request, provider: str) -> str: user = _current_user(request) if not user or not user.get("id"): return "" with get_conn() as conn: row = conn.execute( "SELECT access_token FROM user_oauth_tokens WHERE user_id=? AND provider=? " "ORDER BY updated_at DESC LIMIT 1", (user["id"], provider), ).fetchone() return row["access_token"] if row else ""