flowdeck/app/routers/imports.py

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