diff --git a/README.md b/README.md index 3fd5b1b1..7a04fb41 100644 --- a/README.md +++ b/README.md @@ -1,284 +1,297 @@ -

- OpenMagpie -

- -

- Open-source social listening -

- -

- License -

- -

- openmagpie.ai | - Quickstart | - CLI reference | - Changelog -

- ---- - -

- magpie CLI tour: the feed's sources, the watch's semantic filter, then the matches with scores and links -

- -> [!TIP] -> **CLI not for you?** A UI / hosted version is on the way. Star the repo for updates, or join the waitlist at [openmagpie.ai](https://www.openmagpie.ai/). - -## What it does - -You scan Reddit, Hacker News, and a few RSS feeds looking for someone hitting a problem your product solves or asking a question you can answer well. Getting there while the conversation is happening is how you build a brand and a community around what you know. OpenMagpie watches the threads for you so you spend your time on engagement instead of searching. - -You curate sources into a feed, write a natural-language description of what's relevant (for example, "someone frustrated with manual social monitoring and asking for alternatives"), and a local LLM run via any OpenAI-compatible runner (e.g. Ollama, vLLM, LM Studio) scores each new post against it. Matches go to a webhook or your logs (more integrations coming); everything else is dropped. You read the hits instead of the firehose. - -## Where it listens - -OpenMagpie listens wherever communities are having those conversations. - -- **Public discussion (today):** Reddit, Hacker News, and any RSS or Atom feed (news, blogs, Substack publications, and forums that publish feeds). -- **Communities you're in (roadmap):** Slack workspaces and LinkedIn you already belong to, so you catch relevant threads in the groups where you participate, no admin or app install required. - -## Quickstart - -One command for your first real match (needs Docker and uv; it clones the repo and runs the quickstart for you): - -```bash -curl -fsSL https://openmagpie.ai | sh -``` - -Once setup finishes, keep processing new posts in the background: - -```bash -make up-jobs # run the schedulers in the background -tail -f .jobs/*.log # watch them work -``` - -The quickstart from curl to first matches: prerequisites check, LLM setup prompts, personalizing the listener (which subreddits, what to flag, how strict), then the seeded backlog scoring and the ready summary - -Prefer to not generate seed data? `SKIP_DATA_SEED=1` brings up the stack without sample data: - -```bash -curl -fsSL https://openmagpie.ai | SKIP_DATA_SEED=1 sh -``` - -Prefer to clone first? - -```bash -git clone https://github.com/obris-dev/openmagpie.git -cd openmagpie -./scripts/quickstart/run.sh -``` - -Either way it walks you through your first listener (which subreddits to watch, what to flag in plain language, how strict), seeds it, and once an LLM is reachable runs the pipeline once so the first matches print straight to the logs, tagged `[quickstart]`. Matches show up in the terminal and the CLI activity log, not the web UI yet. Your feed and watch are saved as editable YAML in `config/quickstart/` (see [config/README.md](config/README.md) for editing and reusing them). Want more posts to start with? `DAYS=7 ./scripts/quickstart/seed.sh` backfills a week instead of a day. See [examples/README.md](examples/README.md) for ready-made starters to apply by hand. Or have an AI assistant interview you and build the config: copy the prompt in [Set it up with an AI assistant](examples/README.md#set-it-up-with-an-ai-assistant). - -### Prereq: an OpenAI-compatible LLM endpoint - -OpenMagpie is BYO LLM; the dev stack doesn't bundle one. Whatever you already run almost certainly works, because **Ollama, vLLM, llama.cpp, and LM Studio all expose an OpenAI-compatible `/v1` API** (so do hosted providers like OpenAI, Together, or Groq). OpenMagpie talks to that `/v1` endpoint with the standard OpenAI client, so you just point `ENGINE_BASE_URL` at it. The quickstart validates your endpoint and points you at the model of your choice. - -- **Local.** Any OpenAI-compatible server on your machine works; point `ENGINE_BASE_URL` at its `/v1`: Ollama (`http://host.docker.internal:11434/v1`), vLLM (`:8000/v1`), LM Studio (`:1234/v1`), or llama.cpp (`:8080/v1`). The shipped default is Ollama's `:11434`. New here and want the quickest start? Install [Ollama](https://ollama.com/download), then `ollama pull qwen2.5:7b && ollama serve`. -- **Remote (LAN box, GPU server, cloud).** Set `ENGINE_BASE_URL=http://your-host:11434/v1` in `apps/core/.env`. -- **Hosted API.** Set `ENGINE_BASE_URL=https://api.openai.com/v1` and `ENGINE_API_KEY=...` (local servers leave the key blank). - -Set `ENGINE_MODEL` to the model you want to judge with. A 7B model judges in roughly 1 to 3 seconds on Apple Silicon or a recent NVIDIA GPU; CPU-only works but is slower. - -### Use it - -The quickstart already built and seeded everything and put the `magpie` CLI on your `PATH`. Drive it from there: - -```bash -magpie auth login # browser device flow -magpie feed create # opens $EDITOR on a feed template (sources + retention) -magpie watch create # opens $EDITOR on a watch template (feeds + action chain) -magpie activity summary --action # per-state run breakdown for any action (filter, webhook, log) -``` - -On a headless box (a server you SSH into, no browser), skip the device flow and use a personal access token. Mint one on the server, then sign in with it on the box, the token is pasted (stdin or a hidden prompt, never the command line) and stored in `~/.magpie` at `0600`, persisting across sessions: - -```bash -# on the server (the issue_cli_token management command, via the local stack): -make local-manage CMD="issue_cli_token --email you@example.com --name my-box" -# then, on the box: -magpie auth login --token # paste the printed token at the prompt -``` - -For CI or an ephemeral box, set `MAGPIE_TOKEN=mgp_...` in the environment instead: it's -read on every request, takes precedence over a stored login, and is never persisted (the -`GH_TOKEN` pattern), so no login step. Manage tokens with `magpie auth token list` / -`create` / `revoke` (minting needs a browser login; a token can't mint another). - -A watch's `actions:` chain typically starts with a `semantic_filter` (your natural-language criteria + threshold) followed by a `webhook` or `log` delivery. Pick a backfill window when you create the feed and the first `make local-tick` scores real posts against your criteria immediately, with no wait for the scheduler. - -Full command list: the [magpie CLI reference](apps/cli/README.md). The dev loop runs through `make`: see [make/README.md](make/README.md) or `make help`. - -### Running it continuously - -`make local-tick` runs one pass by hand. For ongoing operation, start the background scheduler. The four pipeline stages each tick on their own cadence (poll feeds, trigger watches, drain runs, flush digests): - -```bash -make up-jobs # start the tickers (a pid + log per stage under .jobs/) -tail -f .jobs/drain.log # watch a stage -make down-jobs # stop them -``` - -Each stage is single-flight: a pass that outruns its interval self-skips the next tick, so loops never stack. Production scheduling is then just a plain cron entry per command on the same cadences, with no flock or singleton infrastructure. Override any cadence inline, e.g. `make up-jobs DRAIN_INTERVAL=30`. - -Run `make help` for the full target list (`make up` / `down`, `make logs`, `make local-test`, `make local-check`, and so on). - -### Upgrading - -The quickstart pins your checkout to the latest release. When a newer one ships, upgrade in place from your install directory: - -```bash -make upgrade # or: ./scripts/upgrade.sh -``` - -It advances the checkout to the latest release tag, rebuilds the stack, applies migrations, and refreshes the `magpie` CLI. **Your data is preserved** (the database volume persists, migrations are additive, and it never re-seeds). For the bleeding edge instead of a release, `OPENMAGPIE_BRANCH=main make upgrade`. - -## How it works - -A `Feed` is a reusable, curated stream (a set of sources plus an item log). A `Watch` subscribes to one or more feeds and runs an ordered **action chain** over each new item: a `semantic_filter` gates the chain (a score below threshold stops it), and downstream `webhook` / `log` actions deliver what passes. One feed can back many watches, so you pay for source polling once. - -```mermaid -graph TD - subgraph Sources - REDDIT[Reddit] - RSS[RSS / Atom feeds] - HN[Hacker News] - SLACK[Slack] - LINKEDIN[LinkedIn] - GITHUB[GitHub] - end - - subgraph OpenMagpie - FEED[Feed
curated streams + item log] - WATCH[Watch
subscribes to feeds] - FILTER[semantic_filter
action] - ENGINE[Relevance engine
BYO LLM] - DELIVER[webhook / log
delivery action] - end - - subgraph Out - WEBHOOK[Webhook] - LOG[Log] - FUTURE["email / Slack (planned)"] - end - - REDDIT --> FEED - RSS --> FEED - HN --> FEED - SLACK -. planned .-> FEED - LINKEDIN -. planned .-> FEED - GITHUB -. planned .-> FEED - - FEED -- "new items" --> WATCH - WATCH -- "action chain" --> FILTER - FILTER --> ENGINE - ENGINE -. "your LLM" .-> LLM["any OpenAI-compatible /v1 API
Ollama | vLLM | llama.cpp | LM Studio | OpenAI"] - FILTER -- "passes -> next action" --> DELIVER - - DELIVER --> WEBHOOK - DELIVER --> LOG - DELIVER -. planned .-> FUTURE -``` - -Delivery is **instant** (per item) or **digest** (a window of items batched into one emission). A `webhook` action POSTs (or PUTs / PATCHes) one self-describing body; instant and digest use the same shape (instant is a one-item batch): - -```json -{ - "watch": {"id": "01K...", "name": "ai-webhook"}, - "action_id": "01K...", - "delivery": "digest", - "window": {"since": "...", "until": "..."}, - "items": [ - { - "key": "reddit_subreddit:abc123", - "source": {"label": "r/ClaudeAI", "kind": "reddit_subreddit"}, - "item": {"title": "...", "url": "..."} - } - ] -} -``` - -`item` is the feed item narrowed to the action's `include_fields`. Each item's `key` is `source:external_id`; delivery is at-least-once, so receivers dedup on it. Every call is recorded as a `WatchActionDelivery` you can inspect: - -```bash -make local-cli ARGS="delivery list --action " # the list: state / HTTP / host / items / attempt -make local-cli ARGS="delivery get " # one call in full, incl. the exact body sent -``` - -See [AGENTS.md](AGENTS.md) for the design conventions (char pointers, typed-blob pattern, the trigger/drain/flush execution model). - -## Why self-host it - -Social listening is a crowded market (Brand24, Mention, Octolens, Syften, and tools like OutX that pair monitoring with AI-drafted replies). They are all closed SaaS behind a paid plan, a trial, or a sales demo, and the few genuinely free options are basic mention notifiers, not full listening. OpenMagpie is the open, self-hostable exception: run it on your own box with your own model for the cost of the hardware. - -- **Open source.** Apache 2.0, the whole stack. Read it, fork it, and extend the connectors and engines yourself. -- **Bring your own LLM.** Relevance is judged by an LLM you run (via any OpenAI-compatible backend like Ollama, vLLM, llama.cpp, LM Studio etc), so your criteria and your matches stay on your infrastructure when you self-host the model. -- **Natural-language matching.** You describe what's relevant in natural language and the model scores each new post on meaning. -- **Auditable.** Every poll, judgement, and delivery is a row you can inspect (`magpie activity summary` / `delivery list`), as a table or `--jsonl` to pipe into `jq` / an LLM, or written to a file with `-o`. - -## What's shipped today - -| Layer | Shipped | -|---|---| -| Connectors | Reddit (`reddit_subreddit`), Hacker News (`hn_feed`, `hn_comment`), RSS/Atom (`rss`) | -| Engines | Any OpenAI-compatible `/v1` API: Ollama, vLLM, llama.cpp, LM Studio, OpenAI, ... | -| Action kinds | `semantic_filter` (LLM-judged), `webhook`, `log` | -| Delivery modes | instant, digest | -| Webhook methods | `POST`, `PUT`, `PATCH` | -| Delivery audit | per-attempt `WatchActionDelivery` | - -## Roadmap - -- **More connectors**: Slack, LinkedIn, GitHub, Bluesky, Mastodon, and X. -- **More engines**: Anthropic, OpenAI, and a keyword engine behind the same `Engine` Protocol. -- **Learns from feedback**: thumbs up/down on past matches become few-shot examples for the next pass. -- **Run-history in the payload**: the upstream filter score and chain provenance as an opt-in webhook field. -- **Branching and parallel chains**: the data model already carries `WatchPath` and dense action ranks; multi-path and DAG branching are post-v1. -- **Retention**: pruning for `WatchActionRun` and `WatchActionDelivery` history. - -## Hosted version - -Self-hosting is free and stays free. A managed hosted version (no infrastructure to run) is in the works as the paid tier. - -Join the waitlist at [openmagpie.ai](https://www.openmagpie.ai/). - -## Project structure - -uv workspace; one root `uv.lock` for everything Python. - -``` -apps/ - core/ Django backend (deployable) - common/ BaseModel (ULID PK + timestamps), ULIDField, locks, db ceilings, /healthz - accounts/ User / Account / UserProfile + services + AccountScopedAPIView mixin - auth_api/ signup / login / logout / me + tokens/* + device-flow handshake (DRF) - sources/ Connectors (Reddit, Hacker News, RSS/Atom) + SourcePayload classes + registry - feeds/ Feed + Source + FeedItem models + poll orchestrator + item log - engine/ Engine Protocol + OpenAICompatEngine + registry (+ probe) - watches/ Watch + WatchFeed + WatchPath + WatchAction + WatchActionRun + WatchActionDelivery - conf/ settings (base/local), urls, wsgi - cli/ magpie CLI (Typer + httpx + Pydantic); distributed as a standalone wheel -packages/ - openmagpie-schema/ Pure Pydantic models shared by core + cli (configs, wire types, feed shapes) -web/ pnpm workspace: apps/{app,marketing,email-render} (Next.js) + packages/{ui,api-utils,auth,tailwind-config} -make/ Per-concern Makefile targets -scripts/ quickstart installer (quickstart/{bootstrap,preflight,run,seed,tick}.sh) + dev tooling (Docker preflight, git hooks, whitespace/branch/length checks, make-help) -``` - -## Documentation - -- [CONTRIBUTING.md](CONTRIBUTING.md): contribution flow, branch naming, running the checks. -- [CHANGELOG.md](CHANGELOG.md): notable changes per release. -- [magpie CLI reference](apps/cli/README.md): install + the full command list. -- [make/README.md](make/README.md): the important dev `make` commands (`make help` for the full list). -- [AGENTS.md](AGENTS.md): cross-cutting design conventions, plus per-area notes: [apps/core](apps/core/AGENTS.md), [apps/cli](apps/cli/AGENTS.md), [web](web/AGENTS.md). - -## Telemetry - -OpenMagpie ships **anonymous, opt-in** usage telemetry, **off by default**. It helps prioritize what to build (a UI? which sources next? is setup too hard?) without ever sending your content. Enable it during `quickstart`, or with `make local-manage CMD="telemetry enable"`; turn it off any time with `make local-manage CMD="telemetry disable"` or `DO_NOT_TRACK=1`. Exactly what is and isn't collected: [apps/core/TELEMETRY.md](apps/core/TELEMETRY.md). - -## License - -OpenMagpie is open source under the [Apache License 2.0](LICENSE), with optional enterprise directories (`**/ee/`) reserved for future commercial features. +

+ OpenMagpie +

+ +

+ Open-source social listening +

+ +

+ License +

+ +

+ openmagpie.ai | + Quickstart | + CLI reference | + Changelog +

+ +--- + +

+ magpie CLI tour: the feed's sources, the watch's semantic filter, then the matches with scores and links +

+ +> [!TIP] +> **CLI not for you?** A UI / hosted version is on the way. Star the repo for updates, or join the waitlist at [openmagpie.ai](https://www.openmagpie.ai/). + +## What it does + +You scan Reddit, Hacker News, and a few RSS feeds looking for someone hitting a problem your product solves or asking a question you can answer well. Getting there while the conversation is happening is how you build a brand and a community around what you know. OpenMagpie watches the threads for you so you spend your time on engagement instead of searching. + +You curate sources into a feed, write a natural-language description of what's relevant (for example, "someone frustrated with manual social monitoring and asking for alternatives"), and a local LLM run via any OpenAI-compatible runner (e.g. Ollama, vLLM, LM Studio) scores each new post against it. Matches go to a webhook or your logs (more integrations coming); everything else is dropped. You read the hits instead of the firehose. + +## Where it listens + +OpenMagpie listens wherever communities are having those conversations. + +- **Public discussion (today):** Reddit, Hacker News, and any RSS or Atom feed (news, blogs, Substack publications, and forums that publish feeds). +- **Communities you're in (roadmap):** Slack workspaces and LinkedIn you already belong to, so you catch relevant threads in the groups where you participate, no admin or app install required. + +## Quickstart + +One command for your first real match (needs Docker and uv; it clones the repo and runs the quickstart for you): + +```bash +curl -fsSL https://openmagpie.ai | sh +``` + +Once setup finishes, keep processing new posts in the background: + +```bash +make up-jobs # run the schedulers in the background +tail -f .jobs/*.log # watch them work +``` + +The quickstart from curl to first matches: prerequisites check, LLM setup prompts, personalizing the listener (which subreddits, what to flag, how strict), then the seeded backlog scoring and the ready summary + +Prefer to not generate seed data? `SKIP_DATA_SEED=1` brings up the stack without sample data: + +```bash +curl -fsSL https://openmagpie.ai | SKIP_DATA_SEED=1 sh +``` + +Prefer to clone first? + +```bash +git clone https://github.com/obris-dev/openmagpie.git +cd openmagpie +./scripts/quickstart/run.sh +``` + +Either way it walks you through your first listener (which subreddits to watch, what to flag in plain language, how strict), seeds it, and once an LLM is reachable runs the pipeline once so the first matches print straight to the logs, tagged `[quickstart]`. Matches show up in the terminal and the CLI activity log, not the web UI yet. Your feed and watch are saved as editable YAML in `config/quickstart/` (see [config/README.md](config/README.md) for editing and reusing them). Want more posts to start with? `DAYS=7 ./scripts/quickstart/seed.sh` backfills a week instead of a day. See [examples/README.md](examples/README.md) for ready-made starters to apply by hand. Or have an AI assistant interview you and build the config: copy the prompt in [Set it up with an AI assistant](examples/README.md#set-it-up-with-an-ai-assistant). + +### Prereq: an OpenAI-compatible LLM endpoint + +OpenMagpie is BYO LLM; the dev stack doesn't bundle one. Whatever you already run almost certainly works, because **Ollama, vLLM, llama.cpp, and LM Studio all expose an OpenAI-compatible `/v1` API** (so do hosted providers like OpenAI, Together, or Groq). OpenMagpie talks to that `/v1` endpoint with the standard OpenAI client, so you just point `ENGINE_BASE_URL` at it. The quickstart validates your endpoint and points you at the model of your choice. + +- **Local.** Any OpenAI-compatible server on your machine works; point `ENGINE_BASE_URL` at its `/v1`: Ollama (`http://host.docker.internal:11434/v1`), vLLM (`:8000/v1`), LM Studio (`:1234/v1`), or llama.cpp (`:8080/v1`). The shipped default is Ollama's `:11434`. New here and want the quickest start? Install [Ollama](https://ollama.com/download), then `ollama pull qwen2.5:7b && ollama serve`. +- **Remote (LAN box, GPU server, cloud).** Set `ENGINE_BASE_URL=http://your-host:11434/v1` in `apps/core/.env`. +- **Hosted API.** Set `ENGINE_BASE_URL=https://api.openai.com/v1` and `ENGINE_API_KEY=...` (local servers leave the key blank). + +Set `ENGINE_MODEL` to the model you want to judge with. A 7B model judges in roughly 1 to 3 seconds on Apple Silicon or a recent NVIDIA GPU; CPU-only works but is slower. + +### Use it + +The quickstart already built and seeded everything and put the `magpie` CLI on your `PATH`. Drive it from there: + +```bash +magpie auth login # browser device flow +magpie feed create # opens $EDITOR on a feed template (sources + retention) +magpie watch create # opens $EDITOR on a watch template (feeds + action chain) +magpie activity summary --action # per-state run breakdown for any action (filter, webhook, log) +``` + +On a headless box (a server you SSH into, no browser), skip the device flow and use a personal access token. Mint one on the server, then sign in with it on the box, the token is pasted (stdin or a hidden prompt, never the command line) and stored in `~/.magpie` at `0600`, persisting across sessions: + +```bash +# on the server (the issue_cli_token management command, via the local stack): +make local-manage CMD="issue_cli_token --email you@example.com --name my-box" +# then, on the box: +magpie auth login --token # paste the printed token at the prompt +``` + +For CI or an ephemeral box, set `MAGPIE_TOKEN=mgp_...` in the environment instead: it's +read on every request, takes precedence over a stored login, and is never persisted (the +`GH_TOKEN` pattern), so no login step. Manage tokens with `magpie auth token list` / +`create` / `revoke` (minting needs a browser login; a token can't mint another). + +A watch's `actions:` chain typically starts with a `semantic_filter` (your natural-language criteria + threshold) followed by a `webhook` or `log` delivery. Pick a backfill window when you create the feed and the first `make local-tick` scores real posts against your criteria immediately, with no wait for the scheduler. + +Full command list: the [magpie CLI reference](apps/cli/README.md). The dev loop runs through `make`: see [make/README.md](make/README.md) or `make help`. + +### Running it continuously + +`make local-tick` runs one pass by hand. For ongoing operation, start the background scheduler. The four pipeline stages each tick on their own cadence (poll feeds, trigger watches, drain runs, flush digests): + +```bash +make up-jobs # start the tickers (a pid + log per stage under .jobs/) +tail -f .jobs/drain.log # watch a stage +make down-jobs # stop them +``` + +Each stage is single-flight: a pass that outruns its interval self-skips the next tick, so loops never stack. Production scheduling is then just a plain cron entry per command on the same cadences, with no flock or singleton infrastructure. Override any cadence inline, e.g. `make up-jobs DRAIN_INTERVAL=30`. + +Run `make help` for the full target list (`make up` / `down`, `make logs`, `make local-test`, `make local-check`, and so on). + +### Upgrading + +The quickstart pins your checkout to the latest release. When a newer one ships, upgrade in place from your install directory: + +```bash +make upgrade # or: ./scripts/upgrade.sh +``` + +It advances the checkout to the latest release tag, rebuilds the stack, applies migrations, and refreshes the `magpie` CLI. **Your data is preserved** (the database volume persists, migrations are additive, and it never re-seeds). For the bleeding edge instead of a release, `OPENMAGPIE_BRANCH=main make upgrade`. + +## How it works + +A `Feed` is a reusable, curated stream (a set of sources plus an item log). A `Watch` subscribes to one or more feeds and runs an ordered **action chain** over each new item: a `semantic_filter` gates the chain (a score below threshold stops it), and downstream `webhook` / `log` actions deliver what passes. One feed can back many watches, so you pay for source polling once. + +```mermaid +graph TD + subgraph Sources + REDDIT[Reddit] + RSS[RSS / Atom feeds] + HN[Hacker News] + SLACK[Slack] + LINKEDIN[LinkedIn] + GITHUB[GitHub] + end + + subgraph OpenMagpie + FEED[Feed
curated streams + item log] + WATCH[Watch
subscribes to feeds] + FILTER[semantic_filter
action] + ENGINE[Relevance engine
BYO LLM] + DELIVER[webhook / log
delivery action] + end + + subgraph Out + WEBHOOK[Webhook] + LOG[Log] + FUTURE["email / Slack (planned)"] + end + + REDDIT --> FEED + RSS --> FEED + HN --> FEED + SLACK -. planned .-> FEED + LINKEDIN -. planned .-> FEED + GITHUB -. planned .-> FEED + + FEED -- "new items" --> WATCH + WATCH -- "action chain" --> FILTER + FILTER --> ENGINE + ENGINE -. "your LLM" .-> LLM["any OpenAI-compatible /v1 API
Ollama | vLLM | llama.cpp | LM Studio | OpenAI"] + FILTER -- "passes -> next action" --> DELIVER + + DELIVER --> WEBHOOK + DELIVER --> LOG + DELIVER -. planned .-> FUTURE +``` + +Delivery is **instant** (per item) or **digest** (a window of items batched into one emission). A `webhook` action POSTs (or PUTs / PATCHes) one self-describing body; instant and digest use the same shape (instant is a one-item batch): + +```json +{ + "watch": {"id": "01K...", "name": "ai-webhook"}, + "action_id": "01K...", + "delivery": "digest", + "window": {"since": "...", "until": "..."}, + "items": [ + { + "key": "reddit_subreddit:abc123", + "source": {"label": "r/ClaudeAI", "kind": "reddit_subreddit"}, + "item": {"title": "...", "url": "..."} + } + ] +} +``` + +`item` is the feed item narrowed to the action's `include_fields`. Each item's `key` is `source:external_id`; delivery is at-least-once, so receivers dedup on it. Every call is recorded as a `WatchActionDelivery` you can inspect: + +```bash +make local-cli ARGS="delivery list --action " # the list: state / HTTP / host / items / attempt +make local-cli ARGS="delivery get " # one call in full, incl. the exact body sent +``` + +See [AGENTS.md](AGENTS.md) for the design conventions (char pointers, typed-blob pattern, the trigger/drain/flush execution model). + +## Why self-host it + +Social listening is a crowded market (Brand24, Mention, Octolens, Syften, and tools like OutX that pair monitoring with AI-drafted replies). They are all closed SaaS behind a paid plan, a trial, or a sales demo, and the few genuinely free options are basic mention notifiers, not full listening. OpenMagpie is the open, self-hostable exception: run it on your own box with your own model for the cost of the hardware. + +- **Open source.** Apache 2.0, the whole stack. Read it, fork it, and extend the connectors and engines yourself. +- **Bring your own LLM.** Relevance is judged by an LLM you run (via any OpenAI-compatible backend like Ollama, vLLM, llama.cpp, LM Studio etc), so your criteria and your matches stay on your infrastructure when you self-host the model. +- **Natural-language matching.** You describe what's relevant in natural language and the model scores each new post on meaning. +- **Auditable.** Every poll, judgement, and delivery is a row you can inspect (`magpie activity summary` / `delivery list`), as a table or `--jsonl` to pipe into `jq` / an LLM, or written to a file with `-o`. + +## What's shipped today + +| Layer | Shipped | +|---|---| +| Connectors | YouTube (`youtube_search`), Reddit (`reddit_subreddit`), Hacker News (`hn_feed`, `hn_comment`), RSS/Atom (`rss`) | +| Engines | Any OpenAI-compatible `/v1` API: Ollama, vLLM, llama.cpp, LM Studio, OpenAI, ... | +| Action kinds | `semantic_filter` (LLM-judged), `webhook`, `log` | +| Delivery modes | instant, digest | +| Webhook methods | `POST`, `PUT`, `PATCH` | +| Delivery audit | per-attempt `WatchActionDelivery` | + +## What we've done + +YouTube listening is the connector added via yt-dlp. What shipped in this branch: + +- **`youtube_search` source kind** — a yt-dlp-based connector that runs YouTube search queries and maps results to a schema-parity `NewVideoPayload`, registered alongside the existing kinds with the same feed/watch/webhook pipeline. +- **No authentication required** — public YouTube search works without credentials; optional cookie file for age-restricted content. +- **Error taxonomy** — 6 error codes (`video_unavailable`, `rate_limited`, `js_runtime_missing`, `network_error`, `date_parse_error`, `yt_dlp_error`) with retry semantics. +- **Watermark-based deduplication** — videos newer than the source's `last_event_at` are surfaced. +- **Metrics extraction** — views, likes, comments mapped from YouTube metadata. +- **Thumbnail media** — full thumbnail URLs attached to payloads for rich display. + +Next up on the roadmap: **Facebook, TikTok, and Instagram connectors** (soon to be added), then Slack, LinkedIn, GitHub, Bluesky, and Mastodon. + +## Roadmap + +- **More connectors**: Slack, LinkedIn, GitHub, Bluesky, Mastodon, and X. +- **More engines**: Anthropic, OpenAI, and a keyword engine behind the same `Engine` Protocol. +- **Learns from feedback**: thumbs up/down on past matches become few-shot examples for the next pass. +- **Run-history in the payload**: the upstream filter score and chain provenance as an opt-in webhook field. +- **Branching and parallel chains**: the data model already carries `WatchPath` and dense action ranks; multi-path and DAG branching are post-v1. +- **Retention**: pruning for `WatchActionRun` and `WatchActionDelivery` history. + +## Hosted version + +Self-hosting is free and stays free. A managed hosted version (no infrastructure to run) is in the works as the paid tier. + +Join the waitlist at [openmagpie.ai](https://www.openmagpie.ai/). + +## Project structure + +uv workspace; one root `uv.lock` for everything Python. + +``` +apps/ + core/ Django backend (deployable) + common/ BaseModel (ULID PK + timestamps), ULIDField, locks, db ceilings, /healthz + accounts/ User / Account / UserProfile + services + AccountScopedAPIView mixin + auth_api/ signup / login / logout / me + tokens/* + device-flow handshake (DRF) + sources/ Connectors (Reddit, Hacker News, RSS/Atom) + SourcePayload classes + registry + feeds/ Feed + Source + FeedItem models + poll orchestrator + item log + engine/ Engine Protocol + OpenAICompatEngine + registry (+ probe) + watches/ Watch + WatchFeed + WatchPath + WatchAction + WatchActionRun + WatchActionDelivery + conf/ settings (base/local), urls, wsgi + cli/ magpie CLI (Typer + httpx + Pydantic); distributed as a standalone wheel +packages/ + openmagpie-schema/ Pure Pydantic models shared by core + cli (configs, wire types, feed shapes) +web/ pnpm workspace: apps/{app,marketing,email-render} (Next.js) + packages/{ui,api-utils,auth,tailwind-config} +make/ Per-concern Makefile targets +scripts/ quickstart installer (quickstart/{bootstrap,preflight,run,seed,tick}.sh) + dev tooling (Docker preflight, git hooks, whitespace/branch/length checks, make-help) +``` + +## Documentation + +- [CONTRIBUTING.md](CONTRIBUTING.md): contribution flow, branch naming, running the checks. +- [CHANGELOG.md](CHANGELOG.md): notable changes per release. +- [magpie CLI reference](apps/cli/README.md): install + the full command list. +- [make/README.md](make/README.md): the important dev `make` commands (`make help` for the full list). +- [AGENTS.md](AGENTS.md): cross-cutting design conventions, plus per-area notes: [apps/core](apps/core/AGENTS.md), [apps/cli](apps/cli/AGENTS.md), [web](web/AGENTS.md). + +## Telemetry + +OpenMagpie ships **anonymous, opt-in** usage telemetry, **off by default**. It helps prioritize what to build (a UI? which sources next? is setup too hard?) without ever sending your content. Enable it during `quickstart`, or with `make local-manage CMD="telemetry enable"`; turn it off any time with `make local-manage CMD="telemetry disable"` or `DO_NOT_TRACK=1`. Exactly what is and isn't collected: [apps/core/TELEMETRY.md](apps/core/TELEMETRY.md). + +## License + +OpenMagpie is open source under the [Apache License 2.0](LICENSE), with optional enterprise directories (`**/ee/`) reserved for future commercial features. diff --git a/apps/core/pyproject.toml b/apps/core/pyproject.toml index 39abc14a..ca2cae8b 100644 --- a/apps/core/pyproject.toml +++ b/apps/core/pyproject.toml @@ -24,6 +24,7 @@ dependencies = [ "python-dotenv>=1.1", "pyyaml>=6.0", # reads the examples/starters/*.yaml in seed_quickstart "trafilatura>=1.7", # HTML -> readable article text for the engine's lazy external-link fetch + "yt-dlp>=2026.07.04", # YouTube search connector (public API only) "ulid>=1.1", ] diff --git a/apps/core/sources/connectors/__init__.py b/apps/core/sources/connectors/__init__.py index 041ba035..54e1557a 100644 --- a/apps/core/sources/connectors/__init__.py +++ b/apps/core/sources/connectors/__init__.py @@ -2,6 +2,7 @@ from .hackernews import HackerNewsCommentConnector, HackerNewsFeedConnector from .reddit import RedditSubRedditConnector from .rss import RssConnector +from .youtube import YouTubeSearchConnector __all__ = [ "Connector", @@ -9,4 +10,5 @@ "HackerNewsFeedConnector", "RedditSubRedditConnector", "RssConnector", + "YouTubeSearchConnector", ] diff --git a/apps/core/sources/connectors/youtube/__init__.py b/apps/core/sources/connectors/youtube/__init__.py new file mode 100644 index 00000000..926ed05d --- /dev/null +++ b/apps/core/sources/connectors/youtube/__init__.py @@ -0,0 +1,7 @@ +from .connector import YouTubeSearchConnector +from .payloads import NewVideoPayload + +__all__ = [ + "YouTubeSearchConnector", + "NewVideoPayload", +] diff --git a/apps/core/sources/connectors/youtube/client.py b/apps/core/sources/connectors/youtube/client.py new file mode 100644 index 00000000..5802a0a9 --- /dev/null +++ b/apps/core/sources/connectors/youtube/client.py @@ -0,0 +1,95 @@ +"""yt-dlp-based YouTube client for search extraction. + +Wraps yt-dlp's YoutubeDL to perform YouTube searches without downloading +video content. Uses extract_flat mode for efficiency and handles errors +via the error taxonomy in errors.py. + +Key patterns (ported from listeningkit Twitter client): +- One YtDlpClient instance per search call; yt-dlp is thread-safe for + read-only extraction operations. +- Search queries use the `ytsearch:` URI scheme. +- Results are returned as dicts (not downloaded), containing metadata. +- No authentication required for public search; cookies optional for + age-restricted content. +""" + +from __future__ import annotations + +import logging +from typing import Any + +import yt_dlp + +from .errors import YouTubeError, map_ytdlp_error + +log = logging.getLogger("sources.youtube") + +# Maximum results per search query. yt-dlp accepts up to 100 but we cap +# lower to match the Twitter connector's default count. +MAX_SEARCH_RESULTS = 50 + + +class YtDlpClient: + """Thin wrapper around yt-dlp for search-only extraction. + + No auth state: YouTube search is public. Optional cookie file can be + passed for age-restricted content (not commonly needed for search). + """ + + def __init__( + self, + *, + quiet: bool = True, + no_warnings: bool = True, + cookie_file: str | None = None, + ) -> None: + self._quiet = quiet + self._no_warnings = no_warnings + self._cookie_file = cookie_file + + def _build_opts(self) -> dict[str, Any]: + opts: dict[str, Any] = { + "quiet": self._quiet, + "no_warnings": self._no_warnings, + "extract_flat": False, # need full metadata for payloads + "skip_download": True, + } + if self._cookie_file: + opts["cookies"] = self._cookie_file + return opts + + def search( + self, + query: str, + count: int = 20, + ) -> list[dict[str, Any]]: + """Run one YouTube search; returns list of video info dicts. + + Args: + query: Search expression (keywords, phrases). + count: Max results to fetch (capped at MAX_SEARCH_RESULTS). + + Returns: + List of video metadata dicts, newest first. + + Raises: + YouTubeError: On extraction failures (mapped from yt-dlp exceptions). + """ + capped_count = min(count, MAX_SEARCH_RESULTS) + search_uri = f"ytsearch{capped_count}:{query}" + + try: + with yt_dlp.YoutubeDL(self._build_opts()) as ydl: + info = ydl.extract_info(search_uri, download=False) + entries = info.get("entries", []) or [] + return [e for e in entries if e is not None] + except Exception as exc: + err = map_ytdlp_error(exc, {"query": query, "count": capped_count}) + log.warning("youtube search failed query=%r code=%s: %s", query, err.code, err.message) + raise YouTubeError( + code=err.code, + message=err.message, + retryable=err.retryable, + action=err.action, + context=err.context, + ) from exc diff --git a/apps/core/sources/connectors/youtube/connector.py b/apps/core/sources/connectors/youtube/connector.py new file mode 100644 index 00000000..c9b23720 --- /dev/null +++ b/apps/core/sources/connectors/youtube/connector.py @@ -0,0 +1,83 @@ +"""YouTube search connector using yt-dlp. + +Polls a `youtube_search` source: one live YouTube search per cycle via +the yt-dlp client, mapping each result video to a `NewVideoPayload` +newer than the source's `since` watermark. + +Error semantics follow the connector contract: any YouTube/yt-dlp +failure is raised as `ConnectorParseError` (a `_RECOVERABLE_ERRORS` +member at the poll seam), so a bad source logs + skips instead of +aborting the feed cycle. The source's watermark stays put on failure, +so the next cycle re-reads from the same point and the external_id +dedup absorbs anything already recorded. +""" + +from __future__ import annotations + +import logging +from collections.abc import Callable, Iterator +from datetime import datetime + +from openmagpie_schema.configs import YouTubeSearchSourceSpec +from sources.payload_registry import register +from sources.payloads import SourcePayload + +from ..base import BaseConnector, ConnectorParseError +from .client import YtDlpClient +from .errors import YouTubeError +from .payloads import NewVideoPayload + +log = logging.getLogger("sources.youtube") + + +class YouTubeSearchConnector(BaseConnector[YouTubeSearchSourceSpec]): + """Polls one YouTube search stream via yt-dlp. + + Live-mode semantics mirror the other connectors: every cycle yields + videos newer than `since` (the Source row's `last_event_at`). There + is no pagination in phase 1: a search returns up to `spec.count` + videos and the connector filters them by the watermark (YouTube's + search ordering is newest-first; a quiet stream needs no backfill + walk). + """ + + kind = YouTubeSearchSourceSpec.SOURCE_KIND + payloads: list[type[SourcePayload]] = [NewVideoPayload] + + # One stateless client; no auth needed for public search. + _client = YtDlpClient() + + def poll( + self, + spec: YouTubeSearchSourceSpec, + since: datetime | None, + field_map: dict[str, str] | None = None, + heartbeat: Callable[[], bool] | None = None, + ) -> Iterator[SourcePayload]: + del field_map + del heartbeat + try: + results = self._client.search(spec.query, spec.count) + except YouTubeError as exc: + log.warning( + "youtube search failed query=%r code=%s retryable=%s: %s", + spec.query, + exc.code, + exc.retryable, + exc.message, + ) + raise ConnectorParseError( + f"youtube search {spec.display()} failed: {exc.code}: {exc.message} ({exc.action})" + ) from exc + + for video in results: + payload = NewVideoPayload.from_video(video) + # Watermark filter: only surface videos strictly newer than the + # cursor (the poll op advances the source watermark to the + # newest seen, so a video at the watermark is already recorded). + if since is not None and payload.occurred_at <= since: + continue + yield payload + + +register(YouTubeSearchConnector.kind, YouTubeSearchConnector.payloads) diff --git a/apps/core/sources/connectors/youtube/errors.py b/apps/core/sources/connectors/youtube/errors.py new file mode 100644 index 00000000..7a9d06a7 --- /dev/null +++ b/apps/core/sources/connectors/youtube/errors.py @@ -0,0 +1,92 @@ +"""Error taxonomy for the YouTube (yt-dlp) connector. + +Maps yt-dlp exceptions to canonical error shapes with retry semantics, +following the same pattern as the Twitter connector's ListenerError. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class YouTubeError: + """Canonical error shape for one YouTube fetch failure.""" + + code: str # stable machine code + message: str # human-readable + retryable: bool # safe to retry with backoff? + action: str # what the ops layer should do + context: dict[str, Any] = field(default_factory=dict) + + +# Error codes for YouTube-specific failures. +YT_DLP_ERROR_CODES: dict[type[Exception], str] = { + # yt-dlp DownloadError subclasses + Exception: "yt_dlp_error", # catch-all +} + + +def map_ytdlp_error(exc: Exception, context: dict[str, Any] | None = None) -> YouTubeError: + """Translate an yt-dlp exception into a canonical YouTubeError.""" + msg = str(exc) + + # Video not available (region-restricted, deleted, private) + if "This video is not available" in msg or "Video unavailable" in msg: + return YouTubeError( + code="video_unavailable", + message=msg, + retryable=False, + action="skip (video no longer available)", + context=context or {}, + ) + + # Rate limiting / throttling + if "rate limited" in msg.lower() or "too many requests" in msg.lower(): + return YouTubeError( + code="rate_limited", + message=msg, + retryable=True, + action="retry with exponential backoff", + context=context or {}, + ) + + # Missing JavaScript runtime (warning only, still works in degraded mode) + if "No supported JavaScript runtime" in msg: + return YouTubeError( + code="js_runtime_missing", + message=msg, + retryable=False, + action="install deno or node; proceeding in degraded mode", + context=context or {}, + ) + + # Network/connection errors + if any(marker in msg.lower() for marker in ["connection", "timeout", "network", "urlopen"]): + return YouTubeError( + code="network_error", + message=msg, + retryable=True, + action="retry with backoff", + context=context or {}, + ) + + # Upload date parsing failures + if "upload_date" in msg.lower() or "date" in msg.lower(): + return YouTubeError( + code="date_parse_error", + message=msg, + retryable=False, + action="use current timestamp as fallback", + context=context or {}, + ) + + # Generic fallback + return YouTubeError( + code="yt_dlp_error", + message=msg, + retryable=True, + action="log and retry with backoff", + context=context or {}, + ) diff --git a/apps/core/sources/connectors/youtube/payloads.py b/apps/core/sources/connectors/youtube/payloads.py new file mode 100644 index 00000000..e3d383ac --- /dev/null +++ b/apps/core/sources/connectors/youtube/payloads.py @@ -0,0 +1,140 @@ +"""YouTube payloads: a video observed via yt-dlp search. + +Maps YouTube video metadata to the openmagpie SourcePayload contract: +the engine judges title + content, so the video's description goes to +content and the uploader's name becomes the within-kind source_slug. +Metrics / refs / media stay on the payload as source-specific fields. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any, ClassVar + +from openmagpie_schema.configs import YouTubeSearchSourceSpec +from sources.payloads import SourcePayload + +# YouTube video URL base. +YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v=" + + +class NewVideoPayload(SourcePayload): + """A single YouTube video observed by a watched search stream. + + `author` is the channel name; `handle` is the channel ID and the + within-kind source slug (grouping items by producing channel). + `content` is the video description (the engine's judgeable body). + The rest is source-specific: `metrics`, `refs` (related video IDs), + `media` (thumbnails), `duration`. + """ + + PAYLOAD_KIND: ClassVar[str] = "new_video" + + author: str = "" + handle: str = "" + duration: int = 0 # seconds + metrics: dict[str, int | None] = {} + refs: dict[str, str | None] = {} + media: list[dict[str, Any]] = [] + + model_config = {"frozen": True, "extra": "ignore"} + + def source_slug(self) -> str | None: + return self.handle or None + + @classmethod + def sample(cls, variant: int = 0) -> NewVideoPayload: + n = variant + 1 + video_id = str(999_000_000_000_000_000 + n) + handle = f"example_channel_{n}" + return cls( + external_id=video_id, + kind=cls.PAYLOAD_KIND, + occurred_at=datetime(2026, 5, 27, 12, 0, tzinfo=UTC), + source=YouTubeSearchSourceSpec.SOURCE_KIND, + title="", + content=f"Example YouTube video {n}: the description text that matched this watch.", + url=f"{YOUTUBE_VIDEO_URL}{video_id}", + author=f"Example Channel {n}", + handle=handle, + duration=60 * n, + metrics={"views": 1000 + n, "likes": 100 + n, "comments": 10 + n}, + refs={}, + media=[], + ) + + @classmethod + def from_video(cls, video: dict[str, Any]) -> NewVideoPayload: + """Map a yt-dlp video info dict to a payload. + + yt-dlp returns videos as plain dicts when extract_info is called + on a search URI. All attributes are accessed via dict get() with + defaults, so the connector's unit tests can hand in lightweight + fakes without importing yt-dlp. + """ + video_id = str(video.get("id") or "") + uploader = str(video.get("uploader") or video.get("channel") or "") + uploader_id = str(video.get("uploader_id") or video.get("channel_id") or "") + description = str(video.get("description") or "") + upload_date = str(video.get("upload_date") or "") + + # Parse upload_date (format: YYYYMMDD) to datetime. + occurred_at = datetime.now(UTC) + if upload_date and len(upload_date) == 8: + try: + occurred_at = datetime.strptime(upload_date, "%Y%m%d").replace(tzinfo=UTC) + except ValueError: + pass + + # Duration in seconds. + duration = int(video.get("duration") or 0) + + # Metrics. + metrics = { + "views": int_or_none(video.get("view_count")), + "likes": int_or_none(video.get("like_count")), + "comments": int_or_none(video.get("comment_count")), + } + + # Media: thumbnails. + media = [] + for thumb in video.get("thumbnails") or []: + url = thumb.get("url") + if url: + media.append({ + "type": "thumbnail", + "url": url, + "width": int_or_none(thumb.get("width")), + "height": int_or_none(thumb.get("height")), + }) + # Fallback to thumbnail field if thumbnails list is empty. + if not media: + thumb_url = video.get("thumbnail") + if thumb_url: + media.append({"type": "thumbnail", "url": thumb_url}) + + return cls( + external_id=video_id, + kind=cls.PAYLOAD_KIND, + occurred_at=occurred_at, + source=YouTubeSearchSourceSpec.SOURCE_KIND, + title=str(video.get("title") or ""), + content=description, + url=str(video.get("webpage_url") or f"{YOUTUBE_VIDEO_URL}{video_id}"), + author=uploader, + handle=uploader_id, + duration=duration, + metrics=metrics, + refs={}, + media=media, + ) + + +def int_or_none(obj: Any) -> int | None: + """Safely convert to int or return None.""" + if obj is None: + return None + try: + return int(obj) + except (ValueError, TypeError): + return None diff --git a/apps/core/sources/registry.py b/apps/core/sources/registry.py index bfa6713d..4228fbfc 100644 --- a/apps/core/sources/registry.py +++ b/apps/core/sources/registry.py @@ -15,6 +15,7 @@ HackerNewsFeedConnector, RedditSubRedditConnector, RssConnector, + YouTubeSearchConnector, ) _REGISTRY: dict[str, Connector[Any]] = { @@ -22,6 +23,7 @@ RssConnector.kind: RssConnector(), HackerNewsFeedConnector.kind: HackerNewsFeedConnector(), HackerNewsCommentConnector.kind: HackerNewsCommentConnector(), + YouTubeSearchConnector.kind: YouTubeSearchConnector(), } # Core kinds captured before any plugin registers; a plugin can't replace one. diff --git a/packages/openmagpie-schema/src/openmagpie_schema/configs.py b/packages/openmagpie-schema/src/openmagpie_schema/configs.py index 34258ef5..faa98098 100644 --- a/packages/openmagpie-schema/src/openmagpie_schema/configs.py +++ b/packages/openmagpie-schema/src/openmagpie_schema/configs.py @@ -1,280 +1,312 @@ -"""Pure typed source specs, keyed by kind. - -SHARED, zero-Django source of truth (imported by core *and* the magpie -CLI). Carries only *shape* + pure transforms. The Django/settings-coupled -*policy* (SSRF / https rules, default engine kind, ...) is NOT here ; it -lives in `core` and runs at the server's validation seam. Splitting shape -from policy is what lets this module be a dependency-free shared package. -""" - -import json -import re -from typing import Annotated, ClassVar, Literal, NamedTuple, get_args -from urllib.parse import urlsplit - -from pydantic import BaseModel, Field, field_validator - -from ._unions import _PLUGIN_MEMBER_LOC_NAMES, KIND_MAX_LENGTH, builtin_union_kinds, reject_builtin_kind - -# ── Source specs (discriminated union over kind) ────────────────────────── - -# Reddit's max subreddit-name length; the slug validator bounds names to it. -MAX_SUBREDDIT_LENGTH = 21 - - -class RedditSubredditSourceSpec(BaseModel): - """Identity of one subreddit source. Bound to RedditSubRedditConnector.""" - - SOURCE_KIND: ClassVar[str] = "reddit_subreddit" - URL_FIELDS: ClassVar[tuple[str, ...]] = () # no operator-supplied URL to SSRF-check - - kind: Literal["reddit_subreddit"] = "reddit_subreddit" - subreddit: str - - @field_validator("subreddit") - @classmethod - def _validate_subreddit(cls, value: str) -> str: - """Validate + normalize the BARE subreddit name. A pasted `r/` or `/r/` - prefix is stripped (callers build the `r//...` request URL and - `display()` re-adds the prefix), the name is held to a URL-safe charset - (letters/digits/underscores, <=MAX_SUBREDDIT_LENGTH chars - nothing like - `/`, `?`, `#`, `+`, or whitespace that would break a request URL), and the - result is lowercased (subreddit names are case-insensitive, so that's the - one canonical identity). - - Deliberately a URL-safe subset, not Reddit's exact naming rule: it's looser - (allows 1-2 char names) and doesn't special-case `u_` user feeds. Unlike - RssSourceSpec.url's validate-only check, this also normalizes.""" - slug = re.sub(r"^/?r/", "", value.strip(), flags=re.IGNORECASE) - if not re.fullmatch(rf"[A-Za-z0-9_]{{1,{MAX_SUBREDDIT_LENGTH}}}", slug): - raise ValueError(f"invalid subreddit {value!r}: letters/digits/underscores, <={MAX_SUBREDDIT_LENGTH} chars") - # Subreddit names are case-insensitive (r/Python and r/python are the same - # sub), so normalize to the one canonical lowercase form. - return slug.lower() - - def display(self) -> str: - return f"r/{self.subreddit}" - - -class RssSourceSpec(BaseModel): - """Identity of one RSS/Atom source by URL. Bound to a generic RSS connector.""" - - SOURCE_KIND: ClassVar[str] = "rss" - # Fields the connector actually FETCHES, so the write-time SSRF gate checks only - # these (not display-only fields like `name`, which the connector never dereferences - # and which shouldn't 400 for containing a private-IP-looking string). - URL_FIELDS: ClassVar[tuple[str, ...]] = ("url",) - - kind: Literal["rss"] = "rss" - url: str - name: str = "" - - @field_validator("url") - @classmethod - def _validate_url_structural(cls, value: str) -> str: - """Structural check only (http/https scheme + host present). - Connector-side reachability / feed-format validation runs at poll - time. An empty URL slips through plain `str` typing and silently - produces a blank source_label downstream; reject it here.""" - parts = urlsplit(value) - if parts.scheme not in {"http", "https"}: - raise ValueError(f"rss URL scheme must be http or https, got {parts.scheme!r}") - if not parts.netloc: - raise ValueError(f"rss URL missing host: {value!r}") - return value - - def display(self) -> str: - return self.name or self.url - - -class _HackerNewsSpec(BaseModel): - """Shared fields for the Algolia-backed Hacker News specs. - - `query` is the server-side keyword pre-filter (Algolia full-text search); - `match` picks AND (default, every word must appear) vs ANY (OR, via - Algolia `optionalWords`). Empty `query` means no pre-filter (fine for the - low-volume story feeds; the comment spec makes it required).""" - - URL_FIELDS: ClassVar[tuple[str, ...]] = () # no operator-supplied URL to SSRF-check - - query: str = "" - match: Literal["all", "any"] = "all" - - -class HackerNewsFeedSourceSpec(_HackerNewsSpec): - """Identity of one Hacker News story feed. Bound to HackerNewsFeedConnector. - - `feed` selects which posts the connector pulls; it maps to an Algolia - HN Search `tags` value (new -> story, show -> show_hn, ask -> ask_hn). - The set is closed to the feeds that map to a single Algolia tag and a - newest-first date order; the ranked Firebase feeds (top / best) have no - Algolia equivalent and are intentionally out of scope here.""" - - SOURCE_KIND: ClassVar[str] = "hn_feed" - - kind: Literal["hn_feed"] = "hn_feed" - feed: Literal["new", "show", "ask"] = "new" - - def display(self) -> str: - return {"new": "Hacker News (new)", "show": "Show HN", "ask": "Ask HN"}.get(self.feed, self.feed) - - -class HackerNewsCommentSourceSpec(_HackerNewsSpec): - """Identity of one Hacker News COMMENT stream. Bound to HackerNewsCommentConnector. - - `tags=comment` unfiltered is the site-wide comment firehose (~20k/day on - average, bursty), so `query` is REQUIRED and NON-BLANK here (it overrides the - base default away). That is the structural guard that keeps an unfiltered - firehose from ever reaching the per-item relevance engine -- a blank or - whitespace-only query would strip to no pre-filter, so it is rejected at the - spec layer, not merely required-to-be-present. Volume past the keyword filter - is further bounded by the connector's page cap.""" - - SOURCE_KIND: ClassVar[str] = "hn_comment" - - kind: Literal["hn_comment"] = "hn_comment" - query: str = Field(min_length=1) # required + non-blank: the firehose guard (see _query_not_blank) - - @field_validator("query") - @classmethod - def _query_not_blank(cls, v: str) -> str: - # min_length=1 rejects "" ; this also rejects a whitespace-only query and - # stores it stripped. A blank query = no pre-filter = the whole firehose. - v = v.strip() - if not v: - raise ValueError("hn_comment requires a non-blank query (the firehose guard)") - return v - - def display(self) -> str: - return f'HN comments: "{self.query}"' - - -# The built-ins as a discriminated union over `kind` (defined before the plugin -# fallback so the built-in kind set can be derived from it below). A built-in kind -# with a malformed spec fails its typed member here and is rejected by the fallback, -# so it surfaces as a validation error rather than being absorbed as a raw blob. -_BuiltinSourceSpec = Annotated[ - RedditSubredditSourceSpec | RssSourceSpec | HackerNewsFeedSourceSpec | HackerNewsCommentSourceSpec, - Field(discriminator="kind"), -] - -# Built-in source kinds, DERIVED from the union members above rather than kept as a -# second hand-maintained list, so the set the plugin fallback rejects can never drift -# from the union (adding a member to _BuiltinSourceSpec extends this for free). Taken -# from each member's `SOURCE_KIND` ClassVar (a plain str, always present) rather than -# the `kind` field default, which is fragile: a multi-value Literal would capture only -# the default, and a member without a default would inject PydanticUndefined. -_BUILTIN_SOURCE_SPECS = get_args(get_args(_BuiltinSourceSpec)[0]) -# Reject-set: the kind values the union DISPATCHES on, via the SAME shared helper the -# action/run unions use (it reads every Literal arg, so a multi-value Literal is -# handled). Sharing the helper is why the SOURCE_KIND cross-pin below is the ONLY -# source-specific piece. -_BUILTIN_SOURCE_KINDS = builtin_union_kinds(_BuiltinSourceSpec) - -# Import-time guards, load-bearing so raised explicitly (a bare `assert` is stripped -# under `python -O`, and this is a shared library): -# (1) If the union is ever refactored down to a single member, `get_args` returns () -# and the reject-set goes empty, silently letting the fallback absorb EVERY -# malformed built-in spec (the exact hole this set closes). Require >=2, one per -# member. -# (2) Source-specific cross-pin: each member also carries a SOURCE_KIND ClassVar (the -# connector registry's key), declared independently of the `kind` Literal. Pin the -# Literal to exactly (SOURCE_KIND,) so the discriminator, the reject-set, and the -# connector key can't diverge (a divergence would let a malformed built-in slip -# past the reject-set). getattr default keeps a missing SOURCE_KIND a curated -# RuntimeError, not a bare AttributeError. -if not (len(_BUILTIN_SOURCE_KINDS) == len(_BUILTIN_SOURCE_SPECS) >= 2): - raise RuntimeError(f"expected >=2 built-in source kinds, one per union member; got {sorted(_BUILTIN_SOURCE_KINDS)}") -for _spec in _BUILTIN_SOURCE_SPECS: - if get_args(_spec.model_fields["kind"].annotation) != (getattr(_spec, "SOURCE_KIND", None),): - raise RuntimeError( - f"{_spec.__name__}: kind Literal {get_args(_spec.model_fields['kind'].annotation)} must be exactly " - f"(SOURCE_KIND,) = ({getattr(_spec, 'SOURCE_KIND', None)!r},); the discriminator, reject-set, and " - f"connector key would otherwise diverge" - ) - - -class PluginSourceSpec(BaseModel): - """Fallback spec member for a plugin (non-built-in) source kind. `kind` is any - non-built-in string; the rest of the spec is an open blob (a fork's typed spec - schema lives in its own contract, and its web/CLI narrow on `kind`). Selected - only when no built-in discriminator matches (the left-to-right union below). - `extra="allow"` keeps every submitted field through `model_dump(mode="json")`, - so `canonical_spec` / `source_identity` (the spec_hash basis) stay stable.""" - - model_config = {"extra": "allow"} - - kind: str = Field(min_length=1, max_length=KIND_MAX_LENGTH) - - @field_validator("kind") - @classmethod - def _not_builtin(cls, v: str) -> str: - return reject_builtin_kind(v, _BUILTIN_SOURCE_KINDS) - - def display(self) -> str: - # No typed shape, so fall back to an operator label if the blob carries one, - # else the kind. A fork's typed spec member supplies a real display(). - extra = self.model_extra or {} - label = extra.get("name") or extra.get("label") - return str(label) if label else self.kind - - -# One source spec, keyed by `kind`: the built-in discriminated union above, then a -# left-to-right fallthrough to the plugin member for any other kind (same discipline -# as WatchActionWire). -SourceSpec = Annotated[_BuiltinSourceSpec | PluginSourceSpec, Field(union_mode="left_to_right")] - -# Pin the fallback member name against `_unions._PLUGIN_MEMBER_LOC_NAMES` (see _nodes): -# a rename that isn't mirrored there silently turns off clean_union_errors' stripping. -if PluginSourceSpec.__name__ not in _PLUGIN_MEMBER_LOC_NAMES: - raise RuntimeError(f"{PluginSourceSpec.__name__} missing from _unions._PLUGIN_MEMBER_LOC_NAMES") - - -def canonical_spec(spec: SourceSpec) -> str: - """Canonical JSON for a source spec: the single identity both the server's - `spec_hash` (which sha256s this) and the magpie CLI's source-diff compare on. - Sorted keys + compact separators make it independent of field-declaration - order, so two specs denote the same source iff this string matches. Pure - shape; keep it byte-stable - a change reshuffles every stored `spec_hash` - (pinned by core's `SpecHashCanonicalTests`).""" - return json.dumps(spec.model_dump(mode="json"), sort_keys=True, separators=(",", ":")) - - -class SourceFields(BaseModel): - """The identity + operator-tunable config shared by both source envelopes - - `SourceInput` (write path) and `SourceWire` (read path). - - `meta` is operator-supplied free-form tags; the recorder copies it onto each - FeedItem the source produces. `field_map` overrides the feed-level - `default_field_map` for a single source; empty means inherit (connectors that - don't read it ignore it). The watermark (`last_event_at`) and server-assigned - fields (`id`, `created_at`) live on the envelopes, since they differ by - direction.""" - - spec: SourceSpec - meta: dict[str, str] = Field(default_factory=dict) - field_map: dict[str, str] = Field(default_factory=dict) - - -class SourceIdentity(NamedTuple): - """A source's full reconcile identity - everything `feed source set` keys on: - the spec (`canonical_spec`, the `spec_hash` basis) PLUS the mutable config it - refreshes, meta + field_map. Excludes last_event_at (watermarks are never - reconciled). Hashable + ordered, so callers diff source sets by plain - equality instead of ad-hoc JSON.""" - - spec: str - meta: tuple[tuple[str, str], ...] - field_map: tuple[tuple[str, str], ...] - - -def source_identity(source: SourceFields) -> SourceIdentity: - """Build the shared `SourceIdentity` from any source envelope (SourceInput or - SourceWire, via their `SourceFields` base) - the ONE definition the magpie - CLI's source-diff compares on and that the server's set_sources reconcile - mirrors (spec_hash + meta/field_map).""" - return SourceIdentity( - spec=canonical_spec(source.spec), - meta=tuple(sorted(source.meta.items())), - field_map=tuple(sorted(source.field_map.items())), - ) +"""Pure typed source specs, keyed by kind. + +SHARED, zero-Django source of truth (imported by core *and* the magpie +CLI). Carries only *shape* + pure transforms. The Django/settings-coupled +*policy* (SSRF / https rules, default engine kind, ...) is NOT here ; it +lives in `core` and runs at the server's validation seam. Splitting shape +from policy is what lets this module be a dependency-free shared package. +""" + +import json +import re +from typing import Annotated, ClassVar, Literal, NamedTuple, get_args +from urllib.parse import urlsplit + +from pydantic import BaseModel, Field, field_validator + +from ._unions import _PLUGIN_MEMBER_LOC_NAMES, KIND_MAX_LENGTH, builtin_union_kinds, reject_builtin_kind + +# ── Source specs (discriminated union over kind) ────────────────────────── + +# Reddit's max subreddit-name length; the slug validator bounds names to it. +MAX_SUBREDDIT_LENGTH = 21 + + +class RedditSubredditSourceSpec(BaseModel): + """Identity of one subreddit source. Bound to RedditSubRedditConnector.""" + + SOURCE_KIND: ClassVar[str] = "reddit_subreddit" + URL_FIELDS: ClassVar[tuple[str, ...]] = () # no operator-supplied URL to SSRF-check + + kind: Literal["reddit_subreddit"] = "reddit_subreddit" + subreddit: str + + @field_validator("subreddit") + @classmethod + def _validate_subreddit(cls, value: str) -> str: + """Validate + normalize the BARE subreddit name. A pasted `r/` or `/r/` + prefix is stripped (callers build the `r//...` request URL and + `display()` re-adds the prefix), the name is held to a URL-safe charset + (letters/digits/underscores, <=MAX_SUBREDDIT_LENGTH chars - nothing like + `/`, `?`, `#`, `+`, or whitespace that would break a request URL), and the + result is lowercased (subreddit names are case-insensitive, so that's the + one canonical identity). + + Deliberately a URL-safe subset, not Reddit's exact naming rule: it's looser + (allows 1-2 char names) and doesn't special-case `u_` user feeds. Unlike + RssSourceSpec.url's validate-only check, this also normalizes.""" + slug = re.sub(r"^/?r/", "", value.strip(), flags=re.IGNORECASE) + if not re.fullmatch(rf"[A-Za-z0-9_]{{1,{MAX_SUBREDDIT_LENGTH}}}", slug): + raise ValueError(f"invalid subreddit {value!r}: letters/digits/underscores, <={MAX_SUBREDDIT_LENGTH} chars") + # Subreddit names are case-insensitive (r/Python and r/python are the same + # sub), so normalize to the one canonical lowercase form. + return slug.lower() + + def display(self) -> str: + return f"r/{self.subreddit}" + + +class RssSourceSpec(BaseModel): + """Identity of one RSS/Atom source by URL. Bound to a generic RSS connector.""" + + SOURCE_KIND: ClassVar[str] = "rss" + # Fields the connector actually FETCHES, so the write-time SSRF gate checks only + # these (not display-only fields like `name`, which the connector never dereferences + # and which shouldn't 400 for containing a private-IP-looking string). + URL_FIELDS: ClassVar[tuple[str, ...]] = ("url",) + + kind: Literal["rss"] = "rss" + url: str + name: str = "" + + @field_validator("url") + @classmethod + def _validate_url_structural(cls, value: str) -> str: + """Structural check only (http/https scheme + host present). + Connector-side reachability / feed-format validation runs at poll + time. An empty URL slips through plain `str` typing and silently + produces a blank source_label downstream; reject it here.""" + parts = urlsplit(value) + if parts.scheme not in {"http", "https"}: + raise ValueError(f"rss URL scheme must be http or https, got {parts.scheme!r}") + if not parts.netloc: + raise ValueError(f"rss URL missing host: {value!r}") + return value + + def display(self) -> str: + return self.name or self.url + + +class _HackerNewsSpec(BaseModel): + """Shared fields for the Algolia-backed Hacker News specs. + + `query` is the server-side keyword pre-filter (Algolia full-text search); + `match` picks AND (default, every word must appear) vs ANY (OR, via + Algolia `optionalWords`). Empty `query` means no pre-filter (fine for the + low-volume story feeds; the comment spec makes it required).""" + + URL_FIELDS: ClassVar[tuple[str, ...]] = () # no operator-supplied URL to SSRF-check + + query: str = "" + match: Literal["all", "any"] = "all" + + +class HackerNewsFeedSourceSpec(_HackerNewsSpec): + """Identity of one Hacker News story feed. Bound to HackerNewsFeedConnector. + + `feed` selects which posts the connector pulls; it maps to an Algolia + HN Search `tags` value (new -> story, show -> show_hn, ask -> ask_hn). + The set is closed to the feeds that map to a single Algolia tag and a + newest-first date order; the ranked Firebase feeds (top / best) have no + Algolia equivalent and are intentionally out of scope here.""" + + SOURCE_KIND: ClassVar[str] = "hn_feed" + + kind: Literal["hn_feed"] = "hn_feed" + feed: Literal["new", "show", "ask"] = "new" + + def display(self) -> str: + return {"new": "Hacker News (new)", "show": "Show HN", "ask": "Ask HN"}.get(self.feed, self.feed) + + +class HackerNewsCommentSourceSpec(_HackerNewsSpec): + """Identity of one Hacker News COMMENT stream. Bound to HackerNewsCommentConnector. + + `tags=comment` unfiltered is the site-wide comment firehose (~20k/day on + average, bursty), so `query` is REQUIRED and NON-BLANK here (it overrides the + base default away). That is the structural guard that keeps an unfiltered + firehose from ever reaching the per-item relevance engine -- a blank or + whitespace-only query would strip to no pre-filter, so it is rejected at the + spec layer, not merely required-to-be-present. Volume past the keyword filter + is further bounded by the connector's page cap.""" + + SOURCE_KIND: ClassVar[str] = "hn_comment" + + kind: Literal["hn_comment"] = "hn_comment" + query: str = Field(min_length=1) # required + non-blank: the firehose guard (see _query_not_blank) + + @field_validator("query") + @classmethod + def _query_not_blank(cls, v: str) -> str: + # min_length=1 rejects "" ; this also rejects a whitespace-only query and + # stores it stripped. A blank query = no pre-filter = the whole firehose. + v = v.strip() + if not v: + raise ValueError("hn_comment requires a non-blank query (the firehose guard)") + return v + + def display(self) -> str: + return f'HN comments: "{self.query}"' + + +class YouTubeSearchSourceSpec(BaseModel): + """Identity of one YouTube search stream. Bound to YouTubeSearchConnector. + + `query` is the search expression (keywords, phrases, operators like + `from:`, `channel:`); it is REQUIRED and NON-BLANK so a source always + carries a server-side pre-filter before any per-item LLM cost. + `count` caps the per-cycle fetch (capped at 50 by yt-dlp for search). + """ + + SOURCE_KIND: ClassVar[str] = "youtube_search" + URL_FIELDS: ClassVar[tuple[str, ...]] = () # no operator-supplied URL to SSRF-check + + kind: Literal["youtube_search"] = "youtube_search" + query: str = Field(min_length=1) + count: int = Field(default=20, ge=1, le=50) + + @field_validator("query") + @classmethod + def _query_not_blank(cls, v: str) -> str: + v = v.strip() + if not v: + raise ValueError("youtube_search requires a non-blank query (the firehose guard)") + return v + + def display(self) -> str: + return f'YouTube search: "{self.query}"' + + +# The built-ins as a discriminated union over `kind` (defined before the plugin +# fallback so the built-in kind set can be derived from it below). A built-in kind +# with a malformed spec fails its typed member here and is rejected by the fallback, +# so it surfaces as a validation error rather than being absorbed as a raw blob. +_BuiltinSourceSpec = Annotated[ + RedditSubredditSourceSpec + | RssSourceSpec + | HackerNewsFeedSourceSpec + | HackerNewsCommentSourceSpec + | YouTubeSearchSourceSpec, + Field(discriminator="kind"), +] + +# Built-in source kinds, DERIVED from the union members above rather than kept as a +# second hand-maintained list, so the set the plugin fallback rejects can never drift +# from the union (adding a member to _BuiltinSourceSpec extends this for free). Taken +# from each member's `SOURCE_KIND` ClassVar (a plain str, always present) rather than +# the `kind` field default, which is fragile: a multi-value Literal would capture only +# the default, and a member without a default would inject PydanticUndefined. +_BUILTIN_SOURCE_SPECS = get_args(get_args(_BuiltinSourceSpec)[0]) +# Reject-set: the kind values the union DISPATCHES on, via the SAME shared helper the +# action/run unions use (it reads every Literal arg, so a multi-value Literal is +# handled). Sharing the helper is why the SOURCE_KIND cross-pin below is the ONLY +# source-specific piece. +_BUILTIN_SOURCE_KINDS = builtin_union_kinds(_BuiltinSourceSpec) + +# Import-time guards, load-bearing so raised explicitly (a bare `assert` is stripped +# under `python -O`, and this is a shared library): +# (1) If the union is ever refactored down to a single member, `get_args` returns () +# and the reject-set goes empty, silently letting the fallback absorb EVERY +# malformed built-in spec (the exact hole this set closes). Require >=2, one per +# member. +# (2) Source-specific cross-pin: each member also carries a SOURCE_KIND ClassVar (the +# connector registry's key), declared independently of the `kind` Literal. Pin the +# Literal to exactly (SOURCE_KIND,) so the discriminator, the reject-set, and the +# connector key can't diverge (a divergence would let a malformed built-in slip +# past the reject-set). getattr default keeps a missing SOURCE_KIND a curated +# RuntimeError, not a bare AttributeError. +if not (len(_BUILTIN_SOURCE_KINDS) == len(_BUILTIN_SOURCE_SPECS) >= 2): + raise RuntimeError(f"expected >=2 built-in source kinds, one per union member; got {sorted(_BUILTIN_SOURCE_KINDS)}") +for _spec in _BUILTIN_SOURCE_SPECS: + if get_args(_spec.model_fields["kind"].annotation) != (getattr(_spec, "SOURCE_KIND", None),): + raise RuntimeError( + f"{_spec.__name__}: kind Literal {get_args(_spec.model_fields['kind'].annotation)} must be exactly " + f"(SOURCE_KIND,) = ({getattr(_spec, 'SOURCE_KIND', None)!r},); the discriminator, reject-set, and " + f"connector key would otherwise diverge" + ) + + +class PluginSourceSpec(BaseModel): + """Fallback spec member for a plugin (non-built-in) source kind. `kind` is any + non-built-in string; the rest of the spec is an open blob (a fork's typed spec + schema lives in its own contract, and its web/CLI narrow on `kind`). Selected + only when no built-in discriminator matches (the left-to-right union below). + `extra="allow"` keeps every submitted field through `model_dump(mode="json")`, + so `canonical_spec` / `source_identity` (the spec_hash basis) stay stable.""" + + model_config = {"extra": "allow"} + + kind: str = Field(min_length=1, max_length=KIND_MAX_LENGTH) + + @field_validator("kind") + @classmethod + def _not_builtin(cls, v: str) -> str: + return reject_builtin_kind(v, _BUILTIN_SOURCE_KINDS) + + def display(self) -> str: + # No typed shape, so fall back to an operator label if the blob carries one, + # else the kind. A fork's typed spec member supplies a real display(). + extra = self.model_extra or {} + label = extra.get("name") or extra.get("label") + return str(label) if label else self.kind + + +# One source spec, keyed by `kind`: the built-in discriminated union above, then a +# left-to-right fallthrough to the plugin member for any other kind (same discipline +# as WatchActionWire). +SourceSpec = Annotated[_BuiltinSourceSpec | PluginSourceSpec, Field(union_mode="left_to_right")] + +# Pin the fallback member name against `_unions._PLUGIN_MEMBER_LOC_NAMES` (see _nodes): +# a rename that isn't mirrored there silently turns off clean_union_errors' stripping. +if PluginSourceSpec.__name__ not in _PLUGIN_MEMBER_LOC_NAMES: + raise RuntimeError(f"{PluginSourceSpec.__name__} missing from _unions._PLUGIN_MEMBER_LOC_NAMES") + + +def canonical_spec(spec: SourceSpec) -> str: + """Canonical JSON for a source spec: the single identity both the server's + `spec_hash` (which sha256s this) and the magpie CLI's source-diff compare on. + Sorted keys + compact separators make it independent of field-declaration + order, so two specs denote the same source iff this string matches. Pure + shape; keep it byte-stable - a change reshuffles every stored `spec_hash` + (pinned by core's `SpecHashCanonicalTests`).""" + return json.dumps(spec.model_dump(mode="json"), sort_keys=True, separators=(",", ":")) + + +class SourceFields(BaseModel): + """The identity + operator-tunable config shared by both source envelopes - + `SourceInput` (write path) and `SourceWire` (read path). + + `meta` is operator-supplied free-form tags; the recorder copies it onto each + FeedItem the source produces. `field_map` overrides the feed-level + `default_field_map` for a single source; empty means inherit (connectors that + don't read it ignore it). The watermark (`last_event_at`) and server-assigned + fields (`id`, `created_at`) live on the envelopes, since they differ by + direction.""" + + spec: SourceSpec + meta: dict[str, str] = Field(default_factory=dict) + field_map: dict[str, str] = Field(default_factory=dict) + + +class SourceIdentity(NamedTuple): + """A source's full reconcile identity - everything `feed source set` keys on: + the spec (`canonical_spec`, the `spec_hash` basis) PLUS the mutable config it + refreshes, meta + field_map. Excludes last_event_at (watermarks are never + reconciled). Hashable + ordered, so callers diff source sets by plain + equality instead of ad-hoc JSON.""" + + spec: str + meta: tuple[tuple[str, str], ...] + field_map: tuple[tuple[str, str], ...] + + +def source_identity(source: SourceFields) -> SourceIdentity: + """Build the shared `SourceIdentity` from any source envelope (SourceInput or + SourceWire, via their `SourceFields` base) - the ONE definition the magpie + CLI's source-diff compares on and that the server's set_sources reconcile + mirrors (spec_hash + meta/field_map).""" + return SourceIdentity( + spec=canonical_spec(source.spec), + meta=tuple(sorted(source.meta.items())), + field_map=tuple(sorted(source.field_map.items())), + )