ห้องทดลองของหลาม #1
สร้าง Nostr Bot ด้วย Python
> อะแฮ่ม ขอชี้แจงไว้ก่อนว่า ผมเขียนเพื่อให้ตัวเองอ่านเพื่อวันไหนจะกลับมาทำต่อจะได้พอจำได้ว่าตัวเองทำอะไรลงไปบ้าง เพราะงั้นบางส่วนในบทความนี้อาจจะไม่ละเอียด หากลองทำตามแล้วติดตรงไหนอยากสอบถาม ติดต่อมาได้ที่ Nostr Address: [kritta@rightshift.to](https://nosta.me/nprofile1qy0hwumn8ghj7mn0wd68ytfsxvhxgmmjv9nxzcm5dae8jtn0wfnj7qgcwaehxw309aex2mrp0yhxumm5daeks6fwwa5kutcqypnd7czk9kfe4k5xzfpkfzv5tf8v78tzx34nm9rcm652xw8nyq7xgdzhmy5) หรือลองค้นเพิ่มเติมใน link ท้ายบทความนะครับ
อย่างที่หลาย ๆ คนน่าจะทราบกันดีอยู่แล้วว่า Nostr เป็น open protocol ที่ใคร ๆ ก็สามารถเข้ามามีส่วนร่วมในการพัฒนาได้ ทำให้มีโปรเจคต่าง ๆ เกิดขึ้นมากมาย โดยบทความในชุดนี้ผมจะหยิบโปรเจคต่าง ๆ ที่น่าสนใจมามาลองเล่น และนำมาเล่าสู่กันฟัง หวังว่าผู้ที่หลงเข้ามาอ่านจะได้ประโยชน์จากสิ่งนี้นะครับ ;)
โดยในวันนี้โปรเจคที่ผมหยิบมาคือ NDK (Nostr development kit) ผมหาไม่เจอว่าใครเป็นคนเริ่มโปรเจค แต่คนที่ดูแล repo นี้หลัก ๆ คือคุณ [yukibtc](https://nosta.me/nprofile1qywhwumn8ghj7mn0wd68ytndw46xjmnewaskcmr9wshxxmmd9uq36amnwvaz7tmjv4kxz7fwd46hg6tw09mkzmrvv46zucm0d5hsqgrgmqgktyvpqzma5slu9rmarlqj24zxdcg3tzrtneamxfhktmzzwg2x3v53) เริ่มต้นเหมือนจะเริ่มจาก 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.setmetadata(Metadata().setname("Testing หลาม ๆ"))
```
หลังจากกำหนดทุกอย่างเรียบร้อยแล้ว เรามาลองสร้างโพสต์แรกกันเลยดีกว่า
```
#tag เพื่อเอาไว้เติมส่วนต่าง ๆ นอกจาก เนื้อหาของโน๊ต เช่นการ mention การใส่ hashtag
# p = mention
# t = hashtag
tag = Tag.parse(["p", "66df60562d939ada8612436489945a4ecf1d62346b3d9478dea8a338f3203c64"])
#ใส่เนื้อหาที่เราค้องการโพสต์
builder = EventBuilder.text_note("สวัสดีชาวทุ่ง ", [tag])
ส่ง event ไปให้ relay โลดดดด
client.sendeventbuilder(builder)
```
แล้วนอกจากโพสต์ตระกูล kind:1 แล้วเรายังโพสต์ kind อื่น ๆ ได้ด้วย
```
# ส่งจ้อความส่วนตัว
receiverpk = PublicKey.frombech32("npubคนรับ")
event = EventBuilder.encrypteddirectmsg(keys, receiverpk, "ข้อความ", None).toevent(keys)
print(event.as_json())
# templateเปล่า
kind = Kind(เลข kind)
content = "..."
tags = []
builder = EventBuilder(kind, content, tags)
# POW
event = builder.topowevent(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()
)
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.geteventsof([filter], timedelta(seconds=30))
```
สองฟังก์ชันนี้เป็นตัวสำคัญในการทำบอทในส่วนต่อไปจะเป็นตัวเสริมต่าง ๆ ที่เพิ่มลูกเล่นให้บอทได้
### Metadata
metadata มีไว้แก้ไขข้อมูลต่าง ๆ ในโปรไฟล์ของเรา
```
metadata = Metadata().set_name("username")\
# 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, nip04decrypt, UnwrappedGift, initlogger, 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()
nip04filter = Filter().pubkey(pk).kind(Kind.fromenum(KindEnum.ENCRYPTEDDIRECTMESSAGE())).since(now)
nip59filter = Filter().pubkey(pk).kind(Kind.fromenum(KindEnum.GIFT_WRAP())).since(
Timestamp.fromsecs(now.assecs() - 60 * 60 * 24 * 7)) # NIP59 have a tweaked timestamp (in the past)
client.subscribe([nip04filter, nip59filter], None)
class NotificationHandler(HandleNotification):
def handle(self, relayurl, subscriptionid, event: Event):
print(f"Received new event from {relayurl}: {event.asjson()}")
if event.kind().matchenum(KindEnum.ENCRYPTEDDIRECT_MESSAGE()):
print("Decrypting NIP04 event")
try:
msg = nip04_decrypt(sk, event.author(), event.content())
print(f"Received new msg: {msg}")
client.senddirectmsg(event.author(), f"Echo: {msg}", event.id())
except Exception as e:
print(f"Error during content NIP04 decryption: {e}")
elif event.kind().matchenum(KindEnum.GIFTWRAP()):
print("Decrypting NIP59 event")
try:
# Extract rumor
unwrappedgift = UnwrappedGift.fromgift_wrap(keys, event)
sender = unwrapped_gift.sender()
rumor: UnsignedEvent = unwrapped_gift.rumor()
# Check timestamp of rumor
if rumor.createdat().assecs() >= now.as_secs():
if rumor.kind().matchenum(KindEnum.SEALEDDIRECT()):
msg = rumor.content()
print(f"Received new msg [sealed]: {msg}")
client.sendsealedmsg(sender, f"Echo: {msg}", None)
else:
print(f"{rumor.as_json()}")
except Exception as e:
print(f"Error during content NIP59 decryption: {e}")
def handlemsg(self, relayurl, msg):
None
abortable = client.handle_notifications(NotificationHandler())
# Optionally, to abort handle notifications look, call abortable.abort()
while True:
time.sleep(5.0)
# abortable.abort()
```
ผมหวังว่าบทความนี้จะมีประโยชน์กับคนอ่าน และคาดหวังที่จะได้เห็น service ต่าง ๆ ที่สร้างสรรค์เกิดขึ้นหลังจากนี้
Link เพิ่มเติมที่สำหรับศึกษาต่อ