This commit is contained in:
Timothy Jaeryang Baek 2026-05-20 13:19:16 +04:00
commit e3c59f1b03
22 changed files with 2071 additions and 0 deletions

9
.dockerignore Normal file
View file

@ -0,0 +1,9 @@
.git
.github
.venv
__pycache__
*.pyc
.pytest_cache
dist
*.egg-info
.DS_Store

133
.github/workflows/docker.yml vendored Normal file
View file

@ -0,0 +1,133 @@
name: Build Docker Image
on:
push:
branches: [main]
pull_request:
branches: [main]
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}
jobs:
build:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
platform:
- linux/amd64
- linux/arm64
permissions:
contents: read
packages: write
steps:
- uses: actions/checkout@v4
- uses: docker/setup-qemu-action@v3
- uses: docker/setup-buildx-action@v3
- uses: docker/login-action@v3
if: github.event_name != 'pull_request'
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- uses: docker/metadata-action@v5
id: meta
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
- uses: docker/build-push-action@v5
id: build
with:
context: .
platforms: ${{ matrix.platform }}
labels: ${{ steps.meta.outputs.labels }}
outputs: type=image,name=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=${{ github.event_name != 'pull_request' }}
cache-from: type=gha,scope=${{ matrix.platform }}
cache-to: type=gha,scope=${{ matrix.platform }},mode=max
- name: Export digest
if: github.event_name != 'pull_request'
run: |
mkdir -p /tmp/digests
digest="${{ steps.build.outputs.digest }}"
touch "/tmp/digests/${digest#sha256:}"
- uses: actions/upload-artifact@v4
if: github.event_name != 'pull_request'
with:
name: digests-${{ matrix.platform == 'linux/amd64' && 'amd64' || 'arm64' }}
path: /tmp/digests/*
if-no-files-found: error
retention-days: 1
merge:
runs-on: ubuntu-latest
if: github.event_name != 'pull_request'
needs: build
permissions:
contents: read
packages: write
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Get version from pyproject.toml
id: version
run: |
VERSION=$(python3 -c "import tomllib; print(tomllib.load(open('pyproject.toml','rb'))['project']['version'])")
echo "version=$VERSION" >> "$GITHUB_OUTPUT"
MAJOR_MINOR=$(echo "$VERSION" | cut -d. -f1,2)
echo "major_minor=$MAJOR_MINOR" >> "$GITHUB_OUTPUT"
if git rev-parse "refs/tags/v$VERSION" >/dev/null 2>&1; then
TAG_SHA=$(git rev-list -n 1 "v$VERSION")
if [ "$TAG_SHA" = "${{ github.sha }}" ]; then
echo "is_new_version=true" >> "$GITHUB_OUTPUT"
else
echo "is_new_version=false" >> "$GITHUB_OUTPUT"
fi
else
echo "is_new_version=true" >> "$GITHUB_OUTPUT"
fi
- uses: actions/download-artifact@v4
with:
path: /tmp/digests
pattern: digests-*
merge-multiple: true
- uses: docker/setup-buildx-action@v3
- uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- uses: docker/metadata-action@v5
id: meta
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
tags: |
type=raw,value=latest,enable={{is_default_branch}}
type=raw,value=${{ steps.version.outputs.version }},enable=${{ steps.version.outputs.is_new_version }}
type=raw,value=${{ steps.version.outputs.major_minor }},enable=${{ steps.version.outputs.is_new_version }}
type=sha
- name: Create manifest list and push
working-directory: /tmp/digests
run: |
docker buildx imagetools create $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< "$DOCKER_METADATA_OUTPUT_JSON") \
$(printf '${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}@sha256:%s ' *)
- name: Inspect image
run: |
docker buildx imagetools inspect ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ steps.meta.outputs.version }}

51
.github/workflows/release.yml vendored Normal file
View file

@ -0,0 +1,51 @@
name: Release
on:
push:
branches: [main]
paths:
- pyproject.toml
permissions:
contents: write
jobs:
release:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Get version from pyproject.toml
id: version
run: |
VERSION=$(python3 -c "import tomllib; print(tomllib.load(open('pyproject.toml','rb'))['project']['version'])")
echo "version=$VERSION" >> "$GITHUB_OUTPUT"
echo "tag=v$VERSION" >> "$GITHUB_OUTPUT"
- name: Check if tag exists
id: check
run: |
if git rev-parse "refs/tags/${{ steps.version.outputs.tag }}" >/dev/null 2>&1; then
echo "exists=true" >> "$GITHUB_OUTPUT"
else
echo "exists=false" >> "$GITHUB_OUTPUT"
fi
- name: Extract changelog
if: steps.check.outputs.exists == 'false'
run: |
VERSION="${{ steps.version.outputs.version }}"
awk "/^## \[${VERSION}\]/{found=1; next} /^## \[/{if(found) exit} found{print}" CHANGELOG.md > /tmp/release-notes.md
- name: Create tag and GitHub Release
if: steps.check.outputs.exists == 'false'
env:
GH_TOKEN: ${{ github.token }}
run: |
git tag "${{ steps.version.outputs.tag }}"
git push origin "${{ steps.version.outputs.tag }}"
gh release create "${{ steps.version.outputs.tag }}" \
--title "${{ steps.version.outputs.tag }}" \
--notes-file /tmp/release-notes.md

13
.gitignore vendored Normal file
View file

@ -0,0 +1,13 @@
__pycache__/
*.pyc
*.pyo
*.egg-info/
dist/
build/
.venv/
.eggs/
*.egg
.pytest_cache/
.DS_Store
Thumbs.db
uv.lock

1
.python-version Normal file
View file

@ -0,0 +1 @@
3.11

24
CHANGELOG.md Normal file
View file

@ -0,0 +1,24 @@
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
## [0.1.0] - Unreleased
### Added
- 🎉 **Initial release** of oikb, a CLI tool for syncing content to Open WebUI Knowledge Bases.
- 🔄 **Incremental sync** (`oikb sync`) — SHA-256 diffing uploads only new and modified files. Supports local directories, GitHub repos, and S3 buckets.
- 👀 **Dry-run preview** (`oikb diff`) — see what would change without uploading.
- 👁️ **Watch mode** (`oikb watch`) — auto-sync on file changes via watchdog with configurable debounce.
- 📋 **List files** (`oikb ls`) — list files in a Knowledge Base.
- ℹ️ **Status** (`oikb status`) — show KB info, file count, and total size.
- 🗑️ **Reset** (`oikb reset`) — delete all files in a Knowledge Base.
- ⚙️ **Configuration** (`oikb config`) — manage URL and API key via config file, env vars (`OPEN_WEBUI_URL`, `OPEN_WEBUI_API_KEY`), or CLI flags.
- 📁 **Declarative config** (`.oikb.yaml`) — multi-source sync from a single project file. `oikb sync` with no args reads it automatically.
- 🙈 **Ignore patterns** (`.oikbignore`) — gitignore-style file exclusion for local directory syncs.
- 🐙 **GitHub connector** (`github:owner/repo`) — sync via GitHub Trees API using blob SHAs as checksums. No local clone needed. Branch and path filtering supported.
- ☁️ **S3 connector** (`s3://bucket/prefix`) — sync via boto3 using ETags as checksums. Paginated listing for large buckets.
- 🐳 **Docker image** — multi-stage build with multi-arch support (amd64 + arm64). Mount source at `/data`, entrypoint is `oikb`.
- 🚀 **CI/CD** — GitHub Actions workflows for automated Docker image builds (GHCR) and GitHub Releases with changelog extraction.

24
Dockerfile Normal file
View file

@ -0,0 +1,24 @@
# ── Build ──
FROM python:3.12-slim AS builder
WORKDIR /app
COPY . .
RUN pip install --no-cache-dir uv && \
uv build --wheel
# ── Runtime ──
FROM python:3.12-slim
LABEL org.opencontainers.image.source="https://github.com/open-webui/oikb"
LABEL org.opencontainers.image.description="CLI tool for syncing content to Open WebUI Knowledge Bases"
# Install oikb from the built wheel.
COPY --from=builder /app/dist/*.whl /tmp/
RUN pip install --no-cache-dir /tmp/*.whl && rm /tmp/*.whl
# Sync source is mounted at /data by convention.
VOLUME ["/data"]
WORKDIR /data
ENTRYPOINT ["oikb"]

21
LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 Open WebUI Inc.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

198
README.md Normal file
View file

@ -0,0 +1,198 @@
# 📚 oikb
A CLI tool that syncs content to [Open WebUI](https://github.com/open-webui/open-webui) Knowledge Bases. Supports local directories, GitHub repos, and S3 buckets.
## Why oikb?
Manually uploading files to Knowledge Bases is tedious. oikb automates it with incremental SHA-256 diffing, so only new and modified files are uploaded. Directory structure is mirrored automatically. Run it once, run it in CI, or leave it watching for changes.
## Getting Started
### Docker (recommended)
```bash
docker run --rm \
-e OPEN_WEBUI_URL=http://host.docker.internal:3000 \
-e OPEN_WEBUI_API_KEY=sk-your-key \
-v ./docs:/data \
ghcr.io/open-webui/oikb sync /data --kb your-kb-id
```
### pip
```bash
pip install oikb
```
Then set your env vars and sync:
```bash
export OPEN_WEBUI_URL=http://localhost:3000
export OPEN_WEBUI_API_KEY=sk-your-api-key
oikb sync ./docs --kb your-kb-id
```
> [!TIP]
> No config file needed. Env vars are enough. You can also use `oikb config set url ...` and `oikb config set token ...` to save them to `~/.config/oikb/config.yaml`.
## Commands
```bash
# Sync a local directory
oikb sync ./docs --kb your-kb-id
# Sync a GitHub repo (no clone needed)
oikb sync github:owner/repo --kb your-kb-id
# Sync an S3 bucket
oikb sync s3://bucket/prefix --kb your-kb-id
# Preview changes without uploading
oikb diff ./docs --kb your-kb-id
# Watch for changes and auto-sync
oikb watch ./docs --kb your-kb-id
# List files in a KB
oikb ls --kb your-kb-id
# Show KB info
oikb status --kb your-kb-id
# Reset a KB
oikb reset --kb your-kb-id
```
### Sources
| Source | Syntax | Install |
|---|---|---|
| Local directory | `./docs`, `/path/to/dir` | included |
| GitHub repo | `github:owner/repo` | included |
| S3 bucket | `s3://bucket/prefix` | `pip install oikb[s3]` |
GitHub sources support `--branch` and `--path` filtering:
```bash
oikb sync github:owner/repo --kb your-kb-id --branch main --path docs/
```
## Configuration
Settings are resolved in this order (highest priority wins):
1. **CLI flags** (`--url`, `--token`)
2. **Environment variables** (`OPEN_WEBUI_URL`, `OPEN_WEBUI_API_KEY`)
3. **Config file** (`~/.config/oikb/config.yaml`)
```bash
oikb config set url http://localhost:3000
oikb config set token sk-your-api-key
oikb config get
```
## Declarative Config (`.oikb.yaml`)
For recurring multi-source syncs, place a `.oikb.yaml` in your project root:
```yaml
sync:
- source: ./docs
kb: project-docs
- source: github:owner/wiki
kb: team-wiki
branch: main
- source: s3://company-docs/engineering
kb: eng-handbook
```
```bash
# Sync all entries
oikb sync
# Sync a specific entry
oikb sync --name project-docs
```
## `.oikbignore`
Place a `.oikbignore` file in your source directory to exclude files (gitignore-style):
```
dist/
build/
*.pyc
*.draft.*
*.zip
```
## Docker
The published image is available at `ghcr.io/open-webui/oikb`.
```bash
# One-shot sync
docker run --rm \
-e OPEN_WEBUI_URL=http://host.docker.internal:3000 \
-e OPEN_WEBUI_API_KEY=sk-your-key \
-v ./docs:/data \
ghcr.io/open-webui/oikb sync /data --kb your-kb-id
# Watch mode (keep running)
docker run --rm -d \
-e OPEN_WEBUI_URL=http://host.docker.internal:3000 \
-e OPEN_WEBUI_API_KEY=sk-your-key \
-v ./docs:/data \
ghcr.io/open-webui/oikb watch /data --kb your-kb-id
```
### Docker Compose
Run as a sidecar alongside Open WebUI. Watches a mounted directory and auto-syncs:
```yaml
services:
open-webui:
image: ghcr.io/open-webui/open-webui:main
ports:
- "3000:8080"
oikb:
image: ghcr.io/open-webui/oikb:latest
environment:
- OPEN_WEBUI_URL=http://open-webui:8080
- OPEN_WEBUI_API_KEY=${OPEN_WEBUI_API_KEY}
volumes:
- ./docs:/data
command: watch /data --kb ${KB_ID}
depends_on:
- open-webui
restart: unless-stopped
```
### GitHub Actions
```yaml
- name: Sync docs to Open WebUI
uses: docker://ghcr.io/open-webui/oikb:latest
with:
args: sync /github/workspace/docs --kb ${{ secrets.KB_ID }}
env:
OPEN_WEBUI_URL: ${{ secrets.OPEN_WEBUI_URL }}
OPEN_WEBUI_API_KEY: ${{ secrets.OPEN_WEBUI_API_KEY }}
```
## How It Works
1. Scan source (local dir, GitHub, S3), compute SHA-256 checksums
2. Send manifest to Open WebUI's `/sync/diff` endpoint
3. Server diffs against stored file hashes
4. Delete stale files, create missing directories
5. Upload only new and modified files
## License
MIT. See [LICENSE](LICENSE) for details.

23
docker-compose.yaml Normal file
View file

@ -0,0 +1,23 @@
services:
open-webui:
image: ghcr.io/open-webui/open-webui:main
ports:
- "3000:8080"
volumes:
- open-webui:/app/backend/data
# One-shot sync on container start, then watch for changes
oikb:
image: ghcr.io/open-webui/oikb:latest
environment:
- OPEN_WEBUI_URL=http://open-webui:8080
- OPEN_WEBUI_API_KEY=${OPEN_WEBUI_API_KEY}
volumes:
- ./docs:/data
command: watch /data --kb ${KB_ID}
depends_on:
- open-webui
restart: unless-stopped
volumes:
open-webui:

29
pyproject.toml Normal file
View file

@ -0,0 +1,29 @@
[project]
name = "oikb"
version = "0.1.0"
description = "CLI tool for syncing content to Open WebUI Knowledge Bases"
readme = "README.md"
authors = [
{ name = "Tim Baek", email = "tim@openwebui.com" }
]
requires-python = ">=3.11"
license = {file = "LICENSE"}
dependencies = [
"click>=8.1",
"httpx>=0.27",
"pyyaml>=6.0",
"watchdog>=4.0",
]
[project.optional-dependencies]
github = ["httpx"]
s3 = ["boto3>=1.34"]
all = ["boto3>=1.34"]
dev = ["pytest>=8.0", "pytest-asyncio>=0.23", "respx>=0.21"]
[project.scripts]
oikb = "oikb.cli:cli"
[build-system]
requires = ["uv_build>=0.8.0,<0.9"]
build-backend = "uv_build"

3
src/oikb/__init__.py Normal file
View file

@ -0,0 +1,3 @@
"""oikb — Open WebUI Knowledge Base CLI."""
__version__ = "0.1.0"

515
src/oikb/cli.py Normal file
View file

@ -0,0 +1,515 @@
"""oikb CLI — command-line interface for Open WebUI Knowledge Base sync."""
from __future__ import annotations
import sys
from pathlib import Path
import click
from oikb import __version__
from oikb.config import get_config, resolve_token, resolve_url, set_config
@click.group()
@click.version_option(version=__version__, prog_name="oikb")
@click.option("-q", "--quiet", is_flag=True, default=False, help="Suppress non-error output.")
@click.pass_context
def cli(ctx: click.Context, quiet: bool):
"""oikb — sync content to Open WebUI Knowledge Bases."""
ctx.ensure_object(dict)
ctx.obj["quiet"] = quiet
# ── Common options ──────────────────────────────────────────────
def common_options(f):
"""Shared --url, --token, --kb options."""
f = click.option("--url", default=None, envvar="OPEN_WEBUI_URL", help="Open WebUI base URL.")(f)
f = click.option("--token", default=None, envvar="OPEN_WEBUI_API_KEY", help="API key.")(f)
f = click.option("--kb", default=None, help="Knowledge Base ID.")(f)
return f
def _make_client(url: str | None, token: str | None):
"""Create an OikbClient from resolved config."""
from oikb.client import OikbClient
return OikbClient(
base_url=resolve_url(url),
token=resolve_token(token),
)
def _resolve_connector(source: str, branch: str | None = None, path: str | None = None):
"""Resolve a source string to the appropriate connector.
Supports:
- Local paths: ./docs, /absolute/path, relative/path
- GitHub: github:owner/repo[/subdir]
- S3: s3://bucket[/prefix]
"""
if source.startswith("github:"):
from oikb.connectors.github import GitHubConnector, parse_github_source
parsed = parse_github_source(source)
return GitHubConnector(
owner=parsed["owner"],
repo=parsed["repo"],
branch=branch,
path=path or parsed.get("path"),
)
if source.startswith("s3://"):
from oikb.connectors.s3 import S3Connector, parse_s3_source
parsed = parse_s3_source(source)
return S3Connector(
bucket=parsed["bucket"],
prefix=path or parsed.get("prefix"),
)
# Default: local filesystem.
from oikb.connectors.filesystem import FilesystemConnector
return FilesystemConnector(source)
def _load_oikb_yaml() -> list[dict] | None:
"""Load .oikb.yaml from the current directory if it exists."""
import yaml
yaml_path = Path(".oikb.yaml")
if not yaml_path.exists():
return None
with open(yaml_path) as f:
data = yaml.safe_load(f)
if not data or "sync" not in data:
return None
return data["sync"]
# ── sync ────────────────────────────────────────────────────────
@cli.command()
@click.argument("source", required=False)
@common_options
@click.option("--branch", default=None, help="Branch for GitHub sources.")
@click.option("--path", "source_path", default=None, help="Subdirectory within the source.")
@click.option("--dry-run", is_flag=True, help="Preview changes without uploading.")
@click.option("-v", "--verbose", is_flag=True, help="Show detailed progress.")
@click.option("--name", default=None, help="Target a specific entry in .oikb.yaml by name/kb.")
@click.pass_context
def sync(
ctx: click.Context,
source: str | None,
url: str | None,
token: str | None,
kb: str | None,
branch: str | None,
source_path: str | None,
dry_run: bool,
verbose: bool,
name: str | None,
):
"""Incremental sync from a source to a Knowledge Base.
SOURCE can be a local directory, github:owner/repo, or s3://bucket/prefix.
If omitted, reads .oikb.yaml from the current directory.
"""
quiet = ctx.obj.get("quiet", False)
from oikb.sync import run_sync
# ── .oikb.yaml mode ──
if source is None:
entries = _load_oikb_yaml()
if not entries:
click.echo(
click.style("No source specified and no .oikb.yaml found.", fg="red"),
err=True,
)
sys.exit(1)
# Filter by --name if specified.
if name:
entries = [e for e in entries if e.get("kb") == name or e.get("source") == name]
if not entries:
click.echo(click.style(f"No entry matching '{name}' in .oikb.yaml", fg="red"), err=True)
sys.exit(1)
has_errors = False
for entry in entries:
entry_source = entry.get("source")
entry_kb = entry.get("kb")
entry_branch = entry.get("branch")
entry_path = entry.get("path")
if not entry_source or not entry_kb:
click.echo(click.style(f"Skipping invalid entry: {entry}", fg="yellow"), err=True)
continue
if not quiet:
click.echo(f"\n{'─' * 40}")
click.echo(f"Syncing: {entry_source} → {entry_kb}")
try:
connector = _resolve_connector(entry_source, entry_branch, entry_path)
client = _make_client(url, token)
result = run_sync(
client=client,
connector=connector,
kb_id=entry_kb,
dry_run=dry_run,
verbose=verbose,
quiet=quiet,
)
client.close()
if not quiet:
prefix = "Dry run" if dry_run else "Done"
click.echo(f" {prefix}: {result.summary()}")
if result.errors:
has_errors = True
except Exception as e:
click.echo(click.style(f" Failed: {e}", fg="red"), err=True)
has_errors = True
if has_errors:
sys.exit(1)
return
# ── Single source mode ──
if not kb:
click.echo(click.style("--kb is required when syncing a single source.", fg="red"), err=True)
sys.exit(1)
try:
connector = _resolve_connector(source, branch, source_path)
except (FileNotFoundError, ImportError, ValueError) as e:
click.echo(click.style(f"Error: {e}", fg="red"), err=True)
sys.exit(1)
try:
client = _make_client(url, token)
except ValueError as e:
click.echo(click.style(str(e), fg="red"), err=True)
sys.exit(1)
try:
result = run_sync(
client=client,
connector=connector,
kb_id=kb,
dry_run=dry_run,
verbose=verbose,
quiet=quiet,
)
except Exception as e:
click.echo(click.style(f"Sync failed: {e}", fg="red"), err=True)
sys.exit(1)
finally:
client.close()
if not quiet:
prefix = "Dry run" if dry_run else "Sync complete"
click.echo(f"{prefix}: {result.summary()}")
if result.errors:
click.echo(click.style(f"\n{len(result.errors)} error(s):", fg="red"), err=True)
for err in result.errors:
click.echo(f" • {err}", err=True)
sys.exit(1)
# ── diff ────────────────────────────────────────────────────────
@cli.command()
@click.argument("source")
@common_options
@click.option("--branch", default=None, help="Branch for GitHub sources.")
@click.option("--path", "source_path", default=None, help="Subdirectory within the source.")
@click.option("-v", "--verbose", is_flag=True, help="Show detailed output.")
@click.pass_context
def diff(
ctx: click.Context,
source: str,
url: str | None,
token: str | None,
kb: str | None,
branch: str | None,
source_path: str | None,
verbose: bool,
):
"""Preview what a sync would do (alias for sync --dry-run)."""
if not kb:
click.echo(click.style("--kb is required.", fg="red"), err=True)
sys.exit(1)
from oikb.sync import run_sync
try:
connector = _resolve_connector(source, branch, source_path)
except (FileNotFoundError, ImportError, ValueError) as e:
click.echo(click.style(f"Error: {e}", fg="red"), err=True)
sys.exit(1)
try:
client = _make_client(url, token)
except ValueError as e:
click.echo(click.style(str(e), fg="red"), err=True)
sys.exit(1)
try:
result = run_sync(
client=client,
connector=connector,
kb_id=kb,
dry_run=True,
verbose=verbose,
)
except Exception as e:
click.echo(click.style(f"Diff failed: {e}", fg="red"), err=True)
sys.exit(1)
finally:
client.close()
click.echo(f"\n{result.summary()}")
# ── watch ───────────────────────────────────────────────────────
@cli.command()
@click.argument("directory")
@common_options
@click.option("--debounce", default=1.0, type=float, help="Seconds to wait after last change (default: 1.0).")
@click.option("-v", "--verbose", is_flag=True, help="Show detailed progress.")
@click.pass_context
def watch(
ctx: click.Context,
directory: str,
url: str | None,
token: str | None,
kb: str | None,
debounce: float,
verbose: bool,
):
"""Watch a local directory and sync on changes.
Runs continuously until interrupted (Ctrl+C).
"""
if not kb:
click.echo(click.style("--kb is required.", fg="red"), err=True)
sys.exit(1)
quiet = ctx.obj.get("quiet", False)
from oikb.watcher import watch_directory
from oikb.connectors.filesystem import FilesystemConnector, DEFAULT_IGNORE
from oikb.sync import run_sync
try:
client = _make_client(url, token)
except ValueError as e:
click.echo(click.style(str(e), fg="red"), err=True)
sys.exit(1)
click.echo(f"Watching {directory} → KB {kb} (Ctrl+C to stop)")
def on_change():
try:
connector = FilesystemConnector(directory)
result = run_sync(
client=client,
connector=connector,
kb_id=kb,
verbose=verbose,
quiet=quiet,
)
if not quiet:
click.echo(f" [{_timestamp()}] {result.summary()}")
except Exception as e:
click.echo(click.style(f" [{_timestamp()}] Sync error: {e}", fg="red"), err=True)
try:
watch_directory(
directory=directory,
on_change=on_change,
debounce_seconds=debounce,
ignore=DEFAULT_IGNORE,
)
except FileNotFoundError as e:
click.echo(click.style(f"Error: {e}", fg="red"), err=True)
sys.exit(1)
finally:
client.close()
# ── reset ───────────────────────────────────────────────────────
@cli.command()
@common_options
@click.option(
"--keep-directories",
is_flag=True,
help="Keep directory structure, only remove files.",
)
@click.confirmation_option(
prompt="This will delete all files in the Knowledge Base. Continue?"
)
def reset(url: str | None, token: str | None, kb: str | None, keep_directories: bool):
"""Reset a Knowledge Base (delete all files)."""
if not kb:
click.echo(click.style("--kb is required.", fg="red"), err=True)
sys.exit(1)
try:
client = _make_client(url, token)
except ValueError as e:
click.echo(click.style(str(e), fg="red"), err=True)
sys.exit(1)
try:
client.reset_kb(kb, include_directories=not keep_directories)
except Exception as e:
click.echo(click.style(f"Reset failed: {e}", fg="red"), err=True)
sys.exit(1)
finally:
client.close()
click.echo("Knowledge Base reset.")
# ── ls ──────────────────────────────────────────────────────────
@cli.command()
@common_options
def ls(url: str | None, token: str | None, kb: str | None):
"""List files in a Knowledge Base."""
if not kb:
click.echo(click.style("--kb is required.", fg="red"), err=True)
sys.exit(1)
try:
client = _make_client(url, token)
except ValueError as e:
click.echo(click.style(str(e), fg="red"), err=True)
sys.exit(1)
try:
files = client.list_kb_files(kb)
except Exception as e:
click.echo(click.style(f"Failed: {e}", fg="red"), err=True)
sys.exit(1)
finally:
client.close()
if not files:
click.echo("(empty)")
return
for f in files:
meta = f.get("meta", {})
name = meta.get("name", f.get("filename", "?"))
size = meta.get("size", 0)
file_hash = meta.get("file_hash", "")[:8]
click.echo(f" {name} ({_format_size(size)}) {file_hash}")
# ── status ──────────────────────────────────────────────────────
@cli.command()
@common_options
def status(url: str | None, token: str | None, kb: str | None):
"""Show Knowledge Base info and file count."""
if not kb:
click.echo(click.style("--kb is required.", fg="red"), err=True)
sys.exit(1)
try:
client = _make_client(url, token)
except ValueError as e:
click.echo(click.style(str(e), fg="red"), err=True)
sys.exit(1)
try:
info = client.get_kb(kb)
files = info.get("files", [])
except Exception as e:
click.echo(click.style(f"Failed: {e}", fg="red"), err=True)
sys.exit(1)
finally:
client.close()
name = info.get("name", kb)
description = info.get("description", "")
file_count = len(files)
total_size = sum(f.get("meta", {}).get("size", 0) for f in files)
click.echo(f"Knowledge Base: {name}")
if description:
click.echo(f" Description: {description}")
click.echo(f" ID: {kb}")
click.echo(f" Files: {file_count}")
click.echo(f" Total size: {_format_size(total_size)}")
updated_at = info.get("updated_at")
if updated_at:
click.echo(f" Updated: {updated_at}")
# ── config ──────────────────────────────────────────────────────
@cli.group()
def config():
"""Manage oikb configuration."""
@config.command("set")
@click.argument("key", type=click.Choice(["url", "token"]))
@click.argument("value")
def config_set(key: str, value: str):
"""Set a config value (url or token)."""
set_config(key, value)
display = value if key == "url" else f"{value[:4]}...{value[-4:]}"
click.echo(f"{key} = {display}")
@config.command("get")
@click.argument("key", required=False)
def config_get(key: str | None):
"""Show config values."""
data = get_config(key)
if isinstance(data, dict):
if not data:
click.echo("(no config set)")
return
for k, v in data.items():
display = v if k == "url" else f"{v[:4]}...{v[-4:]}" if v else ""
click.echo(f"{k} = {display}")
elif data:
click.echo(data)
else:
click.echo("(not set)")
# ── Helpers ─────────────────────────────────────────────────────
def _format_size(size: int) -> str:
"""Human-readable file size."""
for unit in ("B", "KB", "MB", "GB"):
if size < 1024:
return f"{size:.0f} {unit}" if unit == "B" else f"{size:.1f} {unit}"
size /= 1024
return f"{size:.1f} TB"
def _timestamp() -> str:
"""Current time as HH:MM:SS."""
from datetime import datetime
return datetime.now().strftime("%H:%M:%S")

138
src/oikb/client.py Normal file
View file

@ -0,0 +1,138 @@
"""HTTP client wrapping the Open WebUI Knowledge Base sync API."""
from __future__ import annotations
import json
from typing import Any
import httpx
class OikbClient:
"""Stateless HTTP client for the Open WebUI KB API.
All methods are synchronous — httpx handles connection pooling internally.
"""
def __init__(self, base_url: str, token: str, timeout: float = 120.0):
self._base_url = base_url.rstrip("/")
self._http = httpx.Client(
base_url=f"{self._base_url}/api/v1",
headers={"Authorization": f"Bearer {token}"},
timeout=timeout,
)
def __enter__(self) -> OikbClient:
return self
def __exit__(self, *exc: object) -> None:
self.close()
def close(self) -> None:
self._http.close()
# ── Sync API ────────────────────────────────────────────────
def sync_diff(
self,
kb_id: str,
manifest: list[dict[str, Any]],
) -> dict[str, Any]:
"""POST /knowledge/{id}/sync/diff — compute diff from manifest."""
resp = self._http.post(
f"/knowledge/{kb_id}/sync/diff",
json={"manifest": manifest},
)
resp.raise_for_status()
return resp.json()
def sync_cleanup(
self,
kb_id: str,
file_ids: list[str],
dir_ids: list[str] | None = None,
) -> dict[str, Any]:
"""POST /knowledge/{id}/sync/cleanup — remove stale files and dirs."""
payload: dict[str, Any] = {"file_ids": file_ids}
if dir_ids:
payload["dir_ids"] = dir_ids
resp = self._http.post(
f"/knowledge/{kb_id}/sync/cleanup",
json=payload,
)
resp.raise_for_status()
return resp.json()
# ── File upload ─────────────────────────────────────────────
def upload_file(
self,
file_content: bytes,
filename: str,
kb_id: str,
file_hash: str,
directory_id: str | None = None,
) -> dict[str, Any]:
"""POST /files/ — upload a single file to the KB."""
metadata: dict[str, Any] = {
"knowledge_id": kb_id,
"file_hash": file_hash,
}
if directory_id:
metadata["directory_id"] = directory_id
resp = self._http.post(
"/files/",
files={"file": (filename, file_content)},
data={"metadata": json.dumps(metadata)},
)
resp.raise_for_status()
return resp.json()
# ── Directory management ────────────────────────────────────
def create_directory(
self,
kb_id: str,
name: str,
parent_id: str | None = None,
) -> dict[str, Any]:
"""POST /knowledge/{id}/dirs/create — create a directory."""
payload: dict[str, Any] = {"name": name}
if parent_id:
payload["parent_id"] = parent_id
resp = self._http.post(
f"/knowledge/{kb_id}/dirs/create",
json=payload,
)
resp.raise_for_status()
return resp.json()
# ── KB management ───────────────────────────────────────────
def reset_kb(
self,
kb_id: str,
include_directories: bool = True,
) -> dict[str, Any]:
"""POST /knowledge/{id}/reset — reset the KB."""
resp = self._http.post(
f"/knowledge/{kb_id}/reset",
params={"include_directories": include_directories},
)
resp.raise_for_status()
return resp.json()
def get_kb(self, kb_id: str) -> dict[str, Any]:
"""GET /knowledge/{id} — get KB info."""
resp = self._http.get(f"/knowledge/{kb_id}")
resp.raise_for_status()
return resp.json()
def list_kb_files(self, kb_id: str) -> list[dict[str, Any]]:
"""GET /knowledge/{id}/files — list files in a KB."""
resp = self._http.get(f"/knowledge/{kb_id}")
resp.raise_for_status()
data = resp.json()
return data.get("files", [])

79
src/oikb/config.py Normal file
View file

@ -0,0 +1,79 @@
"""Configuration management for oikb.
Resolution order (highest priority first):
1. CLI flags (--url, --token)
2. Environment variables (OPEN_WEBUI_URL, OPEN_WEBUI_API_KEY)
3. Config file (~/.config/oikb/config.yaml)
"""
from __future__ import annotations
import os
from pathlib import Path
import yaml
CONFIG_DIR = Path(os.environ.get("OIKB_CONFIG_DIR", Path.home() / ".config" / "oikb"))
CONFIG_FILE = CONFIG_DIR / "config.yaml"
def _load_config_file() -> dict:
"""Load config from disk. Returns empty dict if file doesn't exist."""
if CONFIG_FILE.exists():
with open(CONFIG_FILE) as f:
return yaml.safe_load(f) or {}
return {}
def _save_config_file(data: dict) -> None:
"""Write config to disk, creating parent directories as needed."""
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
with open(CONFIG_FILE, "w") as f:
yaml.safe_dump(data, f, default_flow_style=False)
def resolve_url(cli_url: str | None = None) -> str:
"""Resolve the Open WebUI base URL.
Priority: cli_url > OPEN_WEBUI_URL > config file > error.
"""
url = cli_url or os.environ.get("OPEN_WEBUI_URL") or _load_config_file().get("url")
if not url:
raise ValueError(
"No Open WebUI URL configured. Set via:\n"
" export OPEN_WEBUI_URL=<base_url>\n"
" oikb config set url <base_url>\n"
" --url <base_url>"
)
return url.rstrip("/")
def resolve_token(cli_token: str | None = None) -> str:
"""Resolve the API token.
Priority: cli_token > OPEN_WEBUI_API_KEY > config file > error.
"""
token = cli_token or os.environ.get("OPEN_WEBUI_API_KEY") or _load_config_file().get("token")
if not token:
raise ValueError(
"No API key configured. Set via:\n"
" export OPEN_WEBUI_API_KEY=<api_key>\n"
" oikb config set token <api_key>\n"
" --token <api_key>"
)
return token
def set_config(key: str, value: str) -> None:
"""Set a key in the config file."""
data = _load_config_file()
data[key] = value
_save_config_file(data)
def get_config(key: str | None = None) -> dict | str | None:
"""Get a key (or all keys) from the config file."""
data = _load_config_file()
if key:
return data.get(key)
return data

View file

@ -0,0 +1,68 @@
"""Base connector interface for oikb content sources."""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class ManifestEntry:
"""A single file in the source manifest.
Attributes:
filename: Basename of the file (e.g. "readme.md").
path: Directory path relative to source root (e.g. "docs/api").
Empty string for root-level files.
checksum: SHA-256 hex digest of the file content.
size: File size in bytes.
"""
filename: str
path: str
checksum: str
size: int
def to_dict(self) -> dict:
return {
"filename": self.filename,
"path": self.path,
"checksum": self.checksum,
"size": self.size,
}
@property
def display_path(self) -> str:
"""Human-readable relative path."""
if self.path:
return f"{self.path}/{self.filename}"
return self.filename
class BaseConnector(ABC):
"""Abstract base for all content source connectors.
Every connector must implement two methods:
- build_manifest(): enumerate all files with checksums
- read_file(): return raw bytes for a specific file
"""
@abstractmethod
def build_manifest(self) -> list[ManifestEntry]:
"""Scan the source and return a manifest of all files.
Returns:
A list of ManifestEntry objects — one per file.
"""
@abstractmethod
def read_file(self, path: str, filename: str) -> bytes:
"""Read raw file content for upload.
Args:
path: Directory path relative to source root.
filename: Basename of the file.
Returns:
Raw bytes of the file content.
"""

View file

@ -0,0 +1,108 @@
"""Filesystem connector — sync a local directory to a Knowledge Base."""
from __future__ import annotations
import hashlib
from pathlib import Path
from oikb.connectors import BaseConnector, ManifestEntry
from oikb.ignore import load_ignore_patterns, should_ignore
# Files and directories to always skip during traversal.
DEFAULT_IGNORE = frozenset({
".git",
".svn",
".hg",
".DS_Store",
"Thumbs.db",
"__pycache__",
".pytest_cache",
"node_modules",
".oikb",
".env",
})
def _sha256(path: Path) -> str:
"""Compute SHA-256 hex digest of a file, reading in 64 KiB chunks."""
h = hashlib.sha256()
with open(path, "rb") as f:
while chunk := f.read(65_536):
h.update(chunk)
return h.hexdigest()
class FilesystemConnector(BaseConnector):
"""Walk a local directory tree and produce a manifest.
Respects:
- Built-in ignore list (hidden files, .git, node_modules, etc.)
- .oikbignore file in the root directory (gitignore-style patterns)
- Custom ignore set passed via constructor
Args:
root: Path to the directory to sync.
ignore: Additional names to skip (files and dirs).
"""
def __init__(
self,
root: str | Path,
ignore: frozenset[str] | None = None,
):
self.root = Path(root).resolve()
self.builtin_ignore = ignore or DEFAULT_IGNORE
self.ignore_patterns = load_ignore_patterns(self.root)
if not self.root.is_dir():
raise FileNotFoundError(f"Not a directory: {self.root}")
def build_manifest(self) -> list[ManifestEntry]:
"""Recursively walk root, computing SHA-256 for each file."""
entries: list[ManifestEntry] = []
self._walk(self.root, "", entries)
# Sort for deterministic output.
entries.sort(key=lambda e: e.display_path)
return entries
def _walk(
self,
directory: Path,
relative_prefix: str,
entries: list[ManifestEntry],
) -> None:
"""Recursive directory walker."""
for child in sorted(directory.iterdir()):
# Always skip built-in ignores.
if child.name in self.builtin_ignore or child.name.startswith("."):
continue
relative_path = (
f"{relative_prefix}/{child.name}" if relative_prefix else child.name
)
# Check .oikbignore patterns.
if self.ignore_patterns and should_ignore(
relative_path, child.name, child.is_dir(), self.ignore_patterns
):
continue
if child.is_dir():
self._walk(child, relative_path, entries)
elif child.is_file():
entries.append(
ManifestEntry(
filename=child.name,
path=relative_prefix,
checksum=_sha256(child),
size=child.stat().st_size,
)
)
def read_file(self, path: str, filename: str) -> bytes:
"""Read a file from the local filesystem."""
if path:
file_path = self.root / path / filename
else:
file_path = self.root / filename
return file_path.read_bytes()

View file

@ -0,0 +1,147 @@
"""GitHub connector — sync a GitHub repo to a Knowledge Base via the API.
Requires: pip install oikb[github]
Uses the GitHub Trees API for checksums (blob SHAs) — no local clone needed.
"""
from __future__ import annotations
import os
from typing import Any
import httpx
from oikb.connectors import BaseConnector, ManifestEntry
class GitHubConnector(BaseConnector):
"""Sync files from a GitHub repository.
Args:
owner: Repository owner (e.g. "open-webui").
repo: Repository name (e.g. "docs").
branch: Branch to sync from (default: repo default branch).
path: Subdirectory to scope to (e.g. "docs/").
token: GitHub personal access token (or GITHUB_TOKEN env var).
"""
def __init__(
self,
owner: str,
repo: str,
branch: str | None = None,
path: str | None = None,
token: str | None = None,
):
self.owner = owner
self.repo = repo
self.branch = branch
self.path = path.strip("/") if path else None
self._token = token or os.environ.get("GITHUB_TOKEN")
headers: dict[str, str] = {"Accept": "application/vnd.github+json"}
if self._token:
headers["Authorization"] = f"Bearer {self._token}"
self._http = httpx.Client(
base_url="https://api.github.com",
headers=headers,
timeout=60.0,
)
def build_manifest(self) -> list[ManifestEntry]:
"""Fetch the repo tree and build a manifest.
Uses the recursive tree API — one request for the full repo structure.
Blob SHAs are used as checksums (they're content-addressable hashes).
"""
ref = self.branch or self._get_default_branch()
# Get recursive tree.
resp = self._http.get(
f"/repos/{self.owner}/{self.repo}/git/trees/{ref}",
params={"recursive": "1"},
)
resp.raise_for_status()
tree = resp.json()
entries: list[ManifestEntry] = []
for item in tree.get("tree", []):
if item["type"] != "blob":
continue
file_path = item["path"]
# Filter by path prefix if specified.
if self.path:
if not file_path.startswith(self.path + "/"):
continue
# Strip the prefix so paths are relative to the scoped dir.
file_path = file_path[len(self.path) + 1:]
parts = file_path.rsplit("/", 1)
if len(parts) == 2:
dir_path, filename = parts
else:
dir_path, filename = "", parts[0]
entries.append(
ManifestEntry(
filename=filename,
path=dir_path,
checksum=item["sha"], # Git blob SHA — content-addressable.
size=item.get("size", 0),
)
)
entries.sort(key=lambda e: e.display_path)
return entries
def read_file(self, path: str, filename: str) -> bytes:
"""Download a file's content via the GitHub Contents API."""
file_path = f"{path}/{filename}" if path else filename
if self.path:
file_path = f"{self.path}/{file_path}"
ref = self.branch or self._get_default_branch()
resp = self._http.get(
f"/repos/{self.owner}/{self.repo}/contents/{file_path}",
params={"ref": ref},
headers={"Accept": "application/vnd.github.raw+json"},
)
resp.raise_for_status()
return resp.content
def _get_default_branch(self) -> str:
"""Fetch the repo's default branch name."""
resp = self._http.get(f"/repos/{self.owner}/{self.repo}")
resp.raise_for_status()
return resp.json()["default_branch"]
def close(self) -> None:
self._http.close()
def parse_github_source(source: str) -> dict[str, str | None]:
"""Parse a github:owner/repo[/path] source string.
Examples:
github:open-webui/docs
github:open-webui/docs/api
github:owner/repo --branch main --path docs/
"""
# Strip "github:" prefix.
source = source.removeprefix("github:")
parts = source.split("/", 2)
if len(parts) < 2:
raise ValueError(f"Invalid GitHub source: {source}. Expected: github:owner/repo")
owner = parts[0]
repo = parts[1]
path = parts[2] if len(parts) > 2 else None
return {"owner": owner, "repo": repo, "path": path}

118
src/oikb/connectors/s3.py Normal file
View file

@ -0,0 +1,118 @@
"""S3 connector — sync an S3 bucket/prefix to a Knowledge Base.
Requires: pip install oikb[s3]
Uses ETags as checksums — S3 computes these server-side.
"""
from __future__ import annotations
from typing import Any
from oikb.connectors import BaseConnector, ManifestEntry
class S3Connector(BaseConnector):
"""Sync files from an S3 bucket.
Args:
bucket: S3 bucket name.
prefix: Key prefix to scope to (e.g. "docs/").
region: AWS region (default: from boto3 config).
"""
def __init__(
self,
bucket: str,
prefix: str | None = None,
region: str | None = None,
):
try:
import boto3
except ImportError:
raise ImportError(
"S3 connector requires boto3. Install with: pip install oikb[s3]"
)
self.bucket = bucket
self.prefix = prefix.strip("/") + "/" if prefix else ""
session_kwargs: dict[str, Any] = {}
if region:
session_kwargs["region_name"] = region
self._s3 = boto3.client("s3", **session_kwargs)
def build_manifest(self) -> list[ManifestEntry]:
"""List objects in the bucket/prefix and build a manifest.
ETags are used as checksums — they're MD5 hashes for single-part uploads,
and composite hashes for multipart. Good enough for change detection.
"""
entries: list[ManifestEntry] = []
paginator = self._s3.get_paginator("list_objects_v2")
page_kwargs: dict[str, Any] = {"Bucket": self.bucket}
if self.prefix:
page_kwargs["Prefix"] = self.prefix
for page in paginator.paginate(**page_kwargs):
for obj in page.get("Contents", []):
key = obj["Key"]
# Skip "directory" markers.
if key.endswith("/"):
continue
# Strip prefix to get relative path.
relative = key[len(self.prefix):] if self.prefix else key
parts = relative.rsplit("/", 1)
if len(parts) == 2:
dir_path, filename = parts
else:
dir_path, filename = "", parts[0]
# ETag comes quoted from S3 — strip quotes.
etag = obj["ETag"].strip('"')
entries.append(
ManifestEntry(
filename=filename,
path=dir_path,
checksum=etag,
size=obj["Size"],
)
)
entries.sort(key=lambda e: e.display_path)
return entries
def read_file(self, path: str, filename: str) -> bytes:
"""Download an object from S3."""
key = self.prefix
if path:
key += f"{path}/{filename}"
else:
key += filename
resp = self._s3.get_object(Bucket=self.bucket, Key=key)
return resp["Body"].read()
def parse_s3_source(source: str) -> dict[str, str | None]:
"""Parse an s3://bucket/prefix source string.
Examples:
s3://my-bucket
s3://my-bucket/docs/engineering
"""
source = source.removeprefix("s3://")
parts = source.split("/", 1)
bucket = parts[0]
prefix = parts[1] if len(parts) > 1 else None
if not bucket:
raise ValueError(f"Invalid S3 source. Expected: s3://bucket[/prefix]")
return {"bucket": bucket, "prefix": prefix}

62
src/oikb/ignore.py Normal file
View file

@ -0,0 +1,62 @@
"""Gitignore-style pattern matching for .oikbignore files."""
from __future__ import annotations
import fnmatch
from pathlib import Path
def load_ignore_patterns(root: Path) -> list[str]:
"""Load patterns from .oikbignore file in the given directory.
Supports gitignore-style patterns:
- Blank lines and lines starting with # are ignored
- Standard glob patterns (*, ?, [])
- Leading / anchors to the root
- Trailing / matches directories only
- ! negates a pattern (not yet supported)
"""
ignore_file = root / ".oikbignore"
if not ignore_file.exists():
return []
patterns: list[str] = []
with open(ignore_file) as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
patterns.append(line)
return patterns
def should_ignore(
relative_path: str,
name: str,
is_dir: bool,
patterns: list[str],
) -> bool:
"""Check if a file or directory should be ignored based on .oikbignore patterns.
Args:
relative_path: Path relative to root (e.g. "docs/api/readme.md").
name: Basename (e.g. "readme.md").
is_dir: Whether the entry is a directory.
patterns: List of gitignore-style patterns.
"""
for pattern in patterns:
# Directory-only patterns (trailing /).
dir_only = pattern.endswith("/")
if dir_only:
if not is_dir:
continue
pattern = pattern.rstrip("/")
# Match against both the basename and the full relative path.
if fnmatch.fnmatch(name, pattern):
return True
if fnmatch.fnmatch(relative_path, pattern):
return True
return False

221
src/oikb/sync.py Normal file
View file

@ -0,0 +1,221 @@
"""Sync orchestrator — diff → cleanup → mkdir → upload."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import click
from oikb.client import OikbClient
from oikb.connectors import BaseConnector
@dataclass
class SyncResult:
"""Summary of a completed sync operation."""
added: int = 0
modified: int = 0
deleted: int = 0
unmodified: int = 0
dirs_created: int = 0
dirs_removed: int = 0
errors: list[str] | None = None
@property
def total_changes(self) -> int:
return self.added + self.modified + self.deleted
def summary(self) -> str:
parts = []
if self.added:
parts.append(f"{self.added} added")
if self.modified:
parts.append(f"{self.modified} modified")
if self.deleted:
parts.append(f"{self.deleted} deleted")
if self.unmodified:
parts.append(f"{self.unmodified} unchanged")
if self.dirs_created:
parts.append(f"{self.dirs_created} dirs created")
if self.dirs_removed:
parts.append(f"{self.dirs_removed} dirs removed")
return ", ".join(parts) if parts else "nothing to do"
def run_sync(
client: OikbClient,
connector: BaseConnector,
kb_id: str,
dry_run: bool = False,
verbose: bool = False,
quiet: bool = False,
) -> SyncResult:
"""Execute a full incremental sync.
Steps:
1. Build manifest from connector
2. POST manifest to /sync/diff
3. Cleanup stale files (delete before upload)
4. Create missing directories
5. Upload added + modified files
6. Return summary
"""
result = SyncResult()
result.errors = []
# ── 1. Build manifest ──────────────────────────────────────
if verbose:
click.echo("Scanning source...", err=True)
manifest = connector.build_manifest()
if verbose:
click.echo(f" {len(manifest)} files found", err=True)
if not manifest:
if not quiet:
click.echo("Source is empty — nothing to sync.", err=True)
return result
# ── 2. Compute diff ────────────────────────────────────────
if verbose:
click.echo("Computing diff...", err=True)
diff = client.sync_diff(kb_id, [e.to_dict() for e in manifest])
added: list[dict[str, Any]] = diff.get("added", [])
modified: list[dict[str, Any]] = diff.get("modified", [])
deleted: list[dict[str, Any]] = diff.get("deleted", [])
unmodified_count: int = diff.get("unmodified_count", 0)
mkdir: list[str] = diff.get("mkdir", [])
rmdir: list[str] = diff.get("rmdir", [])
directory_map: dict[str, str] = diff.get("directory_map", {})
result.unmodified = unmodified_count
# ── Dry run: just print what would happen ──────────────────
if dry_run:
result.added = len(added)
result.modified = len(modified)
result.deleted = len(deleted)
result.dirs_created = len(mkdir)
result.dirs_removed = len(rmdir)
if added:
click.echo(click.style("+ Added:", fg="green"))
for f in added:
_echo_file_entry(f, "+", "green")
if modified:
click.echo(click.style("~ Modified:", fg="yellow"))
for f in modified:
_echo_file_entry(f, "~", "yellow")
if deleted:
click.echo(click.style("- Deleted:", fg="red"))
for f in deleted:
_echo_file_entry(f, "-", "red")
if mkdir:
click.echo(click.style("📁 Dirs to create:", fg="cyan"))
for d in mkdir:
click.echo(f" + {d}")
if rmdir:
click.echo(click.style("📁 Dirs to remove:", fg="cyan"))
for d in rmdir:
click.echo(f" - {d}")
return result
# Nothing to do?
if not added and not modified and not deleted and not mkdir and not rmdir:
return result
# ── 3. Cleanup stale files ─────────────────────────────────
stale_file_ids = [
*[d["file_id"] for d in deleted],
*[m["stale_file_id"] for m in modified],
]
if stale_file_ids or rmdir:
if verbose:
click.echo(
f"Cleaning up {len(stale_file_ids)} files, {len(rmdir)} dirs...",
err=True,
)
client.sync_cleanup(kb_id, stale_file_ids, rmdir if rmdir else None)
result.deleted = len(deleted)
result.dirs_removed = len(rmdir)
# ── 4. Create missing directories ──────────────────────────
for dir_path in mkdir:
segments = dir_path.split("/")
name = segments[-1]
parent_path = "/".join(segments[:-1])
parent_id = directory_map.get(parent_path)
if verbose:
click.echo(f" mkdir {dir_path}", err=True)
resp = client.create_directory(kb_id, name, parent_id)
directory_map[dir_path] = resp.get("id", "")
result.dirs_created += 1
# ── 5. Upload files ────────────────────────────────────────
# Build a lookup from manifest entries for fast access.
manifest_by_key = {(e.path, e.filename): e for e in manifest}
files_to_upload = [
*[(a, "added") for a in added],
*[(m, "modified") for m in modified],
]
for i, (entry, change_type) in enumerate(files_to_upload, 1):
filename = entry["filename"]
path = entry.get("path", "")
display = f"{path}/{filename}" if path else filename
if verbose:
click.echo(f" [{i}/{len(files_to_upload)}] {display}", err=True)
manifest_entry = manifest_by_key.get((path, filename))
if not manifest_entry:
result.errors.append(f"File not in manifest: {display}")
continue
try:
content = connector.read_file(path, filename)
directory_id = directory_map.get(path) if path else None
client.upload_file(
file_content=content,
filename=filename,
kb_id=kb_id,
file_hash=manifest_entry.checksum,
directory_id=directory_id,
)
if change_type == "added":
result.added += 1
else:
result.modified += 1
except Exception as e:
result.errors.append(f"{display}: {e}")
click.echo(
click.style(f" ✗ {display}: {e}", fg="red"),
err=True,
)
return result
def _echo_file_entry(entry: dict, prefix: str, color: str) -> None:
"""Print a file entry with color."""
path = entry.get("path", "")
filename = entry["filename"]
display = f"{path}/{filename}" if path else filename
click.echo(click.style(f" {prefix} {display}", fg=color))

86
src/oikb/watcher.py Normal file
View file

@ -0,0 +1,86 @@
"""File watcher — debounced fswatch integration for live sync."""
from __future__ import annotations
import threading
import time
from pathlib import Path
from typing import Callable
from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer
class _DebouncedHandler(FileSystemEventHandler):
"""Collects filesystem events and fires a callback after a quiet period."""
def __init__(
self,
callback: Callable[[], None],
debounce_seconds: float = 1.0,
ignore: frozenset[str] | None = None,
):
super().__init__()
self._callback = callback
self._debounce = debounce_seconds
self._ignore = ignore or frozenset()
self._timer: threading.Timer | None = None
self._lock = threading.Lock()
def _should_ignore(self, path: str) -> bool:
"""Skip events for ignored files/directories."""
parts = Path(path).parts
return any(part in self._ignore or part.startswith(".") for part in parts)
def on_any_event(self, event: FileSystemEvent) -> None:
if event.is_directory:
return
if self._should_ignore(event.src_path):
return
with self._lock:
if self._timer is not None:
self._timer.cancel()
self._timer = threading.Timer(self._debounce, self._callback)
self._timer.daemon = True
self._timer.start()
def watch_directory(
directory: str | Path,
on_change: Callable[[], None],
debounce_seconds: float = 1.0,
ignore: frozenset[str] | None = None,
) -> None:
"""Watch a directory for changes and call on_change after debounce.
Blocks until interrupted (Ctrl+C).
Args:
directory: Path to watch.
on_change: Callback fired after changes settle.
debounce_seconds: Quiet period before triggering sync.
ignore: File/dir names to ignore.
"""
path = Path(directory).resolve()
if not path.is_dir():
raise FileNotFoundError(f"Not a directory: {path}")
handler = _DebouncedHandler(
callback=on_change,
debounce_seconds=debounce_seconds,
ignore=ignore,
)
observer = Observer()
observer.schedule(handler, str(path), recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
pass
finally:
observer.stop()
observer.join()