feat: bulk-import endpoint mit propfind walk

app/bulk.py

@@ -0,0 +1,96 @@
+import logging
+import xml.etree.ElementTree as ET
+from pathlib import PurePosixPath
+from urllib.parse import unquote
+
+import httpx
+from fastapi import APIRouter, BackgroundTasks, Header, status
+from pydantic import BaseModel
+
+from app.config import get_settings
+from app.ingest.extractors import SUPPORTED_TYPES
+from app.ingest.pipeline import process_file
+from app.webhook.auth import verify_secret
+from app.webhook.models import EventType
+
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+class BulkRequest(BaseModel):
+    path: str
+
+
+PROPFIND_BODY = """<?xml version="1.0"?>
+<d:propfind xmlns:d="DAV:"><d:prop><d:resourcetype/></d:prop></d:propfind>
+"""
+
+DAV_NS = {"d": "DAV:"}
+
+
+async def list_files_recursive(base_url: str, user: str, password: str, path: str) -> list[str]:
+    """PROPFIND with Depth: infinity. Returns relative file paths (no folders)."""
+    base = base_url.rstrip("/")
+    rel = path.strip("/")
+    url = f"{base}/{rel}"
+
+    async with httpx.AsyncClient(auth=(user, password), timeout=120.0) as client:
+        response = await client.request(
+            "PROPFIND",
+            url,
+            headers={"Depth": "infinity", "Content-Type": "application/xml"},
+            content=PROPFIND_BODY,
+        )
+    if response.status_code not in (200, 207):
+        raise RuntimeError(f"PROPFIND failed: status={response.status_code}")
+
+    root = ET.fromstring(response.text)
+    base_path_segment = PurePosixPath(httpx.URL(base).path).as_posix()  # "/remote.php/dav/files/u"
+    out: list[str] = []
+    for resp in root.findall("d:response", DAV_NS):
+        href = resp.findtext("d:href", default="", namespaces=DAV_NS)
+        decoded = unquote(href)
+        # Strip the WebDAV base prefix → leaves "Documents/.../file.pdf"
+        if decoded.startswith(base_path_segment):
+            decoded = decoded[len(base_path_segment):]
+        decoded = decoded.lstrip("/")
+        if not decoded or decoded.endswith("/"):
+            continue
+        # Skip directory entries (those have <d:collection/> resourcetype)
+        rt = resp.find("d:propstat/d:prop/d:resourcetype/d:collection", DAV_NS)
+        if rt is not None:
+            continue
+        out.append(decoded)
+    return out
+
+
+@router.post("/bulk-import", status_code=status.HTTP_202_ACCEPTED)
+async def bulk_import(
+    body: BulkRequest,
+    background: BackgroundTasks,
+    x_webhook_secret: str | None = Header(default=None),
+):
+    settings = get_settings()
+    verify_secret(x_webhook_secret, settings.webhook_secret)
+
+    files = await list_files_recursive(
+        settings.nextcloud_webdav_url,
+        settings.nextcloud_user,
+        settings.nextcloud_app_password,
+        body.path,
+    )
+
+    dispatched = 0
+    for f in files:
+        ext = PurePosixPath(f).suffix.lstrip(".").lower()
+        if ext not in SUPPORTED_TYPES:
+            continue
+        background.add_task(process_file, f, EventType.CREATED)
+        dispatched += 1
+
+    logger.info(
+        "bulk dispatch",
+        extra={"event": "bulk_dispatch", "path": body.path, "dispatched": dispatched, "total_listed": len(files)},
+    )
+    return {"status": "accepted", "dispatched": dispatched}

app/main.py

@@ -9,6 +9,7 @@ from app.ingest.embedder import embedding_dimension
 from app.logging_setup import setup_logging
 from app.qdrant_store import ensure_collection
 from app.webhook.handler import router as webhook_router
+from app.bulk import router as bulk_router
 
 
 logger = logging.getLogger(__name__)
@@ -34,6 +35,7 @@ async def lifespan(app: FastAPI):
 
 app = FastAPI(title="rag-ingestor", lifespan=lifespan)
 app.include_router(webhook_router)
+app.include_router(bulk_router)
 
 
 @app.get("/health")