ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(adapters/feishu): 8095->5665; unify request/response wrappers, card builders, comment timeline selectors; dead symbols dropped; compact docs

Browse Source
This commit is contained in:
Teknium
2026-09-02 14:06:33 -07:00
parent 33c74a26c6
commit 513ba5b846
4 changed files with 1719 additions and 4149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/platforms/feishu/feishu_comment_rules.py
+106 -193
View File
@@ -11,6 +11,7 @@ from __future__ import annotations
import json
import logging
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
@@ -20,22 +21,11 @@ from hermes_constants import get_hermes_home
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
#
# Uses the canonical ``get_hermes_home()`` helper (HERMES_HOME-aware and
# profile-safe). Resolved at import time; this module is lazy-imported by
# the Feishu comment event handler, which runs long after profile overrides
# have been applied, so freezing paths here is safe.
# Resolved at import time: this module is lazy-imported by the comment event handler,
# long after profile/HERMES_HOME overrides have been applied, so freezing is safe.
RULES_FILE = get_hermes_home() / "feishu_comment_rules.json"
PAIRING_FILE = get_hermes_home() / "feishu_comment_pairing.json"
# ---------------------------------------------------------------------------
# Data models
# ---------------------------------------------------------------------------
_VALID_POLICIES = ("allowlist", "pairing")
@@ -62,15 +52,11 @@ class ResolvedCommentRule:
enabled: bool
policy: str
allow_from: frozenset
match_source: str # e.g. "exact:docx:xxx" | "wildcard" | "top" | "default"
match_source: str # e.g. "exact:docx:xxx" | "wildcard" | "top"
# ---------------------------------------------------------------------------
# Mtime-cached file loading
# ---------------------------------------------------------------------------
class _MtimeCache:
"""Generic mtime-based file cache. ``stat()`` per access, re-read only on change."""
"""Mtime-based JSON file cache: ``stat()`` per access, re-read only on change."""
def __init__(self, path: Path):
self._path = path
@@ -79,16 +65,13 @@ class _MtimeCache:
def load(self) -> dict:
try:
st = self._path.stat()
mtime = st.st_mtime
mtime = self._path.stat().st_mtime
except FileNotFoundError:
self._mtime = 0.0
self._data = {}
return {}
if mtime == self._mtime and self._data is not None:
return self._data
try:
with open(self._path, "r", encoding="utf-8") as f:
data = json.load(f)
@@ -97,7 +80,6 @@ class _MtimeCache:
except (json.JSONDecodeError, OSError):
logger.warning("[Feishu-Rules] Failed to read %s, using empty config", self._path)
data = {}
self._mtime = mtime
self._data = data
return data
@@ -107,30 +89,30 @@ _rules_cache = _MtimeCache(RULES_FILE)
_pairing_cache = _MtimeCache(PAIRING_FILE)
# ---------------------------------------------------------------------------
# Config parsing
# ---------------------------------------------------------------------------
# --- Config parsing ---
def _parse_frozenset(raw: Any) -> Optional[frozenset]:
"""Parse a list of strings into a frozenset; return None if key absent."""
if raw is None:
return None
"""Parse a list of strings into a frozenset; None if absent or not a list."""
if isinstance(raw, (list, tuple)):
return frozenset(str(u).strip() for u in raw if str(u).strip())
return None
def _parse_policy(raw: Any, default: Optional[str]) -> Optional[str]:
"""Normalize a policy value; unknown/invalid values fall back to *default*."""
if raw is None:
return default
policy = str(raw).strip().lower()
return policy if policy in _VALID_POLICIES else default
def _parse_document_rule(raw: dict) -> CommentDocumentRule:
enabled = raw.get("enabled")
if enabled is not None:
enabled = bool(enabled)
policy = raw.get("policy")
if policy is not None:
policy = str(policy).strip().lower()
if policy not in _VALID_POLICIES:
policy = None
allow_from = _parse_frozenset(raw.get("allow_from"))
return CommentDocumentRule(enabled=enabled, policy=policy, allow_from=allow_from)
return CommentDocumentRule(
enabled=None if enabled is None else bool(enabled),
policy=_parse_policy(raw.get("policy"), None),
allow_from=_parse_frozenset(raw.get("allow_from")),
)
def load_config() -> CommentsConfig:
@@ -138,93 +120,56 @@ def load_config() -> CommentsConfig:
raw = _rules_cache.load()
if not raw:
return CommentsConfig()
documents: Dict[str, CommentDocumentRule] = {}
raw_docs = raw.get("documents", {})
if isinstance(raw_docs, dict):
for key, rule_raw in raw_docs.items():
if isinstance(rule_raw, dict):
documents[str(key)] = _parse_document_rule(rule_raw)
policy = str(raw.get("policy", "pairing")).strip().lower()
if policy not in _VALID_POLICIES:
policy = "pairing"
documents = {
str(key): _parse_document_rule(rule_raw)
for key, rule_raw in (raw_docs.items() if isinstance(raw_docs, dict) else ())
if isinstance(rule_raw, dict)
}
return CommentsConfig(
enabled=raw.get("enabled", True),
policy=policy,
allow_from=_parse_frozenset(raw.get("allow_from")) or frozenset(),
documents=documents,
policy=_parse_policy(raw.get("policy", "pairing"), "pairing"),
allow_from=_parse_frozenset(raw.get("allow_from")) or frozenset(), documents=documents,
)
# ---------------------------------------------------------------------------
# Rule resolution (§8.4 field-by-field fallback)
# ---------------------------------------------------------------------------
# --- Rule resolution (field-by-field fallback) ---
def has_wiki_keys(cfg: CommentsConfig) -> bool:
"""Check if any document rule key starts with 'wiki:'."""
return any(k.startswith("wiki:") for k in cfg.documents)
def resolve_rule(
cfg: CommentsConfig,
file_type: str,
file_token: str,
wiki_token: str = "",
) -> ResolvedCommentRule:
def resolve_rule(cfg: CommentsConfig, file_type: str, file_token: str, wiki_token: str = "") -> ResolvedCommentRule:
"""Resolve effective rule: exact doc → wiki key → wildcard → top-level → defaults."""
exact_key = f"{file_type}:{file_token}"
exact = cfg.documents.get(exact_key)
exact_src = f"exact:{exact_key}"
if exact is None and wiki_token:
wiki_key = f"wiki:{wiki_token}"
exact = cfg.documents.get(wiki_key)
exact_src = f"exact:{wiki_key}"
wildcard = cfg.documents.get("*")
layers = []
if exact is not None:
layers.append((exact, exact_src))
if wildcard is not None:
layers.append((wildcard, "wildcard"))
exact_key = f"wiki:{wiki_token}"
exact = cfg.documents.get(exact_key)
layers = [(exact, f"exact:{exact_key}"), (cfg.documents.get("*"), "wildcard")]
def _pick(field_name: str):
for layer, source in layers:
val = getattr(layer, field_name)
if val is not None:
return val, source
# First non-None document-layer value wins; otherwise the top-level value (even if None).
for layer, src in layers:
if layer is not None and getattr(layer, field_name) is not None:
return getattr(layer, field_name), src
return getattr(cfg, field_name), "top"
enabled, en_src = _pick("enabled")
policy, pol_src = _pick("policy")
allow_from, _ = _pick("allow_from")
# match_source = highest-priority tier that contributed any field
# match_source = highest-priority tier that contributed enabled or policy
priority_order = {"exact": 0, "wildcard": 1, "top": 2}
best_src = min(
[en_src, pol_src],
key=lambda s: priority_order.get(s.split(":")[0], 3),
)
return ResolvedCommentRule(
enabled=enabled,
policy=policy,
allow_from=allow_from,
match_source=best_src,
)
best_src = min([en_src, pol_src], key=lambda s: priority_order.get(s.split(":")[0], 3))
return ResolvedCommentRule(enabled=enabled, policy=policy, allow_from=allow_from, match_source=best_src)
# ---------------------------------------------------------------------------
# Pairing store
# ---------------------------------------------------------------------------
# --- Pairing store ---
def _load_pairing_approved() -> set:
"""Return set of approved user open_ids (mtime-cached)."""
data = _pairing_cache.load()
approved = data.get("approved", {})
approved = _pairing_cache.load().get("approved", {})
if isinstance(approved, dict):
return set(approved.keys())
if isinstance(approved, list):
@@ -238,49 +183,46 @@ def _save_pairing(data: dict) -> None:
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
tmp.replace(PAIRING_FILE)
# Invalidate cache so next load picks up change
_pairing_cache._mtime = 0.0
_pairing_cache._mtime = 0.0 # invalidate so the next load re-reads
_pairing_cache._data = None
def _mutate_pairing(user_open_id: str, add: bool) -> bool:
"""Add/remove *user_open_id* in the approved dict; True when the store actually changed."""
data = _pairing_cache.load()
approved = data.get("approved", {})
if not isinstance(approved, dict):
if not add:
return False
approved = {}
if (user_open_id in approved) == add:
return False
if add:
approved[user_open_id] = {"approved_at": time.time()}
else:
del approved[user_open_id]
data["approved"] = approved
_save_pairing(data)
return True
def pairing_add(user_open_id: str) -> bool:
"""Add a user to the pairing-approved list. Returns True if newly added."""
data = _pairing_cache.load()
approved = data.get("approved", {})
if not isinstance(approved, dict):
approved = {}
if user_open_id in approved:
return False
approved[user_open_id] = {"approved_at": time.time()}
data["approved"] = approved
_save_pairing(data)
return True
return _mutate_pairing(user_open_id, add=True)
def pairing_remove(user_open_id: str) -> bool:
"""Remove a user from the pairing-approved list. Returns True if removed."""
data = _pairing_cache.load()
approved = data.get("approved", {})
if not isinstance(approved, dict):
return False
if user_open_id not in approved:
return False
del approved[user_open_id]
data["approved"] = approved
_save_pairing(data)
return True
return _mutate_pairing(user_open_id, add=False)
def pairing_list() -> Dict[str, Any]:
"""Return the approved dict {user_open_id: {approved_at: ...}}."""
data = _pairing_cache.load()
approved = data.get("approved", {})
approved = _pairing_cache.load().get("approved", {})
return dict(approved) if isinstance(approved, dict) else {}
# ---------------------------------------------------------------------------
# Access check (public API for feishu_comment.py)
# ---------------------------------------------------------------------------
# --- Access check (public API for feishu_comment.py) ---
def is_user_allowed(rule: ResolvedCommentRule, user_open_id: str) -> bool:
"""Check if user passes the resolved rule's policy gate."""
@@ -291,32 +233,20 @@ def is_user_allowed(rule: ResolvedCommentRule, user_open_id: str) -> bool:
return False
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
# --- CLI ---
def _print_status() -> None:
cfg = load_config()
print(f"Rules file: {RULES_FILE}")
print(f" exists: {RULES_FILE.exists()}")
print(f"Pairing file: {PAIRING_FILE}")
print(f" exists: {PAIRING_FILE.exists()}")
print()
print("Top-level:")
print(f" enabled: {cfg.enabled}")
print(f" policy: {cfg.policy}")
print(f" allow_from: {sorted(cfg.allow_from) if cfg.allow_from else '[]'}")
print()
print(f"Rules file: {RULES_FILE}\n exists: {RULES_FILE.exists()}")
print(f"Pairing file: {PAIRING_FILE}\n exists: {PAIRING_FILE.exists()}\n")
print(f"Top-level:\n enabled: {cfg.enabled}\n policy: {cfg.policy}")
print(f" allow_from: {sorted(cfg.allow_from) if cfg.allow_from else '[]'}\n")
if cfg.documents:
print(f"Document rules ({len(cfg.documents)}):")
for key, rule in sorted(cfg.documents.items()):
parts = []
if rule.enabled is not None:
parts.append(f"enabled={rule.enabled}")
if rule.policy is not None:
parts.append(f"policy={rule.policy}")
if rule.allow_from is not None:
parts.append(f"allow_from={sorted(rule.allow_from)}")
fields = (("enabled", rule.enabled), ("policy", rule.policy),
("allow_from", sorted(rule.allow_from) if rule.allow_from is not None else None))
parts = [f"{name}={value}" for name, value in fields if value is not None]
print(f" [{key}] {', '.join(parts) if parts else '(empty — inherits all)'}")
else:
print("Document rules: (none)")
@@ -324,8 +254,7 @@ def _print_status() -> None:
approved = pairing_list()
print(f"Pairing approved ({len(approved)}):")
for uid, meta in sorted(approved.items()):
ts = meta.get("approved_at", 0)
print(f" {uid} (approved_at={ts})")
print(f" {uid} (approved_at={meta.get('approved_at', 0)})")
def _do_check(doc_key: str, user_open_id: str) -> None:
@@ -334,28 +263,46 @@ def _do_check(doc_key: str, user_open_id: str) -> None:
if len(parts) != 2:
print(f"Error: doc_key must be 'fileType:fileToken', got '{doc_key}'")
return
file_type, file_token = parts
rule = resolve_rule(cfg, file_type, file_token)
rule = resolve_rule(cfg, parts[0], parts[1])
allowed = is_user_allowed(rule, user_open_id)
print(f"Document: {doc_key}")
print(f"User: {user_open_id}")
print("Resolved rule:")
print(f" enabled: {rule.enabled}")
print(f" policy: {rule.policy}")
print(f"Document: {doc_key}\nUser: {user_open_id}\nResolved rule:")
print(f" enabled: {rule.enabled}\n policy: {rule.policy}")
print(f" allow_from: {sorted(rule.allow_from) if rule.allow_from else '[]'}")
print(f" match_source: {rule.match_source}")
print(f"Result: {'ALLOWED' if allowed else 'DENIED'}")
print(f" match_source: {rule.match_source}\nResult: {'ALLOWED' if allowed else 'DENIED'}")
def _pairing_cmd(args: list) -> int:
"""Handle ``pairing <add|remove|list> [user]``; returns the exit code."""
if len(args) < 2:
print("Usage: pairing <add|remove|list> [args]")
return 1
sub = args[1]
if sub == "list":
approved = pairing_list()
if not approved:
print("(no approved users)")
for uid, meta in sorted(approved.items()):
print(f" {uid} approved_at={meta.get('approved_at', '?')}")
return 0
ops = {"add": (pairing_add, "Added: {}", "Already approved: {}"),
"remove": (pairing_remove, "Removed: {}", "Not in approved list: {}")}
if sub not in ops:
print(f"Unknown pairing subcommand: {sub}")
return 1
if len(args) < 3:
print(f"Usage: pairing {sub} <user_open_id>")
return 1
fn, ok_msg, noop_msg = ops[sub]
print((ok_msg if fn(args[2]) else noop_msg).format(args[2]))
return 0
def _main() -> int:
import sys
try:
from hermes_cli.env_loader import load_hermes_dotenv
load_hermes_dotenv()
except Exception:
pass
usage = (
"Usage: python -m gateway.platforms.feishu_comment_rules <command> [args]\n"
"\n"
@@ -370,53 +317,20 @@ def _main() -> int:
" Edit this JSON file directly to configure policies and document rules.\n"
" Changes take effect on the next comment event (no restart needed).\n"
)
args = sys.argv[1:]
if not args:
print(usage)
return 1
cmd = args[0]
if cmd == "status":
_print_status()
elif cmd == "check":
if len(args) < 3:
print("Usage: check <fileType:fileToken> <user_open_id>")
return 1
_do_check(args[1], args[2])
elif cmd == "pairing":
if len(args) < 2:
print("Usage: pairing <add|remove|list> [args]")
return 1
sub = args[1]
if sub == "add":
if len(args) < 3:
print("Usage: pairing add <user_open_id>")
return 1
if pairing_add(args[2]):
print(f"Added: {args[2]}")
else:
print(f"Already approved: {args[2]}")
elif sub == "remove":
if len(args) < 3:
print("Usage: pairing remove <user_open_id>")
return 1
if pairing_remove(args[2]):
print(f"Removed: {args[2]}")
else:
print(f"Not in approved list: {args[2]}")
elif sub == "list":
approved = pairing_list()
if not approved:
print("(no approved users)")
for uid, meta in sorted(approved.items()):
print(f" {uid} approved_at={meta.get('approved_at', '?')}")
else:
print(f"Unknown pairing subcommand: {sub}")
return 1
return _pairing_cmd(args)
else:
print(f"Unknown command: {cmd}\n")
print(usage)
@@ -425,5 +339,4 @@ def _main() -> int:
if __name__ == "__main__":
import sys
sys.exit(_main())