-
Notifications
You must be signed in to change notification settings - Fork 2
/
nostr_ln_bot.py
88 lines (76 loc) · 2.92 KB
/
nostr_ln_bot.py
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
78
79
80
81
82
83
84
85
86
87
88
# nostr_ln_bot.py
from nostr import PrivateKey, SimpleRelayManager, Event, EventKind
from dvms import Nostr
from lndgrpc import LNDClient
from config import NOSTR_RELAYS, PRIVATE_KEY_HEX, LND_GRPC_HOST, LND_MACAROON_PATH, LND_TLS_CERT_PATH
def main():
# Generate keys
private_key = PrivateKey.from_hex(PRIVATE_KEY_HEX)
public_key = private_key.public_key.hex()
# Connect to relays
relay_manager = SimpleRelayManager()
for relay in NOSTR_RELAYS:
relay_manager.add_relay(relay)
relay_manager.run_in_thread()
# Connect to LND
lnd = LNDClient(
LND_GRPC_HOST,
macaroon_path=LND_MACAROON_PATH,
cert_path=LND_TLS_CERT_PATH
)
def recommend_channels():
channels = lnd.list_channels()
recommended = []
for channel in channels.channels:
if channel.remote_balance > 1000000:
if channel.local_balance / channel.remote_balance > 0.5:
if channel.local_balance / channel.remote_balance < 2:
recommended.append(channel)
return recommended
def handle_nostr_event(event):
if event.kind == EventKind.TEXT_NOTE and 'recommend channels' in event.content:
recommendations = recommend_channels()
recommendations_str = "Recommended Channels:\n" + "\n".join([str(ch) for ch in recommendations])
response_event = Event(kind=EventKind.TEXT_NOTE, content=recommendations_str)
response_event.sign(private_key.hex())
relay_manager.publish(response_event)
while True:
event = relay_manager.get_event()
if event:
handle_nostr_event(event)
if __name__ == "__main__":
main()
# Generate keys
private_key = PrivateKey.from_hex(PRIVATE_KEY_HEX)
public_key = private_key.public_key.hex()
# Connect to relays
relay_manager = SimpleRelayManager()
for relay in NOSTR_RELAYS:
relay_manager.add_relay(relay)
relay_manager.run_in_thread()
# Connect to LND
lnd = LNDClient(
LND_GRPC_HOST,
macaroon_path=LND_MACAROON_PATH,
cert_path=LND_TLS_CERT_PATH
)
def recommend_channels():
channels = lnd.list_channels()
recommended = []
for channel in channels.channels:
if channel.capacity > 1000000:
if channel.local_balance > 1000000:
if channel.remote_balance > 1000000:
recommended.append(channel)
return recommended
def handle_nostr_event(event):
if event.kind == EventKind.TEXT_NOTE and 'recommend channels' in event.content:
recommendations = recommend_channels()
recommendations_str = "Recommended Channels:\n" + "\n".join([str(ch) for ch in recommendations])
response_event = Event(kind=EventKind.TEXT_NOTE, content=recommendations_str)
response_event.sign(private_key.hex())
relay_manager.publish(response_event)
while True:
event = relay_manager.get_event()
if event:
handle_nostr_event(event)