Let's NiP05 scan the #spambots

Better coders can finetune I suppose

Assumptions and Setup

Nostr Relay: You have a WebSocket connection to a Nostr relay (e.g., wss://relay.damus.io) to fetch events.

NIP-05 Provider Data: You have a database or API with member public keys and their NIP-05 registration timestamps.

Libraries: Uses websocket for Nostr relay communication, json for event parsing, and random for selecting replies. You may need additional libraries like requests for NIP-05 verification checks.

Database: A simple SQLite database is assumed for storing member data, but you can adapt to your setup (e.g., MySQL, API).

Bot Detection: Basic bot detection checks for identical or nonsensical replies using simple text comparison. You can enhance this with NLP libraries like nltk or textblob for advanced analysis.

Script

import websocket

import json

import time

import random

import sqlite3

from datetime import datetime, timedelta

import requests

import re

from collections import Counter

# Configuration

RELAY_URL = "wss://relay.damus.io" # Replace with your relay

DB_PATH = "nip05_members.db" # SQLite database for NIP-05 members

TIME_WINDOW = 7 # Days to check for recent activity

REPLY_SAMPLE_SIZE = 5 # Number of replies to analyze

SIMILARITY_THRESHOLD = 0.9 # Jaccard similarity threshold for identical replies

# Initialize database

def init_db():

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

cursor.execute("""

CREATE TABLE IF NOT EXISTS members (

pubkey TEXT PRIMARY KEY,

nip05_address TEXT,

registration_timestamp INTEGER

)

""")

# Example: Populate with dummy data (replace with your actual data)

cursor.execute("INSERT OR IGNORE INTO members VALUES (?, ?, ?)",

("b0635d6a9851d3aed0cd6c495b282167acf761729078d975fc341b22650b07b9",

"bob@example.com", int(time.time() - 86400 * 30)))

conn.commit()

return conn, cursor

# Connect to Nostr relay

def connect_to_relay():

ws = websocket.WebSocket()

ws.connect(RELAY_URL)

return ws

# Fetch events from relay

def fetch_events(ws, pubkey, since_timestamp, kind=1):

subscription_id = f"scan_{pubkey[:8]}"

filter = {

"authors": [pubkey],

"kinds": [kind], # kind=1 for posts/replies

"since": since_timestamp

}

ws.send(json.dumps(["REQ", subscription_id, filter]))

events = []

timeout = time.time() + 10 # 10-second timeout

while time.time() < timeout:

try:

message = json.loads(ws.recv())

if message[0] == "EVENT" and message[1] == subscription_id:

events.append(message[2])

except websocket.WebSocketTimeoutException:

break

ws.send(json.dumps(["CLOSE", subscription_id]))

return events

# Calculate Jaccard similarity between two texts

def jaccard_similarity(text1, text2):

if not text1 or not text2:

return 0.0

set1 = set(re.findall(r'\w+', text1.lower()))

set2 = set(re.findall(r'\w+', text2.lower()))

intersection = len(set1 & set2)

union = len(set1 | set2)

return intersection / union if union > 0 else 0.0

# Check if text appears nonsensical (basic heuristic)

def is_nonsense(text):

# Basic check: too short, repetitive characters, or random strings

if len(text) < 10:

return True

if re.match(r'^(.)\1{3,}$', text): # Repeated characters

return True

# Add more checks (e.g., entropy, common spam phrases)

return False

# Check NIP-05 registration status

def check_nip05_status(pubkey, nip05_address):

try:

local_part, domain = nip05_address.split('@')