Skip to content

Commit 2fd9319

Browse files
tjbcktjbck
committed
fix: split messaging connectors by day for incremental sync (v0.1.2)
1 parent 08a1bd1 commit 2fd9319

6 files changed

Lines changed: 163 additions & 65 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
44

55
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
66

7+
## [0.1.2] - 2025-05-20
8+
9+
### Changed
10+
11+
- Messaging connectors (Slack, Discord, Teams) now split messages by day for truly incremental sync. Past days are immutable so their checksums never change.
12+
713
## [0.1.1] - 2025-05-20
814

915
### Added

‎pyproject.toml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
[project]
22
name = "oikb"
3-
version = "0.1.1"
4-
description = "CLI tool for syncing content to Open WebUI Knowledge Bases"
3+
version = "0.1.2"
4+
description = "Sync anything to Open WebUI Knowledge Bases"
55
readme = "README.md"
66
authors = [
77
{ name = "Tim Baek", email = "tim@openwebui.com" }

‎src/oikb/__init__.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,3 @@
11
"""oikb — Open WebUI Knowledge Base CLI."""
22

3-
__version__ = "0.1.1"
3+
__version__ = "0.1.2"

‎src/oikb/connectors/discord.py‎

Lines changed: 46 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,15 @@
11
"""Discord connector — sync channel messages to a Knowledge Base.
22
33
Auth via DISCORD_TOKEN env var (Bot token with Message Content intent).
4+
Messages are split by day for incremental sync.
45
"""
56

67
from __future__ import annotations
78

89
import hashlib
910
import os
11+
from collections import defaultdict
12+
from datetime import datetime
1013
from typing import Any
1114

1215
import httpx
@@ -15,9 +18,9 @@
1518

1619

1720
class DiscordConnector(BaseConnector):
18-
"""Sync messages from a Discord channel."""
21+
"""Sync messages from a Discord channel, one file per day."""
1922

20-
def __init__(self, channel_id: str, token: str | None = None, limit: int = 1000):
23+
def __init__(self, channel_id: str, token: str | None = None, limit: int = 5000):
2124
self.channel_id = channel_id
2225
self.limit = limit
2326
self._token = token or os.environ.get("DISCORD_TOKEN")
@@ -29,28 +32,51 @@ def __init__(self, channel_id: str, token: str | None = None, limit: int = 1000)
2932
headers={"Authorization": f"Bot {self._token}"},
3033
timeout=30.0,
3134
)
32-
self._text: str = ""
35+
self._channel_name: str = ""
36+
self._daily_texts: dict[str, str] = {}
3337

3438
def build_manifest(self) -> list[ManifestEntry]:
3539
messages = self._fetch_messages()
3640
if not messages:
3741
return []
3842

39-
lines = []
40-
for msg in reversed(messages):
41-
author = msg.get("author", {}).get("username", "unknown")
42-
content = msg.get("content", "")
43-
ts = msg.get("timestamp", "")
44-
lines.append(f"[{ts}] {author}: {content}")
45-
46-
self._text = "\n".join(lines)
47-
checksum = hashlib.sha256(self._text.encode()).hexdigest()[:16]
48-
49-
# Get channel name.
5043
info = self._http.get(f"/channels/{self.channel_id}")
51-
name = info.json().get("name", self.channel_id) if info.status_code == 200 else self.channel_id
44+
self._channel_name = (
45+
info.json().get("name", self.channel_id)
46+
if info.status_code == 200
47+
else self.channel_id
48+
)
5249

53-
return [ManifestEntry(filename=f"{name}.txt", path="", checksum=checksum, size=len(self._text.encode()))]
50+
# Group by day.
51+
by_day: dict[str, list[dict]] = defaultdict(list)
52+
for msg in messages:
53+
ts = msg.get("timestamp", "")[:10] # YYYY-MM-DD from ISO string.
54+
if not ts:
55+
continue
56+
by_day[ts].append(msg)
57+
58+
entries: list[ManifestEntry] = []
59+
for day, day_msgs in sorted(by_day.items()):
60+
lines = []
61+
for msg in sorted(day_msgs, key=lambda m: m.get("timestamp", "")):
62+
author = msg.get("author", {}).get("username", "unknown")
63+
content = msg.get("content", "")
64+
lines.append(f"[{msg.get('timestamp', '')}] {author}: {content}")
65+
66+
text = "\n".join(lines)
67+
self._daily_texts[day] = text
68+
checksum = hashlib.sha256(text.encode()).hexdigest()[:16]
69+
70+
entries.append(
71+
ManifestEntry(
72+
filename=f"{self._channel_name}_{day}.txt",
73+
path="",
74+
checksum=checksum,
75+
size=len(text.encode()),
76+
)
77+
)
78+
79+
return entries
5480

5581
def _fetch_messages(self) -> list[dict]:
5682
messages: list[dict] = []
@@ -69,7 +95,10 @@ def _fetch_messages(self) -> list[dict]:
6995
return messages
7096

7197
def read_file(self, path: str, filename: str) -> bytes:
72-
return self._text.encode("utf-8")
98+
for day, text in self._daily_texts.items():
99+
if day in filename:
100+
return text.encode("utf-8")
101+
raise FileNotFoundError(f"Day not found: {filename}")
73102

74103
def close(self) -> None:
75104
self._http.close()

‎src/oikb/connectors/slack.py‎

Lines changed: 53 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,15 @@
11
"""Slack connector — sync channel history to a Knowledge Base.
22
33
Auth via SLACK_TOKEN env var (Bot User OAuth Token with channels:history scope).
4+
Messages are split by day for incremental sync — past days never change.
45
"""
56

67
from __future__ import annotations
78

89
import hashlib
910
import os
11+
from collections import defaultdict
12+
from datetime import datetime, timezone
1013
from typing import Any
1114

1215
import httpx
@@ -15,9 +18,9 @@
1518

1619

1720
class SlackConnector(BaseConnector):
18-
"""Sync messages from a Slack channel."""
21+
"""Sync messages from a Slack channel, one file per day."""
1922

20-
def __init__(self, channel_id: str, token: str | None = None, limit: int = 1000):
23+
def __init__(self, channel_id: str, token: str | None = None, limit: int = 5000):
2124
self.channel_id = channel_id
2225
self.limit = limit
2326
self._token = token or os.environ.get("SLACK_TOKEN")
@@ -29,27 +32,61 @@ def __init__(self, channel_id: str, token: str | None = None, limit: int = 1000)
2932
headers={"Authorization": f"Bearer {self._token}"},
3033
timeout=30.0,
3134
)
32-
self._messages: list[dict] = []
35+
self._channel_name: str = ""
36+
self._daily_texts: dict[str, str] = {}
3337

3438
def build_manifest(self) -> list[ManifestEntry]:
35-
self._messages = self._fetch_history()
36-
if not self._messages:
39+
messages = self._fetch_history()
40+
if not messages:
3741
return []
3842

39-
text = self._format_messages(self._messages)
40-
checksum = hashlib.sha256(text.encode()).hexdigest()[:16]
41-
4243
# Get channel name.
4344
info = self._http.get("/conversations.info", params={"channel": self.channel_id})
44-
name = info.json().get("channel", {}).get("name", self.channel_id) if info.status_code == 200 else self.channel_id
45+
self._channel_name = (
46+
info.json().get("channel", {}).get("name", self.channel_id)
47+
if info.status_code == 200
48+
else self.channel_id
49+
)
4550

46-
return [ManifestEntry(filename=f"{name}.txt", path="", checksum=checksum, size=len(text.encode()))]
51+
# Group messages by date.
52+
by_day: dict[str, list[dict]] = defaultdict(list)
53+
for msg in messages:
54+
ts = float(msg.get("ts", "0"))
55+
day = datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d")
56+
by_day[day].append(msg)
57+
58+
entries: list[ManifestEntry] = []
59+
for day, day_msgs in sorted(by_day.items()):
60+
lines = []
61+
for msg in sorted(day_msgs, key=lambda m: m.get("ts", "")):
62+
user = msg.get("user", "unknown")
63+
text = msg.get("text", "")
64+
ts = msg.get("ts", "")
65+
lines.append(f"[{ts}] {user}: {text}")
66+
67+
content = "\n".join(lines)
68+
self._daily_texts[day] = content
69+
checksum = hashlib.sha256(content.encode()).hexdigest()[:16]
70+
71+
entries.append(
72+
ManifestEntry(
73+
filename=f"{self._channel_name}_{day}.txt",
74+
path="",
75+
checksum=checksum,
76+
size=len(content.encode()),
77+
)
78+
)
79+
80+
return entries
4781

4882
def _fetch_history(self) -> list[dict]:
4983
messages: list[dict] = []
5084
cursor = None
5185
while len(messages) < self.limit:
52-
params: dict[str, Any] = {"channel": self.channel_id, "limit": min(200, self.limit - len(messages))}
86+
params: dict[str, Any] = {
87+
"channel": self.channel_id,
88+
"limit": min(200, self.limit - len(messages)),
89+
}
5390
if cursor:
5491
params["cursor"] = cursor
5592
resp = self._http.get("/conversations.history", params=params)
@@ -61,17 +98,12 @@ def _fetch_history(self) -> list[dict]:
6198
break
6299
return messages
63100

64-
def _format_messages(self, messages: list[dict]) -> str:
65-
lines = []
66-
for msg in reversed(messages):
67-
user = msg.get("user", "unknown")
68-
text = msg.get("text", "")
69-
ts = msg.get("ts", "")
70-
lines.append(f"[{ts}] {user}: {text}")
71-
return "\n".join(lines)
72-
73101
def read_file(self, path: str, filename: str) -> bytes:
74-
return self._format_messages(self._messages).encode("utf-8")
102+
# Extract date from filename: channel_YYYY-MM-DD.txt
103+
for day, text in self._daily_texts.items():
104+
if day in filename:
105+
return text.encode("utf-8")
106+
raise FileNotFoundError(f"Day not found: {filename}")
75107

76108
def close(self) -> None:
77109
self._http.close()

‎src/oikb/connectors/teams.py‎

Lines changed: 55 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,14 @@
22
33
Auth via Microsoft Graph API using app credentials.
44
Set TEAMS_TENANT_ID, TEAMS_CLIENT_ID, TEAMS_CLIENT_SECRET env vars.
5+
Messages are split by day for incremental sync.
56
"""
67

78
from __future__ import annotations
89

910
import hashlib
1011
import os
12+
from collections import defaultdict
1113
from typing import Any
1214

1315
import httpx
@@ -16,7 +18,7 @@
1618

1719

1820
class TeamsConnector(BaseConnector):
19-
"""Sync messages from a Microsoft Teams channel."""
21+
"""Sync messages from a Microsoft Teams channel, one file per day."""
2022

2123
def __init__(self, team_id: str, channel_id: str,
2224
tenant_id: str | None = None, client_id: str | None = None, client_secret: str | None = None):
@@ -40,45 +42,74 @@ def __init__(self, team_id: str, channel_id: str,
4042
headers={"Authorization": f"Bearer {token_resp.json()['access_token']}"},
4143
timeout=30.0,
4244
)
43-
self._text: str = ""
45+
self._channel_name: str = ""
46+
self._daily_texts: dict[str, str] = {}
4447

4548
def build_manifest(self) -> list[ManifestEntry]:
49+
messages = self._fetch_messages()
50+
if not messages:
51+
return []
52+
53+
info = self._http.get(f"/teams/{self.team_id}/channels/{self.channel_id}")
54+
self._channel_name = (
55+
info.json().get("displayName", self.channel_id)
56+
if info.status_code == 200
57+
else self.channel_id
58+
)
59+
60+
by_day: dict[str, list[dict]] = defaultdict(list)
61+
for msg in messages:
62+
ts = msg.get("createdDateTime", "")[:10]
63+
if not ts:
64+
continue
65+
by_day[ts].append(msg)
66+
67+
entries: list[ManifestEntry] = []
68+
for day, day_msgs in sorted(by_day.items()):
69+
lines = []
70+
for msg in sorted(day_msgs, key=lambda m: m.get("createdDateTime", "")):
71+
sender = msg.get("from", {}).get("user", {}).get("displayName", "unknown")
72+
body = msg.get("body", {}).get("content", "")
73+
lines.append(f"[{msg.get('createdDateTime', '')}] {sender}: {body}")
74+
75+
text = "\n".join(lines)
76+
self._daily_texts[day] = text
77+
checksum = hashlib.sha256(text.encode()).hexdigest()[:16]
78+
79+
entries.append(
80+
ManifestEntry(
81+
filename=f"{self._channel_name}_{day}.txt",
82+
path="",
83+
checksum=checksum,
84+
size=len(text.encode()),
85+
)
86+
)
87+
88+
return entries
89+
90+
def _fetch_messages(self) -> list[dict]:
4691
messages: list[dict] = []
4792
url = f"/teams/{self.team_id}/channels/{self.channel_id}/messages"
48-
while url and len(messages) < 1000:
93+
while url and len(messages) < 5000:
4994
resp = self._http.get(url)
5095
resp.raise_for_status()
5196
data = resp.json()
5297
messages.extend(data.get("value", []))
53-
url = data.get("@odata.nextLink", "").replace("https://graph.microsoft.com/v1.0", "") if data.get("@odata.nextLink") else None
54-
55-
if not messages:
56-
return []
57-
58-
lines = []
59-
for msg in messages:
60-
sender = msg.get("from", {}).get("user", {}).get("displayName", "unknown")
61-
body = msg.get("body", {}).get("content", "")
62-
ts = msg.get("createdDateTime", "")
63-
lines.append(f"[{ts}] {sender}: {body}")
64-
65-
self._text = "\n".join(lines)
66-
checksum = hashlib.sha256(self._text.encode()).hexdigest()[:16]
67-
68-
info = self._http.get(f"/teams/{self.team_id}/channels/{self.channel_id}")
69-
name = info.json().get("displayName", self.channel_id) if info.status_code == 200 else self.channel_id
70-
71-
return [ManifestEntry(filename=f"{name}.txt", path="", checksum=checksum, size=len(self._text.encode()))]
98+
next_link = data.get("@odata.nextLink", "")
99+
url = next_link.replace("https://graph.microsoft.com/v1.0", "") if next_link else None
100+
return messages
72101

73102
def read_file(self, path: str, filename: str) -> bytes:
74-
return self._text.encode("utf-8")
103+
for day, text in self._daily_texts.items():
104+
if day in filename:
105+
return text.encode("utf-8")
106+
raise FileNotFoundError(f"Day not found: {filename}")
75107

76108
def close(self) -> None:
77109
self._http.close()
78110

79111

80112
def parse_teams_source(source: str) -> dict[str, str | None]:
81-
"""Parse teams:team-id/channel-id."""
82113
source = source.removeprefix("teams:")
83114
parts = source.split("/", 1)
84115
if len(parts) < 2:

0 commit comments

Comments
 (0)