Initial commit.

chat_archive/README.md

@@ -0,0 +1,104 @@
+# chat_archive
+
+This package contains an Owlbot module that writes Owncast chat activity
+and stream context updates to JSON files for later processing.
+
+## Behavior
+
+An archive is the JSON file this module writes for one live stream. The module
+starts an archive when Owlbot sees a live stream, including streams that are
+already live when the module loads.
+
+While an archive is active, the module records chat messages, user joins, user
+parts, name changes, stream title updates, and message visibility updates in the
+order Owlbot dispatches them. Events outside an active stream are ignored.
+
+Archive files are stream-scoped. When a stream ends, the active archive is
+finalized, and the next live stream starts a new archive file. Module unload
+also finalizes the active archive.
+
+## Configuration
+
+```yaml
+modules:
+  chat_archive:
+    enabled: true
+    archive_dir: "data/chat_archives"
+    pretty_json: true
+```
+
+- `archive_dir`: directory where archive files are written. The module creates
+  it if needed.
+- `pretty_json`: when true, write indented JSON. When false, write compact JSON.
+
+## Archive Format
+
+Archive filenames are based on the detected stream start time when available:
+
+```text
+chat-archive-20260526T120000Z.json
+```
+
+If a file already exists for that timestamp, the module adds a suffix such as
+`chat-archive-20260526T120000Z-2.json`.
+
+Each archive is a single JSON document with these top-level fields:
+
+- `archive_start_reason`: one of `module_startup` or `stream_started_event`.
+- `archive_started_at`: ISO 8601 timestamp for when the archive file was
+  started.
+- `archive_updated_at`: ISO 8601 timestamp for the most recent archive write.
+- `archive_stopped_at`: ISO 8601 timestamp for when the archive was finalized,
+  or `null` while the archive is active.
+- `archive_stop_reason`: reason the archive was finalized, or `null` while the
+  archive is active. Current values are `stream_stopped_event`,
+  `module_teardown`, and `replaced_by_stream_started_event`.
+- `stream_title`: stream title when the archive started, or `null` if
+  unavailable.
+- `stream_stopped_at`: stream stop timestamp recorded when finalized by a
+  stream stop event, or `null` otherwise.
+- `event_count`: number of objects in `events`.
+- `events`: ordered list of captured activity events.
+
+Each object in `events` contains:
+
+- `event_type`: one of `CHAT`, `USER_JOINED`, `USER_PARTED`, `NAME_CHANGE`,
+  `STREAM_TITLE_UPDATED`, or `VISIBILITY-UPDATE`.
+- `archived_at`: ISO 8601 timestamp for when the event was written.
+- `event_data`: parsed Owlbot event data converted to JSON-compatible values.
+  The exact fields depend on `event_type`.
+
+Abridged example archive:
+
+```json
+{
+  "archive_start_reason": "stream_started_event",
+  "archive_started_at": "2026-05-26T12:00:01.000000+00:00",
+  "archive_updated_at": "2026-05-26T12:01:00.000000+00:00",
+  "archive_stopped_at": null,
+  "archive_stop_reason": null,
+  "stream_title": "Game Night",
+  "stream_stopped_at": null,
+  "event_count": 1,
+  "events": [
+    {
+      "event_type": "CHAT",
+      "archived_at": "2026-05-26T12:01:00.000000+00:00",
+      "event_data": {
+        "message_id": "msg-1",
+        "raw_body": "hello chat",
+        "body": "hello chat",
+        "is_visible": true
+      }
+    }
+  ]
+}
+```
+
+Owlbot modules receive parsed event dataclasses rather than the untouched
+Owncast webhook payload. `event_data` stores that parsed event data after
+converting values such as datetimes and sets into JSON-compatible values.
+
+The archive is rewritten after each captured event through a temporary sibling
+file and atomic replace. For an existing archive file, readers should see either
+the previous complete JSON document or the next complete JSON document.

chat_archive/__init__.py

@@ -0,0 +1,332 @@
+# Copyright 2026 Logan Fick
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Archive Owncast chat activity to JSON files."""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import asdict, dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, Literal, cast
+
+import orjson
+from owlbot.api import (
+    ChatEvent,
+    Event,
+    EventContext,
+    EventType,
+    ModuleContext,
+    NameChangedEvent,
+    Priority,
+    StreamStartedEvent,
+    StreamStoppedEvent,
+    StreamTitleUpdatedEvent,
+    UserJoinedEvent,
+    UserPartedEvent,
+    VisibilityUpdateEvent,
+    on_event,
+    on_setup,
+    on_teardown,
+)
+
+_STATE_KEY = "chat_archive_state"
+
+type ArchiveStartReason = Literal["module_startup", "stream_started_event"]
+type ArchiveStopReason = Literal[
+    "module_teardown",
+    "replaced_by_stream_started_event",
+    "stream_stopped_event",
+]
+
+
+@dataclass(slots=True)
+class ArchiveState:
+    """Runtime state for the chat archive module."""
+
+    archive_dir: Path
+    pretty_json: bool
+    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
+    current_path: Path | None = None
+    current_data: dict[str, Any] | None = None
+
+
+def _get_state(ctx: ModuleContext) -> ArchiveState:
+    """Return the typed ArchiveState stored in Owlbot's ctx.state dict."""
+    state = ctx.state.get(_STATE_KEY)
+    if not isinstance(state, ArchiveState):
+        msg = "chat_archive state is not initialized"
+        raise RuntimeError(msg)  # noqa: TRY004  # state error, not a type error
+    return state
+
+
+@on_setup
+async def setup(ctx: ModuleContext) -> None:
+    """Initialize config/runtime state and start an archive if the stream is live."""
+    ctx.config.register_defaults(
+        {
+            "archive_dir": "data/chat_archives",
+            "pretty_json": True,
+        }
+    )
+
+    archive_dir_value = ctx.config.get("archive_dir", "data/chat_archives")
+    pretty_json = bool(ctx.config.get("pretty_json", True))
+
+    state = ArchiveState(
+        archive_dir=Path(str(archive_dir_value)),
+        pretty_json=pretty_json,
+    )
+    state.archive_dir.mkdir(parents=True, exist_ok=True)
+
+    ctx.state[_STATE_KEY] = state
+
+    try:
+        status = await ctx.owncast_client.get_status()
+    except Exception:  # noqa: BLE001  # best-effort startup check; stream events still work
+        ctx.logger.warning(
+            "Could not check stream status. Chat archiving will start on stream start.",
+            exc_info=True,
+        )
+        return
+
+    if status.get("online", False):
+        last_connect_time = status.get("lastConnectTime")
+        started_at = datetime.now(UTC)
+        if isinstance(last_connect_time, str) and last_connect_time:
+            try:
+                started_at = datetime.fromisoformat(last_connect_time)
+            except ValueError:
+                ctx.logger.warning(
+                    "Owncast status returned lastConnectTime in unexpected "
+                    "format: %r. Falling back to the current time.",
+                    last_connect_time,
+                )
+        async with state.lock:
+            path = await _start_archive(
+                ctx,
+                state,
+                archive_start_reason="module_startup",
+                stream_started_at=started_at,
+                stream_title=status.get("streamTitle"),
+            )
+        ctx.logger.info("Stream is already live. Chat archive started: %s", path)
+    else:
+        ctx.logger.info("Stream is offline. Chat archiving will start on stream start.")
+
+
+@on_event(EventType.STREAM_STARTED)
+async def handle_stream_started(ctx: EventContext[StreamStartedEvent]) -> None:
+    """Create a new archive when Owncast reports that the stream went live."""
+    state = _get_state(ctx.module)
+    started_at = ctx.event.timestamp or datetime.now(UTC)
+
+    async with state.lock:
+        path = await _start_archive(
+            ctx.module,
+            state,
+            archive_start_reason="stream_started_event",
+            stream_started_at=started_at,
+            stream_title=ctx.event.stream_title,
+        )
+
+    ctx.logger.info("Chat archive started: %s", path)
+
+
+@on_event(EventType.CHAT, priority=Priority.HIGHEST)
+async def archive_chat_message(ctx: EventContext[ChatEvent]) -> None:
+    """Append an incoming chat message to the active archive file."""
+    await _archive_event(ctx.module, EventType.CHAT, ctx.event)
+
+
+@on_event(EventType.VISIBILITY_UPDATE, priority=Priority.HIGHEST)
+async def archive_visibility_update(
+    ctx: EventContext[VisibilityUpdateEvent],
+) -> None:
+    """Append a message visibility update to the active archive file."""
+    await _archive_event(ctx.module, EventType.VISIBILITY_UPDATE, ctx.event)
+
+
+@on_event(EventType.USER_JOINED, priority=Priority.HIGHEST)
+async def archive_user_joined(ctx: EventContext[UserJoinedEvent]) -> None:
+    """Append a user join event to the active archive file."""
+    await _archive_event(ctx.module, EventType.USER_JOINED, ctx.event)
+
+
+@on_event(EventType.USER_PARTED, priority=Priority.HIGHEST)
+async def archive_user_parted(ctx: EventContext[UserPartedEvent]) -> None:
+    """Append a user part event to the active archive file."""
+    await _archive_event(ctx.module, EventType.USER_PARTED, ctx.event)
+
+
+@on_event(EventType.NAME_CHANGE, priority=Priority.HIGHEST)
+async def archive_name_change(ctx: EventContext[NameChangedEvent]) -> None:
+    """Append a name change event to the active archive file."""
+    await _archive_event(ctx.module, EventType.NAME_CHANGE, ctx.event)
+
+
+@on_event(EventType.STREAM_TITLE_UPDATED, priority=Priority.HIGHEST)
+async def archive_stream_title_update(
+    ctx: EventContext[StreamTitleUpdatedEvent],
+) -> None:
+    """Append a stream title update event to the active archive file."""
+    await _archive_event(ctx.module, EventType.STREAM_TITLE_UPDATED, ctx.event)
+
+
+@on_event(EventType.STREAM_STOPPED)
+async def handle_stream_stopped(ctx: EventContext[StreamStoppedEvent]) -> None:
+    """Finalize the active archive when Owncast reports that the stream stopped."""
+    state = _get_state(ctx.module)
+
+    async with state.lock:
+        path = await _finalize_archive(
+            ctx.module,
+            state,
+            reason="stream_stopped_event",
+            stream_stopped_at=ctx.event.timestamp or datetime.now(UTC),
+        )
+
+    if path is not None:
+        ctx.logger.info("Chat archive finalized after stream stopped: %s", path)
+
+
+@on_teardown
+async def teardown(ctx: ModuleContext) -> None:
+    """Finalize any active archive when Owlbot unloads the module."""
+    state = _get_state(ctx)
+
+    async with state.lock:
+        await _finalize_archive(ctx, state, reason="module_teardown")
+
+
+async def _start_archive(
+    ctx: ModuleContext,
+    state: ArchiveState,
+    *,
+    archive_start_reason: ArchiveStartReason,
+    stream_started_at: datetime,
+    stream_title: str | None,
+) -> Path:
+    """Open a new archive file, finalizing any active file first."""
+    if state.current_data is not None:
+        await _finalize_archive(ctx, state, reason="replaced_by_stream_started_event")
+
+    archive_started_at = datetime.now(UTC)
+
+    if stream_started_at.tzinfo is None:
+        stream_started_at = stream_started_at.replace(tzinfo=UTC)
+    timestamp = stream_started_at.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ")
+    stem = f"chat-archive-{timestamp}"
+    path = state.archive_dir / f"{stem}.json"
+    index = 2
+    while path.exists():
+        path = state.archive_dir / f"{stem}-{index}.json"
+        index += 1
+
+    data: dict[str, Any] = {
+        "archive_start_reason": archive_start_reason,
+        "archive_started_at": archive_started_at.isoformat(),
+        "archive_updated_at": archive_started_at.isoformat(),
+        "archive_stopped_at": None,
+        "archive_stop_reason": None,
+        "stream_title": stream_title,
+        "stream_stopped_at": None,
+        "event_count": 0,
+        "events": [],
+    }
+
+    await _write_json(path, data, pretty_json=state.pretty_json)
+
+    state.current_path = path
+    state.current_data = data
+
+    return path
+
+
+async def _archive_event(
+    ctx: ModuleContext,
+    event_type: EventType,
+    event: Event,
+) -> None:
+    """Record one event and rewrite the active archive, if a stream is live."""
+    state = _get_state(ctx)
+
+    async with state.lock:
+        data = state.current_data
+        path = state.current_path
+        if data is None or path is None:
+            return
+
+        archived_at = datetime.now(UTC)
+        events = cast("list[dict[str, Any]]", data["events"])
+        events.append(
+            {
+                "event_type": event_type.value,
+                "archived_at": archived_at.isoformat(),
+                "event_data": asdict(event),
+            },
+        )
+        data["event_count"] = len(events)
+        data["archive_updated_at"] = archived_at.isoformat()
+        await _write_json(path, data, pretty_json=state.pretty_json)
+
+
+async def _finalize_archive(
+    ctx: ModuleContext,
+    state: ArchiveState,
+    *,
+    reason: ArchiveStopReason,
+    stream_stopped_at: datetime | None = None,
+) -> Path | None:
+    """Mark the active archive complete and write its final JSON state."""
+    data = state.current_data
+    path = state.current_path
+    if data is None or path is None:
+        return None
+
+    stopped_at = datetime.now(UTC)
+
+    data["archive_stopped_at"] = stopped_at.isoformat()
+    data["archive_stop_reason"] = reason
+    data["archive_updated_at"] = stopped_at.isoformat()
+    if stream_stopped_at is not None:
+        data["stream_stopped_at"] = stream_stopped_at.isoformat()
+
+    await _write_json(path, data, pretty_json=state.pretty_json)
+    ctx.logger.info("Wrote chat archive: %s", path)
+    state.current_path = None
+    state.current_data = None
+    return path
+
+
+async def _write_json(path: Path, data: dict[str, Any], *, pretty_json: bool) -> None:
+    """Atomically write a valid JSON document to disk without blocking the loop."""
+
+    def json_default(value: object) -> object:
+        if isinstance(value, (set, frozenset)):
+            return sorted(value, key=str)
+
+        raise TypeError(f"Type is not JSON serializable: {type(value).__name__}")
+
+    def write_sync() -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        option = orjson.OPT_INDENT_2 if pretty_json else 0
+        payload = orjson.dumps(data, option=option, default=json_default) + b"\n"
+
+        tmp_path = path.with_name(f".{path.name}.tmp")
+        tmp_path.write_bytes(payload)
+        tmp_path.replace(path)
+
+    await asyncio.to_thread(write_sync)