May 02, 2024 3 min read หลามมันร้าย

ห้องทดลองของหลาม #1

สร้าง 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):
                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 เพิ่มเติมที่สำหรับศึกษาต่อ