"""文字おこしくん HTTP API の最小のクライアント（Python 3.10+、標準ライブラリだけ）。 export MOJIOKOSHI_API_KEY=mjk_live_... # サーバー側にだけ置く（ブラウザ・リポジトリに置かない） python mojiokoshi_http.py sessions [active|ended] # 自分のセッションの一覧（新しい順） python mojiokoshi_http.py transcript # 書き出し（1 発話 1 行のテキスト） python mojiokoshi_http.py token # transcript:read のセッショントークンを発行 python mojiokoshi_http.py live # 稼働中の最新のセッションをライブで購読（mojiokoshi_subscribe.py が要る） python mojiokoshi_http.py usage [from] [to] # 利用量（UTC の YYYY-MM-DD） 守っている決まり: エラーは problem details の code で見分ける / 429 rate_limited は Retry-After だけ待ってやり直す （quota_exceeded は翌日まで戻らないのでやり直さない）/ トークンの発行は Idempotency-Key 付き。 仕様: https://moji.kaitoranma.com/docs/openapi.yaml """ import json import os import sys import time import urllib.error import urllib.parse import urllib.request import uuid DEFAULT_BASE_URL = "https://moji.kaitoranma.com" class MojiokoshiHttpError(Exception): def __init__(self, status: int, problem: dict | None): self.status = status self.problem = problem self.code = problem.get("code") if problem else f"http_{status}" super().__init__(f"{self.code}: {problem.get('detail')}" if problem else f"HTTP {status}") class MojiokoshiHttp: def __init__(self, api_key: str, base_url: str = DEFAULT_BASE_URL, max_retries: int = 2): self.api_key = api_key self.base_url = base_url.rstrip("/") self.max_retries = max_retries def _request(self, method: str, path: str, body: dict | None = None, headers: dict | None = None) -> bytes: data = json.dumps(body).encode("utf-8") if body is not None else None all_headers = {"Authorization": f"Bearer {self.api_key}", **(headers or {})} if data is not None: all_headers["Content-Type"] = "application/json" for attempt in range(self.max_retries + 1): request = urllib.request.Request(self.base_url + path, data=data, method=method, headers=all_headers) try: with urllib.request.urlopen(request, timeout=30) as response: return response.read() except urllib.error.HTTPError as error: try: problem = json.loads(error.read().decode("utf-8")) except ValueError: problem = None retry_after = int(error.headers.get("Retry-After") or 0) if error.code == 429 and problem and problem.get("code") == "rate_limited" and attempt < self.max_retries and retry_after > 0: time.sleep(retry_after) continue raise MojiokoshiHttpError(error.code, problem) from None raise AssertionError("unreachable") def _json(self, method: str, path: str, body: dict | None = None, headers: dict | None = None) -> dict: return json.loads(self._request(method, path, body, headers).decode("utf-8")) def list_sessions(self, status: str | None = None, limit: int | None = None, cursor: str | None = None) -> dict: """自分のテナントのセッション（新しい順）。transcripts:read。続きは nextCursor を cursor に渡す。""" query = {k: v for k, v in {"status": status, "limit": limit, "cursor": cursor}.items() if v is not None} suffix = f"?{urllib.parse.urlencode(query)}" if query else "" return self._json("GET", f"/v1/sessions{suffix}") def get_session(self, session_id: str) -> dict: return self._json("GET", f"/v1/sessions/{urllib.parse.quote(session_id, safe='')}") def export_transcript(self, session_id: str) -> dict: """書き出し（JSON）。transcripts:read。""" return self._json("GET", f"/v1/sessions/{urllib.parse.quote(session_id, safe='')}/transcript") def export_transcript_text(self, session_id: str, tz: str = "Asia/Tokyo") -> str: query = urllib.parse.urlencode({"format": "text", "tz": tz}) return self._request("GET", f"/v1/sessions/{urllib.parse.quote(session_id, safe='')}/transcript?{query}").decode("utf-8") def mint_read_token(self, session_id: str, ttl_seconds: int = 3600) -> dict: """ブラウザ・OBS・ボットに渡す transcript:read のセッショントークン（既定 1 時間、最大 24 時間）。""" return self._json( "POST", f"/v1/sessions/{urllib.parse.quote(session_id, safe='')}/tokens", {"scope": "transcript:read", "ttlSeconds": ttl_seconds}, {"Idempotency-Key": str(uuid.uuid4())}, ) def get_usage(self, date_from: str | None = None, date_to: str | None = None) -> dict: """日ごとの利用量と今日の上限。usage:read。""" query = {k: v for k, v in {"from": date_from, "to": date_to}.items() if v is not None} suffix = f"?{urllib.parse.urlencode(query)}" if query else "" return self._json("GET", f"/v1/usage{suffix}") def subscribe_url(self) -> str: return ("ws" + self.base_url[len("http"):] if self.base_url.startswith("http") else self.base_url) + "/v1/subscribe" def main(argv: list[str]) -> int: api = MojiokoshiHttp(os.environ["MOJIOKOSHI_API_KEY"], os.environ.get("MOJIOKOSHI_BASE_URL", DEFAULT_BASE_URL)) command, args = (argv[0], argv[1:]) if argv else ("sessions", []) if command == "sessions": for s in api.list_sessions(status=args[0] if args else None, limit=20)["sessions"]: print(s["sessionId"], "live" if s["live"] else " ", s["label"]) elif command == "transcript": print(api.export_transcript_text(args[0]), end="") elif command == "token": print(json.dumps(api.mint_read_token(args[0]))) elif command == "usage": report = api.get_usage(*args[:2]) today = report["today"] # limitAudioSeconds は 1 日の上限が無ければ None（公式のサーバーはこちら）。 limit = today["limitAudioSeconds"] suffix = "" if limit is None else f" / {limit / 60:.0f}" print(f"今日 {today['audioSeconds'] / 60:.1f}{suffix} 分（プラン {report['plan']}）") elif command == "live": import asyncio from mojiokoshi_subscribe import run # https://moji.kaitoranma.com/docs/examples/mojiokoshi_subscribe.py live = api.list_sessions(status="active", limit=1)["sessions"] if not live: print("稼働中のセッションがありません", file=sys.stderr) return 1 token = api.mint_read_token(live[0]["sessionId"])["token"] asyncio.run(run(api.subscribe_url(), token)) else: print(__doc__, file=sys.stderr) return 2 return 0 if __name__ == "__main__": try: sys.exit(main(sys.argv[1:])) except MojiokoshiHttpError as error: print(f"エラー: {error}", file=sys.stderr) sys.exit(1)