feat(import): v5.6.0 unified data import (phases 0-5)
- unified importer framework (app/services/importers/): normalized model, registry, common pipeline (hierarchy, attachments, collections, dedup), async jobs, dry-run preview, column->type mapping - Phase 1: Obsidian, Notion, Logseq/Roam, HTML (Apple Notes/Bear/Ulysses/ OneNote), Google Keep, generic Markdown - Phase 2: typed CSV/TSV, Excel (openpyxl), generic JSON - Phase 3: Word .docx (python-docx), PDF (pypdf), HTML folders - Phase 4: Raindrop, Pocket, Readwise, Shaarli, Netscape bookmarks, .ics, OPML, Standard Notes, Gitea/GitHub issues (+labels/milestones) - Phase 5: incremental re-sync (skip/update/duplicate), partial-error resume, forge repo files, URL web clipper (SSRF guard), batch multi-file + UI queue, Notion relation resolution, exportable JSON reports - /import wizard, API /api/import/*, migration 9 (import_items, import_jobs) - fix: property values stored by property id (correct DB view rendering) - deps: openpyxl, beautifulsoup4, PyYAML, python-docx, pypdf - 43 import tests; full suite 491 green; ruff clean - bump version 5.11.5
This commit is contained in:
@@ -0,0 +1,378 @@
|
||||
"""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)
|
||||
async 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 jinja2 import Environment, FileSystemLoader
|
||||
|
||||
env = Environment(loader=FileSystemLoader("app/templates"))
|
||||
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")
|
||||
async 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")
|
||||
async 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")
|
||||
async def import_jobs(request: Request):
|
||||
return {"jobs": list_jobs()}
|
||||
|
||||
|
||||
@router.get("/jobs/{job_id}")
|
||||
async 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")
|
||||
async 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 ""
|
||||
Reference in New Issue
Block a user