-
@ 123🦈ปลาฉลามขึ้นบก
2024-05-02 06:49:46สร้าง Nostr Bot ด้วย Python
อะแฮ่ม ขอชี้แจงไว้ก่อนว่า ผมเขียนเพื่อให้ตัวเองอ่านเพื่อวันไหนจะกลับมาทำต่อจะได้พอจำได้ว่าตัวเองทำอะไรลงไปบ้าง เพราะงั้นบางส่วนในบทความนี้อาจจะไม่ละเอียด หากลองทำตามแล้วติดตรงไหนอยากสอบถาม ติดต่อมาได้ที่ Nostr Address: kritta@rightshift.to หรือลองค้นเพิ่มเติมใน link ท้ายบทความนะครับ
อย่างที่หลาย ๆ คนน่าจะทราบกันดีอยู่แล้วว่า Nostr เป็น open protocol ที่ใคร ๆ ก็สามารถเข้ามามีส่วนร่วมในการพัฒนาได้ ทำให้มีโปรเจคต่าง ๆ เกิดขึ้นมากมาย โดยบทความในชุดนี้ผมจะหยิบโปรเจคต่าง ๆ ที่น่าสนใจมามาลองเล่น และนำมาเล่าสู่กันฟัง หวังว่าผู้ที่หลงเข้ามาอ่านจะได้ประโยชน์จากสิ่งนี้นะครับ ;)
โดยในวันนี้โปรเจคที่ผมหยิบมาคือ NDK (Nostr development kit) ผมหาไม่เจอว่าใครเป็นคนเริ่มโปรเจค แต่คนที่ดูแล repo นี้หลัก ๆ คือคุณ yukibtc เริ่มต้นเหมือนจะเริ่มจาก RUST แต่ตอนนี้เหมือนจะแตกไป swift, java, python เอาจริง ๆ ผมไม่รู้หรอกว่าใครเป็นคนทำภาษาไหนเพราะ contributors เขาเยอะมาก แต่ก็นั่นแหละ ขอบคุณที่สร้างอะไรสนุก ๆ แบบนี้ออกมาให้ได้เล่นนะครับ
โดยอย่างแรกที่เราต้องเริ่มคือการสั่ง pip install ตัว nostr sdk เพื่อใช่งาน สำหรับคนที่ไม่มี python ในเครื่องก็ไปลง python ก่อนด้วยนะ หรือจะใช้ online ผ่าน google colab ลองเล่นดูก่อนก็ได้
pip install pip install nostr-sdkจากนั้นเราก็จะสามารถใช้งาน Nostr_sdk ได้แล้ว!!! โดยการที่เราจะเข้ามาใช้งาน Nostr ได้นั้นเราจำเป็นต้องมี keys เพื่อเข้าสู่ระบบเสียก่อนงั้นเรามาเริ่มจากการสร้าง keys กันก่อน``` from nostr_sdk import Keys
เพียงคำสั่งนี้คำสั่งเดียวก็ได้ keys แล้วงั้นเหรอ!!
keys = Keys.generate()
แยก keys ออกเป็น secret key (sk) และ public key (pk)
sk = keys.secret_key() pk = keys.public_key()
ไหน ๆ ขอดู keys หน่อยสิ้
print(f"public key: {pk.to_bech32()}") print(f"Secret key: {sk.to_bech32()}")
output:
public key: npub1wkxaxzmmamc6h8n6ev7yq3y5qmqnyxmu0xmrllcepxup9tktuzrsu646r0
Secret key: nsec160gefyqkderqlnr545ps4d5th6pex3ducqgcev69z0rstqakkv9scvat97
``` note ถ้าสร้าง keys เสร็จแล้วเอาไปเก็บไว้ในพวก dot env จะปลอดภัยและสะดวกในการใช้ต่อมากกว่า
แล้วหลังจากได้ keys มาแล้วเราจำเป็นต้องกำหนด signer, client และ relay ที่เราจะใช้ในการรับ event ของเรา
```
กำหนด keys ที่เราพึ่งสร้างให้เป็นตัว sign event
signer = NostrSigner.keys(keys)
นำเข้า key ที่มีอยู่แล้ว
app_keys = Keys.parse("nsec......")
signer = NostrSigner.keys(app_keys)
หรือใช้ NIP46 signer
uri = NostrConnectUri.parse("bunker://.. or nostrconnect://..")
nip46 = Nip46Signer(uri, app_keys, timedelta(seconds=60), None)
signer = NostrSigner.nip46(nip46)
กำหนด client ให้ใช้ signer ตัวนี้ (feel like log in)
client = Client(signer)
เพิ่ม relays ที่จะเก็บ event
client.add_relays(["wss://relay.damus.io", "wss://siamstr.com", "wss://siamstr.com","wss://relay.notoshi.win"]) client.connect()
ตั้งชื่อให้ account เราสักหน่อยเพื่อเช็คด้วยว่า เราเชื่อมต่อ relay ต่าง ๆ ผ่านมั้ย
client.set_metadata(Metadata().set_name("Testing หลาม ๆ"))
``` หลังจากกำหนดทุกอย่างเรียบร้อยแล้ว เรามาลองสร้างโพสต์แรกกันเลยดีกว่า
```
tag เพื่อเอาไว้เติมส่วนต่าง ๆ นอกจาก เนื้อหาของโน๊ต เช่นการ mention การใส่ hashtag
p = mention
t = hashtag
tag = Tag.parse(["p", "66df60562d939ada8612436489945a4ecf1d62346b3d9478dea8a338f3203c64"])
ใส่เนื้อหาที่เราค้องการโพสต์
builder = EventBuilder.text_note("สวัสดีชาวทุ่ง ", [tag]) ส่ง event ไปให้ relay โลดดดด client.send_event_builder(builder) ``` แล้วนอกจากโพสต์ตระกูล kind:1 แล้วเรายังโพสต์ kind อื่น ๆ ได้ด้วย
```
ส่งจ้อความส่วนตัว
receiver_pk = PublicKey.from_bech32("npubคนรับ") event = EventBuilder.encrypted_direct_msg(keys, receiver_pk, "ข้อความ", None).to_event(keys) print(event.as_json())
templateเปล่า
kind = Kind(เลข kind) content = "..." tags = [] builder = EventBuilder(kind, content, tags)
POW
event = builder.to_pow_event(keys, 20) print(f"POW event: {event.as_json()}")
``` ส่วนตัวผมมองว่าส่วนนี้แหละคือส่วนที่สนุกที่สุดของวันนี้ เพราะเป็นจุดที่เราสามารถนำมันออกไปต่อยอดได้มากที่สุด เช่นการเชื่อมต่อกับ service อื่น ๆ เช่น mempool.space เพื่อส่งค่าฟี bitcoin ให้เราผ่านแชท, ทำเกมง่าย ๆ เล่นกับเพื่อน ๆ หน้า timeline อย่าง cowdle หรือ หวย อย่างที่เห็นกันไปในช่วงก่อนหน้านี้ หรือใช้ทำงานกรรมกรแทนเรา เช่นการแจก badges ที่ทาง rightshift ได้ทำไปก่อนหน้า, bot relay notoshi, zapbot และอีกต่าง ๆ มากมาย
filter
ตัว filter เป็นคำสั่งที่ช่วยเรากรอง event ที่จะขอจาก relay ใช้เพื่อรับเฉพาะ event ที่เราต้องการเท่านั้น
```
f = (Filter() .pubkey(keys.public_key()) .kinds([Kind(0), Kind.from_enum(KindEnum.TEXT_NOTE())]) .custom_tag(SingleLetterTag.lowercase(Alphabet.J), ["test"]) ) print(f.as_json())
output: {"kinds":[0,1],"#j":["test"],"#p":["758dd30b7beef1ab9e7acb3c40449406c1321b7c79b63fff1909b812aecbe087"]}
f = f.kind(Kind(4)).custom_tag(SingleLetterTag.lowercase(Alphabet.J), ["append-new"]) print(f.as_json()) {"kinds":[0,1,4],"#j":["test","append-new"],"#p":["758dd30b7beef1ab9e7acb3c40449406c1321b7c79b63fff1909b812aecbe087"]}
ตัวอย่างเช่นรับเฉพาะ event ของคนที่ใช้ notoshi relay
filter =Filter().kind(Kind(10002)).custom_tag(SingleLetterTag.lowercase(Alphabet.R), ["wss://relay.notoshi.win"]) events = client.get_events_of([filter], timedelta(seconds=30))
``` สองฟังก์ชันนี้เป็นตัวสำคัญในการทำบอทในส่วนต่อไปจะเป็นตัวเสริมต่าง ๆ ที่เพิ่มลูกเล่นให้บอทได้
Metadata
metadata มีไว้แก้ไขข้อมูลต่าง ๆ ในโปรไฟล์ของเรา
``` metadata = Metadata().set_name("username")\ .set_display_name("My Username")\ .set_about("Description")\ .set_picture("https://example.com/avatar.png")\ .set_banner("https://example.com/banner.png")\ .set_nip05("username@example.com")\ .set_lud16("username@example.com")
name = ชื้อผู้ใช่
display_name = ชื่อที่จะแสดงให้คนอื่นเห็น (ถ้าช่องนี้ว่างมักจะโชว์ชื่อที่ใว่ในช่อง name)
about = bio
picture = รูปโปรไฟล์
banner = รูปปก
nip05 = Nostr addr
lud16 = Lightning addr
```
NWC
NWC หรือ Nostr wallet connection มีไว้ใช้ในการเชื่อมต่อกับกระเป๋า ln ของเราเพื่อคุมกระเป๋าของเราผ่าน Nostr
```
นำ NWC uri มาวาง
uri = NostrWalletConnectUri.parse("nostr+walletconnect://..")
สร้าง client ในรูปแบบที่เพิ่มการ zap
keys = Keys.generate() signer = NostrSigner.keys(keys) zapper = NostrZapper.nwc(uri) client = ClientBuilder().signer(signer).zapper(zapper).build()
client.add_relay("wss://relay.damus.io") client.connect()
pk = PublicKey.from_bech32(" npub คนรับ") client.zap(ZapEntity.public_key(pk), 1000, None)
```
Bot template
```
from nostr_sdk import Client, NostrSigner, Keys, Event, UnsignedEvent, Filter, \ HandleNotification, Timestamp, nip04_decrypt, UnwrappedGift, init_logger, LogLevel, Kind, KindEnum import time
init_logger(LogLevel.DEBUG)
sk = SecretKey.from_bech32("nsec1ufnus6pju578ste3v90xd5m2decpuzpql2295m3sknqcjzyys9ls0qlc85")
keys = Keys(sk)
OR
keys = Keys.parse("nsec1ufnus6pju578ste3v90xd5m2decpuzpql2295m3sknqcjzyys9ls0qlc85")
sk = keys.secret_key() pk = keys.public_key() print(f"Bot public key: {pk.to_bech32()}")
signer = NostrSigner.keys(keys) client = Client(signer)
client.add_relay("wss://relay.damus.io") client.add_relay("wss://nostr.mom") client.add_relay("wss://nostr.oxtr.dev") client.connect()
now = Timestamp.now()
nip04_filter = Filter().pubkey(pk).kind(Kind.from_enum(KindEnum.ENCRYPTED_DIRECT_MESSAGE())).since(now) nip59_filter = Filter().pubkey(pk).kind(Kind.from_enum(KindEnum.GIFT_WRAP())).since( Timestamp.from_secs(now.as_secs() - 60 * 60 * 24 * 7)) # NIP59 have a tweaked timestamp (in the past) client.subscribe([nip04_filter, nip59_filter], None)
class NotificationHandler(HandleNotification): def handle(self, relay_url, subscription_id, event: Event): print(f"Received new event from {relay_url}: {event.as_json()}") if event.kind().match_enum(KindEnum.ENCRYPTED_DIRECT_MESSAGE()): print("Decrypting NIP04 event") try: msg = nip04_decrypt(sk, event.author(), event.content()) print(f"Received new msg: {msg}") client.send_direct_msg(event.author(), f"Echo: {msg}", event.id()) except Exception as e: print(f"Error during content NIP04 decryption: {e}") elif event.kind().match_enum(KindEnum.GIFT_WRAP()): print("Decrypting NIP59 event") try: # Extract rumor unwrapped_gift = UnwrappedGift.from_gift_wrap(keys, event) sender = unwrapped_gift.sender() rumor: UnsignedEvent = unwrapped_gift.rumor()
# Check timestamp of rumor if rumor.created_at().as_secs() >= now.as_secs(): if rumor.kind().match_enum(KindEnum.SEALED_DIRECT()): msg = rumor.content() print(f"Received new msg [sealed]: {msg}") client.send_sealed_msg(sender, f"Echo: {msg}", None) else: print(f"{rumor.as_json()}") except Exception as e: print(f"Error during content NIP59 decryption: {e}") def handle_msg(self, relay_url, msg): Noneabortable = client.handle_notifications(NotificationHandler())
Optionally, to abort handle notifications look, call abortable.abort()
while True: time.sleep(5.0) # abortable.abort() ```
ผมหวังว่าบทความนี้จะมีประโยชน์กับคนอ่าน และคาดหวังที่จะได้เห็น service ต่าง ๆ ที่สร้างสรรค์เกิดขึ้นหลังจากนี้ Link เพิ่มเติมที่สำหรับศึกษาต่อ
- https://github.com/rust-nostr/nostr/tree/master
- https://github.com/nostr-protocol/nips