diff --git a/src/oikb/cli.py b/src/oikb/cli.py
index b7bffbc..eaa3eec 100644
--- a/src/oikb/cli.py
+++ b/src/oikb/cli.py
@@ -269,6 +269,13 @@ def _resolve_connector(source: str, branch: str | None = None, path: str | None
)
# Default: local filesystem.
+ if ":" not in source and source.isascii() and source.isupper() and source.isalnum():
+ raise ValueError(
+ f"Unrecognized source {source!r}. Bare uppercase names map to "
+ f"/data/{source} on the filesystem. For Confluence spaces use "
+ f"confluence:{source}."
+ )
+
from oikb.connectors.filesystem import FilesystemConnector
return FilesystemConnector(source)
@@ -397,46 +404,38 @@ def sync(
sys.exit(1)
has_errors = False
- for entry in entries:
- entry_source = entry.get("source")
- entry_kb = entry.get("kb-id")
- entry_branch = entry.get("branch")
- entry_path = entry.get("path")
- entry_filter = entry.get("filter", {})
-
- if not entry_source or not entry_kb:
- click.echo(click.style(f"Skipping invalid entry (needs source + kb-id): {entry}", fg="yellow"), err=True)
+ from oikb.kb_sync import group_entries_by_kb, run_entries_sync, sources_label
+
+ for group in group_entries_by_kb(entries):
+ entry_kb = group[0].get("kb-id")
+ label = sources_label(group)
+
+ if not entry_kb or any(not e.get("source") for e in group):
+ click.echo(
+ click.style(
+ f"Skipping invalid group (needs source + kb-id): {group}",
+ fg="yellow",
+ ),
+ err=True,
+ )
continue
try:
- connector = _resolve_connector(entry_source, entry_branch, entry_path)
client = _make_client(url, token)
if not quiet:
click.echo(f"\n{'─' * 40}")
- click.echo(f"Syncing: {entry_source} → {entry_kb}")
-
- mf = None
- inc = entry_filter.get("include")
- exc = entry_filter.get("exclude")
- ms = entry_filter.get("max-size") or max_file_size
- if inc or exc or ms:
- from oikb.sync import build_manifest_filter, parse_size
- mf = build_manifest_filter(
- include=inc,
- exclude=exc,
- max_size=parse_size(ms),
- )
-
- result = run_sync(
- client=client,
- connector=connector,
- kb_id=entry_kb,
+ click.echo(f"Syncing: {label} → {entry_kb}")
+
+ result = run_entries_sync(
+ client,
+ group,
+ resolve_connector=_resolve_connector,
dry_run=dry_run,
verbose=verbose,
quiet=quiet,
- manifest_filter=mf,
- concurrency=entry.get("concurrency", concurrency),
+ concurrency=concurrency,
+ max_file_size=max_file_size,
)
if not quiet:
diff --git a/src/oikb/connectors/composite.py b/src/oikb/connectors/composite.py
new file mode 100644
index 0000000..384038c
--- /dev/null
+++ b/src/oikb/connectors/composite.py
@@ -0,0 +1,46 @@
+"""Composite connector — merge multiple sources into one KB manifest."""
+
+from __future__ import annotations
+
+from oikb.connectors import BaseConnector, ManifestEntry
+
+
+class CompositeConnector(BaseConnector):
+ """Route read_file calls across connectors with a pre-merged manifest.
+
+ Used when several .oikb.yaml sources target the same kb-id. Each source
+ sync alone would drop files from the others (OWUI diff is KB-wide).
+ """
+
+ def __init__(self, parts: list[tuple[BaseConnector, list[ManifestEntry]]]):
+ self._parts = parts
+ self._route: dict[tuple[str, str], BaseConnector] = {}
+ self._manifest: list[ManifestEntry] = []
+
+ for connector, manifest in parts:
+ for entry in manifest:
+ key = (entry.path, entry.filename)
+ if key in self._route:
+ raise ValueError(
+ f"Duplicate manifest path across sources: {entry.display_path}"
+ )
+ self._route[key] = connector
+ self._manifest.append(entry)
+
+ def build_manifest(self) -> list[ManifestEntry]:
+ return list(self._manifest)
+
+ def read_file(self, path: str, filename: str) -> bytes:
+ connector = self._route.get((path, filename))
+ if not connector:
+ display = f"{path}/{filename}" if path else filename
+ raise FileNotFoundError(f"File not in merged manifest: {display}")
+ return connector.read_file(path, filename)
+
+ def close(self) -> None:
+ closed: set[int] = set()
+ for connector, _ in self._parts:
+ token = id(connector)
+ if token not in closed:
+ connector.close()
+ closed.add(token)
diff --git a/src/oikb/connectors/confluence.py b/src/oikb/connectors/confluence.py
index c7c2d28..c413b83 100644
--- a/src/oikb/connectors/confluence.py
+++ b/src/oikb/connectors/confluence.py
@@ -1,7 +1,11 @@
"""Confluence connector — sync a Confluence space to a Knowledge Base.
-Uses the Confluence Cloud REST API v2. Pages are exported as plain text.
-Auth via CONFLUENCE_URL, CONFLUENCE_USER, CONFLUENCE_TOKEN env vars.
+Supports Confluence REST API v1 (self-hosted) and v2 (Cloud). Pages are
+exported as plain text. Select the API via CONFLUENCE_API_VERSION (default v2).
+
+Auth via CONFLUENCE_URL, CONFLUENCE_USER, and CONFLUENCE_TOKEN env vars:
+ - Server/Data Center PAT: set CONFLUENCE_TOKEN only (Bearer auth)
+ - Cloud API token: set CONFLUENCE_USER (email) + CONFLUENCE_TOKEN (Basic auth)
"""
from __future__ import annotations
@@ -17,24 +21,164 @@
from oikb.connectors import BaseConnector, ManifestEntry
-def _storage_to_text(storage_html: str) -> str:
- """Convert Confluence storage format (XHTML) to plain text."""
- # Strip all HTML tags.
+BASE_ENDPOINTS = {
+ "v1": "/rest/api",
+ "v2": "/wiki/api/v2",
+}
+
+
+_INVALID_PATH_CHARS = re.compile(r'[<>:"/\\|?*]')
+
+
+def _sanitize_path_segment(name: str) -> str:
+ """Sanitize a single path segment (ancestor dir or filename stem)."""
+ cleaned = _INVALID_PATH_CHARS.sub("_", name).strip()
+ return cleaned or "_"
+
+
+def _ancestor_dir_path_v1(page: dict[str, Any]) -> str:
+ """Build KB directory path from Confluence v1 ancestors list."""
+ segments = [
+ _sanitize_path_segment(a.get("title", ""))
+ for a in page.get("ancestors") or []
+ if a.get("title")
+ ]
+ return "/".join(segments)
+
+
+def _ancestor_dir_path_v2(
+ page_id: str, pages_by_id: dict[str, dict[str, Any]]
+) -> str:
+ """Build KB directory path by walking Confluence v2 parentId chain."""
+ segments: list[str] = []
+ current = pages_by_id.get(str(page_id))
+ if not current:
+ return ""
+
+ parent_id = current.get("parentId")
+ visited: set[str] = set()
+ while parent_id:
+ pid = str(parent_id)
+ if pid in visited:
+ break
+ visited.add(pid)
+ parent = pages_by_id.get(pid)
+ if not parent:
+ break
+ title = parent.get("title", "")
+ if title:
+ segments.append(_sanitize_path_segment(title))
+ parent_id = parent.get("parentId")
+
+ segments.reverse()
+ return "/".join(segments)
+
+
+def _parse_api_version(api_version: str | None) -> str:
+ version = (api_version or os.environ.get("CONFLUENCE_API_VERSION", "v2")).lower()
+ if version not in BASE_ENDPOINTS:
+ valid = ", ".join(sorted(BASE_ENDPOINTS))
+ raise ValueError(
+ f"Invalid Confluence API version {version!r}. "
+ f"Expected one of: {valid}. Set CONFLUENCE_API_VERSION=v1 or v2."
+ )
+ return version
+
+
+def _storage_to_text(storage_html: str, title: str = "") -> str:
+ """Convert Confluence storage format (XHTML) to plain text.
+
+ Confluence link-only index pages store targets in XML attributes
+ (ri:content-title, href) rather than visible text. Plain tag stripping
+ yields an empty string and Open WebUI rejects the upload with 400.
+ """
+ parts: list[str] = []
+
+ # Page / attachment link targets live in attributes, not element text.
+ for pattern in (
+ r'ri:content-title="([^"]*)"',
+ r'ri:filename="([^"]*)"',
+ r'ri:url="([^"]*)"',
+ ):
+ for match in re.finditer(pattern, storage_html):
+ value = html.unescape(match.group(1)).strip()
+ if value:
+ parts.append(value)
+
+ for match in re.finditer(r'href="([^"#][^"]*)"', storage_html):
+ value = html.unescape(match.group(1)).strip()
+ if value:
+ parts.append(value)
+
+ for match in re.finditer(
+ r"([^<]*)",
+ storage_html,
+ ):
+ value = html.unescape(match.group(1)).strip()
+ if value:
+ parts.append(value)
+
+ # Remaining visible text after stripping tags/macros.
text = re.sub(r"<[^>]+>", " ", storage_html)
text = html.unescape(text)
- # Collapse whitespace.
text = re.sub(r"\s+", " ", text).strip()
- return text
+ if text:
+ parts.append(text)
+
+ # De-dupe while preserving order.
+ seen: set[str] = set()
+ lines: list[str] = []
+ for part in parts:
+ if part not in seen:
+ seen.add(part)
+ lines.append(part)
+
+ if lines:
+ return "\n".join(lines)
+
+ return title.strip()
+
+
+def _finalize_page_text(
+ text: str,
+ *,
+ title: str,
+ page_id: str,
+ space_key: str,
+ base_url: str,
+ path: str = "",
+ filename: str = "",
+) -> str:
+ """Ensure non-empty, unique text for Open WebUI vector deduplication.
+
+ OWUI hashes document chunks and rejects duplicates already in the KB.
+ The unique id must be in the *first* lines so chunking differs for pages
+ that share boilerplate body text.
+ """
+ body = text.strip() or title.strip() or f"Confluence page {page_id}"
+ display = f"{path}/{filename}" if path else (filename or title)
+ uid = f"confluence:{space_key}:{page_id}"
+ if display:
+ uid = f"{uid}:{display}"
+
+ header = f"# {title}\n{uid}\n"
+ source = uid
+ if base_url:
+ source = (
+ f"{uid} {base_url.rstrip('/')}/pages/viewpage.action?pageId={page_id}"
+ )
+ return f"{header}\n{body}\n\n---\n{source}"
class ConfluenceConnector(BaseConnector):
- """Sync pages from a Confluence Cloud space.
+ """Sync pages from a Confluence space.
Args:
- space_key: Confluence space key (e.g. "ENG").
- base_url: Confluence instance URL (or CONFLUENCE_URL env var).
- user: Confluence user email (or CONFLUENCE_USER env var).
- token: Confluence API token (or CONFLUENCE_TOKEN env var).
+ space_key: Confluence space key (e.g. "ENG").
+ base_url: Confluence instance URL (or CONFLUENCE_URL env var).
+ user: Confluence user email/username (or CONFLUENCE_USER env var).
+ token: API token or PAT (or CONFLUENCE_TOKEN env var).
+ api_version: REST API version, "v1" or "v2" (or CONFLUENCE_API_VERSION env var).
"""
def __init__(
@@ -43,12 +187,16 @@ def __init__(
base_url: str | None = None,
user: str | None = None,
token: str | None = None,
+ api_version: str | None = None,
):
self.space_key = space_key
self._base_url = (base_url or os.environ.get("CONFLUENCE_URL", "")).rstrip("/")
self._user = user or os.environ.get("CONFLUENCE_USER", "")
self._token = token or os.environ.get("CONFLUENCE_TOKEN", "")
+ self._api_version = _parse_api_version(api_version)
+
+ headers = {"Accept": "application/json"}
if not self._base_url:
raise ValueError(
@@ -61,19 +209,62 @@ def __init__(
" export CONFLUENCE_TOKEN="
)
+ if not self._user:
+ headers["Authorization"] = f"Bearer {self._token}"
+
self._http = httpx.Client(
- base_url=f"{self._base_url}/wiki",
+ base_url=f"{self._base_url}{BASE_ENDPOINTS[self._api_version]}",
auth=(self._user, self._token) if self._user else None,
- headers={"Accept": "application/json"},
+ headers=headers,
timeout=60.0,
+ follow_redirects=False,
)
- # Cache page content for read_file.
- self._page_cache: dict[str, str] = {}
+ # (path, filename) -> Confluence page id
+ self._page_cache: dict[tuple[str, str], str] = {}
def build_manifest(self) -> list[ManifestEntry]:
"""List all pages in the space and build a manifest."""
+ if self._api_version == "v2":
+ return self._build_manifest_v2()
+ return self._build_manifest_v1()
+
+ def _build_manifest_v1(self) -> list[ManifestEntry]:
entries: list[ManifestEntry] = []
+ used_keys: set[tuple[str, str]] = set()
+ start = 0
+ limit = 250
+
+ while True:
+ params: dict[str, Any] = {
+ "spaceKey": self.space_key,
+ "type": "page",
+ "limit": limit,
+ "start": start,
+ "expand": "ancestors,version",
+ }
+
+ resp = self._http.get("/content", params=params)
+ resp.raise_for_status()
+ data = resp.json()
+
+ results = data.get("results", [])
+ if not results:
+ break
+
+ for page in results:
+ dir_path = _ancestor_dir_path_v1(page)
+ self._add_page_entry(entries, page, dir_path, used_keys)
+
+ if len(results) < limit:
+ break
+ start += len(results)
+
+ entries.sort(key=lambda e: e.display_path)
+ return entries
+
+ def _build_manifest_v2(self) -> list[ManifestEntry]:
+ all_pages: list[dict[str, Any]] = []
cursor = None
while True:
@@ -82,65 +273,115 @@ def build_manifest(self) -> list[ManifestEntry]:
params["cursor"] = cursor
resp = self._http.get(
- f"/api/v2/spaces/{self.space_key}/pages",
+ f"/spaces/{self.space_key}/pages",
params=params,
)
resp.raise_for_status()
data = resp.json()
- for page in data.get("results", []):
- page_id = page["id"]
- title = page["title"]
- version = page.get("version", {}).get("number", 0)
-
- # Use version number as part of checksum.
- checksum = hashlib.sha256(
- f"{page_id}:v{version}".encode()
- ).hexdigest()[:16]
-
- # Sanitize title for filename.
- filename = re.sub(r'[<>:"/\\|?*]', "_", title) + ".txt"
+ all_pages.extend(data.get("results", []))
- entries.append(
- ManifestEntry(
- filename=filename,
- path="",
- checksum=checksum,
- size=0,
- )
- )
-
- # Store page ID for later retrieval.
- self._page_cache[filename] = page_id
-
- # Handle pagination.
next_link = data.get("_links", {}).get("next")
if not next_link:
break
- # Extract cursor from next link.
cursor_match = re.search(r"cursor=([^&]+)", next_link)
cursor = cursor_match.group(1) if cursor_match else None
if not cursor:
break
+ pages_by_id = {str(page["id"]): page for page in all_pages}
+ entries: list[ManifestEntry] = []
+ used_keys: set[tuple[str, str]] = set()
+ for page in all_pages:
+ dir_path = _ancestor_dir_path_v2(str(page["id"]), pages_by_id)
+ self._add_page_entry(entries, page, dir_path, used_keys)
+
entries.sort(key=lambda e: e.display_path)
return entries
+ def _unique_path_filename(
+ self,
+ page: dict[str, Any],
+ dir_path: str,
+ used_keys: set[tuple[str, str]],
+ ) -> tuple[str, str]:
+ """Return unique (path, filename) using page id when titles collide."""
+ page_id = str(page["id"])
+ filename = _sanitize_path_segment(page.get("title", "Untitled")) + ".txt"
+ key = (dir_path, filename)
+ if key in used_keys:
+ stem = filename.removesuffix(".txt")
+ filename = f"{stem}_{page_id}.txt"
+ key = (dir_path, filename)
+ used_keys.add(key)
+ return dir_path, filename
+
+ def _space_prefixed_path(self, dir_path: str) -> str:
+ """Prefix ancestor path with Confluence space key for multi-space KBs."""
+ if dir_path:
+ return f"{self.space_key}/{dir_path}"
+ return self.space_key
+
+ def _add_page_entry(
+ self,
+ entries: list[ManifestEntry],
+ page: dict[str, Any],
+ dir_path: str,
+ used_keys: set[tuple[str, str]],
+ ) -> None:
+ page_id = str(page["id"])
+ version = page.get("version", {}).get("number", 0)
+
+ checksum = hashlib.sha256(
+ f"{page_id}:v{version}".encode()
+ ).hexdigest()[:16]
+
+ dir_path = self._space_prefixed_path(dir_path)
+ dir_path, filename = self._unique_path_filename(page, dir_path, used_keys)
+
+ entries.append(
+ ManifestEntry(
+ filename=filename,
+ path=dir_path,
+ checksum=checksum,
+ size=0,
+ )
+ )
+
+ self._page_cache[(dir_path, filename)] = page_id
+
def read_file(self, path: str, filename: str) -> bytes:
"""Fetch a page's content and return as text."""
- page_id = self._page_cache.get(filename)
+ page_id = self._page_cache.get((path, filename))
if not page_id:
- raise FileNotFoundError(f"Page not found: {filename}")
+ raise FileNotFoundError(f"Page not found: {path}/{filename}" if path else filename)
+
+ if self._api_version == "v2":
+ resp = self._http.get(
+ f"/pages/{page_id}",
+ params={"body-format": "storage"},
+ )
+ else:
+ resp = self._http.get(
+ f"/content/{page_id}",
+ params={"expand": "body.storage,version"},
+ )
- resp = self._http.get(
- f"/api/v2/pages/{page_id}",
- params={"body-format": "storage"},
- )
resp.raise_for_status()
data = resp.json()
storage = data.get("body", {}).get("storage", {}).get("value", "")
- text = _storage_to_text(storage)
+ title = data.get("title", "")
+ text = _storage_to_text(storage, title=title)
+ text = _finalize_page_text(
+ text,
+ title=title,
+ page_id=page_id,
+ space_key=self.space_key,
+ base_url=self._base_url,
+ path=path,
+ filename=filename,
+ )
return text.encode("utf-8")
def close(self) -> None:
diff --git a/src/oikb/daemon.py b/src/oikb/daemon.py
index c385d68..69c6844 100644
--- a/src/oikb/daemon.py
+++ b/src/oikb/daemon.py
@@ -187,14 +187,36 @@ async def history_endpoint(
)
async def trigger_sync(identifier: str, dry_run: bool = False):
"""Triggers an immediate sync matching the given alias or Knowledge Base ID. The sync runs asynchronously in the background. Use get_sync_status to check progress. Set dry_run=true to preview changes without uploading."""
- for entry in _entries:
- if entry.get("name") == identifier or entry.get("kb-id") == identifier:
- if dry_run:
- result = await _run_entry(entry, dry_run=True)
- return {"dry_run": True, "name": entry.get("name"), "kb_id": entry.get("kb-id"), "result": result}
- asyncio.create_task(_run_entry(entry))
- return {"triggered": True, "name": entry.get("name"), "kb_id": entry.get("kb-id")}
- return {"triggered": False, "error": f"No entry matching '{identifier}'"}
+ from oikb.kb_sync import group_entries_by_kb
+
+ matched = [
+ entry
+ for entry in _entries
+ if entry.get("name") == identifier or entry.get("kb-id") == identifier
+ ]
+ if not matched:
+ return {"triggered": False, "error": f"No entry matching '{identifier}'"}
+
+ kb_id = matched[0]["kb-id"]
+ group = next(g for g in group_entries_by_kb(_entries) if g[0]["kb-id"] == kb_id)
+
+ if dry_run:
+ result = await _run_kb_group(group, dry_run=True)
+ return {
+ "dry_run": True,
+ "name": identifier,
+ "kb_id": kb_id,
+ "sources": [e.get("source") for e in group],
+ "result": result,
+ }
+
+ asyncio.create_task(_run_kb_group(group))
+ return {
+ "triggered": True,
+ "name": identifier,
+ "kb_id": kb_id,
+ "sources": [e.get("source") for e in group],
+ }
# ── Scheduler ────────────────────────────────────────────────────
@@ -240,75 +262,61 @@ async def _send_notification(entry: dict, payload: dict) -> None:
async def _run_entry(entry: dict, dry_run: bool = False) -> dict | None:
- """Run a single sync for an entry.
+ """Run a single yaml entry (delegates to kb group sync)."""
+ return await _run_kb_group([entry], dry_run=dry_run)
+
- Uses a per-KB lock to prevent overlapping syncs to the same
- Knowledge Base (e.g. webhook fires while a scheduled sync is running).
- """
- kb_id = entry["kb-id"]
+async def _run_kb_group(entries: list[dict], dry_run: bool = False) -> dict | None:
+ """Run one or more sources into the same Knowledge Base."""
+ kb_id = entries[0]["kb-id"]
- # Get or create a lock for this KB.
if kb_id not in _sync_locks:
_sync_locks[kb_id] = asyncio.Lock()
lock = _sync_locks[kb_id]
if lock.locked():
- log.info(f"Skipping {entry.get('source', '?')} — sync already running for {kb_id}")
- return {"skipped": True, "reason": "sync already running"} if dry_run else None
+ from oikb.kb_sync import sources_label
+
+ log.info(
+ f"Waiting for {sources_label(entries)} — sync already running for {kb_id}"
+ )
async with lock:
- return await _run_entry_locked(entry, dry_run=dry_run)
+ return await _run_kb_group_locked(entries, dry_run=dry_run)
-async def _run_entry_locked(entry: dict, dry_run: bool = False) -> dict | None:
- """Inner sync logic, called under the per-KB lock."""
+async def _run_kb_group_locked(entries: list[dict], dry_run: bool = False) -> dict | None:
+ """Inner sync logic for a kb-id group."""
from oikb.cli import _make_client, _resolve_connector
- from oikb.sync import run_sync
+ from oikb.kb_sync import run_entries_sync, sources_label
- source = entry["source"]
- kb_id = entry["kb-id"]
+ label = sources_label(entries)
+ kb_id = entries[0]["kb-id"]
started_at = time.time()
- _scheduler_state[source] = {
- **_scheduler_state.get(source, {}),
- "name": entry.get("name", source),
- "status": "running",
- "started_at": started_at,
- }
+ for entry in entries:
+ source = entry["source"]
+ _scheduler_state[source] = {
+ **_scheduler_state.get(source, {}),
+ "name": entry.get("name", source),
+ "status": "running",
+ "started_at": started_at,
+ }
try:
- connector = _resolve_connector(
- source,
- branch=entry.get("branch"),
- path=entry.get("path"),
- )
client = _make_client(
- url=entry.get("url"),
- token=entry.get("token"),
+ url=entries[0].get("url"),
+ token=entries[0].get("token"),
)
- mf = None
- entry_filter = entry.get("filter", {})
- inc = entry_filter.get("include")
- exc = entry_filter.get("exclude")
- ms = entry_filter.get("max-size")
- if inc or exc or ms:
- from oikb.sync import build_manifest_filter, parse_size
- mf = build_manifest_filter(
- include=inc,
- exclude=exc,
- max_size=parse_size(ms),
- )
-
result = await asyncio.to_thread(
- run_sync,
- client=client,
- connector=connector,
- kb_id=kb_id,
+ run_entries_sync,
+ client,
+ entries,
+ resolve_connector=_resolve_connector,
dry_run=dry_run,
quiet=True,
- manifest_filter=mf,
- concurrency=entry.get("concurrency", 1),
+ concurrency=entries[0].get("concurrency", 1),
)
client.close()
@@ -325,119 +333,122 @@ async def _run_entry_locked(entry: dict, dry_run: bool = False) -> dict | None:
duration_ms = int(duration_s * 1000)
status = "success" if not result.errors else "partial"
- _scheduler_state[source] = {
- "name": entry.get("name", source),
- "status": status,
- "last_sync": time.time(),
- "duration_ms": duration_ms,
- "files_added": result.added,
- "files_modified": result.modified,
- "files_deleted": result.deleted,
- "unmodified": result.unmodified,
- "errors": result.errors or [],
- }
-
- record_sync(
- source=source,
- status=status,
- duration_seconds=duration_s,
- files_added=result.added,
- files_modified=result.modified,
- files_deleted=result.deleted,
- )
+ for entry in entries:
+ source = entry["source"]
+ _scheduler_state[source] = {
+ "name": entry.get("name", source),
+ "status": status,
+ "last_sync": time.time(),
+ "duration_ms": duration_ms,
+ "files_added": result.added,
+ "files_modified": result.modified,
+ "files_deleted": result.deleted,
+ "unmodified": result.unmodified,
+ "errors": result.errors or [],
+ }
- if _history:
- await asyncio.to_thread(
- _history.log,
+ record_sync(
source=source,
- kb_id=kb_id,
- status="success",
- started_at=started_at,
+ status=status,
+ duration_seconds=duration_s,
files_added=result.added,
files_modified=result.modified,
files_deleted=result.deleted,
- unmodified=result.unmodified,
)
+ if _history:
+ await asyncio.to_thread(
+ _history.log,
+ source=source,
+ kb_id=kb_id,
+ status="success",
+ started_at=started_at,
+ files_added=result.added,
+ files_modified=result.modified,
+ files_deleted=result.deleted,
+ unmodified=result.unmodified,
+ )
+
+ await _send_notification(entry, {
+ "source": source,
+ "kb_id": kb_id,
+ "status": status,
+ "duration_ms": duration_ms,
+ "summary": result.summary(),
+ "files_added": result.added,
+ "files_modified": result.modified,
+ "files_deleted": result.deleted,
+ "errors": result.errors or [],
+ })
+
log.info(
- f"Synced {source} -> {kb_id}: {result.summary()} ({duration_ms}ms)"
+ f"Synced {label} -> {kb_id}: {result.summary()} ({duration_ms}ms)"
)
- await _send_notification(entry, {
- "source": source,
- "kb_id": kb_id,
- "status": status,
- "duration_ms": duration_ms,
- "summary": result.summary(),
- "files_added": result.added,
- "files_modified": result.modified,
- "files_deleted": result.deleted,
- "errors": result.errors or [],
- })
-
except Exception as e:
- _scheduler_state[source] = {
- "name": entry.get("name", source),
- "status": "error",
- "last_sync": time.time(),
- "error": str(e),
- }
- record_sync(
- source=source,
- status="error",
- duration_seconds=time.time() - started_at,
- )
- if _history:
- await asyncio.to_thread(
- _history.log,
+ for entry in entries:
+ source = entry["source"]
+ _scheduler_state[source] = {
+ "name": entry.get("name", source),
+ "status": "error",
+ "last_sync": time.time(),
+ "error": str(e),
+ }
+ record_sync(
source=source,
- kb_id=kb_id,
status="error",
- started_at=started_at,
- error=str(e),
+ duration_seconds=time.time() - started_at,
)
- log.error(f"Sync failed for {source}: {e}")
-
- await _send_notification(entry, {
- "source": source,
- "kb_id": kb_id,
- "status": "error",
- "error": str(e),
- })
-
-
-async def _schedule_entry(entry: dict) -> None:
- """Run sync for one .oikb.yaml entry on a loop.
-
- Supports both simple intervals ('30m', '1h') and cron expressions
- ('0 */6 * * *', '0 6 * * 1-5').
- """
- raw_interval = entry.get("interval", "30m")
- source = entry.get("source", "unknown")
+ if _history:
+ await asyncio.to_thread(
+ _history.log,
+ source=source,
+ kb_id=kb_id,
+ status="error",
+ started_at=started_at,
+ error=str(e),
+ )
+ log.error(f"Sync failed for {source}: {e}")
+ await _send_notification(entry, {
+ "source": source,
+ "kb_id": kb_id,
+ "status": "error",
+ "error": str(e),
+ })
+
+
+async def _schedule_kb_group(entries: list[dict]) -> None:
+ """Run merged sync for all sources sharing a kb-id."""
+ from oikb.kb_sync import sources_label
+
+ raw_interval = entries[0].get("interval", "30m")
+ label = sources_label(entries)
use_cron = _is_cron(str(raw_interval))
if use_cron:
- log.info(f"Scheduling {source} with cron: {raw_interval}")
+ log.info(f"Scheduling {label} with cron: {raw_interval}")
else:
interval = parse_interval(raw_interval)
- log.info(f"Scheduling {source} every {interval}s")
+ log.info(f"Scheduling {label} every {interval}s")
while not _shutdown_event.is_set():
- await _run_entry(entry)
+ await _run_kb_group(entries)
- # Compute wait until next run.
if use_cron:
delay = _next_cron_delay(str(raw_interval))
- _scheduler_state.setdefault(source, {})["next_sync_in"] = f"{int(delay)}s"
else:
- delay = float(interval)
- _scheduler_state.setdefault(source, {})["next_sync_in"] = f"{interval}s"
+ delay = float(parse_interval(raw_interval))
+
+ for entry in entries:
+ _scheduler_state.setdefault(entry["source"], {})["next_sync_in"] = (
+ f"{int(delay)}s" if use_cron else f"{delay}s"
+ )
try:
await asyncio.wait_for(_shutdown_event.wait(), timeout=delay)
- break # Shutdown requested.
+ break
except asyncio.TimeoutError:
- pass # Time elapsed, run again.
+ pass
async def _run_scheduler(entries: list[dict]) -> None:
@@ -445,11 +456,14 @@ async def _run_scheduler(entries: list[dict]) -> None:
global _shutdown_event
_shutdown_event = asyncio.Event()
+ from oikb.kb_sync import group_entries_by_kb
+
loop = asyncio.get_event_loop()
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(sig, _shutdown_event.set)
- tasks = [asyncio.create_task(_schedule_entry(e)) for e in entries]
+ groups = group_entries_by_kb(entries)
+ tasks = [asyncio.create_task(_schedule_kb_group(group)) for group in groups]
await _shutdown_event.wait()
for task in tasks:
diff --git a/src/oikb/kb_sync.py b/src/oikb/kb_sync.py
new file mode 100644
index 0000000..ade2737
--- /dev/null
+++ b/src/oikb/kb_sync.py
@@ -0,0 +1,109 @@
+"""Helpers for syncing one or more .oikb.yaml entries into a Knowledge Base."""
+
+from __future__ import annotations
+
+from collections import OrderedDict
+from typing import Any, Callable
+
+from oikb.connectors import BaseConnector, ManifestEntry
+from oikb.connectors.composite import CompositeConnector
+from oikb.sync import SyncResult, build_manifest_filter, parse_size, run_sync
+
+
+def group_entries_by_kb(entries: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
+ """Group yaml entries that share the same kb-id."""
+ groups: OrderedDict[str, list[dict[str, Any]]] = OrderedDict()
+ for entry in entries:
+ groups.setdefault(entry["kb-id"], []).append(entry)
+ return list(groups.values())
+
+
+def manifest_filter_for_entry(
+ entry: dict[str, Any],
+ max_file_size: str | None = None,
+) -> Callable | None:
+ entry_filter = entry.get("filter", {})
+ include = entry_filter.get("include")
+ exclude = entry_filter.get("exclude")
+ max_size = entry_filter.get("max-size") or max_file_size
+ if not include and not exclude and not max_size:
+ return None
+ return build_manifest_filter(
+ include=include,
+ exclude=exclude,
+ max_size=parse_size(max_size),
+ )
+
+
+def build_connector_for_entries(
+ entries: list[dict[str, Any]],
+ resolve_connector: Callable[..., BaseConnector],
+ max_file_size: str | None = None,
+) -> BaseConnector:
+ """Build a single connector for one yaml entry or a merged composite."""
+ if len(entries) == 1:
+ entry = entries[0]
+ return resolve_connector(
+ entry["source"],
+ entry.get("branch"),
+ entry.get("path"),
+ )
+
+ parts: list[tuple[BaseConnector, list[ManifestEntry]]] = []
+ for entry in entries:
+ connector = resolve_connector(
+ entry["source"],
+ entry.get("branch"),
+ entry.get("path"),
+ )
+ manifest = connector.build_manifest()
+ manifest_filter = manifest_filter_for_entry(entry, max_file_size)
+ if manifest_filter:
+ manifest = manifest_filter(manifest)
+ parts.append((connector, manifest))
+
+ return CompositeConnector(parts)
+
+
+def sources_label(entries: list[dict[str, Any]]) -> str:
+ if len(entries) == 1:
+ return entries[0].get("source", "?")
+ return "+".join(entry.get("source", "?") for entry in entries)
+
+
+def run_entries_sync(
+ client: Any,
+ entries: list[dict[str, Any]],
+ *,
+ resolve_connector: Callable[..., BaseConnector],
+ dry_run: bool = False,
+ verbose: bool = False,
+ quiet: bool = False,
+ concurrency: int = 1,
+ max_file_size: str | None = None,
+) -> SyncResult:
+ """Sync one kb-id group (single source or merged composite)."""
+ if len(entries) == 1:
+ entry = entries[0]
+ connector = build_connector_for_entries(entries, resolve_connector, max_file_size)
+ return run_sync(
+ client=client,
+ connector=connector,
+ kb_id=entry["kb-id"],
+ dry_run=dry_run,
+ verbose=verbose,
+ quiet=quiet,
+ manifest_filter=manifest_filter_for_entry(entry, max_file_size),
+ concurrency=entry.get("concurrency", concurrency),
+ )
+
+ connector = build_connector_for_entries(entries, resolve_connector, max_file_size)
+ return run_sync(
+ client=client,
+ connector=connector,
+ kb_id=entries[0]["kb-id"],
+ dry_run=dry_run,
+ verbose=verbose,
+ quiet=quiet,
+ concurrency=max(entry.get("concurrency", concurrency) for entry in entries),
+ )
diff --git a/src/oikb/sync.py b/src/oikb/sync.py
index 0842d3f..f219221 100644
--- a/src/oikb/sync.py
+++ b/src/oikb/sync.py
@@ -347,7 +347,7 @@ def _upload_one(
time.sleep(2 ** attempt)
last_err = e
continue
- last_err = e
+ last_err = _http_error_detail(e)
break
except Exception as e:
last_err = e
@@ -412,6 +412,20 @@ def _tally(outcome: str | None) -> None:
return result
+def _http_error_detail(exc: httpx.HTTPStatusError) -> Exception:
+ """Attach Open WebUI response body to upload errors when available."""
+ detail = exc.response.text.strip()
+ try:
+ payload = exc.response.json()
+ if isinstance(payload, dict) and payload.get("detail"):
+ detail = str(payload["detail"])
+ except Exception:
+ pass
+ if detail:
+ return RuntimeError(f"{exc} — {detail}")
+ return exc
+
+
def _echo_file_entry(entry: dict, prefix: str, color: str) -> None:
"""Print a file entry with color."""
path = entry.get("path", "")