examples/monitor_relays.py at main · BigBrotr/examples · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
"""Monitor Relays.

Monitors a list of Nostr relays and publishes NIP-66 events to a target relay:
- Kind 0: monitor profile
- Kind 10166: monitor announcement (interval, timeout, enabled networks)
- Kind 30166: relay discovery (NIP-11 info + NIP-66 checks per relay)

Requires GeoLite2 databases for geolocation and ASN lookups.
"""

import asyncio
import json

import geoip2.database
from nostr_sdk import Client, Keys, NostrSigner, RelayUrl

from bigbrotr.models.constants import NetworkType
from bigbrotr.models.relay import Relay
from bigbrotr.nips.event_builders import (
    build_monitor_announcement,
    build_profile_event,
    build_relay_discovery,
)
from bigbrotr.nips.nip11 import Nip11, Nip11Selection
from bigbrotr.nips.nip66 import Nip66, Nip66Dependencies, Nip66Selection

RELAYS = [
    "wss://relay.damus.io",
    "wss://nos.lol",
    "wss://relay.nostr.band",
]
KEYS = Keys.generate()
PUBLISH_TO = "wss://nos.lol"
# Download from https://github.com/P3TERX/GeoLite.mmdb
CITY_DB = "GeoLite2-City.mmdb"
ASN_DB = "GeoLite2-ASN.mmdb"


async def main() -> None:
    city_reader = geoip2.database.Reader(CITY_DB)
    asn_reader = geoip2.database.Reader(ASN_DB)
    deps = Nip66Dependencies(city_reader=city_reader, asn_reader=asn_reader)

    signer = NostrSigner.keys(KEYS)
    client = Client(signer)
    await client.add_relay(RelayUrl.parse(PUBLISH_TO))
    await client.connect()

    await client.send_event_builder(build_profile_event(name="My Monitor"))
    await client.send_event_builder(
        build_monitor_announcement(
            interval=3600,
            timeout_ms=10000,
            enabled_networks=[NetworkType.CLEARNET],
            nip11_selection=Nip11Selection(),
            nip66_selection=Nip66Selection(),
        )
    )

    for url in RELAYS:
        relay = Relay(url)
        nip11 = await Nip11.create(relay)
        nip66 = await Nip66.create(relay, deps=deps)
        await client.send_event_builder(build_relay_discovery(relay, nip11, nip66))
        print(f"\n{'='*60}")
        print(f"Published 30166 for {url}")
        print(f"{'='*60}")
        print(f"\nNIP-11:\n{json.dumps(nip11.model_dump(), indent=2, default=str)}")
        print(f"\nNIP-66:\n{json.dumps(nip66.model_dump(), indent=2, default=str)}")

    await client.disconnect()
    city_reader.close()
    asn_reader.close()


if __name__ == "__main__":
    asyncio.run(main())