FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
pythonostr/main.py at main · plebnet-dev/pythonostr · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
This repository was archived by the owner on May 22, 2024. It is now read-only.
plebnet-dev
/
pythonostr
Public archive
Notifications
You must be signed in to change notification settings
Fork
0
Star
0
Code
Issues
0
Pull requests
0
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Issues
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
pythonostr
/
main.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
118 lines (98 loc) · 3.36 KB
Breadcrumbs
pythonostr
/
main.py
Copy path
File metadata and controls
118 lines (98 loc) · 3.36 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from
nostr
.
client
.
client
import
NostrClient
from
nostr
.
event
import
Event
from
nostr
.
key
import
PublicKey
import
asyncio
import
threading
import
time
import
datetime
def
print_status
(
client
):
print
(
""
)
for
relay
in
client
.
relay_manager
.
relays
.
values
():
connected_text
=
"🟢"
if
relay
.
connected
else
"🔴"
status_text
=
f"
{
connected_text
}
⬆️
{
relay
.
num_sent_events
}
⬇️
{
relay
.
num_received_events
}
⚠️
{
relay
.
error_counter
}
⏱️
{
relay
.
ping
}
ms -
{
relay
.
url
.
split
(
'//'
)[
1
]
}
"
print
(
status_text
)
async
def
dm
():
print
(
"This is an example NIP-04 DM flow"
)
pk
=
input
(
"Enter your privatekey to post from (enter nothing for a random one): "
)
def
callback
(
event
:
Event
,
decrypted_content
):
"""
Callback to trigger when a DM is received.
"""
print
(
f"
\n
From
{
event
.
public_key
[:
3
]
}
..
{
event
.
public_key
[
-
3
:]
}
:
{
decrypted_content
}
"
)
client
=
NostrClient
(
privatekey_hex
=
pk
)
if
not
pk
:
print
(
f"Your private key:
{
client
.
private_key
.
bech32
()
}
"
)
print
(
f"Your public key:
{
client
.
public_key
.
bech32
()
}
"
)
t
=
threading
.
Thread
(
target
=
client
.
get_dm
,
args
=
(
client
.
public_key
,
callback
),
daemon
=
True
)
t
.
start
()
pubkey_to_str
=
(
input
(
"Enter other pubkey to DM to (enter nothing to DM yourself): "
)
or
client
.
public_key
.
hex
()
)
if
pubkey_to_str
.
startswith
(
"npub"
):
pubkey_to
=
PublicKey
().
from_npub
(
pubkey_to_str
)
else
:
pubkey_to
=
PublicKey
(
bytes
.
fromhex
(
pubkey_to_str
))
print
(
f"Sending DMs to
{
pubkey_to
.
bech32
()
}
"
)
while
True
:
print_status
(
client
)
await
asyncio
.
sleep
(
1
)
msg
=
input
(
"
\n
Enter message: "
)
client
.
dm
(
msg
,
pubkey_to
)
async
def
post
():
print
(
"This posts and reads a nostr note"
)
pk
=
input
(
"Enter your privatekey to post from (enter nothing for a random one): "
)
def
callback
(
event
:
Event
):
"""
Callback to trigger when post appers.
"""
print
(
f"
\n
From
{
event
.
public_key
[:
3
]
}
..
{
event
.
public_key
[
-
3
:]
}
:
{
event
.
content
}
"
)
sender_client
=
NostrClient
(
privatekey_hex
=
pk
)
# await asyncio.sleep(1)
pubkey_to_str
=
(
input
(
"Enter other pubkey (enter nothing to read your own posts, enter * for all): "
)
or
sender_client
.
public_key
.
hex
()
)
if
pubkey_to_str
==
"*"
:
pubkey_to
=
None
elif
pubkey_to_str
.
startswith
(
"npub"
):
pubkey_to
=
PublicKey
().
from_npub
(
pubkey_to_str
)
else
:
pubkey_to
=
PublicKey
(
bytes
.
fromhex
(
pubkey_to_str
))
print
(
f"Subscribing to posts by
{
pubkey_to
.
bech32
()
if
pubkey_to
else
'everyone'
}
"
)
filters
=
{
"since"
:
int
(
time
.
mktime
(
(
datetime
.
datetime
.
now
()
-
datetime
.
timedelta
(
hours
=
1
)).
timetuple
()
)
)
}
t
=
threading
.
Thread
(
target
=
sender_client
.
get_post
,
args
=
(
pubkey_to
,
callback
,
filters
,
),
daemon
=
True
,
)
t
.
start
()
while
True
:
print_status
(
sender_client
)
await
asyncio
.
sleep
(
1
)
msg
=
input
(
"
\n
Enter post: "
)
sender_client
.
post
(
msg
)
if
input
(
"Enter '1' for DM, '2' for Posts (Default: 1): "
)
==
"2"
:
# make a post and subscribe to posts
asyncio
.
run
(
post
())
else
:
# write a DM and receive DMs
asyncio
.
run
(
dm
())
Back
|
FazBrowse Home
|
New Git URL