feat(farcaster): 新增Farcaster去中心化社交协议通道插件
该提交完整实现了ForcePilot的Farcaster通道插件，包含： 1. 基础配置适配器与多账户支持 2. Neynar API客户端封装，带重试机制与签名验证 3. 消息去重处理模块 4. 网关服务与健康检查、轮询降级逻辑 5. Webhook回调端点与事件解析 6. 收发消息、 reactions、媒体发送等完整交互能力 7. 插件元数据与系统集成适配
This commit is contained in:
parent
59c6caaa64
commit
7c6186e9a4
503
backend/package/yuxi/channel/extensions/farcaster/__init__.py
Normal file
503
backend/package/yuxi/channel/extensions/farcaster/__init__.py
Normal file
@ -0,0 +1,503 @@
|
||||
import logging
|
||||
|
||||
import httpx
|
||||
|
||||
from yuxi.channel.capabilities import ChannelCapabilities
|
||||
from yuxi.channel.context import ChannelContext
|
||||
from yuxi.channel.extensions.base import BaseChannelPlugin
|
||||
from yuxi.channel.extensions.farcaster.client import NeynarClient, NeynarError
|
||||
from yuxi.channel.extensions.farcaster.config import FarcasterConfigAdapter
|
||||
from yuxi.channel.extensions.farcaster.defaults import DEDUPE_TTL_S, DEDUPE_MAX_TRACKED
|
||||
from yuxi.channel.extensions.farcaster.dedupe import FarcasterDeduplicator
|
||||
from yuxi.channel.extensions.farcaster.direct_cast_client import DirectCastClient, DirectCastError
|
||||
from yuxi.channel.extensions.farcaster.gateway import FarcasterGateway, FarcasterGatewayError
|
||||
from yuxi.channel.extensions.farcaster.outbound import FarcasterOutboundAdapter
|
||||
from yuxi.channel.extensions.farcaster.webhook import (
|
||||
register_gateway_state,
|
||||
unregister_gateway_state,
|
||||
_dispatch_webhook_event,
|
||||
)
|
||||
from yuxi.channel.plugins.registry import ChannelPluginRegistry
|
||||
from yuxi.channel.protocols import ClassifiedError, ErrorSeverity
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FarcasterPlugin(BaseChannelPlugin):
|
||||
id = "farcaster"
|
||||
name = "Farcaster"
|
||||
order = 105
|
||||
label = "Farcaster"
|
||||
aliases = ["fc"]
|
||||
resolve_reply_to_mode = "thread"
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._config = FarcasterConfigAdapter()
|
||||
self._gateway = FarcasterGateway()
|
||||
self._outbound: FarcasterOutboundAdapter | None = None
|
||||
self._dedupe = FarcasterDeduplicator()
|
||||
|
||||
@property
|
||||
def capabilities(self) -> ChannelCapabilities:
|
||||
return ChannelCapabilities(
|
||||
chat_types=["direct"],
|
||||
message_types=["text"],
|
||||
interactions=["recast"],
|
||||
reactions=True,
|
||||
typing_indicator=False,
|
||||
threads=True,
|
||||
edit=False,
|
||||
unsend=True,
|
||||
reply=True,
|
||||
media=True,
|
||||
native_commands=False,
|
||||
polls=False,
|
||||
streaming=True,
|
||||
streaming_mode="block",
|
||||
block_streaming=True,
|
||||
block_streaming_chunk_min_chars=240,
|
||||
block_streaming_chunk_max_chars=300,
|
||||
block_streaming_chunk_break_preference="paragraph",
|
||||
)
|
||||
|
||||
# ── ConfigProtocol ──────────────────────────────────
|
||||
|
||||
def list_account_ids(self, config: dict) -> list[str]:
|
||||
return self._config.list_account_ids(config)
|
||||
|
||||
async def resolve_account(self, account_id: str) -> dict:
|
||||
return await self._config.resolve_account(account_id)
|
||||
|
||||
def is_configured(self, account: dict) -> bool:
|
||||
return self._config.is_configured(account)
|
||||
|
||||
def is_enabled(self, account: dict) -> bool:
|
||||
return self._config.is_enabled(account)
|
||||
|
||||
def disabled_reason(self, account: dict) -> str:
|
||||
return self._config.disabled_reason(account)
|
||||
|
||||
def describe_account(self, account: dict) -> dict:
|
||||
return self._config.describe_account(account)
|
||||
|
||||
async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None:
|
||||
account = await self._config.resolve_account(account_id)
|
||||
return account.get("allow_from", [])
|
||||
|
||||
def config_schema(self) -> dict:
|
||||
return self._config.config_schema()
|
||||
|
||||
# ── GatewayProtocol ─────────────────────────────────
|
||||
|
||||
async def start(self, ctx: ChannelContext) -> object:
|
||||
account_id = getattr(ctx, "account_id", "default")
|
||||
self._config.update_config(ctx.config)
|
||||
account = await self._config.resolve_account(account_id)
|
||||
|
||||
if not account.get("configured"):
|
||||
raise FarcasterGatewayError("Farcaster account not configured: missing api_key")
|
||||
|
||||
self._outbound = FarcasterOutboundAdapter(self._client_getter, account_id=account_id)
|
||||
|
||||
handle = await self._gateway.start(ctx)
|
||||
|
||||
register_gateway_state(account_id, ctx, self._gateway, self._dedupe)
|
||||
|
||||
logger.info(
|
||||
"Farcaster plugin started: account=%s fid=%s fname=%s",
|
||||
account_id,
|
||||
handle.get("bot_fid"),
|
||||
handle.get("bot_fname"),
|
||||
)
|
||||
return handle
|
||||
|
||||
async def stop(self, ctx: ChannelContext) -> None:
|
||||
account_id = getattr(ctx, "account_id", "default")
|
||||
unregister_gateway_state(account_id)
|
||||
await self._gateway.stop(ctx)
|
||||
self._outbound = None
|
||||
|
||||
# ── OutboundProtocol ────────────────────────────────
|
||||
|
||||
@property
|
||||
def delivery_mode(self):
|
||||
if self._outbound is None:
|
||||
from yuxi.channel.protocols import OutboundDeliveryMode
|
||||
|
||||
return OutboundDeliveryMode.DIRECT
|
||||
return self._outbound.delivery_mode
|
||||
|
||||
@property
|
||||
def text_chunk_limit(self) -> int | None:
|
||||
if self._outbound is None:
|
||||
return 320
|
||||
return self._outbound.text_chunk_limit
|
||||
|
||||
async def send_text(
|
||||
self,
|
||||
target_id: str,
|
||||
content: str,
|
||||
*,
|
||||
reply_to_id: str | None = None,
|
||||
thread_id: str | None = None,
|
||||
account_id: str | None = None,
|
||||
) -> None:
|
||||
if self._outbound is None:
|
||||
logger.error("Farcaster send_text: outbound not initialized")
|
||||
return
|
||||
resolved = await self._config.resolve_account(account_id or "default")
|
||||
channel_id = resolved.get("channel_id", "")
|
||||
await self._outbound.send_text(
|
||||
target_id,
|
||||
content,
|
||||
reply_to_id=reply_to_id,
|
||||
thread_id=thread_id,
|
||||
account_id=account_id,
|
||||
channel_id=channel_id or None,
|
||||
)
|
||||
|
||||
async def send_media(
|
||||
self,
|
||||
target_id: str,
|
||||
media_url: str,
|
||||
media_type: str,
|
||||
reply_to_id: str | None = None,
|
||||
thread_id: str | None = None,
|
||||
account_id: str | None = None,
|
||||
) -> None:
|
||||
if self._outbound is None:
|
||||
return
|
||||
await self._outbound.send_media(
|
||||
target_id,
|
||||
media_url,
|
||||
media_type,
|
||||
reply_to_id=reply_to_id,
|
||||
thread_id=thread_id,
|
||||
account_id=account_id,
|
||||
)
|
||||
|
||||
async def send_recast(
|
||||
self,
|
||||
target_hash: str,
|
||||
*,
|
||||
account_id: str | None = None,
|
||||
) -> None:
|
||||
if self._outbound is None:
|
||||
logger.error("Farcaster send_recast: outbound not initialized")
|
||||
return
|
||||
await self._outbound.send_reaction(target_hash, reaction_type="recast")
|
||||
|
||||
async def delete_cast(
|
||||
self,
|
||||
cast_hash: str,
|
||||
*,
|
||||
account_id: str | None = None,
|
||||
) -> None:
|
||||
client = self._client_getter(account_id or "default")
|
||||
if client is None:
|
||||
logger.error("Farcaster delete_cast: no active client")
|
||||
return
|
||||
try:
|
||||
await client.delete_cast(cast_hash)
|
||||
logger.info("Farcaster cast deleted: hash=%s", cast_hash)
|
||||
except NeynarError as e:
|
||||
logger.error("Farcaster delete_cast failed: %s", e)
|
||||
|
||||
async def send_direct_cast(
|
||||
self,
|
||||
recipient_fid: int,
|
||||
content: str,
|
||||
*,
|
||||
account_id: str | None = None,
|
||||
) -> None:
|
||||
account = await self._config.resolve_account(account_id or "default")
|
||||
api_key = account.get("api_key", "")
|
||||
signer_uuid = account.get("signer_uuid", "")
|
||||
if not api_key or not signer_uuid:
|
||||
logger.error("Farcaster send_direct_cast: missing api_key or signer_uuid")
|
||||
return
|
||||
dc_client = DirectCastClient(api_key=api_key, signer_uuid=signer_uuid)
|
||||
try:
|
||||
await dc_client.send_direct_cast(recipient_fid, content)
|
||||
logger.info("Farcaster direct cast sent to fid=%s", recipient_fid)
|
||||
except DirectCastError as e:
|
||||
logger.error("Farcaster send_direct_cast failed: %s", e)
|
||||
finally:
|
||||
await dc_client.close()
|
||||
|
||||
async def follow_user(
|
||||
self,
|
||||
target_fid: int,
|
||||
*,
|
||||
account_id: str | None = None,
|
||||
) -> dict:
|
||||
client = self._client_getter(account_id or "default")
|
||||
if client is None:
|
||||
raise NeynarError("No active Farcaster client")
|
||||
return await client.follow_user(target_fid)
|
||||
|
||||
async def unfollow_user(
|
||||
self,
|
||||
target_fid: int,
|
||||
*,
|
||||
account_id: str | None = None,
|
||||
) -> dict:
|
||||
client = self._client_getter(account_id or "default")
|
||||
if client is None:
|
||||
raise NeynarError("No active Farcaster client")
|
||||
return await client.unfollow_user(target_fid)
|
||||
|
||||
async def search_users(
|
||||
self,
|
||||
query: str,
|
||||
*,
|
||||
limit: int = 10,
|
||||
account_id: str | None = None,
|
||||
) -> list[dict]:
|
||||
client = self._client_getter(account_id or "default")
|
||||
if client is None:
|
||||
raise NeynarError("No active Farcaster client")
|
||||
return await client.search_users(query, limit=limit)
|
||||
|
||||
async def block_user(
|
||||
self,
|
||||
target_fid: int,
|
||||
*,
|
||||
account_id: str | None = None,
|
||||
) -> dict:
|
||||
client = self._client_getter(account_id or "default")
|
||||
if client is None:
|
||||
raise NeynarError("No active Farcaster client")
|
||||
return await client.block_user(target_fid)
|
||||
|
||||
async def unblock_user(
|
||||
self,
|
||||
target_fid: int,
|
||||
*,
|
||||
account_id: str | None = None,
|
||||
) -> dict:
|
||||
client = self._client_getter(account_id or "default")
|
||||
if client is None:
|
||||
raise NeynarError("No active Farcaster client")
|
||||
return await client.unblock_user(target_fid)
|
||||
|
||||
async def mute_user(
|
||||
self,
|
||||
target_fid: int,
|
||||
*,
|
||||
account_id: str | None = None,
|
||||
) -> dict:
|
||||
client = self._client_getter(account_id or "default")
|
||||
if client is None:
|
||||
raise NeynarError("No active Farcaster client")
|
||||
return await client.mute_user(target_fid)
|
||||
|
||||
async def unmute_user(
|
||||
self,
|
||||
target_fid: int,
|
||||
*,
|
||||
account_id: str | None = None,
|
||||
) -> dict:
|
||||
client = self._client_getter(account_id or "default")
|
||||
if client is None:
|
||||
raise NeynarError("No active Farcaster client")
|
||||
return await client.unmute_user(target_fid)
|
||||
|
||||
# ── StatusProtocol ──────────────────────────────────
|
||||
|
||||
async def probe(self, account: dict | None = None) -> bool:
|
||||
api_key = ""
|
||||
if account:
|
||||
api_key = account.get("api_key", "")
|
||||
else:
|
||||
resolved = await self._config.resolve_account("default")
|
||||
api_key = resolved.get("api_key", "")
|
||||
if not api_key:
|
||||
return False
|
||||
|
||||
client = NeynarClient(api_key=api_key)
|
||||
try:
|
||||
await client.get_me()
|
||||
return True
|
||||
except NeynarError:
|
||||
return False
|
||||
finally:
|
||||
await client.close()
|
||||
|
||||
def build_summary(self, snapshot: object) -> dict:
|
||||
return {"channel": "farcaster", "accounts": self._gateway.account_count}
|
||||
|
||||
# ── SecurityProtocol ────────────────────────────────
|
||||
|
||||
async def check_allowlist(self, peer_id: str, channel_type: str, account_id: str = "default") -> bool:
|
||||
account = await self._config.resolve_account(account_id)
|
||||
allow_from = account.get("allow_from", [])
|
||||
if not allow_from:
|
||||
return False
|
||||
fid = peer_id.replace("farcaster:", "").strip()
|
||||
return fid in allow_from or str(fid) in allow_from
|
||||
|
||||
def resolve_dm_policy(self) -> dict:
|
||||
return {"mode": "pairing", "allow_from": []}
|
||||
|
||||
# ── AgentPromptProtocol ─────────────────────────────
|
||||
|
||||
@property
|
||||
def channel_format_instructions(self) -> str | None:
|
||||
return (
|
||||
"You are connected to Farcaster, a decentralized social protocol. "
|
||||
"Messages are public Casts visible to everyone on the network. "
|
||||
"Each Cast is limited to 320 UTF-8 bytes. Longer responses will be automatically "
|
||||
"split into multiple Cast replies in the same thread. "
|
||||
"No markdown formatting is supported — use plain text only. "
|
||||
"URLs placed in Casts are automatically rendered as embeds (images, links) by clients. "
|
||||
"Use @username to mention other users. "
|
||||
"Farcaster community appreciates concise, Web3-native communication style. "
|
||||
"Keep answers brief, direct, and informative."
|
||||
)
|
||||
|
||||
# ── DedupeProtocol ──────────────────────────────────
|
||||
|
||||
def is_duplicate(self, key: str) -> bool:
|
||||
return self._dedupe.is_duplicate(key)
|
||||
|
||||
def mark_seen(self, key: str) -> None:
|
||||
self._dedupe.is_duplicate(key)
|
||||
|
||||
def reset(self) -> None:
|
||||
self._dedupe.reset()
|
||||
|
||||
@property
|
||||
def ttl_seconds(self) -> int:
|
||||
return DEDUPE_TTL_S
|
||||
|
||||
@property
|
||||
def max_entries(self) -> int:
|
||||
return DEDUPE_MAX_TRACKED
|
||||
|
||||
# ── InboundHandlerProtocol ──────────────────────────
|
||||
|
||||
def resolve_event_type(self, raw_event: dict) -> str:
|
||||
return raw_event.get("type", "unknown")
|
||||
|
||||
def parse_to_unified(
|
||||
self,
|
||||
raw_event: dict,
|
||||
account_id: str,
|
||||
) -> object | None:
|
||||
from yuxi.channel.extensions.farcaster.monitor import FarcasterMonitor
|
||||
|
||||
event_type = raw_event.get("type", "")
|
||||
event_data = raw_event.get("data", {})
|
||||
|
||||
handle = self._gateway.get_handle(account_id)
|
||||
bot_fid = handle.get("bot_fid", 0) if handle else 0
|
||||
bot_fname = handle.get("bot_fname", "") if handle else ""
|
||||
|
||||
monitor = FarcasterMonitor(bot_fid=bot_fid, bot_fname=bot_fname)
|
||||
|
||||
if event_type == "cast.created":
|
||||
return monitor.convert_cast_to_unified(event_data, account_id=account_id)
|
||||
elif event_type == "reaction.created":
|
||||
return monitor.convert_reaction_to_unified(event_data, account_id=account_id)
|
||||
return None
|
||||
|
||||
async def handle_raw_event(
|
||||
self,
|
||||
event: dict,
|
||||
account: dict,
|
||||
) -> object | None:
|
||||
return self.parse_to_unified(event, account.get("account_id", "default"))
|
||||
|
||||
# ── MentionsProtocol ────────────────────────────────
|
||||
|
||||
def extract_mentions(self, raw_message: dict) -> list[str]:
|
||||
from yuxi.channel.extensions.farcaster.monitor import FarcasterMonitor
|
||||
|
||||
text = raw_message.get("text", "")
|
||||
if not text:
|
||||
return []
|
||||
monitor = FarcasterMonitor()
|
||||
return monitor.extract_mentions(text)
|
||||
|
||||
# ── FormatProtocol ──────────────────────────────────
|
||||
|
||||
def markdown_to_native(self, md_text: str) -> dict | str:
|
||||
return _strip_markdown_formatting(md_text)
|
||||
|
||||
def native_to_markdown(self, native_content: dict | str) -> str:
|
||||
if isinstance(native_content, dict):
|
||||
return native_content.get("text", "")
|
||||
return str(native_content)
|
||||
|
||||
# ── WebhookProtocol ─────────────────────────────────
|
||||
|
||||
def verify_signature(self, body: bytes, signature: str, secret: str) -> bool:
|
||||
import hashlib
|
||||
import hmac
|
||||
|
||||
expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
|
||||
return hmac.compare_digest(expected, signature)
|
||||
|
||||
async def handle_webhook(
|
||||
self,
|
||||
request: object,
|
||||
account_id: str | None = None,
|
||||
) -> object:
|
||||
return await _dispatch_webhook_event(request, account_id or "default")
|
||||
|
||||
def resolve_webhook_auth_bypass_paths(self) -> list[str]:
|
||||
return ["/farcaster/webhook"]
|
||||
|
||||
# ── ErrorHandlingProtocol ───────────────────────────
|
||||
|
||||
def classify_error(self, error: BaseException) -> ClassifiedError:
|
||||
severity = self._classify_severity(error)
|
||||
return ClassifiedError(severity=severity, original_error=error, error_message=str(error))
|
||||
|
||||
def is_retryable(self, error: BaseException) -> bool:
|
||||
return self._classify_severity(error) == ErrorSeverity.RETRYABLE
|
||||
|
||||
def _classify_severity(self, error: BaseException) -> ErrorSeverity:
|
||||
if isinstance(error, NeynarError):
|
||||
return ErrorSeverity.RETRYABLE
|
||||
if isinstance(error, FarcasterGatewayError):
|
||||
return ErrorSeverity.FATAL
|
||||
if isinstance(error, httpx.HTTPStatusError):
|
||||
status = error.response.status_code
|
||||
if status == 429:
|
||||
return ErrorSeverity.RATE_LIMITED
|
||||
if status in (401, 403):
|
||||
return ErrorSeverity.FORBIDDEN
|
||||
if status >= 500:
|
||||
return ErrorSeverity.RETRYABLE
|
||||
return ErrorSeverity.FATAL
|
||||
if isinstance(error, (httpx.TimeoutException, httpx.ConnectError, httpx.RemoteProtocolError)):
|
||||
return ErrorSeverity.NETWORK
|
||||
classified = super().classify_error(error)
|
||||
return classified.severity if hasattr(classified, "severity") else ErrorSeverity.FATAL
|
||||
|
||||
# ── Helpers ─────────────────────────────────────────
|
||||
|
||||
def _client_getter(self, account_id: str = "default"):
|
||||
handle = self._gateway.get_handle(account_id)
|
||||
return handle.get("client") if handle else None
|
||||
|
||||
|
||||
def _strip_markdown_formatting(text: str) -> str:
|
||||
import re
|
||||
|
||||
text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
|
||||
text = re.sub(r"\*(.+?)\*", r"\1", text)
|
||||
text = re.sub(r"~~(.+?)~~", r"\1", text)
|
||||
text = re.sub(r"`([^`]+)`", r"\1", text)
|
||||
text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE)
|
||||
text = re.sub(r"^[-*+]\s+", "", text, flags=re.MULTILINE)
|
||||
text = re.sub(r"^>\s?", "", text, flags=re.MULTILINE)
|
||||
text = re.sub(r"!?\[([^\]]*)\]\([^)]+\)", r"\1", text)
|
||||
return text.strip()
|
||||
|
||||
|
||||
farcaster_plugin = FarcasterPlugin()
|
||||
ChannelPluginRegistry.register(farcaster_plugin)
|
||||
341
backend/package/yuxi/channel/extensions/farcaster/client.py
Normal file
341
backend/package/yuxi/channel/extensions/farcaster/client.py
Normal file
@ -0,0 +1,341 @@
|
||||
import asyncio
|
||||
import hashlib
|
||||
import hmac
|
||||
import secrets
|
||||
import uuid
|
||||
from functools import wraps
|
||||
|
||||
import httpx
|
||||
|
||||
from yuxi.channel.extensions.farcaster.defaults import (
|
||||
CLIENT_TIMEOUT_S,
|
||||
RETRY_MAX_ATTEMPTS,
|
||||
RETRY_BASE_DELAY_S,
|
||||
RETRY_MAX_DELAY_S,
|
||||
)
|
||||
|
||||
NEYNAR_BASE_URL = "https://api.neynar.com/v2/farcaster"
|
||||
|
||||
logger = __import__("logging").getLogger(__name__)
|
||||
|
||||
|
||||
class NeynarError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def _retry_on_network_error(max_attempts: int = RETRY_MAX_ATTEMPTS):
|
||||
def decorator(func):
|
||||
@wraps(func)
|
||||
async def wrapper(self, *args, **kwargs):
|
||||
last_exc = None
|
||||
for attempt in range(1, max_attempts + 1):
|
||||
try:
|
||||
return await func(self, *args, **kwargs)
|
||||
except httpx.HTTPStatusError as e:
|
||||
if e.response.status_code < 500:
|
||||
raise
|
||||
last_exc = e
|
||||
except (httpx.TimeoutException, httpx.ConnectError, httpx.RemoteProtocolError) as e:
|
||||
last_exc = e
|
||||
|
||||
if attempt < max_attempts:
|
||||
delay = min(RETRY_BASE_DELAY_S * (2 ** (attempt - 1)), RETRY_MAX_DELAY_S)
|
||||
logger.warning(
|
||||
"NeynarClient %s attempt %d/%d failed, retrying in %.1fs: %s",
|
||||
func.__name__,
|
||||
attempt,
|
||||
max_attempts,
|
||||
delay,
|
||||
last_exc,
|
||||
)
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
raise last_exc or NeynarError("Retry exhausted")
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
class NeynarClient:
|
||||
def __init__(
|
||||
self,
|
||||
api_key: str,
|
||||
signer_uuid: str | None = None,
|
||||
webhook_secret: str | None = None,
|
||||
timeout: float = CLIENT_TIMEOUT_S,
|
||||
):
|
||||
self._api_key = api_key
|
||||
self._signer_uuid = signer_uuid
|
||||
self._webhook_secret = webhook_secret or secrets.token_hex(32)
|
||||
self._client: httpx.AsyncClient | None = None
|
||||
self._timeout = timeout
|
||||
|
||||
async def _ensure_client(self) -> httpx.AsyncClient:
|
||||
if self._client is None:
|
||||
self._client = httpx.AsyncClient(
|
||||
base_url=NEYNAR_BASE_URL,
|
||||
headers={
|
||||
"accept": "application/json",
|
||||
"x-api-key": self._api_key,
|
||||
},
|
||||
timeout=httpx.Timeout(self._timeout),
|
||||
)
|
||||
return self._client
|
||||
|
||||
async def get_me(self) -> dict:
|
||||
client = await self._ensure_client()
|
||||
resp = await client.get("/user/me")
|
||||
resp.raise_for_status()
|
||||
return resp.json()["result"]["user"]
|
||||
|
||||
async def get_user_by_fid(self, fid: int) -> dict:
|
||||
client = await self._ensure_client()
|
||||
resp = await client.get("/user/bulk", params={"fids": str(fid)})
|
||||
resp.raise_for_status()
|
||||
users = resp.json().get("users", [])
|
||||
if not users:
|
||||
raise NeynarError(f"User not found: fid={fid}")
|
||||
return users[0]
|
||||
|
||||
async def get_user_by_fname(self, fname: str) -> dict:
|
||||
client = await self._ensure_client()
|
||||
resp = await client.get("/user/bulk", params={"fnames": fname})
|
||||
resp.raise_for_status()
|
||||
users = resp.json().get("users", [])
|
||||
if not users:
|
||||
raise NeynarError(f"User not found: fname={fname}")
|
||||
return users[0]
|
||||
|
||||
async def get_cast(self, cast_hash: str) -> dict:
|
||||
client = await self._ensure_client()
|
||||
resp = await client.get("/cast", params={"identifier": cast_hash, "type": "hash"})
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
async def get_notifications(self, fid: int, cursor: str | None = None, limit: int = 25) -> dict:
|
||||
client = await self._ensure_client()
|
||||
params = {"fid": fid, "limit": limit}
|
||||
if cursor:
|
||||
params["cursor"] = cursor
|
||||
resp = await client.get("/notifications/channel", params=params)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def post_cast(
|
||||
self,
|
||||
text: str,
|
||||
*,
|
||||
parent_hash: str | None = None,
|
||||
embeds: list[dict] | None = None,
|
||||
channel_id: str | None = None,
|
||||
) -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
|
||||
body: dict = {"signer_uuid": self._signer_uuid, "text": text, "idem": str(uuid.uuid4())}
|
||||
if parent_hash:
|
||||
body["parent"] = parent_hash
|
||||
if embeds:
|
||||
body["embeds"] = embeds
|
||||
if channel_id:
|
||||
body["channel_id"] = channel_id
|
||||
|
||||
client = await self._ensure_client()
|
||||
resp = await client.post("/cast", json=body)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
if not data.get("success"):
|
||||
raise NeynarError(f"Cast publish failed: {data}")
|
||||
return data
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def delete_cast(self, cast_hash: str) -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
|
||||
client = await self._ensure_client()
|
||||
resp = await client.delete(
|
||||
"/cast",
|
||||
json={"signer_uuid": self._signer_uuid, "target_hash": cast_hash},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def post_reaction(self, target_hash: str, reaction_type: str = "like") -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
|
||||
body = {
|
||||
"signer_uuid": self._signer_uuid,
|
||||
"type": reaction_type,
|
||||
"target": target_hash,
|
||||
"idem": str(uuid.uuid4()),
|
||||
}
|
||||
client = await self._ensure_client()
|
||||
resp = await client.post("/reaction", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def delete_reaction(self, target_hash: str, reaction_type: str = "like") -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
|
||||
body = {
|
||||
"signer_uuid": self._signer_uuid,
|
||||
"type": reaction_type,
|
||||
"target": target_hash,
|
||||
}
|
||||
client = await self._ensure_client()
|
||||
resp = await client.delete("/reaction", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
async def register_webhook(self, url: str, filters: dict) -> dict:
|
||||
body = {
|
||||
"name": "ForcePilot Farcaster Bot",
|
||||
"url": url,
|
||||
"webhook_secret": self._webhook_secret,
|
||||
"subscription": filters,
|
||||
}
|
||||
client = await self._ensure_client()
|
||||
resp = await client.post("/webhook", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
async def get_webhooks(self) -> list[dict]:
|
||||
client = await self._ensure_client()
|
||||
resp = await client.get("/webhook")
|
||||
resp.raise_for_status()
|
||||
return resp.json().get("webhooks", [])
|
||||
|
||||
async def delete_webhook(self, webhook_id: str) -> None:
|
||||
client = await self._ensure_client()
|
||||
resp = await client.delete("/webhook", params={"webhook_id": webhook_id})
|
||||
resp.raise_for_status()
|
||||
|
||||
async def validate_frame(self, message_bytes: str) -> dict:
|
||||
body = {"message_bytes_in_hex": message_bytes}
|
||||
client = await self._ensure_client()
|
||||
resp = await client.post("/frame/validate", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def follow_user(self, target_fid: int) -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
client = await self._ensure_client()
|
||||
body = {
|
||||
"signer_uuid": self._signer_uuid,
|
||||
"target_fid": target_fid,
|
||||
"idem": str(uuid.uuid4()),
|
||||
}
|
||||
resp = await client.post("/user/follow", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def unfollow_user(self, target_fid: int) -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
client = await self._ensure_client()
|
||||
resp = await client.delete(
|
||||
"/user/follow",
|
||||
json={
|
||||
"signer_uuid": self._signer_uuid,
|
||||
"target_fid": target_fid,
|
||||
},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
async def get_storage_usage(self, fid: int) -> dict:
|
||||
client = await self._ensure_client()
|
||||
resp = await client.get("/storage/usage", params={"fid": fid})
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
async def search_users(self, query: str, limit: int = 10) -> list[dict]:
|
||||
client = await self._ensure_client()
|
||||
params: dict = {"q": query}
|
||||
if limit:
|
||||
params["limit"] = limit
|
||||
resp = await client.get("/user/search", params=params)
|
||||
resp.raise_for_status()
|
||||
result = resp.json()
|
||||
return result.get("result", {}).get("users", [])
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def block_user(self, target_fid: int) -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
client = await self._ensure_client()
|
||||
body = {
|
||||
"signer_uuid": self._signer_uuid,
|
||||
"blocked_fid": target_fid,
|
||||
"idem": str(uuid.uuid4()),
|
||||
}
|
||||
resp = await client.post("/block", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def unblock_user(self, target_fid: int) -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
client = await self._ensure_client()
|
||||
body = {
|
||||
"signer_uuid": self._signer_uuid,
|
||||
"blocked_fid": target_fid,
|
||||
}
|
||||
resp = await client.delete("/block", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def mute_user(self, target_fid: int) -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
client = await self._ensure_client()
|
||||
body = {
|
||||
"signer_uuid": self._signer_uuid,
|
||||
"muted_fid": target_fid,
|
||||
"idem": str(uuid.uuid4()),
|
||||
}
|
||||
resp = await client.post("/mute", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
@_retry_on_network_error()
|
||||
async def unmute_user(self, target_fid: int) -> dict:
|
||||
if not self._signer_uuid:
|
||||
raise NeynarError("signer_uuid required for write operations")
|
||||
client = await self._ensure_client()
|
||||
body = {
|
||||
"signer_uuid": self._signer_uuid,
|
||||
"muted_fid": target_fid,
|
||||
}
|
||||
resp = await client.delete("/mute", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
def verify_webhook_signature(self, body: bytes, signature: str) -> bool:
|
||||
expected = hmac.new(
|
||||
self._webhook_secret.encode(),
|
||||
body,
|
||||
hashlib.sha256,
|
||||
).hexdigest()
|
||||
return hmac.compare_digest(expected, signature)
|
||||
|
||||
@property
|
||||
def webhook_secret(self) -> str:
|
||||
return self._webhook_secret
|
||||
|
||||
async def close(self) -> None:
|
||||
if self._client is not None:
|
||||
await self._client.aclose()
|
||||
self._client = None
|
||||
172
backend/package/yuxi/channel/extensions/farcaster/config.py
Normal file
172
backend/package/yuxi/channel/extensions/farcaster/config.py
Normal file
@ -0,0 +1,172 @@
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ENV_NEYNAR_API_KEY = "NEYNAR_API_KEY"
|
||||
ENV_NEYNAR_SIGNER_UUID = "NEYNAR_SIGNER_UUID"
|
||||
|
||||
|
||||
class FarcasterConfigAdapter:
|
||||
def __init__(self):
|
||||
self._config: dict = {}
|
||||
|
||||
def update_config(self, config: dict) -> None:
|
||||
self._config = config
|
||||
|
||||
def list_account_ids(self, config: dict) -> list[str]:
|
||||
self._config = config
|
||||
fc_cfg = self._resolve_fc_config(config)
|
||||
accounts = fc_cfg.get("accounts", {}) if isinstance(fc_cfg, dict) else {}
|
||||
if accounts:
|
||||
return list(accounts.keys())
|
||||
return ["default"]
|
||||
|
||||
async def resolve_account(self, account_id: str) -> dict:
|
||||
fc_cfg = self._get_farcaster_config()
|
||||
accounts = fc_cfg.get("accounts", {}) if isinstance(fc_cfg, dict) else {}
|
||||
raw = accounts.get(account_id, {}) if isinstance(accounts, dict) else {}
|
||||
base = {k: v for k, v in fc_cfg.items() if k != "accounts"}
|
||||
|
||||
api_key = (
|
||||
raw.get("api_key")
|
||||
or base.get("api_key", "")
|
||||
or _read_file(raw.get("api_key_file", "") or base.get("api_key_file", ""))
|
||||
or os.environ.get(ENV_NEYNAR_API_KEY, "")
|
||||
)
|
||||
signer_uuid = (
|
||||
raw.get("signer_uuid") or base.get("signer_uuid", "") or os.environ.get(ENV_NEYNAR_SIGNER_UUID, "")
|
||||
)
|
||||
|
||||
return {
|
||||
"account_id": account_id,
|
||||
"name": raw.get("name", account_id),
|
||||
"enabled": raw.get("enabled", base.get("enabled", True)),
|
||||
"configured": bool(api_key),
|
||||
"api_key": api_key,
|
||||
"signer_uuid": signer_uuid,
|
||||
"fid": raw.get("fid", base.get("fid", 0)),
|
||||
"fname": raw.get("fname", base.get("fname", "")),
|
||||
"dm_policy": raw.get("dm_policy", base.get("dm_policy", "pairing")),
|
||||
"allow_from": raw.get("allow_from", base.get("allow_from", [])),
|
||||
"webhook_base_url": raw.get("webhook_base_url", base.get("webhook_base_url", "")),
|
||||
"channel_id": raw.get("channel_id", base.get("channel_id", "")),
|
||||
}
|
||||
|
||||
def is_configured(self, account: dict) -> bool:
|
||||
return bool(account.get("api_key"))
|
||||
|
||||
def is_enabled(self, account: dict) -> bool:
|
||||
return account.get("enabled", True)
|
||||
|
||||
def disabled_reason(self, account: dict) -> str:
|
||||
if not account.get("api_key"):
|
||||
return "Neynar API Key is required"
|
||||
if not account.get("signer_uuid"):
|
||||
return "Signer UUID is required for write operations"
|
||||
return ""
|
||||
|
||||
def describe_account(self, account: dict) -> dict:
|
||||
return {
|
||||
"account_id": account.get("account_id", ""),
|
||||
"name": account.get("name", ""),
|
||||
"configured": account.get("configured", False),
|
||||
"fname": account.get("fname", ""),
|
||||
}
|
||||
|
||||
def config_schema(self) -> dict:
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"enabled": {"type": "boolean", "default": True, "title": "启用"},
|
||||
"api_key": {
|
||||
"type": "string",
|
||||
"title": "Neynar API Key",
|
||||
"description": "从 dev.neynar.com 获取的 API Key",
|
||||
"sensitive": True,
|
||||
},
|
||||
"signer_uuid": {
|
||||
"type": "string",
|
||||
"title": "Signer UUID",
|
||||
"description": "Neynar 托管签名的 UUID（写操作必需）",
|
||||
"sensitive": True,
|
||||
},
|
||||
"fid": {
|
||||
"type": "integer",
|
||||
"title": "Bot FID",
|
||||
"description": "Bot 的 Farcaster FID",
|
||||
},
|
||||
"fname": {
|
||||
"type": "string",
|
||||
"title": "Bot Username",
|
||||
"description": "Bot 的 Farcaster 用户名（用于 @提及匹配）",
|
||||
},
|
||||
"channel_id": {
|
||||
"type": "string",
|
||||
"title": "默认频道 ID",
|
||||
"description": "Farcaster Channel ID，所有公开 Cast 默认发布到此频道",
|
||||
},
|
||||
"dm_policy": {
|
||||
"type": "string",
|
||||
"enum": ["pairing", "allowlist", "open", "disabled"],
|
||||
"default": "pairing",
|
||||
"title": "DM 安全策略",
|
||||
"description": "pairing=配对码模式, allowlist=白名单, open=开放, disabled=禁用",
|
||||
},
|
||||
"allow_from": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"title": "允许列表",
|
||||
"description": 'FID 白名单（如 ["1234", "5678"])',
|
||||
},
|
||||
"webhook_base_url": {
|
||||
"type": "string",
|
||||
"title": "Webhook 回调地址",
|
||||
"description": "Neynar Webhook 推送的公网地址（如 https://forcepilot.example.com）",
|
||||
},
|
||||
"default_account": {"type": "string", "default": "default"},
|
||||
"accounts": {
|
||||
"type": "object",
|
||||
"additionalProperties": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"enabled": {"type": "boolean", "default": True},
|
||||
"api_key": {"type": "string", "sensitive": True},
|
||||
"signer_uuid": {"type": "string", "sensitive": True},
|
||||
"fid": {"type": "integer"},
|
||||
"fname": {"type": "string"},
|
||||
"channel_id": {"type": "string"},
|
||||
"dm_policy": {
|
||||
"type": "string",
|
||||
"enum": ["pairing", "allowlist", "open", "disabled"],
|
||||
},
|
||||
"allow_from": {"type": "array", "items": {"type": "string"}},
|
||||
"webhook_base_url": {"type": "string"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _resolve_fc_config(config: dict) -> dict:
|
||||
channels = config.get("channels", {})
|
||||
fc_cfg = channels.get("farcaster", {})
|
||||
return fc_cfg if isinstance(fc_cfg, dict) else {}
|
||||
|
||||
def _get_farcaster_config(self) -> dict:
|
||||
return self._resolve_fc_config(self._config)
|
||||
|
||||
|
||||
def _read_file(filepath: str) -> str:
|
||||
if not filepath:
|
||||
return ""
|
||||
try:
|
||||
path = Path(filepath)
|
||||
if path.exists():
|
||||
return path.read_text(encoding="utf-8").strip()
|
||||
except OSError:
|
||||
pass
|
||||
return ""
|
||||
30
backend/package/yuxi/channel/extensions/farcaster/dedupe.py
Normal file
30
backend/package/yuxi/channel/extensions/farcaster/dedupe.py
Normal file
@ -0,0 +1,30 @@
|
||||
import time
|
||||
from collections import OrderedDict
|
||||
|
||||
from yuxi.channel.extensions.farcaster.defaults import DEDUPE_MAX_TRACKED, DEDUPE_TTL_S
|
||||
|
||||
|
||||
class FarcasterDeduplicator:
|
||||
def __init__(self):
|
||||
self._seen: OrderedDict[str, float] = OrderedDict()
|
||||
|
||||
def is_duplicate(self, msg_id: str) -> bool:
|
||||
self._evict_expired()
|
||||
if msg_id in self._seen:
|
||||
return True
|
||||
self._seen[msg_id] = time.time()
|
||||
self._trim()
|
||||
return False
|
||||
|
||||
def reset(self) -> None:
|
||||
self._seen.clear()
|
||||
|
||||
def _evict_expired(self) -> None:
|
||||
now = time.time()
|
||||
expired = [k for k, v in self._seen.items() if now - v > DEDUPE_TTL_S]
|
||||
for key in expired:
|
||||
self._seen.pop(key, None)
|
||||
|
||||
def _trim(self) -> None:
|
||||
while len(self._seen) > DEDUPE_MAX_TRACKED:
|
||||
self._seen.popitem(last=False)
|
||||
@ -0,0 +1,18 @@
|
||||
CAST_BYTE_LIMIT = 320
|
||||
BLOCK_CHUNK_MIN = 240
|
||||
BLOCK_CHUNK_MAX = 300
|
||||
|
||||
WEBHOOK_BODY_MAX_BYTES = 128 * 1024
|
||||
WEBHOOK_BODY_TIMEOUT_S = 5.0
|
||||
|
||||
GATEWAY_HEALTH_CHECK_INTERVAL = 30
|
||||
GATEWAY_FALLBACK_THRESHOLD_S = 120
|
||||
|
||||
DEDUPE_MAX_TRACKED = 10000
|
||||
DEDUPE_TTL_S = 600
|
||||
|
||||
RETRY_MAX_ATTEMPTS = 3
|
||||
RETRY_BASE_DELAY_S = 1.0
|
||||
RETRY_MAX_DELAY_S = 10.0
|
||||
|
||||
CLIENT_TIMEOUT_S = 30.0
|
||||
@ -0,0 +1,57 @@
|
||||
import logging
|
||||
import uuid
|
||||
|
||||
import httpx
|
||||
|
||||
WARPCAST_API_BASE = "https://client.warpcast.com/v2"
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DirectCastError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class DirectCastClient:
|
||||
def __init__(self, api_key: str, signer_uuid: str, timeout: float = 30.0):
|
||||
self._api_key = api_key
|
||||
self._signer_uuid = signer_uuid
|
||||
self._client: httpx.AsyncClient | None = None
|
||||
self._timeout = timeout
|
||||
|
||||
async def _ensure_client(self) -> httpx.AsyncClient:
|
||||
if self._client is None:
|
||||
self._client = httpx.AsyncClient(
|
||||
base_url=WARPCAST_API_BASE,
|
||||
headers={
|
||||
"accept": "application/json",
|
||||
"x-api-key": self._api_key,
|
||||
},
|
||||
timeout=httpx.Timeout(self._timeout),
|
||||
)
|
||||
return self._client
|
||||
|
||||
async def send_direct_cast(
|
||||
self,
|
||||
recipient_fid: int,
|
||||
text: str,
|
||||
*,
|
||||
embeds: list[dict] | None = None,
|
||||
) -> dict:
|
||||
client = await self._ensure_client()
|
||||
body: dict = {
|
||||
"recipient_fid": recipient_fid,
|
||||
"text": text,
|
||||
"signer_uuid": self._signer_uuid,
|
||||
"idem": str(uuid.uuid4()),
|
||||
}
|
||||
if embeds:
|
||||
body["embeds"] = embeds
|
||||
resp = await client.post("/ext-send-direct-cast", json=body)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
async def close(self) -> None:
|
||||
if self._client is not None:
|
||||
await self._client.aclose()
|
||||
self._client = None
|
||||
175
backend/package/yuxi/channel/extensions/farcaster/gateway.py
Normal file
175
backend/package/yuxi/channel/extensions/farcaster/gateway.py
Normal file
@ -0,0 +1,175 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
|
||||
from yuxi.channel.extensions.farcaster.client import NeynarClient, NeynarError
|
||||
from yuxi.channel.extensions.farcaster.defaults import (
|
||||
GATEWAY_FALLBACK_THRESHOLD_S,
|
||||
GATEWAY_HEALTH_CHECK_INTERVAL,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FarcasterGatewayError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class FarcasterGateway:
|
||||
def __init__(self):
|
||||
self._handles: dict[str, dict] = {}
|
||||
|
||||
async def start(self, ctx) -> dict:
|
||||
account_id = getattr(ctx, "account_id", "default")
|
||||
config = getattr(ctx, "config", {}) or {}
|
||||
channels = config.get("channels", {})
|
||||
fc_cfg = channels.get("farcaster", {}) if isinstance(channels, dict) else {}
|
||||
|
||||
api_key = fc_cfg.get("api_key", "")
|
||||
signer_uuid = fc_cfg.get("signer_uuid", "")
|
||||
webhook_base_url = fc_cfg.get("webhook_base_url", "")
|
||||
|
||||
if not api_key:
|
||||
raise FarcasterGatewayError("api_key not configured")
|
||||
|
||||
client = NeynarClient(api_key=api_key, signer_uuid=signer_uuid)
|
||||
|
||||
try:
|
||||
me = await client.get_me()
|
||||
except NeynarError as e:
|
||||
await client.close()
|
||||
raise FarcasterGatewayError(f"Failed to verify API key: {e}") from e
|
||||
|
||||
bot_fid = me.get("fid", 0)
|
||||
bot_fname = me.get("username", "")
|
||||
|
||||
logger.info(
|
||||
"Farcaster bot verified: fid=%s fname=%s display_name=%s",
|
||||
bot_fid,
|
||||
bot_fname,
|
||||
me.get("display_name", ""),
|
||||
)
|
||||
|
||||
handle = {
|
||||
"client": client,
|
||||
"webhook_id": None,
|
||||
"bot_fid": bot_fid,
|
||||
"bot_fname": bot_fname,
|
||||
"fallback_task": None,
|
||||
"last_webhook_event_at": time.time(),
|
||||
"running": True,
|
||||
}
|
||||
|
||||
if webhook_base_url and bot_fid:
|
||||
webhook_url = f"{webhook_base_url.rstrip('/')}/farcaster/webhook?account_id={account_id}"
|
||||
filters = {
|
||||
"cast.created": {
|
||||
"author_fids": [bot_fid],
|
||||
"mentioned_fids": [bot_fid],
|
||||
},
|
||||
"reaction.created": {
|
||||
"author_fids": [bot_fid],
|
||||
},
|
||||
}
|
||||
try:
|
||||
result = await client.register_webhook(webhook_url, filters)
|
||||
webhook_obj = result.get("webhook", {})
|
||||
handle["webhook_id"] = webhook_obj.get("object") if isinstance(webhook_obj, dict) else None
|
||||
logger.info("Neynar webhook registered: id=%s", handle["webhook_id"])
|
||||
except NeynarError as e:
|
||||
logger.warning("Webhook registration failed, will rely on polling fallback: %s", e)
|
||||
|
||||
handle["fallback_task"] = asyncio.create_task(
|
||||
self._health_check_loop(account_id, handle, ctx),
|
||||
name=f"farcaster-health-{account_id}",
|
||||
)
|
||||
|
||||
self._handles[account_id] = handle
|
||||
return handle
|
||||
|
||||
async def stop(self, ctx) -> None:
|
||||
account_id = getattr(ctx, "account_id", "default")
|
||||
handle = self._handles.pop(account_id, None)
|
||||
if handle is None:
|
||||
return
|
||||
|
||||
handle["running"] = False
|
||||
|
||||
fallback_task = handle.get("fallback_task")
|
||||
if fallback_task:
|
||||
fallback_task.cancel()
|
||||
try:
|
||||
await fallback_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
|
||||
webhook_id = handle.get("webhook_id")
|
||||
client = handle.get("client")
|
||||
if webhook_id and client:
|
||||
try:
|
||||
await client.delete_webhook(webhook_id)
|
||||
logger.info("Neynar webhook deleted: id=%s", webhook_id)
|
||||
except NeynarError as e:
|
||||
logger.warning("Failed to delete webhook: %s", e)
|
||||
|
||||
if client:
|
||||
await client.close()
|
||||
logger.info("Farcaster gateway stopped for account %s", account_id)
|
||||
|
||||
def get_handle(self, account_id: str = "default") -> dict | None:
|
||||
return self._handles.get(account_id)
|
||||
|
||||
@property
|
||||
def account_count(self) -> int:
|
||||
return len(self._handles)
|
||||
|
||||
async def _health_check_loop(self, account_id: str, handle: dict, ctx) -> None:
|
||||
STORAGE_WARN_THRESHOLD = 0.9
|
||||
while handle.get("running", False):
|
||||
await asyncio.sleep(GATEWAY_HEALTH_CHECK_INTERVAL)
|
||||
elapsed = time.time() - handle.get("last_webhook_event_at", 0)
|
||||
if elapsed > GATEWAY_FALLBACK_THRESHOLD_S:
|
||||
logger.warning(
|
||||
"Farcaster webhook unhealthy for %s (%ss since last event), falling back to polling",
|
||||
account_id,
|
||||
int(elapsed),
|
||||
)
|
||||
await self._poll_notifications(handle, ctx)
|
||||
|
||||
client = handle.get("client")
|
||||
bot_fid = handle.get("bot_fid")
|
||||
if client and bot_fid:
|
||||
try:
|
||||
storage = await client.get_storage_usage(bot_fid)
|
||||
used = storage.get("used", 0)
|
||||
limit = storage.get("limit", 0)
|
||||
if limit > 0 and used / limit > STORAGE_WARN_THRESHOLD:
|
||||
logger.warning(
|
||||
"Farcaster storage nearly exhausted: fid=%s used=%s limit=%s ratio=%.1f%%",
|
||||
bot_fid,
|
||||
used,
|
||||
limit,
|
||||
used / limit * 100,
|
||||
)
|
||||
except NeynarError:
|
||||
pass
|
||||
|
||||
async def _poll_notifications(self, handle: dict, ctx) -> None:
|
||||
client = handle.get("client")
|
||||
bot_fid = handle.get("bot_fid")
|
||||
if not client or not bot_fid:
|
||||
return
|
||||
|
||||
try:
|
||||
result = await client.get_notifications(bot_fid, limit=25)
|
||||
notifications = result.get("notifications", [])
|
||||
for notif in notifications:
|
||||
event_type = notif.get("type", "")
|
||||
if event_type in ("cast-reply", "mentions"):
|
||||
cast = notif.get("cast", {})
|
||||
if cast:
|
||||
queue = getattr(ctx, "queue", None)
|
||||
if queue is not None:
|
||||
await queue.put({"type": "cast.created", "data": cast})
|
||||
except NeynarError as e:
|
||||
logger.error("Polling fallback failed: %s", e)
|
||||
132
backend/package/yuxi/channel/extensions/farcaster/monitor.py
Normal file
132
backend/package/yuxi/channel/extensions/farcaster/monitor.py
Normal file
@ -0,0 +1,132 @@
|
||||
import logging
|
||||
import re
|
||||
from datetime import datetime
|
||||
|
||||
from yuxi.channel.message.models import MessageType, PeerInfo, UnifiedMessage
|
||||
from yuxi.channel.routing.models import PeerKind
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FarcasterMonitor:
|
||||
def __init__(self, bot_fid: int = 0, bot_fname: str = ""):
|
||||
self._bot_fid = bot_fid
|
||||
self._bot_fname = bot_fname
|
||||
|
||||
def convert_cast_to_unified(
|
||||
self,
|
||||
cast: dict,
|
||||
*,
|
||||
account_id: str = "default",
|
||||
) -> UnifiedMessage | None:
|
||||
text = cast.get("text", "")
|
||||
author = cast.get("author", {})
|
||||
author_fid = author.get("fid")
|
||||
|
||||
if author_fid is None:
|
||||
logger.warning("Farcaster cast missing author.fid, skipping")
|
||||
return None
|
||||
|
||||
if author_fid == self._bot_fid:
|
||||
return None
|
||||
|
||||
clean_text = _strip_bot_mentions(text, self._bot_fname, self._bot_fid) if self._bot_fname else text
|
||||
|
||||
display_name = author.get("display_name") or author.get("username") or f"fid:{author_fid}"
|
||||
|
||||
timestamp = None
|
||||
ts_str = cast.get("timestamp", "")
|
||||
if ts_str:
|
||||
try:
|
||||
timestamp = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
embeds = cast.get("embeds", [])
|
||||
media_urls = []
|
||||
for embed in embeds:
|
||||
if isinstance(embed, dict) and "url" in embed:
|
||||
media_urls.append(embed["url"])
|
||||
|
||||
return UnifiedMessage(
|
||||
msg_id=_make_msg_id("cast", cast.get("hash", "")),
|
||||
channel_type="farcaster",
|
||||
account_id=account_id,
|
||||
content=clean_text or text,
|
||||
sender=PeerInfo(
|
||||
kind=PeerKind.DIRECT,
|
||||
id=f"farcaster:{author_fid}",
|
||||
display_name=display_name,
|
||||
),
|
||||
message_type=MessageType.TEXT,
|
||||
media_urls=media_urls if media_urls else None,
|
||||
timestamp=timestamp,
|
||||
reply_to_id=cast.get("parent_hash"),
|
||||
raw_payload=cast,
|
||||
metadata={
|
||||
"cast_hash": cast.get("hash", ""),
|
||||
"thread_hash": cast.get("thread_hash"),
|
||||
"author_fid": author_fid,
|
||||
"author_fname": author.get("username", ""),
|
||||
"pfp_url": author.get("pfp_url"),
|
||||
"channel": cast.get("channel"),
|
||||
},
|
||||
)
|
||||
|
||||
def convert_reaction_to_unified(
|
||||
self,
|
||||
reaction: dict,
|
||||
*,
|
||||
account_id: str = "default",
|
||||
) -> UnifiedMessage | None:
|
||||
reaction_type = reaction.get("reaction_type", "")
|
||||
user = reaction.get("user", {})
|
||||
cast = reaction.get("cast", {})
|
||||
user_fid = user.get("fid")
|
||||
|
||||
if user_fid is None:
|
||||
return None
|
||||
|
||||
if user_fid == self._bot_fid:
|
||||
return None
|
||||
|
||||
display_name = user.get("display_name") or user.get("username") or f"fid:{user_fid}"
|
||||
cast_hash = cast.get("hash", "")
|
||||
|
||||
return UnifiedMessage(
|
||||
msg_id=_make_msg_id("reaction", f"{cast_hash}:{user_fid}:{reaction_type}"),
|
||||
channel_type="farcaster",
|
||||
account_id=account_id,
|
||||
content=f"[{reaction_type}] cast/{cast_hash}",
|
||||
sender=PeerInfo(
|
||||
kind=PeerKind.DIRECT,
|
||||
id=f"farcaster:{user_fid}",
|
||||
display_name=display_name,
|
||||
),
|
||||
message_type=MessageType.EVENT,
|
||||
raw_payload=reaction,
|
||||
timestamp=datetime.now(datetime.UTC),
|
||||
metadata={
|
||||
"reaction_type": reaction_type,
|
||||
"cast_hash": cast_hash,
|
||||
"cast_text": cast.get("text", ""),
|
||||
"author_fid": user_fid,
|
||||
"author_fname": user.get("username", ""),
|
||||
},
|
||||
)
|
||||
|
||||
def extract_mentions(self, text: str) -> list[str]:
|
||||
return re.findall(r"@(\w+)", text)
|
||||
|
||||
def strip_mentions(self, text: str) -> str:
|
||||
return re.sub(r"@\w+", "", text).strip()
|
||||
|
||||
|
||||
def _strip_bot_mentions(text: str, bot_fname: str, bot_fid: int) -> str:
|
||||
text = re.sub(rf"@?{re.escape(bot_fname)}\b", "", text)
|
||||
text = re.sub(rf"@fid:{bot_fid}", "", text)
|
||||
return text.strip()
|
||||
|
||||
|
||||
def _make_msg_id(prefix: str, hash_val: str) -> str:
|
||||
return f"{prefix}:{hash_val}"
|
||||
105
backend/package/yuxi/channel/extensions/farcaster/outbound.py
Normal file
105
backend/package/yuxi/channel/extensions/farcaster/outbound.py
Normal file
@ -0,0 +1,105 @@
|
||||
import logging
|
||||
|
||||
from yuxi.channel.extensions.farcaster.client import NeynarClient, NeynarError
|
||||
from yuxi.channel.extensions.farcaster.defaults import CAST_BYTE_LIMIT
|
||||
from yuxi.channel.protocols import OutboundDeliveryMode
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FarcasterOutboundAdapter:
|
||||
delivery_mode = OutboundDeliveryMode.DIRECT
|
||||
text_chunk_limit = CAST_BYTE_LIMIT
|
||||
|
||||
def __init__(self, client_getter, account_id: str = "default"):
|
||||
self._client_getter = client_getter
|
||||
self._account_id = account_id
|
||||
|
||||
def _get_client(self) -> NeynarClient:
|
||||
client = self._client_getter(self._account_id)
|
||||
if client is None:
|
||||
raise NeynarError("No active Farcaster client")
|
||||
return client
|
||||
|
||||
async def send_text(
|
||||
self,
|
||||
target_id: str,
|
||||
content: str,
|
||||
*,
|
||||
reply_to_id: str | None = None,
|
||||
thread_id: str | None = None,
|
||||
account_id: str | None = None,
|
||||
channel_id: str | None = None,
|
||||
) -> None:
|
||||
text = content
|
||||
if len(text.encode("utf-8")) > CAST_BYTE_LIMIT:
|
||||
text = _trim_to_byte_limit(text, CAST_BYTE_LIMIT)
|
||||
|
||||
parent_hash = reply_to_id or thread_id
|
||||
|
||||
client = self._get_client()
|
||||
try:
|
||||
result = await client.post_cast(
|
||||
text=text,
|
||||
parent_hash=parent_hash,
|
||||
channel_id=channel_id,
|
||||
)
|
||||
cast_hash = result.get("cast", {}).get("hash", "")
|
||||
logger.info(
|
||||
"Farcaster cast sent: hash=%s to=%s parent=%s channel=%s",
|
||||
cast_hash,
|
||||
target_id,
|
||||
parent_hash,
|
||||
channel_id,
|
||||
)
|
||||
except NeynarError as e:
|
||||
logger.error("Farcaster send_text failed: %s", e)
|
||||
|
||||
async def send_media(
|
||||
self,
|
||||
target_id: str,
|
||||
media_url: str,
|
||||
media_type: str,
|
||||
reply_to_id: str | None = None,
|
||||
thread_id: str | None = None,
|
||||
account_id: str | None = None,
|
||||
channel_id: str | None = None,
|
||||
) -> None:
|
||||
captions = {"image": "🖼️ Image", "video": "🎬 Video", "audio": "🎵 Audio", "file": "📎 File"}
|
||||
prefix = captions.get(media_type, media_url)
|
||||
|
||||
text = prefix
|
||||
if len(text.encode("utf-8")) > CAST_BYTE_LIMIT:
|
||||
text = _trim_to_byte_limit(text, CAST_BYTE_LIMIT)
|
||||
|
||||
parent_hash = reply_to_id or thread_id
|
||||
|
||||
client = self._get_client()
|
||||
try:
|
||||
await client.post_cast(
|
||||
text=text,
|
||||
parent_hash=parent_hash,
|
||||
embeds=[{"url": media_url}],
|
||||
channel_id=channel_id,
|
||||
)
|
||||
except NeynarError as e:
|
||||
logger.error("Farcaster send_media failed: type=%s url=%s error=%s", media_type, media_url, e)
|
||||
|
||||
async def send_reaction(
|
||||
self,
|
||||
target_hash: str,
|
||||
reaction_type: str = "like",
|
||||
) -> None:
|
||||
client = self._get_client()
|
||||
try:
|
||||
await client.post_reaction(target_hash, reaction_type)
|
||||
except NeynarError as e:
|
||||
logger.error("Farcaster send_reaction failed: %s", e)
|
||||
|
||||
|
||||
def _trim_to_byte_limit(text: str, limit: int) -> str:
|
||||
encoded = text.encode("utf-8")
|
||||
if len(encoded) <= limit:
|
||||
return text
|
||||
trimmed = encoded[: limit - 3]
|
||||
return trimmed.decode("utf-8", errors="ignore") + "..."
|
||||
@ -0,0 +1,28 @@
|
||||
{
|
||||
"id": "farcaster",
|
||||
"name": "Farcaster",
|
||||
"version": "1.0.0",
|
||||
"description": "Farcaster decentralized social protocol channel plugin for ForcePilot",
|
||||
"author": "ForcePilot",
|
||||
"order": 105,
|
||||
"enabled": true,
|
||||
"capabilities": {
|
||||
"chat_types": ["direct"],
|
||||
"message_types": ["text"],
|
||||
"interactions": ["recast"],
|
||||
"reactions": true,
|
||||
"typing_indicator": false,
|
||||
"threads": true,
|
||||
"edit": false,
|
||||
"unsend": true,
|
||||
"reply": true,
|
||||
"media": true,
|
||||
"native_commands": false,
|
||||
"polls": false,
|
||||
"streaming": true,
|
||||
"streaming_mode": "block",
|
||||
"block_streaming": true,
|
||||
"block_streaming_chunk_min_chars": 240,
|
||||
"block_streaming_chunk_max_chars": 300
|
||||
}
|
||||
}
|
||||
99
backend/package/yuxi/channel/extensions/farcaster/types.py
Normal file
99
backend/package/yuxi/channel/extensions/farcaster/types.py
Normal file
@ -0,0 +1,99 @@
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class WebhookEventType:
|
||||
CAST_CREATED = "cast.created"
|
||||
REACTION_CREATED = "reaction.created"
|
||||
|
||||
|
||||
class CastAuthor(BaseModel):
|
||||
fid: int
|
||||
username: str | None = None
|
||||
display_name: str | None = None
|
||||
pfp_url: str | None = None
|
||||
follower_count: int = 0
|
||||
following_count: int = 0
|
||||
|
||||
|
||||
class CastEmbed(BaseModel):
|
||||
url: str | None = None
|
||||
|
||||
|
||||
class CastFrame(BaseModel):
|
||||
url: str | None = None
|
||||
|
||||
|
||||
class CastReactionSummary(BaseModel):
|
||||
likes_count: int = 0
|
||||
recasts_count: int = 0
|
||||
|
||||
|
||||
class CastData(BaseModel):
|
||||
hash: str
|
||||
thread_hash: str | None = None
|
||||
parent_hash: str | None = None
|
||||
parent_url: str | None = None
|
||||
author: CastAuthor
|
||||
text: str = ""
|
||||
timestamp: str | None = None
|
||||
embeds: list[CastEmbed] = Field(default_factory=list)
|
||||
reactions: CastReactionSummary | None = None
|
||||
replies: dict | None = None
|
||||
mentioned_profiles: list[dict] = Field(default_factory=list)
|
||||
channel: dict | None = None
|
||||
|
||||
|
||||
class CastCreatedData(BaseModel):
|
||||
object: str = "cast"
|
||||
hash: str
|
||||
thread_hash: str | None = None
|
||||
parent_hash: str | None = None
|
||||
author: CastAuthor
|
||||
text: str = ""
|
||||
timestamp: str | None = None
|
||||
embeds: list[CastEmbed] = Field(default_factory=list)
|
||||
reactions: CastReactionSummary | None = None
|
||||
replies: dict | None = None
|
||||
mentioned_profiles: list[dict] = Field(default_factory=list)
|
||||
channel: dict | None = None
|
||||
|
||||
|
||||
class ReactionUser(BaseModel):
|
||||
fid: int
|
||||
username: str | None = None
|
||||
display_name: str | None = None
|
||||
pfp_url: str | None = None
|
||||
|
||||
|
||||
class ReactionCast(BaseModel):
|
||||
hash: str
|
||||
author: CastAuthor | None = None
|
||||
text: str | None = None
|
||||
|
||||
|
||||
class ReactionCreatedData(BaseModel):
|
||||
object: str = "reaction"
|
||||
reaction_type: str = "like"
|
||||
cast: ReactionCast
|
||||
user: ReactionUser
|
||||
|
||||
|
||||
class WebhookEvent(BaseModel):
|
||||
created_at: int = 0
|
||||
type: str
|
||||
data: dict
|
||||
model_config = {"extra": "allow"}
|
||||
|
||||
|
||||
class ResolvedFarcasterAccount(BaseModel):
|
||||
account_id: str
|
||||
name: str = ""
|
||||
enabled: bool = True
|
||||
configured: bool = False
|
||||
api_key: str = ""
|
||||
signer_uuid: str = ""
|
||||
fid: int = 0
|
||||
fname: str = ""
|
||||
dm_policy: str = "pairing"
|
||||
allow_from: list[str] = Field(default_factory=list)
|
||||
webhook_base_url: str = ""
|
||||
155
backend/package/yuxi/channel/extensions/farcaster/webhook.py
Normal file
155
backend/package/yuxi/channel/extensions/farcaster/webhook.py
Normal file
@ -0,0 +1,155 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from urllib.parse import parse_qs
|
||||
|
||||
from fastapi import APIRouter, Header, Request
|
||||
from fastapi.responses import PlainTextResponse
|
||||
|
||||
from yuxi.channel.extensions.farcaster.defaults import (
|
||||
WEBHOOK_BODY_MAX_BYTES,
|
||||
WEBHOOK_BODY_TIMEOUT_S,
|
||||
)
|
||||
from yuxi.channel.extensions.farcaster.monitor import FarcasterMonitor
|
||||
|
||||
router = APIRouter(prefix="/farcaster/webhook", tags=["farcaster"])
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_GATEWAY_STATE: dict = {}
|
||||
|
||||
|
||||
def register_gateway_state(account_id: str, ctx, gateway, dedupe) -> None:
|
||||
handle = gateway.get_handle(account_id)
|
||||
bot_fid = handle.get("bot_fid", 0) if handle else 0
|
||||
bot_fname = handle.get("bot_fname", "") if handle else ""
|
||||
|
||||
monitor = FarcasterMonitor(bot_fid=bot_fid, bot_fname=bot_fname)
|
||||
_GATEWAY_STATE[account_id] = {
|
||||
"ctx": ctx,
|
||||
"gateway": gateway,
|
||||
"monitor": monitor,
|
||||
"dedupe": dedupe,
|
||||
}
|
||||
|
||||
|
||||
def unregister_gateway_state(account_id: str) -> None:
|
||||
_GATEWAY_STATE.pop(account_id, None)
|
||||
|
||||
|
||||
async def _dispatch_webhook_event(request: Request, account_id: str = "default") -> PlainTextResponse:
|
||||
body = await _read_body(request)
|
||||
if body is None:
|
||||
return PlainTextResponse("", status_code=413)
|
||||
|
||||
x_neynar_signature = request.headers.get("X-Neynar-Signature", "")
|
||||
|
||||
state = _GATEWAY_STATE.get(account_id)
|
||||
if state is None:
|
||||
logger.warning("Farcaster webhook: no active gateway state for account=%s", account_id)
|
||||
return PlainTextResponse("", status_code=503)
|
||||
|
||||
gateway = state["gateway"]
|
||||
handle = gateway.get_handle(account_id)
|
||||
if handle is None or not handle.get("running", False):
|
||||
return PlainTextResponse("", status_code=503)
|
||||
|
||||
client = handle.get("client")
|
||||
if x_neynar_signature and client:
|
||||
if not client.verify_webhook_signature(body, x_neynar_signature):
|
||||
logger.warning("Farcaster webhook: invalid signature")
|
||||
return PlainTextResponse("", status_code=401)
|
||||
|
||||
handle["last_webhook_event_at"] = time.time()
|
||||
|
||||
try:
|
||||
payload = json.loads(body)
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("Farcaster webhook: invalid JSON body")
|
||||
return PlainTextResponse("", status_code=400)
|
||||
|
||||
event_type = payload.get("type", "")
|
||||
event_data = payload.get("data", {})
|
||||
|
||||
ctx = state["ctx"]
|
||||
monitor = state["monitor"]
|
||||
dedupe = state["dedupe"]
|
||||
|
||||
if event_type == "cast.created":
|
||||
await _handle_cast_event(ctx, event_data, handle, monitor, dedupe, account_id)
|
||||
elif event_type == "reaction.created":
|
||||
unified = monitor.convert_reaction_to_unified(event_data, account_id=account_id)
|
||||
if unified is not None and not dedupe.is_duplicate(unified.msg_id):
|
||||
queue = getattr(ctx, "queue", None)
|
||||
if queue is not None:
|
||||
try:
|
||||
queue.put_nowait(unified)
|
||||
except asyncio.QueueFull:
|
||||
logger.warning("Farcaster queue full, dropping reaction: %s", unified.msg_id)
|
||||
else:
|
||||
logger.debug("Farcaster unhandled event type: %s", event_type)
|
||||
|
||||
return PlainTextResponse("ok")
|
||||
|
||||
|
||||
@router.post("")
|
||||
async def farcaster_webhook(
|
||||
request: Request,
|
||||
x_neynar_signature: str = Header("", alias="X-Neynar-Signature"),
|
||||
):
|
||||
account_id = "default"
|
||||
parsed = parse_qs(str(request.url.query))
|
||||
if "account_id" in parsed:
|
||||
account_id = parsed["account_id"][0]
|
||||
|
||||
return await _dispatch_webhook_event(request, account_id)
|
||||
|
||||
|
||||
async def _handle_cast_event(
|
||||
ctx, cast_data: dict, handle: dict, monitor: FarcasterMonitor, dedupe, account_id: str = "default"
|
||||
) -> None:
|
||||
text = cast_data.get("text", "")
|
||||
author = cast_data.get("author", {})
|
||||
bot_fname = handle.get("bot_fname", "")
|
||||
bot_fid = handle.get("bot_fid", 0)
|
||||
|
||||
if not _check_mention(text, bot_fname, bot_fid):
|
||||
logger.debug(
|
||||
"Farcaster cast: not mentioned, fid=%s text=%.100s",
|
||||
author.get("fid", "?"),
|
||||
text,
|
||||
)
|
||||
return
|
||||
|
||||
unified = monitor.convert_cast_to_unified(cast_data, account_id=account_id)
|
||||
if unified is None:
|
||||
return
|
||||
|
||||
if dedupe.is_duplicate(unified.msg_id):
|
||||
return
|
||||
|
||||
queue = getattr(ctx, "queue", None)
|
||||
if queue is not None:
|
||||
try:
|
||||
queue.put_nowait(unified)
|
||||
except asyncio.QueueFull:
|
||||
logger.warning("Farcaster queue full, dropping message: %s", unified.msg_id)
|
||||
|
||||
|
||||
def _check_mention(text: str, bot_fname: str, bot_fid: int) -> bool:
|
||||
if not bot_fname:
|
||||
return False
|
||||
return f"@{bot_fname}" in text or f"@fid:{bot_fid}" in text
|
||||
|
||||
|
||||
async def _read_body(request: Request) -> bytes | None:
|
||||
try:
|
||||
body = await asyncio.wait_for(request.body(), timeout=WEBHOOK_BODY_TIMEOUT_S)
|
||||
except TimeoutError:
|
||||
logger.warning("Farcaster webhook: body read timeout")
|
||||
return None
|
||||
if len(body) > WEBHOOK_BODY_MAX_BYTES:
|
||||
logger.warning("Farcaster webhook: body too large (%d bytes)", len(body))
|
||||
return None
|
||||
return body
|
||||
Loading…
Reference in New Issue
Block a user