feat: seven-send 0.1.0 — Library + CLI für SMS und TTS-Anrufe via seven.io

Wiederverwendbarer Baustein (eigenes Paket), nutzbar in beliebigen Programmen,
Cron-Jobs und Alerts:

- client.py: synchroner SevenClient.send_sms / send_voice (Voice als SSML,
  XML-escaped); Result mit ok/code/price/error, best-effort Fehlerbehandlung,
  dry_run-Modus
- cli.py: `seven-send sms|voice` mit --dry-run, --json, sauberen Exit-Codes;
  API-Key aus --api-key oder $SEVEN_API_KEY
- src-Layout, pyproject (console_script seven-send), MIT-Lizenz
- 12 Tests (Client + CLI), httpx gemockt

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,19 @@
+# Python
+__pycache__/
+*.py[cod]
+*.egg-info/
+build/
+dist/
+.eggs/
+
+# venv
+.venv/
+venv/
+
+# Tooling
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+
+# Env/Secrets
+.env

LICENSE

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

README.md

@@ -0,0 +1,69 @@
+# seven-send
+
+Kleine, wiederverwendbare **Library + CLI** zum Verschicken von **SMS** und
+**TTS-Sprachanrufen** über die [seven.io](https://www.seven.io)-API. Bewusst
+provider-fokussiert und schlank — als Baustein für beliebige Programme, Cron-Jobs
+oder Monitoring-Alerts.
+
+## Installation
+
+```bash
+pipx install git+https://kitux.de/forgejo/dschlueter/seven_send.git
+# oder lokal zur Entwicklung:
+pip install -e ".[dev]"
+```
+
+## CLI
+
+```bash
+export SEVEN_API_KEY=dein-key
+
+seven-send sms   --to +4915112345678 --text "Hallo Welt"
+seven-send voice --to +4915112345678 --text "Dies wird vorgelesen." --voice de-de-female
+
+seven-send --dry-run sms --to +49… --text "nur testen"      # nichts senden
+seven-send --json    sms --to +49… --text "..."             # maschinenlesbar
+```
+
+Exit-Codes: `0` Erfolg · `1` Sendefehler · `2` Konfig-/Aufruf-Fehler.
+
+## Library
+
+```python
+from seven_send import SevenClient
+
+c = SevenClient(api_key, sms_from="Notruf", voice_name="de-de-female")
+
+res = c.send_sms("+4915112345678", "Hallo")
+if not res:                      # bool(res) == res.ok
+    print("Fehler:", res.error)
+
+c.send_voice("+4915112345678", "Achtung, dies ist eine Ansage.")
+```
+
+`SevenClient` ist **synchron**. Aus asynchronem Code:
+
+```python
+res = await asyncio.to_thread(c.send_sms, "+49…", "Hallo")
+```
+
+## Konfiguration
+
+| Parameter | Default | Bedeutung |
+|-----------|---------|-----------|
+| `api_key` | – | seven.io API-Key (oder `$SEVEN_API_KEY` für die CLI) |
+| `sms_from` | `Notruf` | Absender-ID der SMS (alphanumerisch, max. 11 Zeichen) |
+| `voice_from` | `None` | Caller-ID für Anrufe (verifizierte Nummer) |
+| `voice_name` | `de-de-female` | SSML-Stimme für die Anruf-Ansage |
+| `dry_run` | `False` | nichts senden, nur protokollieren |
+
+## Test
+
+```bash
+pip install -e ".[dev]"
+pytest -q
+```
+
+## Lizenz
+
+MIT — siehe [LICENSE](LICENSE).

pyproject.toml

@@ -0,0 +1,26 @@
+[build-system]
+requires = ["setuptools>=61"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "seven-send"
+version = "0.1.0"
+description = "Kleine Library + CLI zum Verschicken von SMS und TTS-Sprachanrufen über seven.io"
+readme = "README.md"
+requires-python = ">=3.10"
+license = { text = "MIT" }
+authors = [{ name = "Dieter Schlüter", email = "dieter.schlueter@linix.de" }]
+keywords = ["seven.io", "sms", "voice", "tts", "cli", "notification"]
+dependencies = ["httpx>=0.27"]
+
+[project.optional-dependencies]
+dev = ["pytest>=8"]
+
+[project.scripts]
+seven-send = "seven_send.cli:main"
+
+[project.urls]
+Homepage = "https://kitux.de/forgejo/dschlueter/seven_send"
+
+[tool.setuptools.packages.find]
+where = ["src"]

src/seven_send/__init__.py

@@ -0,0 +1,6 @@
+"""seven-send: SMS und TTS-Sprachanrufe über die seven.io-API — als Library und CLI."""
+
+from seven_send.client import Result, SevenClient, SevenError
+
+__all__ = ["SevenClient", "Result", "SevenError"]
+__version__ = "0.1.0"

src/seven_send/cli.py

@@ -0,0 +1,82 @@
+"""Kommandozeilen-Frontend für seven-send: ``seven-send sms|voice ...``."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import os
+import sys
+
+from seven_send.client import DEFAULT_SMS_FROM, DEFAULT_VOICE_NAME, Result, SevenClient, SevenError
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        prog="seven-send", description="SMS und TTS-Sprachanrufe über seven.io"
+    )
+    p.add_argument(
+        "--api-key", default=os.getenv("SEVEN_API_KEY", ""),
+        help="seven.io API-Key (Default: $SEVEN_API_KEY)",
+    )
+    p.add_argument("--dry-run", action="store_true", help="nichts senden, nur anzeigen")
+    p.add_argument("--json", dest="as_json", action="store_true", help="Ergebnis als JSON ausgeben")
+    p.add_argument("-v", "--verbose", action="store_true")
+
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    sms = sub.add_parser("sms", help="SMS senden")
+    sms.add_argument("--to", required=True, help="Empfänger (E.164, z. B. +4915112345678)")
+    sms.add_argument("--text", required=True)
+    sms.add_argument("--from", dest="sender", default=DEFAULT_SMS_FROM, help="Absender-ID")
+
+    voice = sub.add_parser("voice", help="TTS-Anruf auslösen")
+    voice.add_argument("--to", required=True, help="Empfänger (E.164)")
+    voice.add_argument("--text", required=True, help="vorzulesender Text")
+    voice.add_argument("--from", dest="sender", default=None, help="Caller-ID (verifizierte Nummer)")
+    voice.add_argument("--voice", dest="voice_name", default=DEFAULT_VOICE_NAME, help="SSML-Stimme")
+    voice.add_argument("--no-ssml", dest="ssml", action="store_false", help="Text roh ohne SSML senden")
+    return p
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _build_parser().parse_args(argv)
+    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.WARNING)
+
+    try:
+        client = SevenClient(
+            args.api_key, dry_run=args.dry_run,
+            voice_name=getattr(args, "voice_name", DEFAULT_VOICE_NAME),
+        )
+    except SevenError as e:
+        print(f"Konfig-Fehler: {e}", file=sys.stderr)
+        return 2
+
+    if args.cmd == "sms":
+        res = client.send_sms(args.to, args.text, sender=args.sender)
+    else:  # "voice" — required=True garantiert eines von beiden
+        res = client.send_voice(args.to, args.text, sender=args.sender, ssml=args.ssml)
+
+    _report(res, as_json=args.as_json)
+    return 0 if res.ok else 1
+
+
+def _report(res: Result, *, as_json: bool) -> None:
+    if as_json:
+        print(json.dumps(
+            {"ok": res.ok, "channel": res.channel, "to": res.to,
+             "code": res.code, "price": res.price, "error": res.error},
+            ensure_ascii=False,
+        ))
+        return
+    status = "OK" if res.ok else "FEHLER"
+    line = f"{status}  {res.channel} -> {res.to}"
+    if res.code:
+        line += f"  (code {res.code})"
+    if res.error:
+        line += f"  : {res.error}"
+    print(line)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main())

src/seven_send/client.py

@@ -0,0 +1,120 @@
+"""Synchroner seven.io-Client für SMS und TTS-Sprachanrufe.
+
+Bewusst synchron (httpx.Client) — das ist der kleinste gemeinsame Nenner für
+Cron-Jobs, Skripte und die meisten Programme. Async-Aufrufer (z. B. ein
+FastAPI-Dienst) wrappen die Methoden via ``asyncio.to_thread``.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Any
+from xml.sax.saxutils import escape
+
+import httpx
+
+logger = logging.getLogger("seven_send")
+
+BASE_URL = "https://gateway.seven.io/api"
+DEFAULT_SMS_FROM = "Notruf"
+DEFAULT_VOICE_NAME = "de-de-female"
+
+
+class SevenError(Exception):
+    """Konfigurations-/Programmierfehler (z. B. fehlender API-Key)."""
+
+
+@dataclass
+class Result:
+    """Ergebnis eines Sendeversuchs. ``bool(result)`` == ``result.ok``."""
+
+    ok: bool
+    channel: str  # "sms" | "voice"
+    to: str
+    code: str | None = None  # seven.io-Statuscode ("100" = Erfolg)
+    price: float | None = None
+    raw: Any = None
+    error: str | None = None
+
+    def __bool__(self) -> bool:
+        return self.ok
+
+
+def _wrap_ssml(text: str, voice_name: str) -> str:
+    """Verpackt den Vorlese-Text in SSML und escaped XML-Sonderzeichen."""
+    return f'<speak><voice name="{voice_name}">{escape(text)}</voice></speak>'
+
+
+class SevenClient:
+    def __init__(
+        self,
+        api_key: str,
+        *,
+        sms_from: str = DEFAULT_SMS_FROM,
+        voice_from: str | None = None,
+        voice_name: str = DEFAULT_VOICE_NAME,
+        timeout: float = 15.0,
+        base_url: str = BASE_URL,
+        dry_run: bool = False,
+    ) -> None:
+        if not api_key and not dry_run:
+            raise SevenError("api_key fehlt (oder dry_run=True setzen)")
+        self.api_key = api_key
+        self.sms_from = sms_from
+        self.voice_from = voice_from
+        self.voice_name = voice_name
+        self.timeout = timeout
+        self.base_url = base_url.rstrip("/")
+        self.dry_run = dry_run
+
+    # -- öffentlich ---------------------------------------------------------
+    def send_sms(self, to: str, text: str, *, sender: str | None = None) -> Result:
+        data = {"to": to, "text": text, "from": sender or self.sms_from}
+        return self._post("sms", to, data)
+
+    def send_voice(
+        self, to: str, text: str, *, sender: str | None = None, ssml: bool = True
+    ) -> Result:
+        data = {"to": to, "text": _wrap_ssml(text, self.voice_name) if ssml else text}
+        frm = sender or self.voice_from
+        if frm:
+            data["from"] = frm
+        return self._post("voice", to, data)
+
+    # -- intern -------------------------------------------------------------
+    def _post(self, channel: str, to: str, data: dict) -> Result:
+        if self.dry_run:
+            logger.info("[dry-run] %s -> %s : %r", channel, to, data)
+            return Result(ok=True, channel=channel, to=to, code="dry-run", raw=data)
+        url = f"{self.base_url}/{channel}"
+        try:
+            resp = httpx.post(
+                url, data=data, headers={"X-Api-Key": self.api_key}, timeout=self.timeout
+            )
+        except Exception as e:  # noqa: BLE001 - Netzwerkfehler nie durchschlagen lassen
+            logger.exception("seven.io-Aufruf fehlgeschlagen")
+            return Result(ok=False, channel=channel, to=to, error=str(e))
+        return self._parse(channel, to, resp)
+
+    @staticmethod
+    def _parse(channel: str, to: str, resp: httpx.Response) -> Result:
+        body = resp.text.strip()
+        if resp.status_code >= 300:
+            return Result(
+                ok=False, channel=channel, to=to,
+                error=f"HTTP {resp.status_code}: {body[:200]}", raw=body,
+            )
+        try:
+            data = resp.json()
+        except Exception:  # Legacy: nackter Statuscode als Text
+            first = body.splitlines()[0] if body else ""
+            return Result(ok=(first == "100"), channel=channel, to=to, code=first or None, raw=body)
+        code = str(data.get("success", "")).strip() or None
+        msgs = data.get("messages") or []
+        ok = code in (None, "100") and all(m.get("success", True) for m in msgs)
+        return Result(
+            ok=ok, channel=channel, to=to, code=code,
+            price=data.get("total_price"), raw=data,
+            error=None if ok else f"seven success={code}",
+        )

tests/test_cli.py

@@ -0,0 +1,53 @@
+import json
+
+import seven_send.cli as cli
+import seven_send.client as cl
+
+
+class FakeResp:
+    status_code = 200
+    text = '{"success":"100"}'
+
+    def json(self):
+        return {"success": "100", "messages": [{"success": True}], "total_price": 0.075}
+
+
+def _patch(monkeypatch, cap):
+    def fake_post(url, data=None, headers=None, timeout=None):
+        cap.append({"url": url, "data": data})
+        return FakeResp()
+
+    monkeypatch.setattr(cl.httpx, "post", fake_post)
+
+
+def test_cli_sms(monkeypatch, capsys):
+    cap = []
+    _patch(monkeypatch, cap)
+    rc = cli.main(["--api-key", "k", "sms", "--to", "+49", "--text", "Hi"])
+    assert rc == 0
+    assert cap[0]["url"].endswith("/sms")
+    assert "OK" in capsys.readouterr().out
+
+
+def test_cli_voice_json(monkeypatch, capsys):
+    cap = []
+    _patch(monkeypatch, cap)
+    rc = cli.main(["--api-key", "k", "--json", "voice", "--to", "+49", "--text", "Hi"])
+    assert rc == 0
+    data = json.loads(capsys.readouterr().out)
+    assert data["ok"] is True and data["channel"] == "voice"
+
+
+def test_cli_dry_run(monkeypatch):
+    def boom(*a, **k):
+        raise AssertionError("kein POST")
+
+    monkeypatch.setattr(cl.httpx, "post", boom)
+    assert cli.main(["--dry-run", "sms", "--to", "+49", "--text", "Hi"]) == 0
+
+
+def test_cli_missing_key(monkeypatch, capsys):
+    monkeypatch.delenv("SEVEN_API_KEY", raising=False)
+    rc = cli.main(["sms", "--to", "+49", "--text", "Hi"])
+    assert rc == 2
+    assert "Konfig-Fehler" in capsys.readouterr().err

tests/test_client.py

@@ -0,0 +1,91 @@
+import pytest
+
+import seven_send.client as cl
+from seven_send import SevenClient, SevenError
+
+
+class FakeResp:
+    def __init__(self, status_code=200, payload=None, text=""):
+        self.status_code = status_code
+        self._payload = payload
+        self.text = text
+
+    def json(self):
+        if self._payload is None:
+            raise ValueError("kein JSON")
+        return self._payload
+
+
+def _patch_post(monkeypatch, capture, resp):
+    def fake_post(url, data=None, headers=None, timeout=None):
+        capture.append({"url": url, "data": data, "headers": headers})
+        return resp
+
+    monkeypatch.setattr(cl.httpx, "post", fake_post)
+
+
+def test_send_sms_ok(monkeypatch):
+    cap = []
+    _patch_post(monkeypatch, cap, FakeResp(
+        payload={"success": "100", "messages": [{"success": True}], "total_price": 0.075}))
+    r = SevenClient("key", sms_from="Notruf").send_sms("+4915112345678", "Hallo")
+    assert r.ok and bool(r) is True
+    assert r.channel == "sms" and r.price == 0.075
+    assert cap[0]["url"].endswith("/sms")
+    assert cap[0]["headers"]["X-Api-Key"] == "key"
+    assert cap[0]["data"] == {"to": "+4915112345678", "text": "Hallo", "from": "Notruf"}
+
+
+def test_send_voice_wraps_ssml(monkeypatch):
+    cap = []
+    _patch_post(monkeypatch, cap, FakeResp(payload={"success": "100"}))
+    r = SevenClient("key", voice_name="de-de-female").send_voice("+49", "Test & <x>")
+    assert r.ok
+    txt = cap[0]["data"]["text"]
+    assert txt.startswith('<speak><voice name="de-de-female">')
+    assert "&amp;" in txt and "&lt;x&gt;" in txt
+    assert cap[0]["url"].endswith("/voice")
+
+
+def test_voice_no_ssml(monkeypatch):
+    cap = []
+    _patch_post(monkeypatch, cap, FakeResp(payload={"success": "100"}))
+    SevenClient("key").send_voice("+49", "roh", ssml=False)
+    assert cap[0]["data"]["text"] == "roh"
+
+
+def test_failure_code(monkeypatch):
+    cap = []
+    _patch_post(monkeypatch, cap, FakeResp(payload={"success": "401"}))
+    r = SevenClient("key").send_sms("+49", "x")
+    assert not r.ok and r.code == "401"
+
+
+def test_http_error(monkeypatch):
+    cap = []
+    _patch_post(monkeypatch, cap, FakeResp(status_code=500, text="boom"))
+    r = SevenClient("key").send_sms("+49", "x")
+    assert not r.ok and "HTTP 500" in r.error
+
+
+def test_network_exception(monkeypatch):
+    def boom(*a, **k):
+        raise RuntimeError("offline")
+
+    monkeypatch.setattr(cl.httpx, "post", boom)
+    r = SevenClient("key").send_sms("+49", "x")
+    assert not r.ok and "offline" in r.error
+
+
+def test_dry_run_does_not_post(monkeypatch):
+    def boom(*a, **k):
+        raise AssertionError("kein POST im dry-run")
+
+    monkeypatch.setattr(cl.httpx, "post", boom)
+    r = SevenClient("", dry_run=True).send_sms("+49", "x")
+    assert r.ok and r.code == "dry-run"
+
+
+def test_missing_key_raises():
+    with pytest.raises(SevenError):
+        SevenClient("")