#!/usr/bin/env python3
"""Nostr Notifier - Watch a single npub for mentions, reposts, reactions."""
import asyncio
import json
import logging
import os
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Set
from collections import defaultdict
import hashlib
from nostr_sdk import (
Client,
Filter,
Kind,
KindStandard,
Nip19,
PublicKey,
RelayUrl,
Events
)
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('nostr_notifier.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class NostrNotifier:
"""Monitor a single npub for activity using nostr-sdk."""
def __init__(self, target_npub: str, relays: List[str], telegram_bot_token: str, telegram_chat_id: str):
self.target_npub = target_npub
self.target_pubkey = self._npub_to_pubkey(target_npub)
self.relays = relays
self.telegram_bot_token = telegram_bot_token
self.telegram_chat_id = telegram_chat_id
# Basic deduplication
self.seen_events: set = set()
self.seen_events_max = 10000 # Limit memory usage
self.profile_cache: Dict[str, str] = {}
self.profile_cache_max = 500 # Limit profile cache
# Smart filters
self.thread_cache: Dict[str, str] = {} # root_event_id -> first_notified_event_id
self.thread_cache_max = 5000
self.user_engagement: Dict[str, int] = defaultdict(int) # pubkey -> interaction count
self.user_engagement_max = 1000
self.event_engagement: Dict[str, Dict[str, int]] = defaultdict(lambda: {"reactions": 0, "reposts": 0})
self.notification_timestamps: List[float] = []
self.min_engagement_threshold = 0 # 0 = notify all
self.rate_limit_per_minute = 10 # Max notifications per minute
self.trusted_users: Set[str] = set() # npubs to always notify for
self.blocked_users: Set[str] = set() # npubs to never notify for
self.client = Client()
logger.info(f"Initialized notifier for {target_npub}")
logger.info(f"Target pubkey (hex): {self.target_pubkey.to_hex()}")
logger.info(f"Configured {len(relays)} relays")
logger.info("Smart filters enabled: thread dedup, rate limiting, engagement tracking")
def _npub_to_pubkey(self, npub: str) -> PublicKey:
"""Convert npub to PublicKey object."""
try:
nip19 = Nip19.from_bech32(npub)
result = nip19.as_enum()
if hasattr(result, 'pubkey'):
return result.pubkey
return PublicKey.parse(npub)
except Exception as e:
logger.error(f"Failed to decode npub {npub}: {e}")
return PublicKey.parse("0" * 64)
async def _fetch_profile(self, pubkey: PublicKey) -> Optional[Dict]:
"""Fetch user profile metadata from relays."""
try:
filter_obj = Filter().authors([pubkey]).kinds([Kind.from_std(KindStandard.METADATA)])
events = await self.client.fetch_events(filter_obj, timedelta(seconds=5))
if not events.is_empty():
event = events.first()
try:
metadata = json.loads(event.content())
return metadata
except json.JSONDecodeError:
pass
except Exception as e:
logger.debug(f"Failed to fetch profile for {pubkey.to_hex()[:12]}: {e}")
return None
async def _get_author_identifier(self, author_pubkey: PublicKey) -> str:
"""Get author's identifier - prefer nip-05, fallback to display name, then truncated pubkey."""
pubkey_hex = author_pubkey.to_hex()
# Check cache first
if pubkey_hex in self.profile_cache:
return self.profile_cache[pubkey_hex]
# Ensure we're connected before fetching
if not self.client:
logger.debug(f"Client not connected, using truncated pubkey for {pubkey_hex[:12]}")
identifier = f"@{pubkey_hex[:12]}..."
self.profile_cache[pubkey_hex] = identifier
return identifier
# Fetch profile to get nip-05 or display name
profile = await self._fetch_profile(author_pubkey)
if profile:
# Priority: nip-05 > display_name > name > truncated pubkey
if 'nip05' in profile and profile['nip05']:
identifier = f"@{profile['nip05']}"
logger.debug(f"Found nip-05: {identifier}")
elif 'display_name' in profile and profile['display_name']:
identifier = f"@{profile['display_name'].replace(' ', '_')}"
logger.debug(f"Found display_name: {identifier}")
elif 'name' in profile and profile['name']:
identifier = f"@{profile['name'].replace(' ', '_')}"
logger.debug(f"Found name: {identifier}")
else:
identifier = f"@{pubkey_hex[:12]}..."
else:
identifier = f"@{pubkey_hex[:12]}..."
# Cache the result
self.profile_cache[pubkey_hex] = identifier
return identifier
def _get_filters(self) -> List[Filter]:
"""Create Nostr filters for events we care about."""
return [
# Mentions: events with 'p' tag pointing to our pubkey
Filter().kinds([Kind.from_std(KindStandard.TEXT_NOTE)]).p_tags([self.target_pubkey]),
# Direct posts by target (for context)
Filter().kinds([Kind.from_std(KindStandard.TEXT_NOTE)]).authors([self.target_pubkey]),
# Reactions to target's posts
Filter().kinds([Kind.from_std(KindStandard.REACTION)]),
# Reposts of target's posts
Filter().kinds([Kind.from_std(KindStandard.REPOST)]),
]
def _check_rate_limit(self) -> bool:
"""Check if we're within rate limits. Returns True if OK to notify."""
now = datetime.now().timestamp()
# Remove timestamps older than 1 minute
self.notification_timestamps = [ts for ts in self.notification_timestamps if now - ts < 60]
if len(self.notification_timestamps) >= self.rate_limit_per_minute:
logger.debug(f"Rate limit hit: {len(self.notification_timestamps)}/{self.rate_limit_per_minute} per minute")
return False
self.notification_timestamps.append(now)
return True
def _get_root_event_id(self, event) -> Optional[str]:
"""Find the root event ID for thread tracking."""
try:
# Look for 'e' tags with 'root' or 'reply' marker
for tag in event.tags():
if str(tag.kind()) == 'e':
# Check if this is the root (look for replacement event first)
event_id = tag.event_id().to_hex()
# Simple heuristic: first 'e' tag is usually the root or parent
return event_id
except Exception as e:
logger.debug(f"Error finding root event: {e}")
return event.id().to_hex() # Fallback to self
def _should_notify_thread(self, event) -> bool:
"""Check if we should notify for this thread (avoid spam)."""
root_id = self._get_root_event_id(event)
event_id = event.id().to_hex()
# Clean up cache if too large
if len(self.thread_cache) > self.thread_cache_max:
keys = list(self.thread_cache.keys())[:self.thread_cache_max // 5]
for k in keys:
self.thread_cache.pop(k, None)
# If we've already notified for this thread, skip
if root_id in self.thread_cache:
logger.debug(f"Thread already notified: {root_id[:12]}")
return False
# Cache this thread
self.thread_cache[root_id] = event_id
return True
def _should_notify_user(self, author_pubkey: PublicKey) -> bool:
"""Check user-based filters (trusted/blocked lists)."""
pubkey_hex = author_pubkey.to_hex()
# Check blocked users
if pubkey_hex in self.blocked_users:
logger.debug(f"Blocked user: {pubkey_hex[:12]}")
return False
# Trusted users always pass
if pubkey_hex in self.trusted_users:
return True
return True
def _should_notify_engagement(self, event_type: str, event_id: str) -> bool:
"""Check engagement-based filters."""
if self.min_engagement_threshold == 0:
return True # No threshold set
# Track engagement
if event_type == "reaction":
self.event_engagement[event_id]["reactions"] += 1
elif event_type == "repost":
self.event_engagement[event_id]["reposts"] += 1
total_engagement = self.event_engagement[event_id]["reactions"] + self.event_engagement[event_id]["reposts"]
if total_engagement >= self.min_engagement_threshold:
return True
logger.debug(f"Engagement threshold not met: {total_engagement}/{self.min_engagement_threshold}")
return False
def _apply_smart_filters(self, event_type: str, author_pubkey: PublicKey, event) -> bool:
"""Apply all smart filters. Returns True if should notify."""
event_id = event.id().to_hex()
# 1. Rate limiting
if not self._check_rate_limit():
return False
# 2. User-based filtering
if not self._should_notify_user(author_pubkey):
return False
# 3. Thread deduplication (for mentions and replies)
if event_type in ["mention", "reply"]:
if not self._should_notify_thread(event):
return False
# 4. Engagement threshold
if not self._should_notify_engagement(event_type, event_id):
return False
# Track user engagement
self.user_engagement[author_pubkey.to_hex()] += 1
return True
async def _send_telegram(self, message: str) -> bool:
"""Send message to Telegram."""
if not self.telegram_bot_token:
logger.info(f"📮 [Telegram] {message[:100]}...")
return True
import aiohttp
url = f"https://api.telegram.org/bot{self.telegram_bot_token}/sendMessage"
payload = {
"chat_id": self.telegram_chat_id,
"text": message,
"parse_mode": "Markdown"
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as response:
if response.status == 200:
logger.info("✅ Telegram message sent")
return True
else:
logger.error(f"Telegram error: {response.status}")
return False
except Exception as e:
logger.error(f"Failed to send Telegram message: {e}")
return False
def _format_notification(self, event_type: str, event, author_identifier: str) -> str:
"""Format event as notification message."""
created_at = event.created_at()
# Handle Timestamp object - convert to milliseconds then to seconds
if hasattr(created_at, 'as_u64'):
timestamp = datetime.fromtimestamp(created_at.as_u64() // 1000).strftime("%H:%M")
else:
timestamp = datetime.now().strftime("%H:%M")
event_id = event.id().to_hex()
content = event.content()
if event_type == "mention":
return (
f"*🔔 MENTION*\n\n"
f"⏰ {timestamp}\n"
f"👤 {author_identifier}\n"
f"📝 *{content[:200]}*\n"
f"🔗 `nevent:{event_id}`"
)
elif event_type == "repost":
reposted_event_id = event_id
try:
for tag in event.tags():
if str(tag.kind()) == 'e':
reposted_event_id = tag.event_id().to_hex()
break
except Exception as e:
logger.debug(f"Error iterating tags for repost: {e}")
return (
f"*🔄 REPOST*\n\n"
f"⏰ {timestamp}\n"
f"👤 {author_identifier}\n"
f"📌 Reposted your note\n"
f"🔗 `nevent:{reposted_event_id}`"
)
elif event_type == "reaction":
reaction_content = content or "❤️"
reacted_event_id = event_id
try:
for tag in event.tags():
if str(tag.kind()) == 'e':
reacted_event_id = tag.event_id().to_hex()
break
except Exception as e:
logger.debug(f"Error iterating tags for reaction: {e}")
return (
f"*💬 REACTION*\n\n"
f"⏰ {timestamp}\n"
f"👤 {author_identifier}\n"
f"💭 *{reaction_content}*\n"
f"🔗 `nevent:{reacted_event_id}`"
)
elif event_type == "reply":
return (
f"*💬 REPLY*\n\n"
f"⏰ {timestamp}\n"
f"👤 {author_identifier}\n"
f"📝 *{content[:200]}*\n"
f"🔗 `nevent:{event_id}`"
)
else:
return (
f"*📢 NEW EVENT*\n\n"
f"⏰ {timestamp}\n"
f"👤 {author_identifier}\n"
f"📝 *{content[:200]}*"
)
async def _process_event(self, event) -> None:
"""Process a single event."""
event_id = event.id().to_hex()
# Deduplicate
if event_id in self.seen_events:
return
# Clean up seen_events if too large
if len(self.seen_events) > self.seen_events_max:
# Remove oldest 20% of entries (safe iteration)
to_remove = list(self.seen_events)[:self.seen_events_max // 5]
self.seen_events = self.seen_events - set(to_remove)
# Clean up event_engagement if too large
if len(self.event_engagement) > 10000:
min_engagement = min(self.event_engagement.values(),
key=lambda x: x['reactions'] + x['reposts'],
default=None)
if min_engagement:
to_remove = [k for k, v in self.event_engagement.items()
if v['reactions'] + v['reposts'] == min_engagement]
for k in to_remove[:len(to_remove) // 5]:
del self.event_engagement[k]
self.seen_events.add(event_id)
# Determine event type
event_type = None
author_pubkey = event.author()
kind = event.kind().as_u16()
# Check if it's a mention (p tag)
if kind == Kind.from_std(KindStandard.TEXT_NOTE).as_u16():
try:
for tag in event.tags():
if str(tag.kind()) == 'p':
tagged_pubkey = tag.pubkey().to_hex()
if tagged_pubkey == self.target_pubkey.to_hex():
event_type = "mention"
break
except Exception as e:
logger.debug(f"Error iterating tags: {e}")
# Check if it's a reply (e tag + p tag)
if not event_type and kind == Kind.from_std(KindStandard.TEXT_NOTE).as_u16():
try:
for tag in event.tags():
if str(tag.kind()) == 'e':
event_type = "reply"
break
except Exception as e:
logger.debug(f"Error iterating tags: {e}")
# Check if it's a repost
if kind in [Kind.from_std(KindStandard.REPOST).as_u16(), 30032]:
event_type = "repost"
# Check if it's a reaction
if kind == Kind.from_std(KindStandard.REACTION).as_u16():
event_type = "reaction"
if not event_type:
return
# Apply smart filters
if not self._apply_smart_filters(event_type, author_pubkey, event):
logger.debug(f"Smart filters blocked {event_type} from {author_pubkey.to_hex()[:12]}")
return
# Get author's identifier (nip-05 or pubkey)
author_identifier = await self._get_author_identifier(author_pubkey)
logger.info(f"🎯 {event_type.upper()} from {author_identifier}")
# Format and send notification
message = self._format_notification(event_type, event, author_identifier)
await self._send_telegram(message)
async def _connect_to_relays(self) -> bool:
"""Connect to all configured relays."""
for relay_url in self.relays:
try:
url = RelayUrl.parse(relay_url)
success = await self.client.add_relay(url)
if success:
logger.info(f"✅ Added relay: {relay_url}")
else:
logger.warning(f"⚠️ Relay already added: {relay_url}")
except Exception as e:
logger.error(f"❌ Failed to add relay {relay_url}: {e}")
try:
await self.client.connect()
logger.info(f"✅ Connected to {len(self.relays)} relays")
return True
except Exception as e:
logger.error(f"❌ Failed to connect to relays: {e}")
return False
async def _listen_for_events(self) -> None:
"""Listen for events across all relays."""
logger.info("👂 Starting event listener...")
filters = self._get_filters()
for i, filter_obj in enumerate(filters):
logger.info(f" [{i+1}/{len(filters)}] Filter configured")
try:
while True:
for filter_obj in filters:
try:
events = await self.client.fetch_events(filter_obj, timedelta(seconds=30))
if not events.is_empty():
logger.debug(f"Received {events.len()} events")
events_list = events.to_vec()
for event in events_list:
await self._process_event(event)
except Exception as e:
logger.error(f"Error fetching events: {e}")
await asyncio.sleep(1)
await asyncio.sleep(5)
except asyncio.CancelledError:
logger.info("🛑 Event listener cancelled")
raise
except Exception as e:
logger.error(f"❌ Event listener error: {e}")
raise
async def run(self) -> None:
"""Run the notifier."""
logger.info("🚀 Starting Nostr Notifier (nostr-sdk version)...")
if not await self._connect_to_relays():
logger.error("❌ Failed to connect to relays, exiting")
return
logger.info(f"✅ Notifier running")
logger.info(f"👀 Watching for: mentions, reposts, reactions, replies")
logger.info(f"🎛️ Smart filters: thread dedup, rate limiting ({self.rate_limit_per_minute}/min), engagement tracking")
await self._listen_for_events()
# Smart filter configuration methods
def set_rate_limit(self, notifications_per_minute: int):
"""Set maximum notifications per minute."""
self.rate_limit_per_minute = notifications_per_minute
logger.info(f"Rate limit set to {notifications_per_minute} per minute")
def set_engagement_threshold(self, min_engagement: int):
"""Set minimum engagement threshold (0 = notify all)."""
self.min_engagement_threshold = min_engagement
logger.info(f"Engagement threshold set to {min_engagement}")
def add_trusted_user(self, npub_or_pubkey: str):
"""Add a user to trusted list (always notify)."""
try:
pubkey = self._npub_to_pubkey(npub_or_pubkey)
self.trusted_users.add(pubkey.to_hex())
logger.info(f"Added trusted user: {npub_or_pubkey[:20]}...")
except Exception as e:
logger.error(f"Failed to add trusted user: {e}")
def add_blocked_user(self, npub_or_pubkey: str):
"""Add a user to blocked list (never notify)."""
try:
pubkey = self._npub_to_pubkey(npub_or_pubkey)
self.blocked_users.add(pubkey.to_hex())
logger.info(f"Added blocked user: {npub_or_pubkey[:20]}...")
except Exception as e:
logger.error(f"Failed to add blocked user: {e}")
def get_stats(self) -> Dict:
"""Get notifier statistics."""
return {
"events_seen": len(self.seen_events),
"threads_cached": len(self.thread_cache),
"profiles_cached": len(self.profile_cache),
"users_tracked": len(self.user_engagement),
"trusted_users": len(self.trusted_users),
"blocked_users": len(self.blocked_users),
"rate_limit": self.rate_limit_per_minute,
"engagement_threshold": self.min_engagement_threshold,
}
async def main():
"""Main entry point."""
from config import TARGET_NPUB, RELAYS, TELEGRAM_CHAT_ID, SMART_FILTERS
telegram_token = os.environ.get('TELEGRAM_BOT_TOKEN', '')
if not telegram_token:
logger.warning("⚠️ TELEGRAM_BOT_TOKEN not set - notifications will be logged only")
logger.info("Set it with: export TELEGRAM_BOT_TOKEN='***'")
notifier = NostrNotifier(
target_npub=TARGET_NPUB,
relays=RELAYS,
telegram_bot_token=telegram_token,
telegram_chat_id=TELEGRAM_CHAT_ID
)
# Apply smart filter configuration
if SMART_FILTERS:
notifier.set_rate_limit(SMART_FILTERS.get("rate_limit_per_minute", 10))
notifier.set_engagement_threshold(SMART_FILTERS.get("engagement_threshold", 0))
for npub in SMART_FILTERS.get("trusted_users", []):
notifier.add_trusted_user(npub)
for npub in SMART_FILTERS.get("blocked_users", []):
notifier.add_blocked_user(npub)
logger.info("✅ Smart filters configured")
logger.info(f"📊 Stats: {notifier.get_stats()}")
try:
await notifier.run()
except KeyboardInterrupt:
logger.info("🛑 Stopping notifier...")
except Exception as e:
logger.error(f"❌ Error: {e}")
if __name__ == "__main__":
asyncio.run(main())