From 65ecd1abd9a1eef5de4bdada5ee8fa63e136a938 Mon Sep 17 00:00:00 2001 From: tedk0n <4rfer0@gmail.com> Date: Fri, 5 Jun 2026 03:07:46 +0200 Subject: [PATCH 1/4] add supporting confluence api v1 and authorization by bearer token --- src/oikb/connectors/confluence.py | 152 ++++++++++++++++++++++-------- 1 file changed, 115 insertions(+), 37 deletions(-) diff --git a/src/oikb/connectors/confluence.py b/src/oikb/connectors/confluence.py index c7c2d28..0da0ea4 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,6 +21,23 @@ from oikb.connectors import BaseConnector, ManifestEntry +BASE_ENDPOINTS = { + "v1": "/rest/api", + "v2": "/wiki/api/v2", +} + + +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) -> str: """Convert Confluence storage format (XHTML) to plain text.""" # Strip all HTML tags. @@ -28,13 +49,14 @@ def _storage_to_text(storage_html: str) -> str: 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 +65,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,11 +87,15 @@ 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. @@ -73,6 +103,42 @@ def __init__( 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] = [] + start = 0 + limit = 250 + + while True: + params: dict[str, Any] = { + "spaceKey": self.space_key, + "type": "page", + "limit": limit, + "start": start, + } + + 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: + self._add_page_entry(entries, page) + + 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]: entries: list[ManifestEntry] = [] cursor = None @@ -82,36 +148,14 @@ 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" - - entries.append( - ManifestEntry( - filename=filename, - path="", - checksum=checksum, - size=0, - ) - ) - - # Store page ID for later retrieval. - self._page_cache[filename] = page_id + self._add_page_entry(entries, page) # Handle pagination. next_link = data.get("_links", {}).get("next") @@ -126,16 +170,50 @@ def build_manifest(self) -> list[ManifestEntry]: entries.sort(key=lambda e: e.display_path) return entries + def _add_page_entry( + self, entries: list[ManifestEntry], page: dict[str, Any] + ) -> None: + 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" + + entries.append( + ManifestEntry( + filename=filename, + path="", + checksum=checksum, + size=0, + ) + ) + + # Store page ID for later retrieval. + self._page_cache[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) if not page_id: raise FileNotFoundError(f"Page not found: {filename}") - resp = self._http.get( - f"/api/v2/pages/{page_id}", - params={"body-format": "storage"}, - ) + 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.raise_for_status() data = resp.json() From 624a45132488ebb6535d634c097b96d14a7bc496 Mon Sep 17 00:00:00 2001 From: tedk0n <4rfer0@gmail.com> Date: Mon, 15 Jun 2026 17:17:50 +0200 Subject: [PATCH 2/4] Fix ixport empty pages to owui, fix duplication filename if they in seperate spaces --- src/oikb/connectors/confluence.py | 165 +++++++++++++++++++++++++----- src/oikb/sync.py | 16 ++- 2 files changed, 155 insertions(+), 26 deletions(-) diff --git a/src/oikb/connectors/confluence.py b/src/oikb/connectors/confluence.py index 0da0ea4..3878bc3 100644 --- a/src/oikb/connectors/confluence.py +++ b/src/oikb/connectors/confluence.py @@ -27,6 +27,53 @@ } +_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: @@ -38,14 +85,58 @@ def _parse_api_version(api_version: str | None) -> str: return version -def _storage_to_text(storage_html: str) -> str: - """Convert Confluence storage format (XHTML) to plain text.""" - # Strip all HTML tags. +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() class ConfluenceConnector(BaseConnector): @@ -98,8 +189,8 @@ def __init__( 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.""" @@ -109,6 +200,7 @@ def build_manifest(self) -> list[ManifestEntry]: def _build_manifest_v1(self) -> list[ManifestEntry]: entries: list[ManifestEntry] = [] + used_keys: set[tuple[str, str]] = set() start = 0 limit = 250 @@ -118,6 +210,7 @@ def _build_manifest_v1(self) -> list[ManifestEntry]: "type": "page", "limit": limit, "start": start, + "expand": "ancestors,version", } resp = self._http.get("/content", params=params) @@ -129,7 +222,8 @@ def _build_manifest_v1(self) -> list[ManifestEntry]: break for page in results: - self._add_page_entry(entries, page) + dir_path = _ancestor_dir_path_v1(page) + self._add_page_entry(entries, page, dir_path, used_keys) if len(results) < limit: break @@ -139,7 +233,7 @@ def _build_manifest_v1(self) -> list[ManifestEntry]: return entries def _build_manifest_v2(self) -> list[ManifestEntry]: - entries: list[ManifestEntry] = [] + all_pages: list[dict[str, Any]] = [] cursor = None while True: @@ -154,54 +248,75 @@ def _build_manifest_v2(self) -> list[ManifestEntry]: resp.raise_for_status() data = resp.json() - for page in data.get("results", []): - self._add_page_entry(entries, page) + all_pages.extend(data.get("results", [])) - # 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 _add_page_entry( - self, entries: list[ManifestEntry], page: dict[str, Any] + self, + entries: list[ManifestEntry], + page: dict[str, Any], + dir_path: str, + used_keys: set[tuple[str, str]], ) -> None: - page_id = page["id"] - title = page["title"] + page_id = str(page["id"]) 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" + dir_path, filename = self._unique_path_filename(page, dir_path, used_keys) entries.append( ManifestEntry( filename=filename, - path="", + path=dir_path, checksum=checksum, size=0, ) ) - # Store page ID for later retrieval. - self._page_cache[filename] = page_id + 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( @@ -218,7 +333,7 @@ def read_file(self, path: str, filename: str) -> bytes: data = resp.json() storage = data.get("body", {}).get("storage", {}).get("value", "") - text = _storage_to_text(storage) + text = _storage_to_text(storage, title=data.get("title", "")) return text.encode("utf-8") def close(self) -> None: 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", "") From 1be3084874f44bd804000887c2966628b40f4d86 Mon Sep 17 00:00:00 2001 From: tedk0n <4rfer0@gmail.com> Date: Mon, 15 Jun 2026 18:33:17 +0200 Subject: [PATCH 3/4] add support for multiple sources into one Knoledge base --- src/oikb/cli.py | 52 +++--- src/oikb/connectors/composite.py | 46 +++++ src/oikb/connectors/confluence.py | 38 +++- src/oikb/daemon.py | 292 ++++++++++++++++-------------- src/oikb/kb_sync.py | 109 +++++++++++ 5 files changed, 367 insertions(+), 170 deletions(-) create mode 100644 src/oikb/connectors/composite.py create mode 100644 src/oikb/kb_sync.py diff --git a/src/oikb/cli.py b/src/oikb/cli.py index b7bffbc..a362dd0 100644 --- a/src/oikb/cli.py +++ b/src/oikb/cli.py @@ -397,46 +397,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 3878bc3..67d54f4 100644 --- a/src/oikb/connectors/confluence.py +++ b/src/oikb/connectors/confluence.py @@ -139,6 +139,27 @@ def _storage_to_text(storage_html: str, title: str = "") -> str: return title.strip() +def _finalize_page_text( + text: str, + *, + title: str, + page_id: str, + space_key: str, + base_url: str, +) -> str: + """Ensure non-empty, unique text for Open WebUI vector deduplication. + + OWUI hashes extracted content and rejects duplicates. Multiple Confluence + pages (empty bodies, identical index pages, same content across spaces) + would otherwise share the same hash — including e3b0c442… for "". + """ + body = text.strip() or title.strip() or f"Confluence page {page_id}" + source = f"confluence:{space_key}:{page_id}" + if base_url: + source = f"{source} {base_url.rstrip('/')}/pages/viewpage.action?pageId={page_id}" + return f"{body}\n\n---\n{source}" + + class ConfluenceConnector(BaseConnector): """Sync pages from a Confluence space. @@ -285,6 +306,12 @@ def _unique_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], @@ -299,6 +326,7 @@ def _add_page_entry( 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( @@ -333,7 +361,15 @@ def read_file(self, path: str, filename: str) -> bytes: data = resp.json() storage = data.get("body", {}).get("storage", {}).get("value", "") - text = _storage_to_text(storage, title=data.get("title", "")) + 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, + ) 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), + ) From c94ddab1dc6b05bd0e96c834a32b76ba9080596a Mon Sep 17 00:00:00 2001 From: tedk0n <4rfer0@gmail.com> Date: Mon, 15 Jun 2026 19:09:16 +0200 Subject: [PATCH 4/4] fix duplication --- src/oikb/cli.py | 7 +++++++ src/oikb/connectors/confluence.py | 24 ++++++++++++++++++------ 2 files changed, 25 insertions(+), 6 deletions(-) diff --git a/src/oikb/cli.py b/src/oikb/cli.py index a362dd0..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) diff --git a/src/oikb/connectors/confluence.py b/src/oikb/connectors/confluence.py index 67d54f4..c413b83 100644 --- a/src/oikb/connectors/confluence.py +++ b/src/oikb/connectors/confluence.py @@ -146,18 +146,28 @@ def _finalize_page_text( 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 extracted content and rejects duplicates. Multiple Confluence - pages (empty bodies, identical index pages, same content across spaces) - would otherwise share the same hash — including e3b0c442… for "". + 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}" - source = f"confluence:{space_key}:{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"{source} {base_url.rstrip('/')}/pages/viewpage.action?pageId={page_id}" - return f"{body}\n\n---\n{source}" + source = ( + f"{uid} {base_url.rstrip('/')}/pages/viewpage.action?pageId={page_id}" + ) + return f"{header}\n{body}\n\n---\n{source}" class ConfluenceConnector(BaseConnector): @@ -369,6 +379,8 @@ def read_file(self, path: str, filename: str) -> bytes: page_id=page_id, space_key=self.space_key, base_url=self._base_url, + path=path, + filename=filename, ) return text.encode("utf-8")