From 44023a3d23e3c0c47c1aab6e2adc2d307989b0e5 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Thu, 3 Sep 2026 00:33:14 -0700
Subject: [PATCH] refactor(plugins/teams_pipeline,spotify,security-guidance):
phase-split pipeline, dedupe Graph/Spotify clients, compact docs (3140->2631
LOC, --help/schema/pattern parity verified)
---
plugins/platforms/teams/summary_writer.py | 92 ++----
plugins/security-guidance/__init__.py | 74 ++---
plugins/security-guidance/patterns.py | 53 ++--
plugins/spotify/__init__.py | 34 +--
plugins/spotify/client.py | 73 ++---
plugins/spotify/tools.py | 277 ++++++------------
plugins/teams_pipeline/cli.py | 86 ++----
plugins/teams_pipeline/meetings.py | 124 +++-----
plugins/teams_pipeline/models.py | 104 +++----
plugins/teams_pipeline/pipeline.py | 330 ++++++++--------------
plugins/teams_pipeline/runtime.py | 31 +-
plugins/teams_pipeline/store.py | 46 +--
plugins/teams_pipeline/subscriptions.py | 35 +--
13 files changed, 425 insertions(+), 934 deletions(-)
diff --git a/plugins/platforms/teams/summary_writer.py b/plugins/platforms/teams/summary_writer.py
index 0de2f22d63..ae43fe7fa1 100644
--- a/plugins/platforms/teams/summary_writer.py
+++ b/plugins/platforms/teams/summary_writer.py
@@ -29,6 +29,9 @@ def _parse_bool(value: Any, *, default: bool = False) -> bool:
_LIST_SECTIONS = (("Key decisions", "key_decisions"), ("Action items", "action_items"), ("Risks", "risks"))
+# Env fallbacks for delivery config keys, applied only where nothing else set the key (access_token is a scoped secret).
+_ENV_KEYS = {"delivery_mode": "TEAMS_DELIVERY_MODE", "incoming_webhook_url": "TEAMS_INCOMING_WEBHOOK_URL",
+ "access_token": "TEAMS_GRAPH_ACCESS_TOKEN", "team_id": "TEAMS_TEAM_ID", "channel_id": "TEAMS_CHANNEL_ID", "chat_id": "TEAMS_CHAT_ID"}
class _StaticAccessTokenProvider:
@@ -38,7 +41,6 @@ class _StaticAccessTokenProvider:
self._access_token = str(access_token or "").strip()
async def get_access_token(self, *, force_refresh: bool = False) -> str:
- del force_refresh
if not self._access_token:
raise ValueError("TEAMS_GRAPH_ACCESS_TOKEN is required for graph delivery mode.")
return self._access_token
@@ -54,22 +56,17 @@ class TeamsSummaryWriter:
self, platform_config: PlatformConfig | None = None, *,
graph_client: Any | None = None, transport: httpx.AsyncBaseTransport | None = None,
) -> None:
- self._platform_config = platform_config
- self._graph_client = graph_client
- self._transport = transport
+ self._platform_config, self._graph_client, self._transport = platform_config, graph_client, transport
- async def write_summary(
- self, payload: Any, config: dict[str, Any] | None, existing_record: Optional[dict[str, Any]] = None
- ) -> dict[str, Any]:
+ async def write_summary(self, payload: Any, config: dict[str, Any] | None, existing_record: Optional[dict[str, Any]] = None) -> dict[str, Any]:
merged = self._resolve_delivery_config(config)
if existing_record and not _parse_bool(merged.get("force_resend"), default=False):
return dict(existing_record)
mode = str(merged.get("delivery_mode") or merged.get("mode") or "").strip().lower()
- if not mode:
- if merged.get("incoming_webhook_url"):
- mode = "incoming_webhook"
- elif merged.get("chat_id") or (merged.get("team_id") and merged.get("channel_id")):
- mode = "graph"
+ if not mode and merged.get("incoming_webhook_url"):
+ mode = "incoming_webhook"
+ elif not mode and (merged.get("chat_id") or (merged.get("team_id") and merged.get("channel_id"))):
+ mode = "graph"
if mode == "incoming_webhook":
return await self._write_summary_via_incoming_webhook(payload, merged)
if mode == "graph":
@@ -86,22 +83,14 @@ class TeamsSummaryWriter:
if platform_cfg.home_channel:
merged.setdefault("channel_id", platform_cfg.home_channel.chat_id)
merged.update(dict(config or {}))
- env_defaults = {
- "delivery_mode": os.getenv("TEAMS_DELIVERY_MODE", ""),
- "incoming_webhook_url": os.getenv("TEAMS_INCOMING_WEBHOOK_URL", ""),
- "access_token": _get_scoped_secret("TEAMS_GRAPH_ACCESS_TOKEN", ""),
- "team_id": os.getenv("TEAMS_TEAM_ID", ""),
- "channel_id": os.getenv("TEAMS_CHANNEL_ID", ""),
- "chat_id": os.getenv("TEAMS_CHAT_ID", ""),
- }
- for key, value in env_defaults.items():
+ for key, env in _ENV_KEYS.items():
+ value = _get_scoped_secret(env, "") if key == "access_token" else os.getenv(env, "")
if value and not merged.get(key):
merged[key] = value
return merged
async def _write_summary_via_incoming_webhook(self, payload: Any, config: dict[str, Any]) -> dict[str, Any]:
import httpx # lazy — see module docstring
-
webhook_url = str(config.get("incoming_webhook_url") or "").strip()
if not webhook_url:
raise ValueError("TEAMS_INCOMING_WEBHOOK_URL is required for incoming_webhook mode.")
@@ -109,10 +98,7 @@ class TeamsSummaryWriter:
async with httpx.AsyncClient(timeout=20.0, transport=self._transport) as client:
response = await client.post(webhook_url, json=body)
response.raise_for_status()
- return {
- "delivery_mode": "incoming_webhook", "webhook_url": webhook_url,
- "status_code": response.status_code, "delivered": True,
- }
+ return {"delivery_mode": "incoming_webhook", "webhook_url": webhook_url, "status_code": response.status_code, "delivered": True}
async def _write_summary_via_graph(self, payload: Any, config: dict[str, Any]) -> dict[str, Any]:
graph_client = self._build_graph_client(config)
@@ -127,69 +113,43 @@ class TeamsSummaryWriter:
raise ValueError("Graph delivery mode requires chat_id, or both team_id and channel_id.")
path = f"/teams/{quote(team_id, safe='')}/channels/{quote(channel_id, safe='')}/messages"
target = {"target_type": "channel", "team_id": team_id, "channel_id": channel_id}
- response = await graph_client.post_json(
- path,
- json_body={"body": {"contentType": "html", "content": self._render_summary_html(payload)}},
- )
- return {
- "delivery_mode": "graph", **target,
- "message_id": (response or {}).get("id"), "web_url": (response or {}).get("webUrl"),
- }
+ response = await graph_client.post_json(path, json_body={"body": {"contentType": "html", "content": self._render_summary_html(payload)}})
+ return {"delivery_mode": "graph", **target, "message_id": (response or {}).get("id"), "web_url": (response or {}).get("webUrl")}
def _build_graph_client(self, config: dict[str, Any]) -> Any:
if self._graph_client is not None:
return self._graph_client
from tools.microsoft_graph_auth import MicrosoftGraphTokenProvider
from tools.microsoft_graph_client import MicrosoftGraphClient
-
access_token = str(config.get("access_token") or "").strip()
- if access_token:
- return MicrosoftGraphClient(_StaticAccessTokenProvider(access_token), transport=self._transport)
- return MicrosoftGraphClient(MicrosoftGraphTokenProvider.from_env(), transport=self._transport)
+ provider = _StaticAccessTokenProvider(access_token) if access_token else MicrosoftGraphTokenProvider.from_env()
+ return MicrosoftGraphClient(provider, transport=self._transport)
def _render_summary_markdown(self, payload: Any) -> str:
- lines = [
- f"**{self._title(payload)}**",
- "",
- f"Summary: {self._text(getattr(payload, 'summary', None), 'No summary available.')}",
- ]
+ lines = [f"**{self._title(payload)}**", "", f"Summary: {self._text(getattr(payload, 'summary', None), 'No summary available.')}"]
for heading, attr in _LIST_SECTIONS:
lines += ["", f"{heading}:", *self._bullet_lines(getattr(payload, attr, None))]
return "\n".join(lines)
def _render_summary_html(self, payload: Any) -> str:
- sections = [
- ("Summary", [self._text(getattr(payload, "summary", None), "No summary available.")]),
- *((heading, list(getattr(payload, attr, None) or [])) for heading, attr in _LIST_SECTIONS),
- ]
- blocks = [f"{html.escape(self._title(payload))}
"]
- for heading, items in sections:
- blocks.append(f"{html.escape(heading)}
")
- if len(items) == 1 and heading == "Summary":
- blocks.append(f"{html.escape(str(items[0]))}
")
- continue
- if items:
- rendered = "".join(f"{html.escape(str(item))}" for item in items if str(item).strip())
- blocks.append(rendered and f"" or "None
")
- else:
- blocks.append("None
")
+ summary = html.escape(self._text(getattr(payload, "summary", None), "No summary available."))
+ blocks = [f"{html.escape(self._title(payload))}
", "Summary
", f"{summary}
"]
+ for heading, attr in _LIST_SECTIONS:
+ rendered = "".join(f"{html.escape(str(item))}" for item in (getattr(payload, attr, None) or []) if str(item).strip())
+ blocks += [f"{html.escape(heading)}
", f"" if rendered else "None
"]
return "".join(blocks)
@staticmethod
def _title(payload: Any) -> str:
- title = getattr(payload, "title", None)
- if title:
+ if title := getattr(payload, "title", None):
return str(title)
meeting_ref = getattr(payload, "meeting_ref", None)
- meeting_id = getattr(meeting_ref, "meeting_id", None) if meeting_ref else None
- return f"Meeting {meeting_id or 'summary'}"
+ return f"Meeting {(getattr(meeting_ref, 'meeting_id', None) if meeting_ref else None) or 'summary'}"
@staticmethod
def _text(value: Any, default: str) -> str:
- text = str(value or "").strip()
- return text or default
+ return str(value or "").strip() or default
@classmethod
def _bullet_lines(cls, values: Any) -> list[str]:
- items = [str(item).strip() for item in (values or []) if str(item).strip()]
- return [f"- {item}" for item in items] or ["- None"]
+ return [f"- {str(item).strip()}" for item in (values or []) if str(item).strip()] or ["- None"]
diff --git a/plugins/security-guidance/__init__.py b/plugins/security-guidance/__init__.py
index 387a008524..999cffa6ec 100644
--- a/plugins/security-guidance/__init__.py
+++ b/plugins/security-guidance/__init__.py
@@ -1,11 +1,10 @@
"""security-guidance plugin — fast pattern-matched security warnings on file writes.
-Scans content written by ``write_file`` / ``patch`` / ``skill_manage`` for known dangerous
-code patterns and appends a ``⚠️ Security guidance`` block to the tool result; the file is
-still written and the model self-corrects next turn. Warn (not block) by default because
-patterns have a real false-positive rate (``eval(`` in a tokenizer, ECB in a test fixture);
-``SECURITY_GUIDANCE_BLOCK=1`` refuses the write instead, ``SECURITY_GUIDANCE_DISABLE=1`` is
-a kill switch. Pattern data is ``patterns.py`` (Apache-2.0 fork, see LICENSE / NOTICE).
+Scans content written by ``write_file`` / ``patch`` / ``skill_manage`` for known dangerous patterns
+and appends a ``⚠️ Security guidance`` block to the tool result; the file is still written and the
+model self-corrects next turn. Warn (not block) by default because patterns have a real false-positive
+rate (``eval(`` in a tokenizer, ECB in a test fixture); ``SECURITY_GUIDANCE_BLOCK=1`` refuses the write
+instead, ``SECURITY_GUIDANCE_DISABLE=1`` is a kill switch. Pattern data: ``patterns.py`` (Apache-2.0 fork).
"""
from __future__ import annotations
@@ -42,24 +41,14 @@ def _compile_rules() -> List[Dict[str, Any]]:
"""Pre-compile regexes once; substrings stay plain (``in`` beats a literal regex)."""
compiled: List[Dict[str, Any]] = []
for rule in _patterns.SECURITY_PATTERNS:
- regex = None
- re_src = rule.get("regex")
- if re_src:
- try:
- regex = re.compile(re_src)
- except re.error as err:
- logger.warning(
- "security-guidance: skipping rule %s — invalid regex %r: %s",
- rule["ruleName"], re_src, err,
- )
- continue
+ try:
+ regex = re.compile(rule["regex"]) if rule.get("regex") else None
+ except re.error as err:
+ logger.warning("security-guidance: skipping rule %s — invalid regex %r: %s", rule["ruleName"], rule["regex"], err)
+ continue
compiled.append({
- "ruleName": rule["ruleName"],
- "reminder": rule["reminder"],
- "path_filter": rule.get("path_filter"),
- "path_check": rule.get("path_check"),
- "substrings": tuple(rule.get("substrings", ())),
- "regex": regex,
+ "ruleName": rule["ruleName"], "reminder": rule["reminder"], "path_filter": rule.get("path_filter"),
+ "path_check": rule.get("path_check"), "substrings": tuple(rule.get("substrings", ())), "regex": regex,
})
return compiled
@@ -68,11 +57,8 @@ _COMPILED: List[Dict[str, Any]] = _compile_rules()
def _rule_matches(entry: Dict[str, Any], path: str, content: str) -> bool:
- """One rule against one write. Path predicates are best-effort: an exception is a non-match.
-
- path_check rules fire on the path ALONE (e.g. "you're editing a workflow file") and never
- pattern-match content; path_filter gates content rules to relevant file types.
- """
+ """One rule against one write; a raising path predicate is a non-match. path_check rules fire on
+ the path ALONE and never scan content; path_filter gates content rules to relevant file types."""
try:
if entry["path_check"] is not None:
return bool(entry["path_check"](path))
@@ -80,9 +66,7 @@ def _rule_matches(entry: Dict[str, Any], path: str, content: str) -> bool:
return False
except Exception:
return False
- return any(sub in content for sub in entry["substrings"]) or (
- entry["regex"] is not None and bool(entry["regex"].search(content))
- )
+ return any(sub in content for sub in entry["substrings"]) or (entry["regex"] is not None and bool(entry["regex"].search(content)))
def _scan_content(path: str, content: str) -> List[Tuple[str, str]]:
@@ -99,14 +83,8 @@ def _scan_args(tool_name: str, args: Any) -> List[Tuple[str, str]]:
if _env_flag("SECURITY_GUIDANCE_DISABLE") or spec is None or not isinstance(args, dict):
return []
path_key, content_keys = spec
- path = args.get(path_key) or ""
- if not isinstance(path, str):
- path = ""
- findings: List[Tuple[str, str]] = []
- for val in (args.get(ck) for ck in content_keys):
- if isinstance(val, str) and val:
- findings.extend(_scan_content(path, val))
- return findings
+ path = raw_path if isinstance(raw_path := args.get(path_key), str) else ""
+ return [finding for val in (args.get(ck) for ck in content_keys) if isinstance(val, str) and val for finding in _scan_content(path, val)]
def _format_warning_block(findings: List[Tuple[str, str]]) -> str:
@@ -125,29 +103,19 @@ def _format_warning_block(findings: List[Tuple[str, str]]) -> str:
def _on_pre_tool_call(tool_name: str = "", args: Any = None, **_: Any) -> Optional[Dict[str, str]]:
"""Block mode only: refuse the write if any pattern matches (None = let it through)."""
- if not _env_flag("SECURITY_GUIDANCE_BLOCK"):
- return None
- findings = _scan_args(tool_name, args)
+ findings = _scan_args(tool_name, args) if _env_flag("SECURITY_GUIDANCE_BLOCK") else []
if not findings:
return None
return {
"action": "block",
- "message": (
- "security-guidance refused this write: "
- + _format_warning_block(findings)
- + "\n\nTo override, unset SECURITY_GUIDANCE_BLOCK and retry."
- ),
+ "message": "security-guidance refused this write: " + _format_warning_block(findings) + "\n\nTo override, unset SECURITY_GUIDANCE_BLOCK and retry.",
}
-def _on_transform_tool_result(
- tool_name: str = "", args: Any = None, result: Any = None, **_: Any,
-) -> Optional[str]:
+def _on_transform_tool_result(tool_name: str = "", args: Any = None, result: Any = None, **_: Any) -> Optional[str]:
"""Warn mode: append the warning block to the result string (None = unchanged)."""
# In block mode pre_tool_call already handled it — the tool didn't run, no result to wrap.
- if _env_flag("SECURITY_GUIDANCE_BLOCK") or not isinstance(result, str):
- return None
- findings = _scan_args(tool_name, args)
+ findings = [] if _env_flag("SECURITY_GUIDANCE_BLOCK") or not isinstance(result, str) else _scan_args(tool_name, args)
if not findings:
return None
# Don't decorate error results — the model already has bigger problems.
diff --git a/plugins/security-guidance/patterns.py b/plugins/security-guidance/patterns.py
index 296a947f99..287f8359fa 100644
--- a/plugins/security-guidance/patterns.py
+++ b/plugins/security-guidance/patterns.py
@@ -138,25 +138,18 @@ def _rule(name, reminder, **triggers):
return {"ruleName": name, "reminder": reminder, **triggers}
-# Security patterns configuration. Regex notes:
-# - eval / exec lookbehinds exclude `.` so method calls (model.eval(), redis.eval()) don't match.
-# - pickle matches deserialization only (load/loads/Unpickler); pickle.dump is not the RCE
-# surface, and `pkl_load` needs a word boundary so similarly named safe loaders don't match.
-# - script_src_without_sri: negative lookahead after ` o[k], root); for computation use a safe expression parser. NEVER interpolate untrusted strings into new Function() bodies.",
substrings=["new Function"]),
@@ -172,18 +165,14 @@ SECURITY_PATTERNS = [
_rule("innerHTML_xss",
"⚠️ Security Warning: Setting innerHTML with untrusted content can lead to XSS vulnerabilities. Use textContent for plain text or safe DOM methods for HTML content. If you need HTML support, consider using an HTML sanitizer library such as DOMPurify.",
substrings=[".innerHTML =", ".innerHTML="]),
- _rule("pickle_deserialization", _UNSAFE_DESERIALIZATION_REMINDER, path_filter=_PY_ONLY,
- regex=r"(?]{0,400}integrity\s*=)"
- r"[^>]{0,200}src\s*=\s*[\x22\x27](?:https?:)?//"
- r"[^\x22\x27]{1,300}[\x22\x27]"
- r"[^>]{0,100}>")),
- _rule("torch_unsafe_load", _UNSAFE_TORCH_LOAD_REMINDER,
- regex=r"(?:\btorch\.load|\.torch_load)\s*\((?![^)\n]{0,200}weights_only\s*=\s*True)"),
- _rule("yaml_unsafe_load_variants", _UNSAFE_YAML_LOAD_REMINDER,
- regex=r"(?:\byaml\.unsafe_load|\.yaml_unsafe_load)\s*\("),
+ regex=r"]{0,400}integrity\s*=)[^>]{0,200}src\s*=\s*[\x22\x27](?:https?:)?//[^\x22\x27]{1,300}[\x22\x27][^>]{0,100}>"),
+ _rule("torch_unsafe_load", _UNSAFE_TORCH_LOAD_REMINDER, regex=r"(?:\btorch\.load|\.torch_load)\s*\((?![^)\n]{0,200}weights_only\s*=\s*True)"),
+ _rule("yaml_unsafe_load_variants", _UNSAFE_YAML_LOAD_REMINDER, regex=r"(?:\byaml\.unsafe_load|\.yaml_unsafe_load)\s*\("),
_rule("pickle_wrapper_load", _UNSAFE_DESERIALIZATION_REMINDER,
regex=r"\bjoblib\.load\s*\(|\b(?:pd|pandas)\.read_pickle\s*\(|\.cloudpickle_load\s*\(|\b(?:np|numpy)\.load\s*\([^)\n]{0,200}allow_pickle\s*=\s*True"),
]
diff --git a/plugins/spotify/__init__.py b/plugins/spotify/__init__.py
index d0acfc09b5..230367c619 100644
--- a/plugins/spotify/__init__.py
+++ b/plugins/spotify/__init__.py
@@ -12,36 +12,20 @@ and ``kind: backend`` bundled plugins auto-load with no ``plugins.enabled`` opt-
from __future__ import annotations
-from plugins.spotify.tools import (
- SPOTIFY_ALBUMS_SCHEMA,
- SPOTIFY_DEVICES_SCHEMA,
- SPOTIFY_LIBRARY_SCHEMA,
- SPOTIFY_PLAYBACK_SCHEMA,
- SPOTIFY_PLAYLISTS_SCHEMA,
- SPOTIFY_QUEUE_SCHEMA,
- SPOTIFY_SEARCH_SCHEMA,
- _check_spotify_available,
- _handle_spotify_albums,
- _handle_spotify_devices,
- _handle_spotify_library,
- _handle_spotify_playback,
- _handle_spotify_playlists,
- _handle_spotify_queue,
- _handle_spotify_search,
-)
+from plugins.spotify import tools as _t
_TOOLS = (
- ("spotify_playback", SPOTIFY_PLAYBACK_SCHEMA, _handle_spotify_playback, "🎵"),
- ("spotify_devices", SPOTIFY_DEVICES_SCHEMA, _handle_spotify_devices, "🔈"),
- ("spotify_queue", SPOTIFY_QUEUE_SCHEMA, _handle_spotify_queue, "📻"),
- ("spotify_search", SPOTIFY_SEARCH_SCHEMA, _handle_spotify_search, "🔎"),
- ("spotify_playlists", SPOTIFY_PLAYLISTS_SCHEMA, _handle_spotify_playlists, "📚"),
- ("spotify_albums", SPOTIFY_ALBUMS_SCHEMA, _handle_spotify_albums, "💿"),
- ("spotify_library", SPOTIFY_LIBRARY_SCHEMA, _handle_spotify_library, "❤️"),
+ ("spotify_playback", _t.SPOTIFY_PLAYBACK_SCHEMA, _t._handle_spotify_playback, "🎵"),
+ ("spotify_devices", _t.SPOTIFY_DEVICES_SCHEMA, _t._handle_spotify_devices, "🔈"),
+ ("spotify_queue", _t.SPOTIFY_QUEUE_SCHEMA, _t._handle_spotify_queue, "📻"),
+ ("spotify_search", _t.SPOTIFY_SEARCH_SCHEMA, _t._handle_spotify_search, "🔎"),
+ ("spotify_playlists", _t.SPOTIFY_PLAYLISTS_SCHEMA, _t._handle_spotify_playlists, "📚"),
+ ("spotify_albums", _t.SPOTIFY_ALBUMS_SCHEMA, _t._handle_spotify_albums, "💿"),
+ ("spotify_library", _t.SPOTIFY_LIBRARY_SCHEMA, _t._handle_spotify_library, "❤️"),
)
def register(ctx) -> None:
"""Register all Spotify tools. Called once by the plugin loader."""
for name, schema, handler, emoji in _TOOLS:
- ctx.register_tool(name=name, toolset="spotify", schema=schema, handler=handler, check_fn=_check_spotify_available, emoji=emoji)
+ ctx.register_tool(name=name, toolset="spotify", schema=schema, handler=handler, check_fn=_t._check_spotify_available, emoji=emoji)
diff --git a/plugins/spotify/client.py b/plugins/spotify/client.py
index e87e1f8fda..50262abc04 100644
--- a/plugins/spotify/client.py
+++ b/plugins/spotify/client.py
@@ -14,12 +14,8 @@ import httpx
from hermes_cli.auth import AuthError, resolve_spotify_runtime_credentials
-class SpotifyError(RuntimeError):
- """Base Spotify tool error."""
-
-
-class SpotifyAuthRequiredError(SpotifyError):
- """Raised when the user needs to authenticate with Spotify first."""
+class SpotifyError(RuntimeError): """Base Spotify tool error."""
+class SpotifyAuthRequiredError(SpotifyError): """Raised when the user needs to authenticate with Spotify first."""
class SpotifyAPIError(SpotifyError):
@@ -27,14 +23,10 @@ class SpotifyAPIError(SpotifyError):
def __init__(self, message: str, *, status_code: Optional[int] = None, response_body: Optional[str] = None) -> None:
super().__init__(message)
- self.status_code = status_code
- self.response_body = response_body
- self.path: Optional[str] = None
+ self.status_code, self.response_body, self.path = status_code, response_body, None
-def _empty_204(message: str) -> Dict[str, Any]:
- """Explanatory payload returned instead of a bare 204 for the player endpoints."""
- return {"status_code": 204, "empty": True, "message": message}
+_empty_204 = lambda message: {"status_code": 204, "empty": True, "message": message} # noqa: E731 explanatory stand-in for a bare 204
class SpotifyClient:
@@ -52,22 +44,13 @@ class SpotifyClient:
return str(self._runtime.get("base_url") or "").rstrip("/")
def request(
- self,
- method: str,
- path: str,
- *,
- params: Optional[Dict[str, Any]] = None,
- json_body: Optional[Dict[str, Any]] = None,
- allow_retry_on_401: bool = True,
- empty_response: Optional[Dict[str, Any]] = None,
+ self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Dict[str, Any]] = None,
+ allow_retry_on_401: bool = True, empty_response: Optional[Dict[str, Any]] = None,
) -> Any:
response = httpx.request(
- method,
- f"{self.base_url}{path}",
+ method, f"{self.base_url}{path}",
headers={"Authorization": f"Bearer {self._runtime['access_token']}", "Content-Type": "application/json"},
- params=_strip_none(params),
- json=_strip_none(json_body) if json_body is not None else None,
- timeout=30.0,
+ params=_strip_none(params), json=_strip_none(json_body) if json_body is not None else None, timeout=30.0,
)
if response.status_code == 401 and allow_retry_on_401:
# One forced token refresh, then retry exactly once.
@@ -76,10 +59,8 @@ class SpotifyClient:
if response.status_code >= 400:
detail = response.text.strip()
message = _friendly_spotify_error_message(
- status_code=response.status_code,
- detail=_extract_spotify_error_detail(response, fallback=detail),
- path=path,
- retry_after=response.headers.get("Retry-After"),
+ status_code=response.status_code, detail=_extract_spotify_error_detail(response, fallback=detail),
+ path=path, retry_after=response.headers.get("Retry-After"),
)
error = SpotifyAPIError(message, status_code=response.status_code, response_body=detail)
error.path = path
@@ -90,9 +71,7 @@ class SpotifyClient:
return response.json()
return {"success": True, "text": response.text}
- # -- player -----------------------------------------------------------
-
- # Player reads return an explanatory payload instead of a bare 204.
+ # -- player: reads return an explanatory payload instead of a bare 204 --------
def get_playback_state(self, *, market: Optional[str] = None) -> Any:
return self.request("GET", "/me/player", params={"market": market}, empty_response=_empty_204(
@@ -125,24 +104,19 @@ def _friendly_spotify_error_message(*, status_code: int, detail: str, path: str,
return "Spotify authentication failed or expired. Run `hermes auth spotify` again."
if status_code == 403:
if is_playback_path:
- return (
- "Spotify rejected this playback request. Playback control usually requires a Spotify Premium account "
- "and an active Spotify Connect device."
- )
+ return ("Spotify rejected this playback request. Playback control usually requires a Spotify Premium account "
+ "and an active Spotify Connect device.")
if "scope" in detail.lower() or "permission" in detail.lower():
return "Spotify rejected the request because the current auth scope is insufficient. Re-run `hermes auth spotify` to refresh permissions."
return "Spotify rejected the request. The account may not have permission for this action."
if status_code == 404:
- if is_playback_path:
- return "Spotify could not find an active playback device or player session for this request."
- return "Spotify resource not found."
+ return "Spotify could not find an active playback device or player session for this request." if is_playback_path else "Spotify resource not found."
if status_code == 429:
return "Spotify rate limit exceeded." + (f" Retry after {retry_after} seconds." if retry_after else "")
return detail or f"Spotify API request failed with status {status_code}."
-def _strip_none(payload: Optional[Dict[str, Any]]) -> Dict[str, Any]:
- return {key: value for key, value in (payload or {}).items() if value is not None}
+_strip_none = lambda payload: {key: value for key, value in (payload or {}).items() if value is not None} # noqa: E731
def _check_type(item_type: str, expected_type: Optional[str]) -> None:
@@ -155,16 +129,13 @@ def normalize_spotify_id(value: str, expected_type: Optional[str] = None) -> str
cleaned = (value or "").strip()
if not cleaned:
raise SpotifyError("Spotify id/uri/url is required.")
- if cleaned.startswith("spotify:"):
- parts = cleaned.split(":")
- if len(parts) >= 3:
- _check_type(parts[1], expected_type)
- return parts[2]
- if "open.spotify.com" in cleaned:
- path_parts = [part for part in urlparse(cleaned).path.split("/") if part]
- if len(path_parts) >= 2:
- _check_type(path_parts[0], expected_type)
- return path_parts[1]
+ # (type, id) segments of a URI or URL; a bare id (or malformed ref) has no segments and passes through.
+ parts = cleaned.split(":")[1:] if cleaned.startswith("spotify:") else []
+ if len(parts) < 2 and "open.spotify.com" in cleaned:
+ parts = [part for part in urlparse(cleaned).path.split("/") if part]
+ if len(parts) >= 2:
+ _check_type(parts[0], expected_type)
+ return parts[1]
return cleaned
diff --git a/plugins/spotify/tools.py b/plugins/spotify/tools.py
index 8301002847..f072f2694c 100644
--- a/plugins/spotify/tools.py
+++ b/plugins/spotify/tools.py
@@ -11,12 +11,7 @@ from typing import Any, Callable, Dict, List, Optional
from hermes_cli.auth import get_auth_status
from plugins.spotify.client import (
- SpotifyClient,
- SpotifyError,
- normalize_spotify_id,
- normalize_spotify_uri,
- normalize_spotify_uris,
-)
+ SpotifyClient, SpotifyError, normalize_spotify_id, normalize_spotify_uri, normalize_spotify_uris)
from tools.registry import tool_error, tool_result
_Handler = Callable[[SpotifyClient, dict, str], str]
@@ -35,6 +30,20 @@ def _spotify_tool_error(exc: Exception) -> str:
return tool_error(f"Spotify tool failed: {type(exc).__name__}: {exc}")
+# Inside a ``_dispatcher`` boundary ``raise SpotifyError(msg)`` renders exactly like ``return tool_error(msg)``.
+def _required(value: Any, message: str) -> Any:
+ if value is None:
+ raise SpotifyError(message)
+ return value
+
+
+def _nonblank(raw: Any, message: str) -> str:
+ value = str(raw or "").strip()
+ if not value:
+ raise SpotifyError(message)
+ return value
+
+
def _coerce_bool(raw: Any, default: bool = False) -> bool:
if isinstance(raw, bool):
return raw
@@ -54,8 +63,7 @@ def _as_list(raw: Any) -> List[str]:
return [str(item).strip() for item in items if str(item).strip()]
-def _offset(args: dict) -> int:
- return max(0, int(args.get("offset") or 0))
+_offset = lambda args: max(0, int(args.get("offset") or 0)) # noqa: E731
def _limit(args: dict, default: int = 20) -> int:
@@ -68,8 +76,7 @@ def _limit(args: dict, default: int = 20) -> int:
return max(1, min(50, value))
-def _ok(action: str, result: Any, **extra: Any) -> str:
- return tool_result({"success": True, "action": action, **extra, "result": result})
+_ok = lambda action, result, **extra: tool_result({"success": True, "action": action, **extra, "result": result}) # noqa: E731
def _dispatcher(tool_name: str, default: str, table: Dict[str, _Handler], prepare: Optional[Callable[[dict], dict]] = None):
@@ -100,21 +107,15 @@ def _dispatcher(tool_name: str, default: str, table: Dict[str, _Handler], prepar
# action -> (flag key reported False, fallback message) when Spotify returns 204/empty.
_EMPTY_PLAYBACK = {
"get_currently_playing": ("is_playing", "Spotify is not currently playing anything."),
- "get_state": ("has_active_device", "No active Spotify playback session was found."),
-}
+ "get_state": ("has_active_device", "No active Spotify playback session was found.")}
def _pb_read(fetch: Callable[..., Any], args: dict, action: str) -> str:
payload = fetch(market=args.get("market"))
if isinstance(payload, dict) and payload.get("empty"):
flag, fallback = _EMPTY_PLAYBACK[action]
- payload = {
- "success": True,
- "action": action,
- flag: False,
- "status_code": payload.get("status_code", 204),
- "message": payload.get("message") or fallback,
- }
+ payload = {"success": True, "action": action, flag: False, "status_code": payload.get("status_code", 204),
+ "message": payload.get("message") or fallback}
return tool_result(payload)
@@ -135,45 +136,31 @@ def _pb_play(client: SpotifyClient, args: dict, action: str) -> str:
return _ok(action, client.request("PUT", "/me/player/play", params={"device_id": args.get("device_id")}, json_body=body))
-def _pb_device_cmd(method: str, path: str, **extra: Any) -> _Handler:
- """Handler for a player command whose only free argument is ``device_id`` (*extra* = fixed params)."""
- return lambda c, a, act: _ok(act, c.request(method, path, params={**extra, "device_id": a.get("device_id")}))
+def _pb_cmd(client: SpotifyClient, args: dict, action: str, method: str, path: str, **extra: Any) -> str:
+ """Player command whose only free argument is ``device_id`` (*extra* = fixed params)."""
+ return _ok(action, client.request(method, path, params={**extra, "device_id": args.get("device_id")}))
-def _pb_seek(client: SpotifyClient, args: dict, action: str) -> str:
- if args.get("position_ms") is None:
- return tool_error("position_ms is required for action='seek'")
- return _pb_device_cmd("PUT", "/me/player/seek", position_ms=int(args["position_ms"]))(client, args, action)
+_pb_device_cmd = lambda method, path: (lambda c, a, act: _pb_cmd(c, a, act, method, path)) # noqa: E731
-def _pb_set_repeat(client: SpotifyClient, args: dict, action: str) -> str:
+def _pb_required_param(path: str, param: str, convert: Callable[[Any], Any]) -> _Handler:
+ """PUT *path* with ``param`` (required, converted) plus device_id."""
+ return lambda c, a, act: _pb_cmd(c, a, act, "PUT", path, **{param: convert(_required(a.get(param), f"{param} is required for action='{act}'"))})
+
+
+def _pb_repeat_state(args: dict) -> str:
state = str(args.get("state") or "").strip().lower()
if state not in {"track", "context", "off"}:
- return tool_error("state must be one of: track, context, off")
- return _pb_device_cmd("PUT", "/me/player/repeat", state=state)(client, args, action)
-
-
-def _pb_set_shuffle(client: SpotifyClient, args: dict, action: str) -> str:
- state = str(_coerce_bool(args.get("state"))).lower()
- return _pb_device_cmd("PUT", "/me/player/shuffle", state=state)(client, args, action)
-
-
-def _pb_set_volume(client: SpotifyClient, args: dict, action: str) -> str:
- if args.get("volume_percent") is None:
- return tool_error("volume_percent is required for action='set_volume'")
- volume = max(0, min(100, int(args["volume_percent"])))
- return _pb_device_cmd("PUT", "/me/player/volume", volume_percent=volume)(client, args, action)
+ raise SpotifyError("state must be one of: track, context, off")
+ return state
def _pb_recently_played(client: SpotifyClient, args: dict, action: str) -> str:
after, before = args.get("after"), args.get("before")
if after and before:
return tool_error("Provide only one of 'after' or 'before'")
- params = {
- "limit": _limit(args),
- "after": int(after) if after is not None else None,
- "before": int(before) if before is not None else None,
- }
+ params = {"limit": _limit(args), "after": int(after) if after is not None else None, "before": int(before) if before is not None else None}
return tool_result(client.request("GET", "/me/player/recently-played", params=params))
@@ -184,40 +171,31 @@ _handle_spotify_playback = _dispatcher("spotify_playback", "get_state", {
"pause": _pb_device_cmd("PUT", "/me/player/pause"),
"next": _pb_device_cmd("POST", "/me/player/next"),
"previous": _pb_device_cmd("POST", "/me/player/previous"),
- "seek": _pb_seek,
- "set_repeat": _pb_set_repeat,
- "set_shuffle": _pb_set_shuffle,
- "set_volume": _pb_set_volume,
+ "seek": _pb_required_param("/me/player/seek", "position_ms", int),
+ "set_repeat": lambda c, a, act: _pb_cmd(c, a, act, "PUT", "/me/player/repeat", state=_pb_repeat_state(a)),
+ "set_shuffle": lambda c, a, act: _pb_cmd(c, a, act, "PUT", "/me/player/shuffle", state=str(_coerce_bool(a.get("state"))).lower()),
+ "set_volume": _pb_required_param("/me/player/volume", "volume_percent", lambda v: max(0, min(100, int(v)))),
"recently_played": _pb_recently_played,
})
# -- spotify_devices / spotify_queue / spotify_search ---------------------------
-def _dev_transfer(client: SpotifyClient, args: dict, action: str) -> str:
- device_id = str(args.get("device_id") or "").strip()
- if not device_id:
- return tool_error("device_id is required for action='transfer'")
- body = {"device_ids": [device_id], "play": _coerce_bool(args.get("play"))}
- return _ok(action, client.request("PUT", "/me/player", json_body=body))
-
-
_handle_spotify_devices = _dispatcher("spotify_devices", "list", {
"list": lambda c, a, act: tool_result(c.request("GET", "/me/player/devices")),
- "transfer": _dev_transfer,
+ "transfer": lambda c, a, act: _ok(act, c.request("PUT", "/me/player", json_body={
+ "device_ids": [_nonblank(a.get("device_id"), "device_id is required for action='transfer'")], "play": _coerce_bool(a.get("play")),
+ })),
})
def _queue_add(client: SpotifyClient, args: dict, action: str) -> str:
uri = normalize_spotify_uri(str(args.get("uri") or ""), None)
- result = client.request("POST", "/me/player/queue", params={"uri": uri, "device_id": args.get("device_id")})
- return _ok(action, result, uri=uri)
+ return _ok(action, client.request("POST", "/me/player/queue", params={"uri": uri, "device_id": args.get("device_id")}), uri=uri)
_handle_spotify_queue = _dispatcher("spotify_queue", "get", {
- "get": lambda c, a, act: tool_result(c.request("GET", "/me/player/queue")),
- "add": _queue_add,
-})
+ "get": lambda c, a, act: tool_result(c.request("GET", "/me/player/queue")), "add": _queue_add})
_SEARCH_TYPES = {"album", "artist", "playlist", "track", "show", "episode", "audiobook"}
@@ -231,10 +209,8 @@ def _handle_spotify_search(args: dict, **kw) -> str:
search_types = [value.lower() for value in raw_types if value.lower() in _SEARCH_TYPES]
if not search_types:
return tool_error("types must contain one or more of: album, artist, playlist, track, show, episode, audiobook")
- params = {
- "q": query, "type": ",".join(search_types), "limit": _limit(args, 10), "offset": _offset(args),
- "market": args.get("market"), "include_external": args.get("include_external"),
- }
+ params = {"q": query, "type": ",".join(search_types), "limit": _limit(args, 10), "offset": _offset(args),
+ "market": args.get("market"), "include_external": args.get("include_external")}
try:
return tool_result(client.request("GET", "/search", params=params))
except Exception as exc:
@@ -243,49 +219,28 @@ def _handle_spotify_search(args: dict, **kw) -> str:
# -- spotify_playlists ---------------------------------------------------------
-def _playlist_path(args: dict, suffix: str = "") -> str:
- return f"/playlists/{normalize_spotify_id(str(args.get('playlist_id') or ''), 'playlist')}{suffix}"
-
-
-def _pl_create(client: SpotifyClient, args: dict, action: str) -> str:
- name = str(args.get("name") or "").strip()
- if not name:
- return tool_error("name is required for action='create'")
- body = {
- "name": name, "public": _coerce_bool(args.get("public")),
- "collaborative": _coerce_bool(args.get("collaborative")), "description": args.get("description"),
- }
- return tool_result(client.request("POST", "/me/playlists", json_body=body))
-
-
-def _pl_remove_items(client: SpotifyClient, args: dict, action: str) -> str:
- path = _playlist_path(args, "/items")
- body = {"items": [{"uri": u} for u in normalize_spotify_uris(_as_list(args.get("uris")))], "snapshot_id": args.get("snapshot_id")}
- return tool_result(client.request("DELETE", path, json_body=body))
+_playlist_path = lambda args, suffix="": f"/playlists/{normalize_spotify_id(str(args.get('playlist_id') or ''), 'playlist')}{suffix}" # noqa: E731
_handle_spotify_playlists = _dispatcher("spotify_playlists", "list", {
"list": lambda c, a, act: tool_result(c.request("GET", "/me/playlists", params={"limit": _limit(a), "offset": _offset(a)})),
"get": lambda c, a, act: tool_result(c.request("GET", _playlist_path(a), params={"market": a.get("market")})),
- "create": _pl_create,
+ "create": lambda c, a, act: tool_result(c.request("POST", "/me/playlists", json_body={
+ "name": _nonblank(a.get("name"), "name is required for action='create'"), "public": _coerce_bool(a.get("public")),
+ "collaborative": _coerce_bool(a.get("collaborative")), "description": a.get("description")})),
"add_items": lambda c, a, act: tool_result(c.request("POST", _playlist_path(a, "/items"), json_body={
- "uris": normalize_spotify_uris(_as_list(a.get("uris"))), "position": a.get("position"),
- })),
- "remove_items": _pl_remove_items,
+ "uris": normalize_spotify_uris(_as_list(a.get("uris"))), "position": a.get("position")})),
+ "remove_items": lambda c, a, act: tool_result(c.request("DELETE", _playlist_path(a, "/items"), json_body={
+ "items": [{"uri": u} for u in normalize_spotify_uris(_as_list(a.get("uris")))], "snapshot_id": a.get("snapshot_id")})),
"update_details": lambda c, a, act: tool_result(c.request("PUT", _playlist_path(a), json_body={
- "name": a.get("name"), "public": a.get("public"), "collaborative": a.get("collaborative"), "description": a.get("description"),
- })),
+ "name": a.get("name"), "public": a.get("public"), "collaborative": a.get("collaborative"), "description": a.get("description")})),
})
# -- spotify_albums ------------------------------------------------------------
-def _page_params(args: dict) -> dict:
- return {"limit": _limit(args), "offset": _offset(args), "market": args.get("market")}
-
-
-def _prepare_album(args: dict) -> dict:
- return {**args, "_path": f"/albums/{normalize_spotify_id(str(args.get('album_id') or args.get('id') or ''), 'album')}"}
+_page_params = lambda args: {"limit": _limit(args), "offset": _offset(args), "market": args.get("market")} # noqa: E731
+_prepare_album = lambda args: {**args, "_path": f"/albums/{normalize_spotify_id(str(args.get('album_id') or args.get('id') or ''), 'album')}"} # noqa: E731
_handle_spotify_albums = _dispatcher("spotify_albums", "get", {
@@ -296,25 +251,18 @@ _handle_spotify_albums = _dispatcher("spotify_albums", "get", {
# -- spotify_library — saved tracks + saved albums, selected by `kind` ---------
-def _lib_list(client: SpotifyClient, args: dict, action: str) -> str:
- return tool_result(client.request("GET", f"/me/{args['kind']}", params=_page_params(args)))
-
-
-def _lib_save(client: SpotifyClient, args: dict, action: str) -> str:
- uris = normalize_spotify_uris(_as_list(args.get("uris") or args.get("items")), args["_item_type"])
- return tool_result(client.request("PUT", "/me/library", params={"uris": ",".join(uris)}))
-
-
-def _lib_remove(client: SpotifyClient, args: dict, action: str) -> str:
+def _lib_remove_uris(args: dict) -> str:
item_type = args["_item_type"]
ids = [normalize_spotify_id(item, item_type) for item in _as_list(args.get("ids") or args.get("items"))]
- if not ids:
- return tool_error("ids/items is required for action='remove'")
- uris = ",".join(f"spotify:{item_type}:{i}" for i in ids)
- return tool_result(client.request("DELETE", "/me/library", params={"uris": uris}))
+ return ",".join(f"spotify:{item_type}:{i}" for i in _required(ids or None, "ids/items is required for action='remove'"))
-_dispatch_library = _dispatcher("spotify_library", "list", {"list": _lib_list, "save": _lib_save, "remove": _lib_remove})
+_dispatch_library = _dispatcher("spotify_library", "list", {
+ "list": lambda c, a, act: tool_result(c.request("GET", f"/me/{a['kind']}", params=_page_params(a))),
+ "save": lambda c, a, act: tool_result(c.request("PUT", "/me/library", params={
+ "uris": ",".join(normalize_spotify_uris(_as_list(a.get("uris") or a.get("items")), a["_item_type"]))})),
+ "remove": lambda c, a, act: tool_result(c.request("DELETE", "/me/library", params={"uris": _lib_remove_uris(a)})),
+})
def _handle_spotify_library(args: dict, **kw) -> str:
@@ -332,82 +280,37 @@ _BOOL = {"type": "boolean"}
_STR_ARRAY = {"type": "array", "items": COMMON_STRING}
-def _strs(*names: str) -> dict:
- return dict.fromkeys(names, COMMON_STRING)
+_strs = lambda *names: dict.fromkeys(names, COMMON_STRING) # noqa: E731
+_enum = lambda *values: {"type": "string", "enum": list(values)} # noqa: E731
+_idesc = lambda text: {"type": "integer", "description": text} # noqa: E731
-def _enum(*values: str) -> dict:
- return {"type": "string", "enum": list(values)}
+def _schema(name: str, description: str, properties: dict, required: tuple = ("action",)) -> dict:
+ return {"name": name, "description": description, "parameters": {"type": "object", "properties": properties, "required": list(required)}}
-def _schema(name: str, description: str, properties: dict, required: list) -> dict:
- return {"name": name, "description": description, "parameters": {"type": "object", "properties": properties, "required": required}}
-
-
-SPOTIFY_PLAYBACK_SCHEMA = _schema(
- "spotify_playback",
- "Control Spotify playback, inspect the active playback state, or fetch recently played tracks.",
- {
- "action": _enum("get_state", "get_currently_playing", "play", "pause", "next", "previous", "seek", "set_repeat", "set_shuffle", "set_volume", "recently_played"),
- **_strs("device_id", "market", "context_uri"),
- "uris": _STR_ARRAY,
- "offset": {"type": "object"},
- "position_ms": _INT,
- "state": {"description": "For set_repeat use track/context/off. For set_shuffle use boolean-like true/false.", "oneOf": [{"type": "string"}, {"type": "boolean"}]},
- "volume_percent": _INT,
- "limit": {"type": "integer", "description": "For recently_played: number of tracks (max 50)"},
- "after": {"type": "integer", "description": "For recently_played: Unix ms cursor (after this timestamp)"},
- "before": {"type": "integer", "description": "For recently_played: Unix ms cursor (before this timestamp)"},
- },
- ["action"],
-)
-
-SPOTIFY_DEVICES_SCHEMA = _schema(
- "spotify_devices",
- "List Spotify Connect devices or transfer playback to a different device.",
- {"action": _enum("list", "transfer"), "device_id": COMMON_STRING, "play": _BOOL},
- ["action"],
-)
-
-SPOTIFY_QUEUE_SCHEMA = _schema(
- "spotify_queue",
- "Inspect the user's Spotify queue or add an item to it.",
- {"action": _enum("get", "add"), **_strs("uri", "device_id")},
- ["action"],
-)
-
+SPOTIFY_PLAYBACK_SCHEMA = _schema("spotify_playback", "Control Spotify playback, inspect the active playback state, or fetch recently played tracks.", {
+ "action": _enum("get_state", "get_currently_playing", "play", "pause", "next", "previous", "seek", "set_repeat", "set_shuffle", "set_volume", "recently_played"),
+ **_strs("device_id", "market", "context_uri"), "uris": _STR_ARRAY, "offset": {"type": "object"}, "position_ms": _INT,
+ "state": {"description": "For set_repeat use track/context/off. For set_shuffle use boolean-like true/false.", "oneOf": [{"type": "string"}, {"type": "boolean"}]},
+ "volume_percent": _INT, "limit": _idesc("For recently_played: number of tracks (max 50)"),
+ "after": _idesc("For recently_played: Unix ms cursor (after this timestamp)"), "before": _idesc("For recently_played: Unix ms cursor (before this timestamp)"),
+})
+SPOTIFY_DEVICES_SCHEMA = _schema("spotify_devices", "List Spotify Connect devices or transfer playback to a different device.",
+ {"action": _enum("list", "transfer"), "device_id": COMMON_STRING, "play": _BOOL})
+SPOTIFY_QUEUE_SCHEMA = _schema("spotify_queue", "Inspect the user's Spotify queue or add an item to it.", {"action": _enum("get", "add"), **_strs("uri", "device_id")})
SPOTIFY_SEARCH_SCHEMA = _schema(
- "spotify_search",
- "Search the Spotify catalog for tracks, albums, artists, playlists, shows, or episodes.",
- {"query": COMMON_STRING, "types": _STR_ARRAY, "type": COMMON_STRING, "limit": _INT, "offset": _INT, **_strs("market", "include_external")},
- ["query"],
-)
-
-SPOTIFY_PLAYLISTS_SCHEMA = _schema(
- "spotify_playlists",
- "List, inspect, create, update, and modify Spotify playlists.",
- {
- "action": _enum("list", "get", "create", "add_items", "remove_items", "update_details"),
- **_strs("playlist_id", "market"), "limit": _INT, "offset": _INT, **_strs("name", "description"),
- "public": _BOOL, "collaborative": _BOOL, "uris": _STR_ARRAY, "position": _INT, "snapshot_id": COMMON_STRING,
- },
- ["action"],
-)
-
-SPOTIFY_ALBUMS_SCHEMA = _schema(
- "spotify_albums",
- "Fetch Spotify album metadata or album tracks.",
- {"action": _enum("get", "tracks"), **_strs("album_id", "id", "market"), "limit": _INT, "offset": _INT},
- ["action"],
-)
-
-SPOTIFY_LIBRARY_SCHEMA = _schema(
- "spotify_library",
- "List, save, or remove the user's saved Spotify tracks or albums. Use `kind` to select which.",
- {
- "kind": {"type": "string", "enum": ["tracks", "albums"], "description": "Which library to operate on"},
- "action": _enum("list", "save", "remove"),
- "limit": _INT, "offset": _INT, "market": COMMON_STRING, "uris": _STR_ARRAY, "ids": _STR_ARRAY, "items": _STR_ARRAY,
- },
- ["kind", "action"],
+ "spotify_search", "Search the Spotify catalog for tracks, albums, artists, playlists, shows, or episodes.",
+ {"query": COMMON_STRING, "types": _STR_ARRAY, "type": COMMON_STRING, "limit": _INT, "offset": _INT, **_strs("market", "include_external")}, ("query",),
)
+SPOTIFY_PLAYLISTS_SCHEMA = _schema("spotify_playlists", "List, inspect, create, update, and modify Spotify playlists.", {
+ "action": _enum("list", "get", "create", "add_items", "remove_items", "update_details"),
+ **_strs("playlist_id", "market"), "limit": _INT, "offset": _INT, **_strs("name", "description"),
+ "public": _BOOL, "collaborative": _BOOL, "uris": _STR_ARRAY, "position": _INT, "snapshot_id": COMMON_STRING})
+SPOTIFY_ALBUMS_SCHEMA = _schema("spotify_albums", "Fetch Spotify album metadata or album tracks.",
+ {"action": _enum("get", "tracks"), **_strs("album_id", "id", "market"), "limit": _INT, "offset": _INT})
+SPOTIFY_LIBRARY_SCHEMA = _schema("spotify_library", "List, save, or remove the user's saved Spotify tracks or albums. Use `kind` to select which.", {
+ "kind": {"type": "string", "enum": ["tracks", "albums"], "description": "Which library to operate on"},
+ "action": _enum("list", "save", "remove"),
+ "limit": _INT, "offset": _INT, "market": COMMON_STRING, "uris": _STR_ARRAY, "ids": _STR_ARRAY, "items": _STR_ARRAY,
+}, ("kind", "action"))
diff --git a/plugins/teams_pipeline/cli.py b/plugins/teams_pipeline/cli.py
index 6ac69b8e17..c844e9c25d 100644
--- a/plugins/teams_pipeline/cli.py
+++ b/plugins/teams_pipeline/cli.py
@@ -11,19 +11,11 @@ from typing import Any, Callable
from hermes_constants import display_hermes_home
from gateway.config import Platform, load_gateway_config
from plugins.teams_pipeline.meetings import (
- enrich_meeting_with_call_record,
- fetch_preferred_transcript_text,
- list_recording_artifacts,
- resolve_meeting_reference,
-)
+ enrich_meeting_with_call_record, fetch_preferred_transcript_text, list_recording_artifacts, resolve_meeting_reference)
from plugins.teams_pipeline.pipeline import TeamsMeetingPipeline
from plugins.teams_pipeline.store import TeamsPipelineStore, resolve_teams_pipeline_store_path
from plugins.teams_pipeline.subscriptions import (
- build_graph_client,
- maintain_graph_subscriptions,
- sync_graph_subscription_record,
- utc_timestamp,
-)
+ build_graph_client, maintain_graph_subscriptions, sync_graph_subscription_record, utc_timestamp)
from tools.microsoft_graph_auth import MicrosoftGraphConfigError, MicrosoftGraphTokenProvider
@@ -62,6 +54,10 @@ def _text(args: argparse.Namespace, name: str) -> str:
return str(getattr(args, name, "") or "").strip()
+def _int_arg(args: argparse.Namespace, name: str, default: int) -> int:
+ return int(getattr(args, name, default) or default)
+
+
def _open_store(args: argparse.Namespace) -> TeamsPipelineStore:
return TeamsPipelineStore(resolve_teams_pipeline_store_path(getattr(args, "store_path", None)))
@@ -98,22 +94,17 @@ def _graph_setup_hint() -> str:
# Graph only emits "created" for the artifact/callRecord collection resources; meetings use "updated".
_CREATED_RESOURCE_PREFIXES = (
- "communications/onlinemeetings/getalltranscripts",
- "communications/onlinemeetings/getallrecordings",
- "communications/callrecords",
-)
+ "communications/onlinemeetings/getalltranscripts", "communications/onlinemeetings/getallrecordings", "communications/callrecords")
def _default_change_type_for_resource(resource: str) -> str:
- normalized = str(resource or "").strip().lower()
- return "created" if normalized.startswith(_CREATED_RESOURCE_PREFIXES) else "updated"
+ return "created" if str(resource or "").strip().lower().startswith(_CREATED_RESOURCE_PREFIXES) else "updated"
def _compact_job(job: dict) -> dict:
payload = dict(job)
summary = dict(payload.get("summary_payload") or {})
- transcript = summary.pop("transcript_text", None)
- if transcript:
+ if transcript := summary.pop("transcript_text", None):
summary["transcript_preview"] = str(transcript)[:240]
payload["summary_payload"] = summary or None
return payload
@@ -121,19 +112,16 @@ def _compact_job(job: dict) -> dict:
def _cmd_validate(args) -> None:
store = _open_store(args)
- env = os.environ
issues: list[str] = []
warnings: list[str] = []
gateway_config = load_gateway_config()
webhook_config = gateway_config.platforms.get(Platform.MSGRAPH_WEBHOOK)
teams_config = gateway_config.platforms.get(Platform("teams"))
-
- graph = {key: bool(env.get(f"MSGRAPH_{key.upper()}")) for key in ("tenant_id", "client_id", "client_secret")}
+ graph = {key: bool(os.environ.get(f"MSGRAPH_{key.upper()}")) for key in ("tenant_id", "client_id", "client_secret")}
webhook_enabled = bool(webhook_config and webhook_config.enabled)
teams_enabled = bool(teams_config and teams_config.enabled)
teams_extra = dict((teams_config.extra or {}) if teams_config else {})
teams_mode = str(teams_extra.get("delivery_mode") or "").strip() or None
-
if not all(graph.values()):
issues.append("Microsoft Graph app-only credentials are incomplete.")
if not webhook_enabled:
@@ -145,8 +133,7 @@ def _cmd_validate(args) -> None:
issues.append("TEAMS_INCOMING_WEBHOOK_URL is required for incoming_webhook mode.")
elif teams_mode == "graph":
# Graph delivery can authenticate with either a dedicated delivery token or the app-only creds.
- has_graph_delivery_token = bool(teams_config.token or teams_extra.get("access_token"))
- if not has_graph_delivery_token and not all(graph.values()):
+ if not (teams_config.token or teams_extra.get("access_token")) and not all(graph.values()):
issues.append("TEAMS_GRAPH_ACCESS_TOKEN or complete MSGRAPH_* app credentials is required for graph delivery mode.")
if not teams_extra.get("team_id"):
issues.append("TEAMS_TEAM_ID is required for graph delivery mode.")
@@ -154,31 +141,26 @@ def _cmd_validate(args) -> None:
issues.append("TEAMS_CHANNEL_ID is required for graph delivery mode.")
else:
warnings.append("TEAMS_DELIVERY_MODE is not set.")
-
_print_json({
"ok": not issues, "issues": issues, "warnings": warnings, "graph_config": graph,
"webhook_enabled": webhook_enabled, "teams_enabled": teams_enabled, "teams_delivery_mode": teams_mode,
- "store_path": str(store.path), "store_stats": store.stats(),
- })
+ "store_path": str(store.path), "store_stats": store.stats()})
def _cmd_list(args) -> None:
jobs = list(_open_store(args).list_jobs().values())
- status = _text(args, "status").lower()
- if status:
+ if status := _text(args, "status").lower():
jobs = [job for job in jobs if str(job.get("status") or "").lower() == status]
jobs.sort(key=lambda item: str((item or {}).get("updated_at") or ""), reverse=True)
- jobs = jobs[: max(1, min(int(getattr(args, "limit", 20) or 20), 100))]
+ jobs = jobs[: max(1, min(_int_arg(args, "limit", 20), 100))]
_print_records("Teams pipeline job(s)", "No Teams meeting pipeline jobs found.", [
(job.get("job_id"), [
("status", job.get("status"), True),
("meeting", (job.get("meeting_ref") or {}).get("meeting_id") or "unknown", True),
("strategy", job.get("selected_artifact_strategy"), False),
("updated", job.get("updated_at"), False),
- ("error", job.get("error_info"), False),
- ])
- for job in jobs
- ])
+ ("error", job.get("error_info"), False)])
+ for job in jobs])
def _cmd_show(args) -> None:
@@ -201,17 +183,13 @@ def _cmd_fetch(args) -> None:
if not meeting_id and not join_web_url:
print("meeting_id or join_web_url is required")
return
-
client = build_graph_client()
meeting_ref = asyncio.run(resolve_meeting_reference(
client, meeting_id=meeting_id, join_web_url=join_web_url,
- tenant_id=_text(args, "tenant_id") or None, organizer_user_id=_text(args, "organizer_user_id") or None,
- ))
+ tenant_id=_text(args, "tenant_id") or None, organizer_user_id=_text(args, "organizer_user_id") or None))
transcript_artifact, transcript_text = asyncio.run(fetch_preferred_transcript_text(client, meeting_ref))
recordings = asyncio.run(list_recording_artifacts(client, meeting_ref))
- call_record = asyncio.run(
- enrich_meeting_with_call_record(client, meeting_ref, call_record_id=_text(args, "call_record_id") or None)
- )
+ call_record = asyncio.run(enrich_meeting_with_call_record(client, meeting_ref, call_record_id=_text(args, "call_record_id") or None))
_print_json({
"meeting_ref": meeting_ref.to_dict(),
"transcript_available": bool(transcript_artifact and transcript_text),
@@ -219,8 +197,7 @@ def _cmd_fetch(args) -> None:
"transcript_preview": (transcript_text or "")[:240] or None,
"recording_count": len(recordings),
"recordings": [recording.to_dict() for recording in recordings[:5]],
- "call_record": call_record.to_dict() if call_record else None,
- })
+ "call_record": call_record.to_dict() if call_record else None})
def _cmd_subscriptions(args) -> None:
@@ -236,10 +213,8 @@ def _cmd_subscriptions(args) -> None:
("resource", sub.get("resource") or "unknown", True),
("changeType", sub.get("changeType") or "unknown", True),
("expires", sub.get("expirationDateTime"), False),
- ("notify", sub.get("notificationUrl"), False),
- ])
- for sub in subscriptions
- ])
+ ("notify", sub.get("notificationUrl"), False)])
+ for sub in subscriptions])
def _cmd_subscribe(args) -> None:
@@ -250,13 +225,11 @@ def _cmd_subscribe(args) -> None:
"notificationUrl": _text(args, "notification_url"),
"resource": resource,
"expirationDateTime": _text(args, "expiration") or utc_timestamp(1),
- "latestSupportedTlsVersion": _text(args, "latest_supported_tls_version") or "v1_2",
- }
+ "latestSupportedTlsVersion": _text(args, "latest_supported_tls_version") or "v1_2"}
if client_state := _text(args, "client_state"):
payload["clientState"] = client_state
if lifecycle_url := _text(args, "lifecycle_notification_url"):
payload["lifecycleNotificationUrl"] = lifecycle_url
-
result = asyncio.run(build_graph_client().post_json("/subscriptions", json_body=payload))
sync_graph_subscription_record(store, result, status="active")
_print_json(result)
@@ -283,10 +256,8 @@ def _cmd_delete_subscription(args) -> None:
def _cmd_maintain_subscriptions(args) -> None:
_print_json(asyncio.run(maintain_graph_subscriptions(
client=build_graph_client(), store=_open_store(args),
- renew_within_hours=int(getattr(args, "renew_within_hours", 24) or 24),
- extend_hours=int(getattr(args, "extend_hours", 24) or 24),
- dry_run=bool(getattr(args, "dry_run", False)), client_state=_text(args, "client_state") or None,
- )))
+ renew_within_hours=_int_arg(args, "renew_within_hours", 24), extend_hours=_int_arg(args, "extend_hours", 24),
+ dry_run=bool(getattr(args, "dry_run", False)), client_state=_text(args, "client_state") or None)))
def _cmd_token_health(args) -> None:
@@ -328,14 +299,12 @@ _SUBCOMMANDS: list[tuple[str, list[str], str, list[tuple[str, dict[str, Any]]],
_opt("--latest-supported-tls-version", default="v1_2"), _STORE_PATH], _cmd_subscribe),
("renew-subscription", [], "Renew a Microsoft Graph subscription",
[_opt("subscription_id"), _opt("--expiration", required=True), _STORE_PATH], _cmd_renew_subscription),
- ("delete-subscription", [], "Delete a Microsoft Graph subscription",
- [_opt("subscription_id"), _STORE_PATH], _cmd_delete_subscription),
+ ("delete-subscription", [], "Delete a Microsoft Graph subscription", [_opt("subscription_id"), _STORE_PATH], _cmd_delete_subscription),
("maintain-subscriptions", [], "Renew near-expiry managed subscriptions",
[_opt("--renew-within-hours", type=int, default=24), _opt("--extend-hours", type=int, default=24),
_opt("--dry-run", action="store_true"), _STORE_PATH, _opt("--client-state", **_EMPTY)], _cmd_maintain_subscriptions),
("token-health", ["token"], "Inspect Graph token health", [_opt("--force-refresh", action="store_true")], _cmd_token_health),
- ("validate", [], "Validate Teams pipeline configuration snapshot", [_STORE_PATH], _cmd_validate),
-]
+ ("validate", [], "Validate Teams pipeline configuration snapshot", [_STORE_PATH], _cmd_validate)]
_ACTIONS = {alias: handler for name, aliases, _help, _options, handler in _SUBCOMMANDS for alias in (name, *aliases)}
@@ -344,5 +313,4 @@ _REQUIRED_ARGS: dict[Callable[[Any], None], tuple[tuple[str, ...], str]] = {
_cmd_show: (("job_id",), "job_id is required"),
_cmd_run: (("job_id",), "job_id is required"),
_cmd_renew_subscription: (("subscription_id", "expiration"), "subscription_id and --expiration are required"),
- _cmd_delete_subscription: (("subscription_id",), "subscription_id is required"),
-}
+ _cmd_delete_subscription: (("subscription_id",), "subscription_id is required")}
diff --git a/plugins/teams_pipeline/meetings.py b/plugins/teams_pipeline/meetings.py
index 97a24cd692..114f1849fe 100644
--- a/plugins/teams_pipeline/meetings.py
+++ b/plugins/teams_pipeline/meetings.py
@@ -16,9 +16,7 @@ from tools.microsoft_graph_client import MicrosoftGraphAPIError, MicrosoftGraphC
# Graph uses both slash keys (users/{id}/...) and quoted keys (users('{id}')/...),
# so every segment pattern has a quoted group and a slash group.
-_USERS_MEETING_RE = re.compile(
- r"(?i)(?:^|/)users(?:\('([^']+)'\)|/([^/'()]+))/onlineMeetings(?:\('([^']+)'\)|/([^/'?]+))"
-)
+_USERS_MEETING_RE = re.compile(r"(?i)(?:^|/)users(?:\('([^']+)'\)|/([^/'()]+))/onlineMeetings(?:\('([^']+)'\)|/([^/'?]+))")
_COMM_MEETING_RE = re.compile(r"(?i)(?:^|/)communications/onlineMeetings(?:\('([^']+)'\)|/([^/'?]+))")
_TRANSCRIPT_RE = re.compile(r"(?i)/transcripts(?:\('([^']+)'\)|/([^/'?]+))")
_RECORDING_RE = re.compile(r"(?i)/recordings(?:\('([^']+)'\)|/([^/'?]+))")
@@ -26,17 +24,10 @@ _RECORDING_RE = re.compile(r"(?i)/recordings(?:\('([^']+)'\)|/([^/'?]+))")
_RESOURCE_SENTINELS = frozenset({"getalltranscripts", "getallrecordings", "transcripts", "recordings"})
-class TeamsMeetingError(RuntimeError):
- """Base class for Teams meeting pipeline failures."""
-
-class TeamsMeetingNotFoundError(TeamsMeetingError):
- """Raised when the meeting cannot be resolved from Graph."""
-
-class TeamsMeetingArtifactNotFoundError(TeamsMeetingError):
- """Raised when a transcript or recording cannot be found."""
-
-class TeamsMeetingPermissionError(TeamsMeetingError):
- """Raised when Graph access is denied for the requested resource."""
+class TeamsMeetingError(RuntimeError): """Base class for Teams meeting pipeline failures."""
+class TeamsMeetingNotFoundError(TeamsMeetingError): """Raised when the meeting cannot be resolved from Graph."""
+class TeamsMeetingArtifactNotFoundError(TeamsMeetingError): """Raised when a transcript or recording cannot be found."""
+class TeamsMeetingPermissionError(TeamsMeetingError): """Raised when Graph access is denied for the requested resource."""
def _match_id(match: re.Match[str] | None, *groups: int) -> str | None:
@@ -57,25 +48,19 @@ def parse_graph_meeting_resource(resource: str) -> dict[str, str | None]:
"organizer_user_id": _match_id(users_match, 1, 2),
"meeting_id": meeting_id,
"transcript_id": _match_id(_TRANSCRIPT_RE.search(text), 1, 2),
- "recording_id": _match_id(_RECORDING_RE.search(text), 1, 2),
- }
+ "recording_id": _match_id(_RECORDING_RE.search(text), 1, 2)}
def looks_like_transcript_id(value: str, *, odata_type: str | None = None) -> bool:
"""True when a Graph id is a callTranscript artifact rather than an onlineMeeting."""
if "calltranscript" in str(odata_type or "").lower():
return True
- text = str(value or "")
- return "transcript" in text.lower() or "transcript" in _decoded_id_hint(text)
+ return "transcript" in str(value or "").lower() or "transcript" in _decoded_id_hint(str(value or ""))
def _decoded_id_hint(value: str) -> str:
- """Best-effort base64 decode of a Graph id for artifact-marker sniffing.
-
- Transcript ids from getAllTranscripts ``resourceData.id`` are base64url blobs whose
- *decoded* payload carries a ``...-TranscriptV2`` suffix while the encoded form has
- no readable marker. Returns lowercase decoded text, or "" when it does not decode.
- """
+ """Best-effort base64 decode of a Graph id: getAllTranscripts ``resourceData.id`` blobs only carry
+ their ``-TranscriptV2`` marker in the *decoded* payload. Lowercase decoded text, or "" if undecodable."""
stripped = value.strip()
if len(stripped) < 16:
return ""
@@ -90,9 +75,7 @@ def _decoded_id_hint(value: str) -> str:
def _meetings_collection(organizer_user_id: str | None) -> str:
"""Organizer-scoped collection when the organizer is known (required for app-only Graph access)."""
- if organizer_user_id:
- return f"/users/{quote(organizer_user_id, safe='')}/onlineMeetings"
- return "/communications/onlineMeetings"
+ return f"/users/{quote(organizer_user_id, safe='')}/onlineMeetings" if organizer_user_id else "/communications/onlineMeetings"
def _meeting_path(meeting_ref: TeamsMeetingRef) -> str:
@@ -102,9 +85,7 @@ def _meeting_path(meeting_ref: TeamsMeetingRef) -> str:
def _wrap_graph_error(exc: MicrosoftGraphAPIError, *, missing_message: str) -> TeamsMeetingError:
if exc.status_code in {401, 403}:
return TeamsMeetingPermissionError(str(exc))
- if exc.status_code == 404:
- return TeamsMeetingNotFoundError(missing_message)
- return TeamsMeetingError(str(exc))
+ return TeamsMeetingNotFoundError(missing_message) if exc.status_code == 404 else TeamsMeetingError(str(exc))
async def _graph(awaitable: Awaitable[Any], *, missing_message: str) -> Any:
@@ -122,22 +103,16 @@ def _parse_organizer_user_id(payload: dict[str, Any]) -> str | None:
return user.get("id") if isinstance(user, dict) else None
-def _normalize_meeting_ref(
- payload: dict[str, Any], *, tenant_id: str | None = None, organizer_user_id: str | None = None
-) -> TeamsMeetingRef:
- metadata = {
- key: payload.get(key)
- for key in ("subject", "startDateTime", "endDateTime", "createdDateTime", "participants")
- if payload.get(key) is not None
- }
+def _normalize_meeting_ref(payload: dict[str, Any], *, tenant_id: str | None = None, organizer_user_id: str | None = None) -> TeamsMeetingRef:
+ metadata = {key: payload.get(key) for key in ("subject", "startDateTime", "endDateTime", "createdDateTime", "participants")
+ if payload.get(key) is not None}
chat = payload.get("chatInfo")
thread_id = str(chat["threadId"]) if isinstance(chat, dict) and chat.get("threadId") else payload.get("threadId")
return TeamsMeetingRef(
meeting_id=str(payload.get("id") or "").strip(),
organizer_user_id=organizer_user_id or _parse_organizer_user_id(payload),
join_web_url=payload.get("joinWebUrl"), calendar_event_id=payload.get("calendarEventId"),
- thread_id=thread_id, tenant_id=tenant_id or payload.get("tenantId"), metadata=metadata,
- )
+ thread_id=thread_id, tenant_id=tenant_id or payload.get("tenantId"), metadata=metadata)
def _normalize_artifact(artifact_type: str, payload: dict[str, Any]) -> MeetingArtifact:
@@ -147,14 +122,11 @@ def _normalize_artifact(artifact_type: str, payload: dict[str, Any]) -> MeetingA
display_name=payload.get("displayName") or payload.get("name"),
content_type=payload.get("contentType") or payload.get("fileMimeType"),
source_url=payload.get("webUrl") or payload.get("contentUrl"),
- download_url=(
- payload.get("@microsoft.graph.downloadUrl") or payload.get("downloadUrl")
- or payload.get("recordingContentUrl") or payload.get("transcriptContentUrl")
- ),
+ download_url=(payload.get("@microsoft.graph.downloadUrl") or payload.get("downloadUrl")
+ or payload.get("recordingContentUrl") or payload.get("transcriptContentUrl")),
created_at=payload.get("createdDateTime"),
available_at=payload.get("lastModifiedDateTime") or payload.get("meetingEndDateTime"),
- size_bytes=payload.get("size"), metadata=dict(payload),
- )
+ size_bytes=payload.get("size"), metadata=dict(payload))
def _transcript_sort_key(artifact: MeetingArtifact) -> tuple[int, int, str]:
@@ -168,15 +140,12 @@ def _transcript_sort_key(artifact: MeetingArtifact) -> tuple[int, int, str]:
async def resolve_meeting_reference(
client: MicrosoftGraphClient, *, meeting_id: str | None = None, join_web_url: str | None = None,
- tenant_id: str | None = None, organizer_user_id: str | None = None,
-) -> TeamsMeetingRef:
+ tenant_id: str | None = None, organizer_user_id: str | None = None) -> TeamsMeetingRef:
if meeting_id and looks_like_transcript_id(meeting_id):
if not join_web_url:
- raise TeamsMeetingError(
- "Refusing to GET /communications/onlineMeetings/{id} with a transcript id. "
- "Graph v1.0 does not support that id format; use the organizer-scoped meeting "
- "id from the notification @odata.id, or a join URL."
- )
+ raise TeamsMeetingError("Refusing to GET /communications/onlineMeetings/{id} with a transcript id. "
+ "Graph v1.0 does not support that id format; use the organizer-scoped meeting "
+ "id from the notification @odata.id, or a join URL.")
meeting_id = None
collection = _meetings_collection(organizer_user_id)
if meeting_id:
@@ -198,14 +167,11 @@ async def resolve_meeting_reference(
return _normalize_meeting_ref(payload, tenant_id=tenant_id, organizer_user_id=organizer_user_id)
-async def _list_artifacts(
- client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef, *, artifact_type: str
-) -> list[MeetingArtifact]:
+async def _list_artifacts(client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef, *, artifact_type: str) -> list[MeetingArtifact]:
collection = f"{artifact_type}s"
payloads = await _graph(
client.collect_paginated(f"{_meeting_path(meeting_ref)}/{collection}"),
- missing_message=f"No {collection} found for Teams meeting {meeting_ref.meeting_id}",
- )
+ missing_message=f"No {collection} found for Teams meeting {meeting_ref.meeting_id}")
return [_normalize_artifact(artifact_type, payload) for payload in payloads if isinstance(payload, dict)]
@@ -215,13 +181,11 @@ list_recording_artifacts = partial(_list_artifacts, artifact_type="recording")
async def _download_artifact(
client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef, artifact: MeetingArtifact, destination: Path, *,
- kind: str, **download_kwargs: Any,
-) -> dict[str, Any]:
+ kind: str, **download_kwargs: Any) -> dict[str, Any]:
path = artifact.download_url or f"{_meeting_path(meeting_ref)}/{kind}s/{quote(artifact.artifact_id, safe='')}/content"
return await _graph(
client.download_to_file(path, destination, **download_kwargs),
- missing_message=f"{kind.capitalize()} {artifact.artifact_id} not found for meeting {meeting_ref.meeting_id}",
- )
+ missing_message=f"{kind.capitalize()} {artifact.artifact_id} not found for meeting {meeting_ref.meeting_id}")
def select_preferred_transcript(candidates: list[MeetingArtifact]) -> MeetingArtifact | None:
@@ -230,25 +194,19 @@ def select_preferred_transcript(candidates: list[MeetingArtifact]) -> MeetingArt
async def download_transcript_text(
- client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef, transcript: MeetingArtifact, *,
- encoding: str = "utf-8",
-) -> str:
+ client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef, transcript: MeetingArtifact, *, encoding: str = "utf-8") -> str:
suffix = Path(transcript.display_name or "transcript.vtt").suffix or ".txt"
with tempfile.TemporaryDirectory(prefix="teams-transcript-", ignore_cleanup_errors=True) as tmp_dir:
destination = Path(tmp_dir) / f"transcript{suffix}"
# Graph's transcript /content endpoint rejects JSON content negotiation.
- await _download_artifact(
- client, meeting_ref, transcript, destination, kind="transcript", headers={"Accept": "text/vtt"}
- )
+ await _download_artifact(client, meeting_ref, transcript, destination, kind="transcript", headers={"Accept": "text/vtt"})
text = destination.read_text(encoding=encoding).strip()
if not text:
raise TeamsMeetingArtifactNotFoundError(f"Transcript {transcript.artifact_id} for meeting {meeting_ref.meeting_id} was empty.")
return text
-async def fetch_preferred_transcript_text(
- client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef
-) -> tuple[MeetingArtifact | None, str | None]:
+async def fetch_preferred_transcript_text(client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef) -> tuple[MeetingArtifact | None, str | None]:
transcript = select_preferred_transcript(await list_transcript_artifacts(client, meeting_ref))
if transcript is None:
return None, None
@@ -259,24 +217,20 @@ async def fetch_preferred_transcript_text(
async def download_recording_artifact(
- client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef, recording: MeetingArtifact, destination: str | Path,
-) -> dict[str, Any]:
+ client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef, recording: MeetingArtifact, destination: str | Path) -> dict[str, Any]:
destination_path = Path(destination)
result = await _download_artifact(client, meeting_ref, recording, destination_path, kind="recording")
return {
"artifact": recording.to_dict(), "path": str(destination_path),
"size_bytes": result.get("size_bytes") or recording.size_bytes,
- "content_type": result.get("content_type") or recording.content_type,
- }
+ "content_type": result.get("content_type") or recording.content_type}
async def enrich_meeting_with_call_record(
- client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef, *, call_record_id: str | None = None,
- allow_permission_errors: bool = True,
+ client: MicrosoftGraphClient, meeting_ref: TeamsMeetingRef, *, call_record_id: str | None = None, allow_permission_errors: bool = True
) -> MeetingArtifact | None:
"""Call records need the extra CallRecords.Read.All scope, so denial is optional-soft (None)."""
- call_record_id = str(call_record_id or meeting_ref.metadata.get("call_record_id") or "")
- if not call_record_id:
+ if not (call_record_id := str(call_record_id or meeting_ref.metadata.get("call_record_id") or "")):
return None
try:
payload = await client.get_json(f"/communications/callRecords/{quote(call_record_id, safe='')}")
@@ -284,19 +238,13 @@ async def enrich_meeting_with_call_record(
if exc.status_code == 404 or (exc.status_code in {401, 403} and allow_permission_errors):
return None
raise _wrap_graph_error(exc, missing_message=f"Call record not found: {call_record_id}") from exc
-
if not isinstance(payload, dict) or not payload.get("id"):
return None
-
- metrics = {
- "version": payload.get("version"), "modalities": payload.get("modalities"),
- "participant_count": len(payload.get("participants") or []), "organizer": _parse_organizer_user_id(payload),
- }
+ metrics = {"version": payload.get("version"), "modalities": payload.get("modalities"),
+ "participant_count": len(payload.get("participants") or []), "organizer": _parse_organizer_user_id(payload)}
if sessions := payload.get("sessions"):
metrics["session_count"] = len(sessions)
-
return MeetingArtifact(
artifact_type="call_record", artifact_id=str(payload["id"]), display_name=payload.get("type") or "call_record",
source_url=payload.get("webUrl"), created_at=payload.get("startDateTime"), available_at=payload.get("endDateTime"),
- metadata={"call_record": payload, "metrics": metrics},
- )
+ metadata={"call_record": payload, "metrics": metrics})
diff --git a/plugins/teams_pipeline/models.py b/plugins/teams_pipeline/models.py
index 3fcbf9801e..bf73043d2c 100644
--- a/plugins/teams_pipeline/models.py
+++ b/plugins/teams_pipeline/models.py
@@ -24,9 +24,7 @@ def _parse_datetime(value: Any) -> datetime | None:
def _serialize_datetime(value: datetime | None) -> str | None:
- if value is None:
- return None
- return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+ return None if value is None else value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
def _camel(name: str) -> str:
@@ -35,37 +33,31 @@ def _camel(name: str) -> str:
def _pick(payload: dict[str, Any], *keys: str) -> Any:
- """Equivalent of ``payload.get(k1) or payload.get(k2) or ...`` (returns the last value when all are falsy)."""
- value = None
- for key in keys:
- value = payload.get(key)
- if value:
- return value
- return value
+ """``payload.get(k1) or payload.get(k2) or ...`` (returns the last value when all are falsy)."""
+ return next((payload.get(key) for key in keys if payload.get(key)), payload.get(keys[-1]))
-def _str(value: Any) -> str:
- return str(value or "").strip()
+_str = lambda value: str(value or "").strip() # noqa: E731
+_list = lambda value: list(value or []) # noqa: E731
+_dict = lambda value: dict(value or {}) # noqa: E731
+_nested = lambda model: (lambda value: model.from_dict(value) if value else None) # noqa: E731
-def _list(value: Any) -> list[Any]:
- return list(value or [])
-
-
-def _dict(value: Any) -> dict[str, Any]:
- return dict(value or {})
-
-
-def _nested(model: type["_Model"]) -> Callable[[Any], Any]:
- return lambda value: model.from_dict(value) if value else None
+def _serialize_value(value: Any) -> Any:
+ """Datetimes -> ISO Z; nested models -> dicts; empty lists/dicts -> None (dropped by ``to_dict``)."""
+ if isinstance(value, datetime):
+ return _serialize_datetime(value)
+ if isinstance(value, list):
+ return [item.to_dict() if isinstance(item, _Model) else item for item in value] or None
+ if isinstance(value, dict):
+ return value or None
+ return value.to_dict() if isinstance(value, _Model) else value
class _Model:
- """Shared snake/camelCase ``from_dict`` and None-dropping ``to_dict`` for the dataclasses below.
-
- ``_ALIASES`` overrides the default ``(snake_name, camelName)`` lookup keys; ``_CONVERT`` post-processes
- the picked raw value; ``_REQUIRED`` string fields must be non-blank; ``_DATETIMES`` are parsed in place.
- """
+ """Shared snake/camelCase ``from_dict`` and None-dropping ``to_dict``. ``_ALIASES`` overrides the default
+ ``(snake_name, camelName)`` lookup keys; ``_CONVERT`` post-processes the picked value; ``_REQUIRED`` string
+ fields must be non-blank; ``_DATETIMES`` are parsed in place."""
_ALIASES: ClassVar[dict[str, tuple[str, ...]]] = {}
_CONVERT: ClassVar[dict[str, Callable[[Any], Any]]] = {}
@@ -74,28 +66,14 @@ class _Model:
@classmethod
def from_dict(cls, payload: dict[str, Any]):
- kwargs = {}
- for spec in fields(cls):
- value = _pick(payload, *cls._ALIASES.get(spec.name, (spec.name, _camel(spec.name))))
- convert = cls._CONVERT.get(spec.name)
- kwargs[spec.name] = convert(value) if convert else value
- return cls(**kwargs)
+ return cls(**{
+ spec.name: cls._CONVERT.get(spec.name, lambda v: v)(_pick(payload, *cls._ALIASES.get(spec.name, (spec.name, _camel(spec.name)))))
+ for spec in fields(cls)
+ })
def to_dict(self) -> dict[str, Any]:
- result: dict[str, Any] = {}
- for spec in fields(self):
- value = getattr(self, spec.name)
- if isinstance(value, datetime):
- value = _serialize_datetime(value)
- elif isinstance(value, list):
- value = [item.to_dict() if isinstance(item, _Model) else item for item in value] or None
- elif isinstance(value, dict):
- value = value or None
- elif isinstance(value, _Model):
- value = value.to_dict()
- if value is not None:
- result[spec.name] = value
- return result
+ out = {spec.name: _serialize_value(getattr(self, spec.name)) for spec in fields(self)}
+ return {key: value for key, value in out.items() if value is not None}
def __post_init__(self) -> None:
for name in self._REQUIRED:
@@ -116,10 +94,7 @@ class GraphSubscription(_Model):
latest_renewal_at: datetime | None = None
status: str | None = None
- _ALIASES = {
- "subscription_id": ("subscription_id", "id"),
- "expiration_datetime": ("expiration_datetime", "expirationDateTime"),
- }
+ _ALIASES = {"subscription_id": ("subscription_id", "id"), "expiration_datetime": ("expiration_datetime", "expirationDateTime")}
_REQUIRED = ("subscription_id", "resource", "change_type", "notification_url")
_CONVERT = dict.fromkeys(_REQUIRED, _str)
_DATETIMES = ("expiration_datetime", "latest_renewal_at")
@@ -159,12 +134,9 @@ class MeetingArtifact(_Model):
metadata: dict[str, Any] = field(default_factory=dict)
_ALIASES = {
- "artifact_id": ("artifact_id", "id"),
- "display_name": ("display_name", "displayName", "name"),
- "source_url": ("source_url", "sourceUrl", "webUrl"),
- "download_url": ("download_url", "downloadUrl", "@microsoft.graph.downloadUrl"),
- "created_at": ("created_at", "createdDateTime"),
- "available_at": ("available_at", "availableDateTime", "lastModifiedDateTime"),
+ "artifact_id": ("artifact_id", "id"), "display_name": ("display_name", "displayName", "name"),
+ "source_url": ("source_url", "sourceUrl", "webUrl"), "download_url": ("download_url", "downloadUrl", "@microsoft.graph.downloadUrl"),
+ "created_at": ("created_at", "createdDateTime"), "available_at": ("available_at", "availableDateTime", "lastModifiedDateTime"),
"size_bytes": ("size_bytes", "size"),
}
_CONVERT = {"artifact_id": _str, "metadata": _dict}
@@ -202,10 +174,8 @@ class TeamsMeetingSummaryPayload(_Model):
# Nested payloads are only read from their snake_case keys.
_ALIASES = {"meeting_ref": ("meeting_ref",), "source_artifacts": ("source_artifacts",)}
_CONVERT = {
- **dict.fromkeys(("participants", "key_decisions", "action_items", "risks"), _list),
- "call_metrics": _dict,
- "meeting_ref": TeamsMeetingRef.from_dict,
- "source_artifacts": lambda value: [MeetingArtifact.from_dict(item) for item in value or []],
+ **dict.fromkeys(("participants", "key_decisions", "action_items", "risks"), _list), "call_metrics": _dict,
+ "meeting_ref": TeamsMeetingRef.from_dict, "source_artifacts": lambda value: [MeetingArtifact.from_dict(item) for item in value or []],
}
_DATETIMES = ("start_time", "end_time")
@@ -232,11 +202,8 @@ class TeamsMeetingPipelineJob(_Model):
_REQUIRED = ("job_id", "event_id", "source_event_type", "dedupe_key", "status")
_CONVERT = {
- **dict.fromkeys(_REQUIRED, _str),
- "retry_count": lambda value: value or 0,
- "meeting_ref": _nested(TeamsMeetingRef),
- "summary_payload": _nested(TeamsMeetingSummaryPayload),
- "error_info": _dict,
+ **dict.fromkeys(_REQUIRED, _str), "retry_count": lambda value: value or 0, "error_info": _dict,
+ "meeting_ref": _nested(TeamsMeetingRef), "summary_payload": _nested(TeamsMeetingSummaryPayload),
}
_DATETIMES = ("created_at", "updated_at")
@@ -245,7 +212,4 @@ class TeamsMeetingPipelineJob(_Model):
self.retry_count = int(self.retry_count)
-__all__ = [
- "ArtifactType", "GraphSubscription", "MeetingArtifact",
- "TeamsMeetingPipelineJob", "TeamsMeetingRef", "TeamsMeetingSummaryPayload",
-]
+__all__ = ["ArtifactType", "GraphSubscription", "MeetingArtifact", "TeamsMeetingPipelineJob", "TeamsMeetingRef", "TeamsMeetingSummaryPayload"]
diff --git a/plugins/teams_pipeline/pipeline.py b/plugins/teams_pipeline/pipeline.py
index 4b55479a04..0b370a1780 100644
--- a/plugins/teams_pipeline/pipeline.py
+++ b/plugins/teams_pipeline/pipeline.py
@@ -24,14 +24,9 @@ from plugins.teams_pipeline.meetings import (
list_recording_artifacts,
looks_like_transcript_id,
parse_graph_meeting_resource,
- resolve_meeting_reference,
-)
+ resolve_meeting_reference)
from plugins.teams_pipeline.models import (
- MeetingArtifact,
- TeamsMeetingPipelineJob,
- TeamsMeetingRef,
- TeamsMeetingSummaryPayload,
-)
+ MeetingArtifact, TeamsMeetingPipelineJob, TeamsMeetingRef, TeamsMeetingSummaryPayload)
from plugins.teams_pipeline.store import TeamsPipelineStore
from tools.transcription_tools import transcribe_audio
@@ -40,26 +35,17 @@ logger = logging.getLogger(__name__)
TERMINAL_PIPELINE_STATES = {"completed", "failed", "retry_scheduled"}
ACTIVE_PIPELINE_STATES = {
"received", "resolving_meeting", "fetching_transcript", "downloading_recording",
- "transcribing_audio", "summarizing", "writing_notion", "writing_linear", "sending_teams",
-}
+ "transcribing_audio", "summarizing", "writing_notion", "writing_linear", "sending_teams"}
_AUDIO_SUFFIXES = {".wav", ".mp3", ".m4a", ".ogg", ".flac", ".aac", ".webm"}
_SUMMARY_SYSTEM_PROMPT = (
"You summarize meeting transcripts. Return only valid JSON with keys: "
- "summary, key_decisions, action_items, risks, confidence, confidence_notes."
-)
+ "summary, key_decisions, action_items, risks, confidence, confidence_notes.")
-class TeamsPipelineError(RuntimeError):
- """Base class for Teams meeting pipeline failures."""
-
-class TeamsPipelineRetryableError(TeamsPipelineError):
- """Raised when the pipeline should be retried later."""
-
-class TeamsPipelineSinkError(TeamsPipelineError):
- """Raised when an output sink fails."""
-
-class TeamsPipelineArtifactNotFoundError(TeamsPipelineRetryableError):
- """Raised when meeting artifacts are not yet available."""
+class TeamsPipelineError(RuntimeError): """Base class for Teams meeting pipeline failures."""
+class TeamsPipelineRetryableError(TeamsPipelineError): """Raised when the pipeline should be retried later."""
+class TeamsPipelineSinkError(TeamsPipelineError): """Raised when an output sink fails."""
+class TeamsPipelineArtifactNotFoundError(TeamsPipelineRetryableError): """Raised when meeting artifacts are not yet available."""
TranscribeFn = Callable[[str, Optional[str]], dict[str, Any]]
@@ -84,17 +70,12 @@ class TeamsPipelineConfig:
def from_dict(cls, payload: Optional[dict[str, Any]]) -> "TeamsPipelineConfig":
data = dict(payload or {})
tmp_dir = data.get("tmp_dir") or data.get("tmpDir")
+ flags = {"transcript_preferred": True, "transcript_required": False, "transcription_fallback": True, "ffmpeg_extract_audio": True}
return cls(
- transcript_preferred=bool(data.get("transcript_preferred", True)),
- transcript_required=bool(data.get("transcript_required", False)),
- transcription_fallback=bool(data.get("transcription_fallback", True)),
- stt_model=data.get("stt_model") or data.get("sttModel"),
- ffmpeg_extract_audio=bool(data.get("ffmpeg_extract_audio", True)),
- transcript_min_chars=int(data.get("transcript_min_chars", 80)),
- tmp_dir=Path(tmp_dir) if tmp_dir else None,
- notion=data.get("notion"), linear=data.get("linear"),
- teams_delivery=data.get("teams_delivery") or data.get("teamsDelivery"),
- )
+ **{name: bool(data.get(name, default)) for name, default in flags.items()},
+ stt_model=data.get("stt_model") or data.get("sttModel"), transcript_min_chars=int(data.get("transcript_min_chars", 80)),
+ tmp_dir=Path(tmp_dir) if tmp_dir else None, notion=data.get("notion"), linear=data.get("linear"),
+ teams_delivery=data.get("teams_delivery") or data.get("teamsDelivery"))
def _rich_text(content: str) -> dict[str, Any]:
@@ -106,12 +87,8 @@ def _bullets(items: list[str]) -> str:
def _sections(payload: TeamsMeetingSummaryPayload) -> list[tuple[str, str]]:
- return [
- ("Summary", payload.summary or ""),
- ("Key Decisions", _bullets(payload.key_decisions)),
- ("Action Items", _bullets(payload.action_items)),
- ("Risks", _bullets(payload.risks)),
- ]
+ return [("Summary", payload.summary or ""), ("Key Decisions", _bullets(payload.key_decisions)),
+ ("Action Items", _bullets(payload.action_items)), ("Risks", _bullets(payload.risks))]
class _HttpSinkWriter:
@@ -140,22 +117,18 @@ class NotionWriter(_HttpSinkWriter):
SECRET_NAME = "NOTION_API_KEY"
async def write_summary(
- self, payload: TeamsMeetingSummaryPayload, config: dict[str, Any],
- existing_record: Optional[dict[str, Any]] = None,
+ self, payload: TeamsMeetingSummaryPayload, config: dict[str, Any], existing_record: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
self._require_api_key()
database_id = str(config.get("database_id") or config.get("databaseId") or "").strip()
page_id = (existing_record or {}).get("page_id")
if not database_id and not page_id:
raise TeamsPipelineSinkError("Notion sink requires database_id or an existing page_id.")
-
headers = {"Authorization": f"Bearer {self.api_key}", "Notion-Version": self.API_VERSION, "Content-Type": "application/json"}
properties = self._build_properties(payload, config)
# Re-runs update the existing page's properties only; body blocks are written once on create.
if page_id:
- record = await self._request(
- "PATCH", f"{self.API_BASE}/pages/{page_id}", headers=headers, body={"properties": properties}
- )
+ record = await self._request("PATCH", f"{self.API_BASE}/pages/{page_id}", headers=headers, body={"properties": properties})
else:
body = {"parent": {"database_id": database_id}, "properties": properties, "children": self._build_blocks(payload)}
record = await self._request("POST", f"{self.API_BASE}/pages", headers=headers, body=body)
@@ -171,43 +144,30 @@ class NotionWriter(_HttpSinkWriter):
return properties
def _build_blocks(self, payload: TeamsMeetingSummaryPayload) -> list[dict[str, Any]]:
- blocks: list[dict[str, Any]] = []
- for heading, body in _sections(payload):
- blocks.append({"object": "block", "type": "heading_2", "heading_2": _rich_text(heading)})
- blocks.append({"object": "block", "type": "paragraph", "paragraph": _rich_text(body or "None")})
- return blocks
+ return [block for heading, body in _sections(payload) for block in (
+ {"object": "block", "type": "heading_2", "heading_2": _rich_text(heading)},
+ {"object": "block", "type": "paragraph", "paragraph": _rich_text(body or "None")})]
class LinearWriter(_HttpSinkWriter):
API_URL = "https://api.linear.app/graphql"
SECRET_NAME = "LINEAR_API_KEY"
- _UPDATE_MUTATION = (
- "mutation($id: String!, $input: IssueUpdateInput!) "
- "{ issueUpdate(id: $id, input: $input) { success issue { id identifier url } } }"
- )
- _CREATE_MUTATION = (
- "mutation($input: IssueCreateInput!) "
- "{ issueCreate(input: $input) { success issue { id identifier url } } }"
- )
+ _UPDATE_MUTATION = "mutation($id: String!, $input: IssueUpdateInput!) { issueUpdate(id: $id, input: $input) { success issue { id identifier url } } }"
+ _CREATE_MUTATION = "mutation($input: IssueCreateInput!) { issueCreate(input: $input) { success issue { id identifier url } } }"
async def write_summary(
- self, payload: TeamsMeetingSummaryPayload, config: dict[str, Any],
- existing_record: Optional[dict[str, Any]] = None,
+ self, payload: TeamsMeetingSummaryPayload, config: dict[str, Any], existing_record: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
self._require_api_key()
headers = {"Authorization": self.api_key, "Content-Type": "application/json"}
team_id = str(config.get("team_id") or config.get("teamId") or "").strip()
- issue_input = {"title": payload.title or f"Meeting Summary: {payload.meeting_ref.meeting_id}",
- "description": _render_summary_markdown(payload)}
- existing_issue_id = (existing_record or {}).get("issue_id")
-
- if existing_issue_id:
+ issue_input = {"title": payload.title or f"Meeting Summary: {payload.meeting_ref.meeting_id}", "description": _render_summary_markdown(payload)}
+ if existing_issue_id := (existing_record or {}).get("issue_id"):
body = {"query": self._UPDATE_MUTATION, "variables": {"id": existing_issue_id, "input": issue_input}}
+ elif not team_id:
+ raise TeamsPipelineSinkError("Linear sink requires team_id when creating a new issue.")
else:
- if not team_id:
- raise TeamsPipelineSinkError("Linear sink requires team_id when creating a new issue.")
body = {"query": self._CREATE_MUTATION, "variables": {"input": {"teamId": team_id, **issue_input}}}
-
payload_json = await self._request("POST", self.API_URL, headers=headers, body=body)
data = payload_json.get("data") or {}
issue = (data.get("issueUpdate") or {}).get("issue") or (data.get("issueCreate") or {}).get("issue")
@@ -223,16 +183,11 @@ class TeamsMeetingPipeline:
self, *, graph_client: Any, store: TeamsPipelineStore,
config: TeamsPipelineConfig | dict[str, Any] | None = None, transcribe_fn: TranscribeFn = transcribe_audio,
summarize_fn: Optional[SummarizeFn] = None, notion_writer: Optional[NotionWriter] = None,
- linear_writer: Optional[LinearWriter] = None, teams_sender: Optional[SinkFn] = None,
- ) -> None:
- self.graph_client = graph_client
- self.store = store
+ linear_writer: Optional[LinearWriter] = None, teams_sender: Optional[SinkFn] = None) -> None:
+ self.graph_client, self.store, self.transcribe_fn = graph_client, store, transcribe_fn
self.config = config if isinstance(config, TeamsPipelineConfig) else TeamsPipelineConfig.from_dict(config)
- self.transcribe_fn = transcribe_fn
self.summarize_fn = summarize_fn or self._generate_summary_payload
- self.notion_writer = notion_writer
- self.linear_writer = linear_writer
- self.teams_sender = teams_sender
+ self.notion_writer, self.linear_writer, self.teams_sender = notion_writer, linear_writer, teams_sender
def create_job_from_notification(self, notification: dict[str, Any]) -> TeamsMeetingPipelineJob:
event_id = TeamsPipelineStore.build_notification_receipt_key(notification)
@@ -243,20 +198,13 @@ class TeamsMeetingPipeline:
resource_data = notification.get("resourceData") or {}
meeting_id, organizer_user_id, extra_metadata = _meeting_ids_from_notification(notification)
job = TeamsMeetingPipelineJob(
- job_id=f"teams-job-{uuid.uuid4().hex[:12]}",
- event_id=event_id,
+ job_id=f"teams-job-{uuid.uuid4().hex[:12]}", event_id=event_id, dedupe_key=event_id, status="received",
source_event_type=str(notification.get("changeType") or "graph.notification"),
- dedupe_key=event_id,
- status="received",
meeting_ref=TeamsMeetingRef(
- meeting_id=str(meeting_id),
- organizer_user_id=organizer_user_id,
+ meeting_id=str(meeting_id), organizer_user_id=organizer_user_id,
tenant_id=resource_data.get("tenantId") or notification.get("tenantId"),
- metadata={
- "notification": dict(notification), "join_web_url": resource_data.get("joinWebUrl"),
- "call_record_id": resource_data.get("callRecordId") or notification.get("callRecordId"),
- **extra_metadata,
- },
+ metadata={"notification": dict(notification), "join_web_url": resource_data.get("joinWebUrl"),
+ "call_record_id": resource_data.get("callRecordId") or notification.get("callRecordId"), **extra_metadata},
),
)
self.store.upsert_job(job.job_id, job.to_dict())
@@ -274,72 +222,20 @@ class TeamsMeetingPipeline:
meeting_ref = job.meeting_ref
if meeting_ref is None:
raise TeamsPipelineError(f"Job {job.job_id} has no meeting_ref.")
-
artifacts: list[MeetingArtifact] = []
try:
- job = self._persist_job(job, status="resolving_meeting")
- notification = meeting_ref.metadata.get("notification") if isinstance(meeting_ref.metadata, dict) else {}
- meeting_id, organizer_user_id = meeting_ref.meeting_id, meeting_ref.organizer_user_id
- # Re-parse the stored notification: older jobs may have persisted a transcript id as meeting_id.
- if isinstance(notification, dict) and notification:
- parsed_id, parsed_org, _extra = _meeting_ids_from_notification(notification)
- if parsed_org:
- organizer_user_id = organizer_user_id or parsed_org
- if parsed_id and not looks_like_transcript_id(parsed_id):
- meeting_id = parsed_id
- resolved_meeting = await resolve_meeting_reference(
- self.graph_client, meeting_id=meeting_id, tenant_id=meeting_ref.tenant_id,
- join_web_url=meeting_ref.join_web_url or meeting_ref.metadata.get("join_web_url"),
- organizer_user_id=organizer_user_id,
- )
- if meeting_ref.metadata:
- resolved_meeting.metadata = {**meeting_ref.metadata, **resolved_meeting.metadata}
- resolved_meeting.organizer_user_id = resolved_meeting.organizer_user_id or meeting_ref.organizer_user_id
- job.meeting_ref = resolved_meeting
- job = self._persist_job(job, meeting_ref=resolved_meeting.to_dict())
-
- transcript_text: str | None = None
- if self.config.transcript_preferred:
- job = self._persist_job(job, status="fetching_transcript")
- transcript_artifact, transcript_text = await fetch_preferred_transcript_text(
- self.graph_client, resolved_meeting
- )
- if transcript_artifact and transcript_text:
- artifacts.append(transcript_artifact)
- if len(transcript_text.strip()) < self.config.transcript_min_chars:
- transcript_text = None
-
- if transcript_text:
- job = self._persist_job(job, selected_artifact_strategy="transcript_first")
- elif self.config.transcript_required:
- raise TeamsPipelineRetryableError(f"Transcript unavailable for meeting {resolved_meeting.meeting_id}.")
- elif not self.config.transcription_fallback:
- raise TeamsPipelineArtifactNotFoundError(
- f"No transcript available and transcription fallback disabled for {resolved_meeting.meeting_id}."
- )
- else:
- job = self._persist_job(job, status="downloading_recording")
- recordings = await list_recording_artifacts(self.graph_client, resolved_meeting)
- if not recordings:
- raise TeamsPipelineRetryableError(f"Recording unavailable for meeting {resolved_meeting.meeting_id}.")
- artifacts.append(recordings[0])
- transcript_text = await self._transcribe_recording(job, resolved_meeting, recordings[0])
- job = self._persist_job(job, selected_artifact_strategy="recording_stt_fallback")
-
+ job, resolved_meeting, notification = await self._resolve_meeting(job, meeting_ref)
+ job, transcript_text = await self._obtain_transcript(job, resolved_meeting, artifacts)
call_record_id = notification.get("callRecordId") or (meeting_ref.metadata or {}).get("call_record_id")
call_record = await enrich_meeting_with_call_record(self.graph_client, resolved_meeting, call_record_id=call_record_id)
if call_record is not None:
artifacts.append(call_record)
-
job = self._persist_job(job, status="summarizing")
- summary_payload = await self.summarize_fn(
- resolved_meeting=resolved_meeting, transcript_text=transcript_text or "", artifacts=artifacts
- )
+ summary_payload = await self.summarize_fn(resolved_meeting=resolved_meeting, transcript_text=transcript_text or "", artifacts=artifacts)
if not isinstance(summary_payload, TeamsMeetingSummaryPayload):
summary_payload = TeamsMeetingSummaryPayload.from_dict(summary_payload)
job.summary_payload = summary_payload
job = self._persist_job(job, summary_payload=summary_payload.to_dict())
-
await self._write_sinks(job, summary_payload)
return self._persist_job(job, status="completed")
except TeamsPipelineRetryableError as exc:
@@ -347,6 +243,55 @@ class TeamsMeetingPipeline:
except Exception as exc:
return self._persist_job(job, status="failed", error_info={"message": str(exc), "type": type(exc).__name__})
+ async def _resolve_meeting(
+ self, job: TeamsMeetingPipelineJob, meeting_ref: TeamsMeetingRef) -> tuple[TeamsMeetingPipelineJob, TeamsMeetingRef, Any]:
+ """Phase 1: resolve the Graph meeting; returns (job, resolved_meeting, stored notification)."""
+ job = self._persist_job(job, status="resolving_meeting")
+ notification = meeting_ref.metadata.get("notification") if isinstance(meeting_ref.metadata, dict) else {}
+ meeting_id, organizer_user_id = meeting_ref.meeting_id, meeting_ref.organizer_user_id
+ # Re-parse the stored notification: older jobs may have persisted a transcript id as meeting_id.
+ if isinstance(notification, dict) and notification:
+ parsed_id, parsed_org, _extra = _meeting_ids_from_notification(notification)
+ if parsed_org:
+ organizer_user_id = organizer_user_id or parsed_org
+ if parsed_id and not looks_like_transcript_id(parsed_id):
+ meeting_id = parsed_id
+ resolved_meeting = await resolve_meeting_reference(
+ self.graph_client, meeting_id=meeting_id, tenant_id=meeting_ref.tenant_id,
+ join_web_url=meeting_ref.join_web_url or meeting_ref.metadata.get("join_web_url"), organizer_user_id=organizer_user_id)
+ if meeting_ref.metadata:
+ resolved_meeting.metadata = {**meeting_ref.metadata, **resolved_meeting.metadata}
+ resolved_meeting.organizer_user_id = resolved_meeting.organizer_user_id or meeting_ref.organizer_user_id
+ job.meeting_ref = resolved_meeting
+ return self._persist_job(job, meeting_ref=resolved_meeting.to_dict()), resolved_meeting, notification
+
+ async def _obtain_transcript(
+ self, job: TeamsMeetingPipelineJob, resolved_meeting: TeamsMeetingRef, artifacts: list[MeetingArtifact]
+ ) -> tuple[TeamsMeetingPipelineJob, str | None]:
+ """Phase 2: transcript first, then the recording->STT fallback (appends chosen artifacts)."""
+ transcript_text: str | None = None
+ if self.config.transcript_preferred:
+ job = self._persist_job(job, status="fetching_transcript")
+ transcript_artifact, transcript_text = await fetch_preferred_transcript_text(self.graph_client, resolved_meeting)
+ if transcript_artifact and transcript_text:
+ artifacts.append(transcript_artifact)
+ if len(transcript_text.strip()) < self.config.transcript_min_chars:
+ transcript_text = None
+ if transcript_text:
+ return self._persist_job(job, selected_artifact_strategy="transcript_first"), transcript_text
+ if self.config.transcript_required:
+ raise TeamsPipelineRetryableError(f"Transcript unavailable for meeting {resolved_meeting.meeting_id}.")
+ if not self.config.transcription_fallback:
+ raise TeamsPipelineArtifactNotFoundError(
+ f"No transcript available and transcription fallback disabled for {resolved_meeting.meeting_id}.")
+ job = self._persist_job(job, status="downloading_recording")
+ recordings = await list_recording_artifacts(self.graph_client, resolved_meeting)
+ if not recordings:
+ raise TeamsPipelineRetryableError(f"Recording unavailable for meeting {resolved_meeting.meeting_id}.")
+ artifacts.append(recordings[0])
+ transcript_text = await self._transcribe_recording(job, resolved_meeting, recordings[0])
+ return self._persist_job(job, selected_artifact_strategy="recording_stt_fallback"), transcript_text
+
def _coerce_job(self, job_or_id: TeamsMeetingPipelineJob | str) -> TeamsMeetingPipelineJob:
if isinstance(job_or_id, TeamsMeetingPipelineJob):
return job_or_id
@@ -362,12 +307,9 @@ class TeamsMeetingPipeline:
return None
def _persist_job(self, job: TeamsMeetingPipelineJob, **updates: Any) -> TeamsMeetingPipelineJob:
- stored = self.store.upsert_job(job.job_id, {**job.to_dict(), **updates})
- return TeamsMeetingPipelineJob.from_dict(stored)
+ return TeamsMeetingPipelineJob.from_dict(self.store.upsert_job(job.job_id, {**job.to_dict(), **updates}))
- async def _transcribe_recording(
- self, job: TeamsMeetingPipelineJob, meeting_ref: TeamsMeetingRef, recording: MeetingArtifact
- ) -> str:
+ async def _transcribe_recording(self, job: TeamsMeetingPipelineJob, meeting_ref: TeamsMeetingRef, recording: MeetingArtifact) -> str:
temp_root = self.config.tmp_dir or (get_hermes_home() / "tmp" / "teams_pipeline")
temp_root.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(dir=str(temp_root), prefix="teams-recording-") as tmp_dir:
@@ -396,57 +338,40 @@ class TeamsMeetingPipeline:
raise TeamsPipelineRetryableError("Recording fallback requires ffmpeg for audio extraction, but ffmpeg was not found.")
audio_path = recording_path.with_suffix(".wav")
proc = await asyncio.create_subprocess_exec(
- ffmpeg, "-y", "-i", str(recording_path), str(audio_path),
- stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
- )
+ ffmpeg, "-y", "-i", str(recording_path), str(audio_path), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
_stdout, stderr = await proc.communicate()
if proc.returncode != 0:
- raise TeamsPipelineRetryableError(
- f"ffmpeg audio extraction failed: {stderr.decode('utf-8', errors='replace').strip()}"
- )
+ raise TeamsPipelineRetryableError(f"ffmpeg audio extraction failed: {stderr.decode('utf-8', errors='replace').strip()}")
return audio_path
async def _generate_summary_payload(
- self, *, resolved_meeting: TeamsMeetingRef, transcript_text: str, artifacts: list[MeetingArtifact]
- ) -> TeamsMeetingSummaryPayload:
+ self, *, resolved_meeting: TeamsMeetingRef, transcript_text: str, artifacts: list[MeetingArtifact]) -> TeamsMeetingSummaryPayload:
prompt = _build_summary_prompt(resolved_meeting, transcript_text, artifacts)
try:
response = await async_call_llm(
task="call", temperature=0.2, max_tokens=900,
- messages=[{"role": "system", "content": _SUMMARY_SYSTEM_PROMPT}, {"role": "user", "content": prompt}],
- )
+ messages=[{"role": "system", "content": _SUMMARY_SYSTEM_PROMPT}, {"role": "user", "content": prompt}])
parsed = _parse_summary_json(extract_content_or_reasoning(response))
except Exception as exc:
logger.info("Teams pipeline LLM summary unavailable, using heuristic summary: %s", exc)
parsed = _heuristic_summary(transcript_text)
-
teams_delivery = self.config.teams_delivery or {}
return TeamsMeetingSummaryPayload(
- meeting_ref=resolved_meeting,
+ meeting_ref=resolved_meeting, transcript_text=transcript_text, source_artifacts=artifacts,
title=str(resolved_meeting.metadata.get("subject") or f"Meeting {resolved_meeting.meeting_id}"),
- start_time=resolved_meeting.metadata.get("startDateTime"),
- end_time=resolved_meeting.metadata.get("endDateTime"),
- participants=_collect_participants(resolved_meeting),
- transcript_text=transcript_text,
- summary=parsed.get("summary"),
- key_decisions=list(parsed.get("key_decisions") or []),
- action_items=list(parsed.get("action_items") or []),
- risks=list(parsed.get("risks") or []),
- call_metrics=_collect_call_metrics(artifacts),
- source_artifacts=artifacts,
- confidence=parsed.get("confidence"), confidence_notes=parsed.get("confidence_notes"),
- notion_target=(self.config.notion or {}).get("database_id"),
- linear_target=(self.config.linear or {}).get("team_id"),
- teams_target=teams_delivery.get("channel_id") or teams_delivery.get("chat_id"),
- )
+ start_time=resolved_meeting.metadata.get("startDateTime"), end_time=resolved_meeting.metadata.get("endDateTime"),
+ participants=_collect_participants(resolved_meeting), call_metrics=_collect_call_metrics(artifacts),
+ summary=parsed.get("summary"), confidence=parsed.get("confidence"), confidence_notes=parsed.get("confidence_notes"),
+ **{key: list(parsed.get(key) or []) for key in ("key_decisions", "action_items", "risks")},
+ notion_target=(self.config.notion or {}).get("database_id"), linear_target=(self.config.linear or {}).get("team_id"),
+ teams_target=teams_delivery.get("channel_id") or teams_delivery.get("chat_id"))
async def _write_sinks(self, job: TeamsMeetingPipelineJob, payload: TeamsMeetingSummaryPayload) -> None:
# Sink order is part of the contract: Notion, then Linear, then Teams delivery.
sinks = (
("notion", "writing_notion", self.config.notion, self.notion_writer),
("linear", "writing_linear", self.config.linear, self.linear_writer),
- ("teams", "sending_teams", self.config.teams_delivery, self.teams_sender),
- )
+ ("teams", "sending_teams", self.config.teams_delivery, self.teams_sender))
for name, status, config, sink in sinks:
if not (config and config.get("enabled") and sink):
continue
@@ -470,11 +395,8 @@ def _collect_participants(meeting_ref: TeamsMeetingRef) -> list[str]:
participants = meeting_ref.metadata.get("participants") or []
if not isinstance(participants, list):
return []
- names = (
- item.get("displayName") or (((item.get("identity") or {}).get("user") or {}).get("displayName"))
- for item in participants
- if isinstance(item, dict)
- )
+ names = (item.get("displayName") or (((item.get("identity") or {}).get("user") or {}).get("displayName"))
+ for item in participants if isinstance(item, dict))
return [str(name) for name in names if name]
@@ -504,41 +426,24 @@ def _resource_data_id_is_artifact(notification: dict[str, Any], resource_data: d
def _meeting_ids_from_notification(notification: dict[str, Any]) -> tuple[str, str | None, dict[str, Any]]:
- """Return (meeting_id, organizer_user_id, extra_metadata) from a Graph change notification.
-
- Ids are taken from parsed resource paths first (odata.id, resource, transcriptContentUrl),
- then from flat fields. resourceData.id is a meeting id only when it is not an artifact id.
- The meeting_id never comes back empty: it falls back to an artifact id, then the receipt key.
- """
+ """Return (meeting_id, organizer_user_id, extra_metadata) from a Graph change notification: parsed
+ resource paths win over flat fields; resourceData.id counts as a meeting id only when it is not an
+ artifact id; meeting_id never comes back empty (falls back to an artifact id, then the receipt key)."""
resource_data = notification.get("resourceData")
resource_data = resource_data if isinstance(resource_data, dict) else {}
odata_type = str(_odata_field(resource_data, "odata.type") or "")
- parsed_paths = [
- parse_graph_meeting_resource(str(_odata_field(resource_data, "odata.id") or "")),
- parse_graph_meeting_resource(str(notification.get("resource") or "")),
- parse_graph_meeting_resource(str(resource_data.get("transcriptContentUrl") or "")),
- ]
-
- first = {
- key: next((parsed[key] for parsed in parsed_paths if parsed.get(key)), None)
- for key in ("organizer_user_id", "meeting_id", "transcript_id", "recording_id")
- }
- organizer_user_id = (
- first["organizer_user_id"]
- or _organizer_user_id_from_payload(resource_data)
- or _organizer_user_id_from_payload(notification)
- )
- meeting_id = first["meeting_id"] or (
- str(resource_data.get("meetingId") or notification.get("meetingId") or "").strip() or None
- )
+ parsed_paths = [parse_graph_meeting_resource(str(raw or "")) for raw in (
+ _odata_field(resource_data, "odata.id"), notification.get("resource"), resource_data.get("transcriptContentUrl"))]
+ first = {key: next((parsed[key] for parsed in parsed_paths if parsed.get(key)), None)
+ for key in ("organizer_user_id", "meeting_id", "transcript_id", "recording_id")}
+ organizer_user_id = first["organizer_user_id"] or _organizer_user_id_from_payload(resource_data) or _organizer_user_id_from_payload(notification)
+ meeting_id = first["meeting_id"] or (str(resource_data.get("meetingId") or notification.get("meetingId") or "").strip() or None)
transcript_id, recording_id = first["transcript_id"], first["recording_id"]
-
resource_data_id = str(resource_data.get("id") or "").strip() or None
if resource_data_id and not _resource_data_id_is_artifact(notification, resource_data):
meeting_id = meeting_id or resource_data_id
elif resource_data_id and not transcript_id and looks_like_transcript_id(resource_data_id, odata_type=odata_type):
transcript_id = resource_data_id
-
meeting_id = meeting_id or transcript_id or recording_id or TeamsPipelineStore.build_notification_receipt_key(notification)
extra_metadata = {k: v for k, v in (("transcript_id", transcript_id), ("recording_id", recording_id)) if v}
return str(meeting_id), organizer_user_id, extra_metadata
@@ -550,8 +455,7 @@ def _build_summary_prompt(meeting_ref: TeamsMeetingRef, transcript_text: str, ar
f"Meeting ID: {meeting_ref.meeting_id}\n"
f"Title: {meeting_ref.metadata.get('subject') or 'Unknown'}\n"
f"Artifacts:\n{chr(10).join(artifact_lines) or '- none'}\n\n"
- f"Transcript:\n{transcript_text[:18000]}"
- )
+ f"Transcript:\n{transcript_text[:18000]}")
def _clean_items(values: Any) -> list[str]:
@@ -571,8 +475,7 @@ def _parse_summary_json(content: str) -> dict[str, Any]:
"summary": str(payload.get("summary") or "").strip(),
**{key: _clean_items(payload.get(key, [])) for key in ("key_decisions", "action_items", "risks")},
"confidence": str(payload.get("confidence") or "medium").strip(),
- "confidence_notes": str(payload.get("confidence_notes") or "").strip(),
- }
+ "confidence_notes": str(payload.get("confidence_notes") or "").strip()}
def _heuristic_summary(transcript_text: str) -> dict[str, Any]:
@@ -581,13 +484,10 @@ def _heuristic_summary(transcript_text: str) -> dict[str, Any]:
return {
"summary": " ".join(lines[:3])[:1200] or "Transcript unavailable or too sparse for a confident summary.",
"key_decisions": [line for line, low in zip(lines, lowered) if "decide" in low or "decision" in low][:6],
- "action_items": [
- line for line, low in zip(lines, lowered) if low.startswith(("action:", "todo:", "next step:", "follow up:"))
- ][:8],
+ "action_items": [line for line, low in zip(lines, lowered) if low.startswith(("action:", "todo:", "next step:", "follow up:"))][:8],
"risks": [line for line, low in zip(lines, lowered) if "risk" in low or "blocker" in low][:6],
"confidence": "low" if len(transcript_text.strip()) < 300 else "medium",
- "confidence_notes": "Generated with heuristic fallback because no LLM summary response was available.",
- }
+ "confidence_notes": "Generated with heuristic fallback because no LLM summary response was available."}
def _render_summary_markdown(payload: TeamsMeetingSummaryPayload) -> str:
diff --git a/plugins/teams_pipeline/runtime.py b/plugins/teams_pipeline/runtime.py
index 40f272b023..5d25afa99c 100644
--- a/plugins/teams_pipeline/runtime.py
+++ b/plugins/teams_pipeline/runtime.py
@@ -19,7 +19,6 @@ _DELIVERY_KEYS = ("incoming_webhook_url", "access_token", "team_id", "channel_id
def _teams_delivery_is_configured(teams_extra: dict[str, Any], teams_delivery: dict[str, Any]) -> bool:
def pick(key: str) -> Any:
return teams_delivery.get(key) or teams_extra.get(key)
-
delivery_mode = str(teams_delivery.get("mode") or pick("delivery_mode") or "").strip().lower()
if delivery_mode == "incoming_webhook":
return bool(pick("incoming_webhook_url"))
@@ -34,11 +33,9 @@ def build_pipeline_runtime_config(gateway_config: Any) -> dict[str, Any]:
teams_config = gateway_config.platforms.get(Platform("teams"))
teams_extra = dict((teams_config.extra or {}) if teams_config else {})
pipeline_config = dict(teams_extra.get("meeting_pipeline") or {})
-
if teams_config and teams_config.enabled:
teams_delivery = dict(pipeline_config.get("teams_delivery") or {})
- delivery_mode = str(teams_extra.get("delivery_mode") or "").strip()
- if delivery_mode:
+ if delivery_mode := str(teams_extra.get("delivery_mode") or "").strip():
teams_delivery["mode"] = delivery_mode
for key in _DELIVERY_KEYS:
value = teams_extra.get(key)
@@ -47,7 +44,6 @@ def build_pipeline_runtime_config(gateway_config: Any) -> dict[str, Any]:
if teams_delivery:
teams_delivery["enabled"] = _teams_delivery_is_configured(teams_extra, teams_delivery)
pipeline_config["teams_delivery"] = teams_delivery
-
return pipeline_config
@@ -59,17 +55,12 @@ def build_pipeline_runtime(gateway: Any) -> TeamsMeetingPipeline:
try:
from plugins.platforms.teams.adapter import TeamsSummaryWriter
except ImportError:
- logger.debug(
- "TeamsSummaryWriter unavailable; Teams outbound delivery remains disabled until the adapter layer is present."
- )
+ logger.debug("TeamsSummaryWriter unavailable; Teams outbound delivery remains disabled until the adapter layer is present.")
else:
teams_sender = TeamsSummaryWriter(platform_config=teams_config)
-
return TeamsMeetingPipeline(
- graph_client=build_graph_client(),
- store=TeamsPipelineStore(resolve_teams_pipeline_store_path()),
- config=pipeline_config,
- teams_sender=teams_sender,
+ graph_client=build_graph_client(), store=TeamsPipelineStore(resolve_teams_pipeline_store_path()),
+ config=pipeline_config, teams_sender=teams_sender,
)
@@ -78,10 +69,8 @@ def bind_gateway_runtime(gateway: Any) -> bool:
adapter = gateway.adapters.get(Platform.MSGRAPH_WEBHOOK)
if adapter is None:
return False
-
if getattr(gateway, "_teams_pipeline_runtime", None) is not None:
return True
-
try:
runtime = build_pipeline_runtime(gateway)
except Exception as exc:
@@ -89,23 +78,17 @@ def bind_gateway_runtime(gateway: Any) -> bool:
gateway._teams_pipeline_runtime_error = str(exc)
logger.warning(
"Teams pipeline runtime unavailable: %s. Installing a drop-scheduler "
- "so Graph notifications ack cleanly without piling up unbound.",
- exc,
+ "so Graph notifications ack cleanly without piling up unbound.", exc,
)
async def _drop(notification: dict[str, Any], event: Any) -> None:
- logger.debug(
- "Dropping Graph notification because runtime is unavailable: id=%s resource=%s",
- notification.get("id"),
- notification.get("resource"),
- )
-
+ logger.debug("Dropping Graph notification because runtime is unavailable: id=%s resource=%s",
+ notification.get("id"), notification.get("resource"))
adapter.set_notification_scheduler(_drop)
return False
async def _schedule(notification: dict[str, Any], event: Any) -> None:
await runtime.run_notification(notification)
-
adapter.set_notification_scheduler(_schedule)
gateway._teams_pipeline_runtime = runtime
gateway._teams_pipeline_runtime_error = None
diff --git a/plugins/teams_pipeline/store.py b/plugins/teams_pipeline/store.py
index c86c90a63c..e2f364ffdd 100644
--- a/plugins/teams_pipeline/store.py
+++ b/plugins/teams_pipeline/store.py
@@ -8,6 +8,7 @@ import os
import threading
from copy import deepcopy
from datetime import datetime, timezone
+from functools import partialmethod
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any, Dict, Optional
@@ -41,9 +42,7 @@ class TeamsPipelineStore:
def _load(self) -> None:
with self._lock:
- if not self.path.exists():
- return
- data = json.loads(self.path.read_text(encoding="utf-8") or "{}")
+ data = json.loads(self.path.read_text(encoding="utf-8") or "{}") if self.path.exists() else None
if isinstance(data, dict):
self._state = {bucket: dict(data.get(bucket) or {}) for bucket in _BUCKETS}
@@ -76,14 +75,14 @@ class TeamsPipelineStore:
self._persist()
return deepcopy(merged)
- def list_subscriptions(self) -> Dict[str, Dict[str, Any]]:
- return self._list("subscriptions")
-
- def get_subscription(self, subscription_id: str) -> Optional[Dict[str, Any]]:
- return self._get("subscriptions", subscription_id)
-
- def upsert_subscription(self, subscription_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
- return self._upsert("subscriptions", "subscription_id", subscription_id, payload)
+ list_subscriptions = partialmethod(_list, "subscriptions")
+ get_subscription = partialmethod(_get, "subscriptions")
+ upsert_subscription = partialmethod(_upsert, "subscriptions", "subscription_id")
+ list_jobs = partialmethod(_list, "jobs")
+ get_job = partialmethod(_get, "jobs")
+ upsert_job = partialmethod(_upsert, "jobs", "job_id")
+ get_sink_record = partialmethod(_get, "sink_records")
+ upsert_sink_record = partialmethod(_upsert, "sink_records", "sink_key")
def delete_subscription(self, subscription_id: str) -> bool:
with self._lock:
@@ -99,35 +98,16 @@ class TeamsPipelineStore:
canonical = json.dumps(notification, sort_keys=True, separators=(",", ":"))
return f"sha256:{hashlib.sha256(canonical.encode('utf-8')).hexdigest()}"
- def record_notification_receipt(
- self, receipt_key: str, payload: Optional[Dict[str, Any]] = None, *, received_at: Optional[str] = None
- ) -> bool:
+ def record_notification_receipt(self, receipt_key: str, payload: Optional[Dict[str, Any]] = None, *, received_at: Optional[str] = None) -> bool:
"""Record a receipt once; returns False when the key was already seen (duplicate delivery)."""
with self._lock:
if receipt_key in self._state["notification_receipts"]:
return False
- self._state["notification_receipts"][receipt_key] = {
- "received_at": received_at or _utc_now_iso(),
- "payload": deepcopy(payload) if isinstance(payload, dict) else payload,
- }
+ self._state["notification_receipts"][receipt_key] = {"received_at": received_at or _utc_now_iso(),
+ "payload": deepcopy(payload) if isinstance(payload, dict) else payload}
self._persist()
return True
def stats(self) -> Dict[str, int]:
with self._lock:
return {bucket: len(self._state[bucket]) for bucket in _BUCKETS}
-
- def upsert_job(self, job_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
- return self._upsert("jobs", "job_id", job_id, payload)
-
- def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
- return self._get("jobs", job_id)
-
- def list_jobs(self) -> Dict[str, Dict[str, Any]]:
- return self._list("jobs")
-
- def upsert_sink_record(self, sink_key: str, payload: Dict[str, Any]) -> Dict[str, Any]:
- return self._upsert("sink_records", "sink_key", sink_key, payload)
-
- def get_sink_record(self, sink_key: str) -> Optional[Dict[str, Any]]:
- return self._get("sink_records", sink_key)
diff --git a/plugins/teams_pipeline/subscriptions.py b/plugins/teams_pipeline/subscriptions.py
index 3142a68afd..22ffc5f25f 100644
--- a/plugins/teams_pipeline/subscriptions.py
+++ b/plugins/teams_pipeline/subscriptions.py
@@ -27,8 +27,7 @@ def utc_timestamp(hours_from_now: int = 0, *, base: datetime | None = None) -> s
def sync_graph_subscription_record(
- store: TeamsPipelineStore, subscription_payload: dict[str, Any], *, status: str | None = None,
- renewed: bool = False,
+ store: TeamsPipelineStore, subscription_payload: dict[str, Any], *, status: str | None = None, renewed: bool = False,
) -> dict[str, Any]:
normalized = GraphSubscription.from_dict(subscription_payload).to_dict()
if status is None:
@@ -46,16 +45,12 @@ def expected_client_state(raw: str | None = None) -> str | None:
return str(raw or "").strip() or None
-def is_managed_subscription(
- store: TeamsPipelineStore, subscription_payload: dict[str, Any], *, expected_client_state_value: str | None,
-) -> bool:
+def is_managed_subscription(store: TeamsPipelineStore, subscription_payload: dict[str, Any], *, expected_client_state_value: str | None) -> bool:
"""A subscription is ours if the store knows it or its clientState matches the configured one."""
subscription_id = str(subscription_payload.get("subscription_id") or subscription_payload.get("id") or "").strip()
if subscription_id and store.get_subscription(subscription_id):
return True
- candidate_state = str(
- subscription_payload.get("client_state") or subscription_payload.get("clientState") or ""
- ).strip()
+ candidate_state = str(subscription_payload.get("client_state") or subscription_payload.get("clientState") or "").strip()
return bool(expected_client_state_value and candidate_state == expected_client_state_value)
@@ -67,14 +62,12 @@ async def maintain_graph_subscriptions(
extend_hours = max(1, int(extend_hours))
managed_client_state = expected_client_state(client_state)
now = _utc_now()
-
remote_subscriptions = await client.collect_paginated("/subscriptions")
remote_ids: set[str] = set()
synced = 0
renewed: list[dict[str, Any]] = []
candidates: list[dict[str, Any]] = []
skipped: list[dict[str, Any]] = []
-
for raw in remote_subscriptions:
if not isinstance(raw, dict):
continue
@@ -84,11 +77,9 @@ async def maintain_graph_subscriptions(
def skip(reason: str, **extra: Any) -> None:
skipped.append({"subscription_id": subscription_id, "reason": reason, **extra})
-
if not is_managed_subscription(store, raw, expected_client_state_value=managed_client_state):
skip("not_managed_by_teams_pipeline")
continue
-
remote_ids.add(subscription_id)
try:
sync_graph_subscription_record(store, raw)
@@ -96,12 +87,10 @@ async def maintain_graph_subscriptions(
except Exception as exc:
skip(f"failed_to_sync_local_store: {exc}")
continue
-
expiration = _parse_datetime(raw.get("expirationDateTime"))
if expiration is None:
skip("missing_expiration")
continue
-
seconds_until_expiry = int((expiration - now).total_seconds())
if seconds_until_expiry < 0:
store.upsert_subscription(subscription_id, {"status": "expired", "expiration_datetime": _iso_z(expiration)})
@@ -110,30 +99,20 @@ async def maintain_graph_subscriptions(
if seconds_until_expiry > threshold_hours * 3600:
skip("not_due", expires_in_seconds=seconds_until_expiry)
continue
-
new_expiration = utc_timestamp(extend_hours, base=max(now, expiration))
- candidate = {
- "subscription_id": subscription_id, "resource": raw.get("resource"),
- "current_expiration": _iso_z(expiration), "new_expiration": new_expiration,
- }
+ candidate = {"subscription_id": subscription_id, "resource": raw.get("resource"),
+ "current_expiration": _iso_z(expiration), "new_expiration": new_expiration}
candidates.append(candidate)
if dry_run:
continue
-
- patched = await client.patch_json(
- f"/subscriptions/{subscription_id}", json_body={"expirationDateTime": new_expiration}
- )
+ patched = await client.patch_json(f"/subscriptions/{subscription_id}", json_body={"expirationDateTime": new_expiration})
merged = {**raw, **(patched or {}), "id": subscription_id, "expirationDateTime": new_expiration}
sync_graph_subscription_record(store, merged, status="active", renewed=True)
renewed.append({**candidate, "result": patched})
-
# Locally-known subscriptions Graph no longer reports are flagged, never deleted.
for subscription_id in store.list_subscriptions():
if subscription_id not in remote_ids:
- store.upsert_subscription(
- subscription_id, {"status": "missing_remote", "last_seen_missing_remote_at": utc_timestamp()}
- )
-
+ store.upsert_subscription(subscription_id, {"status": "missing_remote", "last_seen_missing_remote_at": utc_timestamp()})
return {
"success": True, "dry_run": bool(dry_run), "store_path": str(store.path),
"remote_subscription_count": len(remote_subscriptions), "synced_subscription_count": synced,