diff --git a/nostr_dvm/bot.py b/nostr_dvm/bot.py index de30158..e8590a6 100644 --- a/nostr_dvm/bot.py +++ b/nostr_dvm/bot.py @@ -1,3 +1,4 @@ +import asyncio import json import os import signal @@ -23,15 +24,7 @@ class Bot: job_list: list # This is a simple list just to keep track which events we created and manage, so we don't pay for other requests - def __init__(self, dvm_config, admin_config=None): - self.NAME = "Bot" - dvm_config.DB = "db/" + self.NAME + ".db" - self.dvm_config = dvm_config - nip89config = NIP89Config() - nip89config.NAME = self.NAME - self.dvm_config.NIP89 = nip89config - self.admin_config = admin_config - self.keys = Keys.parse(dvm_config.PRIVATE_KEY) + async def init_bot(self, dvm_config, admin_config=None): wait_for_send = True skip_disconnected_relays = True opts = (Options().wait_for_send(wait_for_send).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)) @@ -48,8 +41,8 @@ class Bot: ', '.join(p.NAME + ":" + p.TASK for p in self.dvm_config.SUPPORTED_DVMS) + "\n") for relay in self.dvm_config.RELAY_LIST: - self.client.add_relay(relay) - self.client.connect() + await self.client.add_relay(relay) + await self.client.connect() zap_filter = Filter().pubkey(pk).kinds([EventDefinitions.KIND_ZAP]).since(Timestamp.now()) dm_filter = Filter().pubkey(pk).kinds([EventDefinitions.KIND_DM]).since(Timestamp.now()) @@ -64,40 +57,54 @@ class Bot: self.client.subscribe([zap_filter, dm_filter, nip59_filter, dvm_filter], None) create_sql_table(self.dvm_config.DB) - admin_make_database_updates(adminconfig=self.admin_config, dvmconfig=self.dvm_config, client=self.client) + await admin_make_database_updates(adminconfig=self.admin_config, dvmconfig=self.dvm_config, client=self.client) + + def __init__(self, dvm_config, admin_config=None): + self.NAME = "Bot" + dvm_config.DB = "db/" + self.NAME + ".db" + self.dvm_config = dvm_config + nip89config = NIP89Config() + nip89config.NAME = self.NAME + self.dvm_config.NIP89 = nip89config + self.admin_config = admin_config + self.keys = Keys.parse(dvm_config.PRIVATE_KEY) + + asyncio.run(self.init_bot(dvm_config, admin_config)) + asyncio.run(self.run_bot(dvm_config)) + # add_sql_table_column(dvm_config.DB) - + async def run_bot(self, dvm_config): class NotificationHandler(HandleNotification): client = self.client dvm_config = self.dvm_config keys = self.keys - def handle(self, relay_url, subscription_id, nostr_event): + async def handle(self, relay_url, subscription_id, nostr_event): if (EventDefinitions.KIND_NIP90_EXTRACT_TEXT.as_u64() + 1000 <= nostr_event.kind().as_u64() <= EventDefinitions.KIND_NIP90_GENERIC.as_u64() + 1000): - handle_nip90_response_event(nostr_event) + await handle_nip90_response_event(nostr_event) elif nostr_event.kind() == EventDefinitions.KIND_FEEDBACK: - handle_nip90_feedback(nostr_event) + await handle_nip90_feedback(nostr_event) elif nostr_event.kind() == EventDefinitions.KIND_ZAP: - handle_zap(nostr_event) + await handle_zap(nostr_event) elif nostr_event.kind() == EventDefinitions.KIND_DM: try: - handle_dm(nostr_event, False) + await handle_dm(nostr_event, False) except Exception as e: print(f"Error during content NIP04 decryption: {e}") elif nostr_event.kind().match_enum(KindEnum.GIFT_WRAP()): try: - handle_dm(nostr_event, True) + await handle_dm(nostr_event, True) except Exception as e: print(f"Error during content NIP59 decryption: {e}") - def handle_msg(self, relay_url, msg): + async def handle_msg(self, relay_url, msg): return - def handle_dm(nostr_event, giftwrap): + async def handle_dm(nostr_event, giftwrap): sender = nostr_event.author().to_hex() if sender == self.keys.public_key().to_hex(): return @@ -131,7 +138,7 @@ class Bot: print(f"Error during content NIP04 decryption: {e}") if decrypted_text != "": - user = get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client, + user = await get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client, config=self.dvm_config) print("[" + self.NAME + "]" + sealed + "Message from " + user.name + ": " + decrypted_text) @@ -217,10 +224,10 @@ class Bot: else: evt = EventBuilder.encrypted_direct_msg(self.keys, PublicKey.parse(sender), message,None).to_event(self.keys) - send_event(evt, client=self.client, dvm_config=dvm_config) + await send_event(evt, client=self.client, dvm_config=dvm_config) elif decrypted_text.startswith("cashuA"): print("Received Cashu token:" + decrypted_text) - cashu_redeemed, cashu_message, total_amount, fees = redeem_cashu(decrypted_text, + cashu_redeemed, cashu_message, total_amount, fees = await redeem_cashu(decrypted_text, self.dvm_config, self.client) print(cashu_message) @@ -255,7 +262,7 @@ class Bot: except Exception as e: print("Error in bot " + str(e)) - def handle_nip90_feedback(nostr_event): + async def handle_nip90_feedback(nostr_event): # print(nostr_event.as_json()) try: is_encrypted = False @@ -303,7 +310,7 @@ class Bot: if status == "success" or status == "error" or status == "processing" or status == "partial" and content != "": entry = next((x for x in self.job_list if x['event_id'] == etag), None) if entry is not None and entry['dvm_key'] == nostr_event.author().to_hex(): - user = get_or_add_user(db=self.dvm_config.DB, npub=entry['npub'], + user = await get_or_add_user(db=self.dvm_config.DB, npub=entry['npub'], client=self.client, config=self.dvm_config) time.sleep(2.0) if entry["giftwrap"]: @@ -328,7 +335,7 @@ class Bot: if entry is not None and entry['is_paid'] is False and entry[ 'dvm_key'] == nostr_event.author().to_hex(): # if we get a bolt11, we pay and move on - user = get_or_add_user(db=self.dvm_config.DB, npub=entry["npub"], + user = await get_or_add_user(db=self.dvm_config.DB, npub=entry["npub"], client=self.client, config=self.dvm_config) if user.balance >= amount: balance = max(user.balance - amount, 0) @@ -345,7 +352,7 @@ class Bot: PublicKey.parse(entry["npub"]), message, None).to_event(self.keys) - send_event(evt, client=self.client, dvm_config=dvm_config) + await send_event(evt, client=self.client, dvm_config=dvm_config) print( "[" + self.NAME + "] Replying " + user.name + " with \"scheduled\" confirmation") @@ -360,14 +367,14 @@ class Bot: str(int(amount - user.balance)) + " Sats, then try again.", None).to_event(self.keys) - send_event(evt, client=self.client, dvm_config=dvm_config) + await send_event(evt, client=self.client, dvm_config=dvm_config) return if len(tag.as_vec()) > 2: bolt11 = tag.as_vec()[2] # else we create a zap else: - user = get_or_add_user(db=self.dvm_config.DB, npub=nostr_event.author().to_hex(), + user = await get_or_add_user(db=self.dvm_config.DB, npub=nostr_event.author().to_hex(), client=self.client, config=self.dvm_config) print("Paying: " + user.name) bolt11 = zaprequest(user.lud16, amount, "Zap", nostr_event, self.keys, @@ -389,7 +396,7 @@ class Bot: except Exception as e: print(e) - def handle_nip90_response_event(nostr_event: Event): + async def handle_nip90_response_event(nostr_event: Event): try: ptag = "" etag = "" @@ -406,7 +413,7 @@ class Bot: if entry is not None and entry[ 'dvm_key'] == nostr_event.author().to_hex(): print(entry) - user = get_or_add_user(db=self.dvm_config.DB, npub=entry['npub'], + user = await get_or_add_user(db=self.dvm_config.DB, npub=entry['npub'], client=self.client, config=self.dvm_config) self.job_list.remove(entry) @@ -436,12 +443,12 @@ class Bot: PublicKey.parse(user.npub), content, None).to_event(self.keys) - send_event(reply_event, client=self.client, dvm_config=dvm_config) + await send_event(reply_event, client=self.client, dvm_config=dvm_config) except Exception as e: print(e) - def handle_zap(zap_event): + async def handle_zap(zap_event): print("[" + self.NAME + "] Zap received") try: invoice_amount, zapped_event, sender, message, anon = parse_zap_event_tags(zap_event, @@ -455,7 +462,7 @@ class Bot: if tag.as_vec()[0] == "e": etag = tag.as_vec()[1] - user = get_or_add_user(self.dvm_config.DB, sender, client=self.client, config=self.dvm_config) + user = await get_or_add_user(self.dvm_config.DB, sender, client=self.client, config=self.dvm_config) entry = next((x for x in self.job_list if x['event_id'] == etag), None) print(entry) @@ -465,7 +472,7 @@ class Bot: print(sender) if entry is not None and entry['is_paid'] is True and entry['dvm_key'] == sender: # if we get a bolt11, we pay and move on - user = get_or_add_user(db=self.dvm_config.DB, npub=entry["npub"], + user = await get_or_add_user(db=self.dvm_config.DB, npub=entry["npub"], client=self.client, config=self.dvm_config) sender = user.npub @@ -692,7 +699,7 @@ class Bot: return None - self.client.handle_notifications(NotificationHandler()) + await self.client.handle_notifications(NotificationHandler()) try: while True: diff --git a/nostr_dvm/dvm.py b/nostr_dvm/dvm.py index 52eb0d9..6b307ea 100644 --- a/nostr_dvm/dvm.py +++ b/nostr_dvm/dvm.py @@ -1,3 +1,4 @@ +import asyncio import json import os import subprocess @@ -34,15 +35,13 @@ class DVM: job_list: list jobs_on_hold_list: list - def __init__(self, dvm_config, admin_config=None): - self.dvm_config = dvm_config - self.admin_config = admin_config - self.keys = Keys.parse(dvm_config.PRIVATE_KEY) + async def init_dvm(self, dvm_config, admin_config=None): wait_for_send = False skip_disconnected_relays = True relaylimits = RelayLimits.disable() - opts = (Options().wait_for_send(wait_for_send).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)) - .skip_disconnected_relays(skip_disconnected_relays).relay_limits(relaylimits)) + opts = ( + Options().wait_for_send(wait_for_send).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)) + .skip_disconnected_relays(skip_disconnected_relays).relay_limits(relaylimits)) signer = NostrSigner.keys(self.keys) self.client = Client.with_opts(signer, opts) @@ -55,8 +54,8 @@ class DVM: ', '.join(p.NAME + ":" + p.TASK for p in self.dvm_config.SUPPORTED_DVMS) + bcolors.ENDC) for relay in self.dvm_config.RELAY_LIST: - self.client.add_relay(relay) - self.client.connect() + await self.client.add_relay(relay) + await self.client.connect() zap_filter = Filter().pubkey(pk).kinds([EventDefinitions.KIND_ZAP]).since(Timestamp.now()) kinds = [EventDefinitions.KIND_NIP90_GENERIC] @@ -64,28 +63,35 @@ class DVM: if dvm.KIND not in kinds: kinds.append(dvm.KIND) dvm_filter = (Filter().kinds(kinds).since(Timestamp.now())) - - self.client.subscribe([dvm_filter, zap_filter], None) - create_sql_table(self.dvm_config.DB) - admin_make_database_updates(adminconfig=self.admin_config, dvmconfig=self.dvm_config, client=self.client) + await admin_make_database_updates(adminconfig=self.admin_config, dvmconfig=self.dvm_config, client=self.client) + await self.client.subscribe([dvm_filter, zap_filter], None) + def __init__(self, dvm_config, admin_config=None): + self.dvm_config = dvm_config + self.admin_config = admin_config + self.keys = Keys.parse(dvm_config.PRIVATE_KEY) + + asyncio.run(self.init_dvm(dvm_config, admin_config)) + asyncio.run(self.run_dvm(dvm_config)) + + async def run_dvm(self, dvm_config): class NotificationHandler(HandleNotification): client = self.client dvm_config = self.dvm_config keys = self.keys - def handle(self, relay_url, subscription_id, nostr_event: Event): + async def handle(self, relay_url, subscription_id, nostr_event: Event): if EventDefinitions.KIND_NIP90_EXTRACT_TEXT.as_u64() <= nostr_event.kind().as_u64() <= EventDefinitions.KIND_NIP90_GENERIC.as_u64(): - handle_nip90_job_event(nostr_event) + await handle_nip90_job_event(nostr_event) elif nostr_event.kind().as_u64() == EventDefinitions.KIND_ZAP.as_u64(): - handle_zap(nostr_event) + await handle_zap(nostr_event) - def handle_msg(self, relay_url, msg): + async def handle_msg(self, relay_url, msg): return - def handle_nip90_job_event(nip90_event): + async def handle_nip90_job_event(nip90_event): # decrypted encrypted events nip90_event = check_and_decrypt_tags(nip90_event, self.dvm_config) # if event is encrypted, but we can't decrypt it (e.g. because its directed to someone else), return @@ -113,11 +119,11 @@ class DVM: # if task is supported, continue, else do nothing. if task_supported: # fetch or add user contacting the DVM from/to local database - user = get_or_add_user(self.dvm_config.DB, nip90_event.author().to_hex(), client=self.client, + user = await get_or_add_user(self.dvm_config.DB, nip90_event.author().to_hex(), client=self.client, config=self.dvm_config, skip_meta=False) # if user is blacklisted for some reason, send an error reaction and return if user.isblacklisted: - send_job_status_reaction(nip90_event, "error", client=self.client, dvm_config=self.dvm_config) + await send_job_status_reaction(nip90_event, "error", client=self.client, dvm_config=self.dvm_config) print("[" + self.dvm_config.NIP89.NAME + "] Request by blacklisted user, skipped") return @@ -130,7 +136,7 @@ class DVM: # If this is a subscription DVM and the Task is directed to us, check for active subscription if dvm_config.NIP88 is not None and p_tag_str == self.dvm_config.PUBLIC_KEY: - send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client, + await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client, "Checking Subscription Status, please wait..", self.dvm_config) # if we stored in the database that the user has an active subscription, we don't need to check it print("User Subscription: " + str(user.subscribed) + " Current time: " + str( @@ -139,14 +145,14 @@ class DVM: if int(user.subscribed) > int(Timestamp.now().as_secs()): print("User subscribed until: " + str(Timestamp.from_secs(user.subscribed).to_human_datetime())) user_has_active_subscription = True - send_job_status_reaction(nip90_event, "subscription-required", True, amount, + await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client, "User subscripton active until " + Timestamp.from_secs(int(user.subscribed)).to_human_datetime().replace( "Z", " ").replace("T", " ") + " GMT", self.dvm_config) # otherwise we check for an active subscription by checking recipie events else: print("[" + self.dvm_config.NIP89.NAME + "] Checking Subscription status") - send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client, + await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client, "I Don't have information about subscription status, checking on the Nostr. This might take a few seconds", self.dvm_config) @@ -155,7 +161,7 @@ class DVM: self.dvm_config.PUBLIC_KEY) if subscription_status["isActive"]: - send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client, + await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client, "User subscripton active until " + Timestamp.from_secs(int( subscription_status[ "validUntil"])).to_human_datetime().replace("Z", @@ -170,7 +176,7 @@ class DVM: self.client, self.dvm_config) else: print("No active subscription found") - send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client, + await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client, "No active subscription found. Manage your subscription at: " + self.dvm_config.SUBSCRIPTION_MANAGEMENT, self.dvm_config) @@ -181,11 +187,11 @@ class DVM: cashu_redeemed = False if cashu != "": print(cashu) - cashu_redeemed, cashu_message, redeem_amount, fees = redeem_cashu(cashu, self.dvm_config, + cashu_redeemed, cashu_message, redeem_amount, fees = await redeem_cashu(cashu, self.dvm_config, self.client, int(amount)) print(cashu_message) if cashu_message != "success": - send_job_status_reaction(nip90_event, "error", False, amount, self.client, cashu_message, + await send_job_status_reaction(nip90_event, "error", False, amount, self.client, cashu_message, self.dvm_config) return # if user is whitelisted or task is free, just do the job @@ -197,14 +203,14 @@ class DVM: ". Starting processing..") if dvm_config.SEND_FEEDBACK_EVENTS: - send_job_status_reaction(nip90_event, "processing", True, 0, + await send_job_status_reaction(nip90_event, "processing", True, 0, content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE, client=self.client, dvm_config=self.dvm_config, user=user) # when we reimburse users on error make sure to not send anything if it was free if user.iswhitelisted or task_is_free: amount = 0 - do_work(nip90_event, amount) + await do_work(nip90_event, amount) # if task is directed to us via p tag and user has balance or is subscribed, do the job and update balance elif (p_tag_str == self.dvm_config.PUBLIC_KEY and ( user.balance >= int( @@ -225,11 +231,11 @@ class DVM: print("[" + self.dvm_config.NIP89.NAME + "] User has active subscription for task: " + task + ". Starting processing.. Balance remains at: " + str(user.balance)) - send_job_status_reaction(nip90_event, "processing", True, 0, + await send_job_status_reaction(nip90_event, "processing", True, 0, content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE, client=self.client, dvm_config=self.dvm_config) - do_work(nip90_event, amount) + await do_work(nip90_event, amount) # else send a payment required event to user elif p_tag_str == "" or p_tag_str == self.dvm_config.PUBLIC_KEY: @@ -238,7 +244,7 @@ class DVM: print( "[" + self.dvm_config.NIP89.NAME + "] Hinting user for Subscription: " + nip90_event.id().to_hex()) - send_job_status_reaction(nip90_event, "subscription-required", + await send_job_status_reaction(nip90_event, "subscription-required", False, 0, client=self.client, dvm_config=self.dvm_config) else: @@ -253,7 +259,7 @@ class DVM: if bid > 0: bid_offer = int(bid / 1000) if bid_offer >= int(amount): - send_job_status_reaction(nip90_event, "payment-required", False, + await send_job_status_reaction(nip90_event, "payment-required", False, int(amount), # bid_offer client=self.client, dvm_config=self.dvm_config) @@ -261,7 +267,7 @@ class DVM: print( "[" + self.dvm_config.NIP89.NAME + "] Requesting payment for Event: " + nip90_event.id().to_hex()) - send_job_status_reaction(nip90_event, "payment-required", + await send_job_status_reaction(nip90_event, "payment-required", False, int(amount), client=self.client, dvm_config=self.dvm_config) @@ -272,13 +278,13 @@ class DVM: # else: # print("[" + self.dvm_config.NIP89.NAME + "] Task " + task + " not supported on this DVM, skipping..") - def handle_zap(zap_event): + async def handle_zap(zap_event): try: invoice_amount, zapped_event, sender, message, anon = parse_zap_event_tags(zap_event, self.keys, self.dvm_config.NIP89.NAME, self.client, self.dvm_config) - user = get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client, config=self.dvm_config) + user = await get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client, config=self.dvm_config) if zapped_event is not None: if zapped_event.kind() == EventDefinitions.KIND_FEEDBACK: @@ -305,7 +311,7 @@ class DVM: # if a reaction by us got zapped print(status) if job_event.kind() == EventDefinitions.KIND_NIP88_SUBSCRIBE_EVENT: - send_job_status_reaction(job_event, "subscription-success", client=self.client, + await send_job_status_reaction(job_event, "subscription-success", client=self.client, dvm_config=self.dvm_config, user=user) @@ -318,7 +324,7 @@ class DVM: user.name)) if amount <= invoice_amount: print("[" + self.dvm_config.NIP89.NAME + "] Payment-request fulfilled...") - send_job_status_reaction(job_event, "processing", client=self.client, + await send_job_status_reaction(job_event, "processing", client=self.client, content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE, dvm_config=self.dvm_config, user=user) indices = [i for i, x in enumerate(self.job_list) if @@ -329,18 +335,18 @@ class DVM: if index > -1: if self.job_list[index].is_processed: self.job_list[index].is_paid = True - check_and_return_event(self.job_list[index].result, job_event) + await check_and_return_event(self.job_list[index].result, job_event) elif not (self.job_list[index]).is_processed: # If payment-required appears before processing self.job_list.pop(index) print("Starting work...") - do_work(job_event, invoice_amount) + await do_work(job_event, invoice_amount) else: print("Job not in List, but starting work...") - do_work(job_event, invoice_amount) + await do_work(job_event, invoice_amount) else: - send_job_status_reaction(job_event, "payment-rejected", + await send_job_status_reaction(job_event, "payment-rejected", False, invoice_amount, client=self.client, dvm_config=self.dvm_config) print("[" + self.dvm_config.NIP89.NAME + "] Invoice was not paid sufficiently") @@ -366,7 +372,7 @@ class DVM: except Exception as e: print("[" + self.dvm_config.NIP89.NAME + "] Error during content decryption: " + str(e)) - def check_event_has_not_unfinished_job_input(nevent, append, client, dvmconfig): + async def check_event_has_not_unfinished_job_input(nevent, append, client, dvmconfig): task_supported, task = check_task_is_supported(nevent, client, config=dvmconfig) if not task_supported: return False @@ -387,14 +393,14 @@ class DVM: if append: job_ = RequiredJobToWatch(event=nevent, timestamp=Timestamp.now().as_secs()) self.jobs_on_hold_list.append(job_) - send_job_status_reaction(nevent, "chain-scheduled", True, 0, + await send_job_status_reaction(nevent, "chain-scheduled", True, 0, client=client, dvm_config=dvmconfig) return False else: return True - def check_and_return_event(data, original_event: Event): + async def check_and_return_event(data, original_event: Event): amount = 0 for x in self.job_list: if x.event == original_event: @@ -403,12 +409,12 @@ class DVM: x.result = data x.is_processed = True if self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and not is_paid: - send_nostr_reply_event(data, original_event.as_json()) - send_job_status_reaction(original_event, "success", amount, + await send_nostr_reply_event(data, original_event.as_json()) + await send_job_status_reaction(original_event, "success", amount, dvm_config=self.dvm_config, ) # or payment-required, or both? elif not self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and not is_paid: - send_job_status_reaction(original_event, "success", amount, + await send_job_status_reaction(original_event, "success", amount, dvm_config=self.dvm_config, ) # or payment-required, or both? @@ -416,7 +422,7 @@ class DVM: self.job_list.remove(x) elif not self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and is_paid: self.job_list.remove(x) - send_nostr_reply_event(data, original_event.as_json()) + await send_nostr_reply_event(data, original_event.as_json()) break task = get_task(original_event, self.client, self.dvm_config) @@ -424,17 +430,17 @@ class DVM: if task == dvm.TASK: try: post_processed = dvm.post_process(data, original_event) - send_nostr_reply_event(post_processed, original_event.as_json()) + await send_nostr_reply_event(post_processed, original_event.as_json()) except Exception as e: print(e) # Zapping back by error in post-processing is a risk for the DVM because work has been done, # but maybe something with parsing/uploading failed. Try to avoid errors here as good as possible - send_job_status_reaction(original_event, "error", + await send_job_status_reaction(original_event, "error", content="Error in Post-processing: " + str(e), dvm_config=self.dvm_config, ) if amount > 0 and self.dvm_config.LNBITS_ADMIN_KEY != "": - user = get_or_add_user(self.dvm_config.DB, original_event.author().to_hex(), + user = await get_or_add_user(self.dvm_config.DB, original_event.author().to_hex(), client=self.client, config=self.dvm_config) print(user.lud16 + " " + str(amount)) bolt11 = zaprequest(user.lud16, amount, "Couldn't finish job, returning sats", @@ -448,7 +454,7 @@ class DVM: except Exception as e: print(e) - def send_nostr_reply_event(content, original_event_as_str): + async def send_nostr_reply_event(content, original_event_as_str): original_event = Event.from_json(original_event_as_str) request_tag = Tag.parse(["request", original_event_as_str]) e_tag = Tag.parse(["e", original_event.id().to_hex()]) @@ -466,7 +472,6 @@ class DVM: if relay_tag is not None: reply_tags.append(relay_tag) - encrypted = False for tag in original_event.tags(): if tag.as_vec()[0] == "encrypted": @@ -489,12 +494,12 @@ class DVM: self.keys) # send_event(reply_event, client=self.client, dvm_config=self.dvm_config) - send_event_outbox(reply_event, client=self.client, dvm_config=self.dvm_config) + await send_event_outbox(reply_event, client=self.client, dvm_config=self.dvm_config) print(bcolors.GREEN + "[" + self.dvm_config.NIP89.NAME + "] " + str( original_event.kind().as_u64() + 1000) + " Job Response event sent: " + reply_event.as_json() + bcolors.ENDC) - def send_job_status_reaction(original_event, status, is_paid=True, amount=0, client=None, + async def send_job_status_reaction(original_event, status, is_paid=True, amount=0, client=None, content=None, dvm_config=None, user=None): @@ -600,13 +605,13 @@ class DVM: keys = Keys.parse(dvm_config.PRIVATE_KEY) reaction_event = EventBuilder(EventDefinitions.KIND_FEEDBACK, str(content), reply_tags).to_event(keys) # send_event(reaction_event, client=self.client, dvm_config=self.dvm_config) - send_event_outbox(reaction_event, client=self.client, dvm_config=self.dvm_config) + await send_event_outbox(reaction_event, client=self.client, dvm_config=self.dvm_config) print(bcolors.YELLOW + "[" + self.dvm_config.NIP89.NAME + "]" + " Sent Kind " + str( EventDefinitions.KIND_FEEDBACK.as_u64()) + " Reaction: " + status + " " + reaction_event.as_json() + bcolors.ENDC) return reaction_event.as_json() - def do_work(job_event, amount): + async def do_work(job_event, amount): if (( EventDefinitions.KIND_NIP90_EXTRACT_TEXT.as_u64() <= job_event.kind().as_u64() <= EventDefinitions.KIND_NIP90_GENERIC.as_u64()) or job_event.kind().as_u64() == EventDefinitions.KIND_DM.as_u64()): @@ -643,25 +648,25 @@ class DVM: else: # Some components might have issues with running code in otuside venv. # We install locally in these cases for now - result = dvm.process(request_form) + result = await dvm.process(request_form) try: post_processed = dvm.post_process(result, job_event) - send_nostr_reply_event(post_processed, job_event.as_json()) + await send_nostr_reply_event(post_processed, job_event.as_json()) except Exception as e: print(bcolors.RED + "[" + self.dvm_config.NIP89.NAME + "] Error: " + str( e) + bcolors.ENDC) - send_job_status_reaction(job_event, "error", content=str(e), + await send_job_status_reaction(job_event, "error", content=str(e), dvm_config=self.dvm_config) except Exception as e: print( bcolors.RED + "[" + self.dvm_config.NIP89.NAME + "] Error: " + str(e) + bcolors.ENDC) # we could send the exception here to the user, but maybe that's not a good idea after all. - send_job_status_reaction(job_event, "error", content=result, + await send_job_status_reaction(job_event, "error", content=result, dvm_config=self.dvm_config) # Zapping back the user on error if amount > 0 and self.dvm_config.LNBITS_ADMIN_KEY != "": - user = get_or_add_user(self.dvm_config.DB, job_event.author().to_hex(), + user = await get_or_add_user(self.dvm_config.DB, job_event.author().to_hex(), client=self.client, config=self.dvm_config) print(user.lud16 + " " + str(amount)) bolt11 = zaprequest(user.lud16, amount, "Couldn't finish job, returning sats", job_event, @@ -677,11 +682,11 @@ class DVM: return - self.client.handle_notifications(NotificationHandler()) + await self.client.handle_notifications(NotificationHandler()) while True: for dvm in self.dvm_config.SUPPORTED_DVMS: - scheduled_result = dvm.schedule(self.dvm_config) + scheduled_result = await dvm.schedule(self.dvm_config) for job in self.job_list: if job.bolt11 != "" and job.payment_hash != "" and not job.payment_hash is None and not job.is_paid: @@ -692,12 +697,12 @@ class DVM: amount = parse_amount_from_bolt11_invoice(job.bolt11) job.is_paid = True - send_job_status_reaction(job.event, "processing", True, 0, + await send_job_status_reaction(job.event, "processing", True, 0, content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE, client=self.client, dvm_config=self.dvm_config) print("[" + self.dvm_config.NIP89.NAME + "] doing work from joblist") - do_work(job.event, amount) + await do_work(job.event, amount) elif ispaid is None: # invoice expired self.job_list.remove(job) @@ -705,9 +710,9 @@ class DVM: self.job_list.remove(job) for job in self.jobs_on_hold_list: - if check_event_has_not_unfinished_job_input(job.event, False, client=self.client, + if await check_event_has_not_unfinished_job_input(job.event, False, client=self.client, dvmconfig=self.dvm_config): - handle_nip90_job_event(nip90_event=job.event) + await handle_nip90_job_event(nip90_event=job.event) try: self.jobs_on_hold_list.remove(job) except: @@ -715,22 +720,5 @@ class DVM: if Timestamp.now().as_secs() > job.timestamp + 60 * 20: # remove jobs to look for after 20 minutes.. self.jobs_on_hold_list.remove(job) - advanced_log = False - if advanced_log: - for url, relay in self.client.relays().items(): - stats = relay.stats() - print(f"Relay: {url}") - print(f"Connected: {relay.is_connected()}") - print(f"Status: {relay.status()}") - print("Stats:") - print(f" Attempts: {stats.attempts()}") - print(f" Success: {stats.success()}") - print(f" Bytes sent: {stats.bytes_sent()}") - print(f" Bytes received: {stats.bytes_received()}") - print(f" Connected at: {stats.connected_at().to_human_datetime()}") - if stats.latency() is not None: - print(f" Latency: {stats.latency().total_seconds() * 1000} ms") - - print("###########################################") time.sleep(1.0) diff --git a/nostr_dvm/interfaces/dvmtaskinterface.py b/nostr_dvm/interfaces/dvmtaskinterface.py index db89544..5307390 100644 --- a/nostr_dvm/interfaces/dvmtaskinterface.py +++ b/nostr_dvm/interfaces/dvmtaskinterface.py @@ -1,3 +1,4 @@ +import asyncio import json import os import subprocess @@ -34,11 +35,14 @@ class DVMTaskInterface: def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, admin_config: AdminConfig = None, options=None, task=None): + if options is None: + self.options = {} + else: + self.options = options self.init(name, dvm_config, admin_config, nip88config, nip89config, task) - self.options = options self.install_dependencies(dvm_config) - def init(self, name, dvm_config, admin_config=None, nip88config=None, nip89config=None, task=None): + def init(self, name, dvm_config, admin_config=None, nip88config=None, nip89config=None, task=None, options=None): self.NAME = name self.PRIVATE_KEY = dvm_config.PRIVATE_KEY if dvm_config.PUBLIC_KEY == "" or dvm_config.PUBLIC_KEY is None: @@ -66,6 +70,14 @@ class DVMTaskInterface: self.dvm_config = dvm_config self.admin_config = admin_config + asyncio.run(self.init_dvm(name, dvm_config, nip89config, nip88config, + admin_config, options)) + + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): + + pass + def install_dependencies(self, dvm_config): if dvm_config.SCRIPT != "": if self.dvm_config.USE_OWN_VENV: @@ -94,7 +106,7 @@ class DVMTaskInterface: nostr_dvm_thread = Thread(target=self.DVM, args=[self.dvm_config, self.admin_config]) nostr_dvm_thread.start() - def schedule(self, dvm_config): + async def schedule(self, dvm_config): """schedule something, e.g. define some time to update or to post, does nothing by default""" pass @@ -115,7 +127,7 @@ class DVMTaskInterface: """Parse input into a request form that will be given to the process method""" pass - def process(self, request_form): + async def process(self, request_form): "Process the data and return the result" pass diff --git a/nostr_dvm/subscription.py b/nostr_dvm/subscription.py index 6ab1c09..27dba05 100644 --- a/nostr_dvm/subscription.py +++ b/nostr_dvm/subscription.py @@ -1,3 +1,4 @@ +import asyncio import json import math import os @@ -24,16 +25,7 @@ from nostr_dvm.utils.zap_utils import create_bolt11_lud16, zaprequest class Subscription: job_list: list - # This is a simple list just to keep track which events we created and manage, so we don't pay for other requests - def __init__(self, dvm_config, admin_config=None): - self.NAME = "Subscription Handler" - dvm_config.DB = "db/" + "subscriptions" + ".db" - self.dvm_config = dvm_config - nip89config = NIP89Config() - nip89config.NAME = self.NAME - self.dvm_config.NIP89 = nip89config - self.admin_config = admin_config - self.keys = Keys.parse(dvm_config.PRIVATE_KEY) + async def init_subscription(self, dvm_config, admin_config=None): wait_for_send = False skip_disconnected_relays = True opts = (Options().wait_for_send(wait_for_send).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)) @@ -49,18 +41,19 @@ class Subscription: pk.to_hex()) + "\n") for relay in self.dvm_config.RELAY_LIST: - self.client.add_relay(relay) - self.client.connect() + await self.client.add_relay(relay) + await self.client.connect() zap_filter = Filter().pubkey(pk).kinds([EventDefinitions.KIND_ZAP]).since(Timestamp.now()) cancel_subscription_filter = Filter().kinds([EventDefinitions.KIND_NIP88_STOP_SUBSCRIPTION_EVENT]).since( Timestamp.now()) authors = [] if admin_config is not None and len(admin_config.USERNPUBS) > 0: - #we might want to limit which services can connect to the subscription handler + # we might want to limit which services can connect to the subscription handler for key in admin_config.USERNPUBS: authors.append(PublicKey.parse(key)) - dvm_filter = Filter().authors(authors).pubkey(pk).kinds([EventDefinitions.KIND_NIP90_DVM_SUBSCRIPTION]).since( + dvm_filter = Filter().authors(authors).pubkey(pk).kinds( + [EventDefinitions.KIND_NIP90_DVM_SUBSCRIPTION]).since( Timestamp.now()) else: # or we don't @@ -68,25 +61,41 @@ class Subscription: [EventDefinitions.KIND_NIP90_DVM_SUBSCRIPTION]).since( Timestamp.now()) - self.client.subscribe([zap_filter, dvm_filter, cancel_subscription_filter], None) + await self.client.subscribe([zap_filter, dvm_filter, cancel_subscription_filter], None) create_subscription_sql_table(dvm_config.DB) + # This is a simple list just to keep track which events we created and manage, so we don't pay for other requests + def __init__(self, dvm_config, admin_config=None): + self.NAME = "Subscription Handler" + dvm_config.DB = "db/" + "subscriptions" + ".db" + self.dvm_config = dvm_config + nip89config = NIP89Config() + nip89config.NAME = self.NAME + self.dvm_config.NIP89 = nip89config + self.admin_config = admin_config + self.keys = Keys.parse(dvm_config.PRIVATE_KEY) + + asyncio.run(self.init_subscription(dvm_config, admin_config)) + asyncio.run(self.run_subscription(dvm_config)) + + async def run_subscription(self, dvm_config): + class NotificationHandler(HandleNotification): client = self.client dvm_config = self.dvm_config keys = self.keys - def handle(self, relay_url, subscription_id, nostr_event: Event): + async def handle(self, relay_url, subscription_id, nostr_event: Event): if nostr_event.kind() == EventDefinitions.KIND_NIP90_DVM_SUBSCRIPTION: - handle_nwc_request(nostr_event) + await handle_nwc_request(nostr_event) elif nostr_event.kind() == EventDefinitions.KIND_NIP88_STOP_SUBSCRIPTION_EVENT: - handle_cancel(nostr_event) + await handle_cancel(nostr_event) - def handle_msg(self, relay_url, msg): + async def handle_msg(self, relay_url, msg): return - def handle_cancel(nostr_event): + async def handle_cancel(nostr_event): print(nostr_event.as_json()) sender = nostr_event.author().to_hex() kind7001eventid = "" @@ -127,7 +136,7 @@ class Subscription: end = start + 60 * 60 * 24 * 356 return end - def send_status_success(original_event, domain): + async def send_status_success(original_event, domain): e_tag = Tag.parse(["e", original_event.id().to_hex()]) p_tag = Tag.parse(["p", original_event.author().to_hex()]) @@ -151,11 +160,11 @@ class Subscription: keys = Keys.parse(dvm_config.PRIVATE_KEY) reaction_event = EventBuilder(EventDefinitions.KIND_FEEDBACK, str(content), reply_tags).to_event(keys) - send_event(reaction_event, client=self.client, dvm_config=self.dvm_config) + await send_event(reaction_event, client=self.client, dvm_config=self.dvm_config) print("[" + self.dvm_config.NIP89.NAME + "]" + ": Sent Kind " + str( EventDefinitions.KIND_FEEDBACK.as_u64()) + " Reaction: " + "success" + " " + reaction_event.as_json()) - def pay_zap_split(nwc, overall_amount, zaps, tier, unit="msats"): + async def pay_zap_split(nwc, overall_amount, zaps, tier, unit="msats"): overallsplit = 0 for zap in zaps: @@ -166,10 +175,10 @@ class Subscription: zapped_amount = 0 for zap in zaps: if zap[1] == "": - #If the client did decide to not add itself to the zap split, or if a slot is left we add the subscription service in the empty space + # If the client did decide to not add itself to the zap split, or if a slot is left we add the subscription service in the empty space zap[1] = Keys.parse(self.dvm_config.PRIVATE_KEY).public_key().to_hex() - name, nip05, lud16 = fetch_user_metadata(zap[1], self.client) + name, nip05, lud16 = await fetch_user_metadata(zap[1], self.client) splitted_amount = math.floor( (int(zap[3]) / overallsplit) * int(overall_amount) / 1000) # invoice = create_bolt11_lud16(lud16, splitted_amount) @@ -187,7 +196,7 @@ class Subscription: print(str(zapped_amount) + "/" + str(overall_amount)) if zapped_amount < overall_amount * 0.8: # TODO how do we handle failed zaps for some addresses? we are ok with 80% for now - #if zapped_amount < int(zaps[0][3]): + # if zapped_amount < int(zaps[0][3]): success = False else: @@ -196,7 +205,7 @@ class Subscription: return success - def make_subscription_zap_recipe(event7001, recipient, subscriber, start, end, tier_dtag): + async def make_subscription_zap_recipe(event7001, recipient, subscriber, start, end, tier_dtag): message = "payed by subscription service" pTag = Tag.parse(["p", recipient]) PTag = Tag.parse(["P", subscriber]) @@ -214,13 +223,14 @@ class Subscription: signer = NostrSigner.keys(self.keys) client = Client(signer) for relay in dvmconfig.RELAY_LIST: - client.add_relay(relay) - client.connect() - recipeid = client.send_event(event) + await client.add_relay(relay) + await client.connect() + recipeid = await client.send_event(event) + await client.disconnect() recipe = recipeid.to_hex() return recipe - def handle_nwc_request(nostr_event): + async def handle_nwc_request(nostr_event): print(nostr_event.as_json()) sender = nostr_event.author().to_hex() if sender == self.keys.public_key().to_hex(): @@ -228,6 +238,8 @@ class Subscription: try: decrypted_text = nip04_decrypt(self.keys.secret_key(), nostr_event.author(), nostr_event.content()) + subscriber = "" + nwc = "" try: jsonevent = json.loads(decrypted_text) for entry in jsonevent: @@ -238,7 +250,7 @@ class Subscription: subscriptionfilter = Filter().kind(EventDefinitions.KIND_NIP88_SUBSCRIBE_EVENT).author( PublicKey.parse(subscriber)).limit(1) - evts = self.client.get_events_of([subscriptionfilter], timedelta(seconds=3)) + evts = await self.client.get_events_of([subscriptionfilter], timedelta(seconds=3)) print(evts) if len(evts) > 0: event7001id = evts[0].id().to_hex() @@ -250,6 +262,7 @@ class Subscription: zaps = [] tier = "DVM" overall_amount = 0 + subscription_event_id = "" for tag in evts[0].tags(): if tag.as_vec()[0] == "amount": overall_amount = int(tag.as_vec()[1]) @@ -315,7 +328,7 @@ class Subscription: print("updated subscription entry before payment") # we attempt to pay the subscription - success = pay_zap_split(nwc, overall_amount, zaps, tier, unit) + success = await pay_zap_split(nwc, overall_amount, zaps, tier, unit) else: start = Timestamp.now().as_secs() @@ -326,8 +339,8 @@ class Subscription: if success: # we create a payment recipe - recipe = make_subscription_zap_recipe(event7001id, recipient, subscriber, start, end, - tier_dtag) + recipe = await make_subscription_zap_recipe(event7001id, recipient, subscriber, start, end, + tier_dtag) print("RECIPE " + recipe) isactivesubscription = True @@ -338,14 +351,15 @@ class Subscription: Timestamp.now().as_secs(), tier) print("updated subscription entry after payment") - send_status_success(nostr_event, "noogle.lol") + await send_status_success(nostr_event, "noogle.lol") keys = Keys.parse(dvm_config.PRIVATE_KEY) message = ("Subscribed to DVM " + tier + ". Renewing on: " + str( - Timestamp.from_secs(end).to_human_datetime().replace("Z", " ").replace("T", " ") + " GMT")) + Timestamp.from_secs(end).to_human_datetime().replace("Z", " ").replace("T", + " ") + " GMT")) evt = EventBuilder.encrypted_direct_msg(keys, PublicKey.parse(subscriber), message, None).to_event(keys) - send_event(evt, client=self.client, dvm_config=dvm_config) + await send_event(evt, client=self.client, dvm_config=dvm_config) @@ -370,15 +384,15 @@ class Subscription: delete_from_subscription_sql_table(dvm_config.DB, subscription.id) print("Delete expired subscription") - def handle_subscription_renewal(subscription): + async def handle_subscription_renewal(subscription): zaps = json.loads(subscription.zaps) - success = pay_zap_split(subscription.nwc, subscription.amount, zaps, subscription.tier, - subscription.unit) + success = await pay_zap_split(subscription.nwc, subscription.amount, zaps, subscription.tier, + subscription.unit) if success: end = infer_subscription_end_time(Timestamp.now().as_secs(), subscription.cadence) - recipe = make_subscription_zap_recipe(subscription.id, subscription.recipent, - subscription.subscriber, subscription.begin, - end, subscription.tier_dtag) + recipe = await make_subscription_zap_recipe(subscription.id, subscription.recipent, + subscription.subscriber, subscription.begin, + end, subscription.tier_dtag) else: end = Timestamp.now().as_secs() recipe = subscription.recipe @@ -403,9 +417,9 @@ class Subscription: evt = EventBuilder.encrypted_direct_msg(keys, PublicKey.parse(subscription.subscriber), message, None).to_event(keys) - send_event(evt, client=self.client, dvm_config=dvm_config) + await send_event(evt, client=self.client, dvm_config=dvm_config) - def check_subscriptions(): + async def check_subscriptions(): try: subscriptions = get_all_subscriptions_from_sql_table(dvm_config.DB) @@ -430,7 +444,7 @@ class Subscription: False, Timestamp.now().as_secs(), subscription.tier) else: - handle_subscription_renewal(subscription) + await handle_subscription_renewal(subscription) else: handle_expired_subscription(subscription) @@ -440,12 +454,12 @@ class Subscription: except Exception as e: print(e) - self.client.handle_notifications(NotificationHandler()) + await self.client.handle_notifications(NotificationHandler()) try: while True: time.sleep(60.0) - check_subscriptions() + await check_subscriptions() except KeyboardInterrupt: print('Stay weird!') os.kill(os.getpid(), signal.SIGTERM) diff --git a/nostr_dvm/tasks/advanced_search.py b/nostr_dvm/tasks/advanced_search.py index a86c8cd..7990d6e 100644 --- a/nostr_dvm/tasks/advanced_search.py +++ b/nostr_dvm/tasks/advanced_search.py @@ -25,11 +25,9 @@ class AdvancedSearch(DVMTaskInterface): FIX_COST: float = 0 dvm_config: DVMConfig - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -54,7 +52,7 @@ class AdvancedSearch(DVMTaskInterface): search = "" max_results = 100 relay = "wss://relay.nostr.band" - + for tag in event.tags(): if tag.as_vec()[0] == 'i': input_type = tag.as_vec()[2] @@ -63,7 +61,7 @@ class AdvancedSearch(DVMTaskInterface): elif tag.as_vec()[0] == 'param': param = tag.as_vec()[1] if param == "user": # check for param type - #user = tag.as_vec()[2] + # user = tag.as_vec()[2] users.append(Tag.parse(["p", tag.as_vec()[2]])) elif param == "users": # check for param type users = json.loads(tag.as_vec()[2]) @@ -85,7 +83,7 @@ class AdvancedSearch(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from nostr_sdk import Filter options = self.set_options(request_form) @@ -96,24 +94,24 @@ class AdvancedSearch(DVMTaskInterface): cli = Client.with_opts(signer, opts) ropts = RelayOptions().ping(False) - cli.add_relay_with_opts(options["relay"], ropts) + await cli.add_relay_with_opts(options["relay"], ropts) - cli.connect() + await cli.connect() - #earch_since_seconds = int(options["since"]) * 24 * 60 * 60 - #dif = Timestamp.now().as_secs() - search_since_seconds - #search_since = Timestamp.from_secs(dif) + # earch_since_seconds = int(options["since"]) * 24 * 60 * 60 + # dif = Timestamp.now().as_secs() - search_since_seconds + # search_since = Timestamp.from_secs(dif) search_since = Timestamp.from_secs(int(options["since"])) - #search_until_seconds = int(options["until"]) * 24 * 60 * 60 - #dif = Timestamp.now().as_secs() - search_until_seconds - #search_until = Timestamp.from_secs(dif) + # search_until_seconds = int(options["until"]) * 24 * 60 * 60 + # dif = Timestamp.now().as_secs() - search_until_seconds + # search_until = Timestamp.from_secs(dif) search_until = Timestamp.from_secs(int(options["until"])) userkeys = [] for user in options["users"]: tag = Tag.parse(user) user = tag.as_vec()[1] - #user = user[1] + # user = user[1] user = str(user).lstrip("@") if str(user).startswith('npub'): userkey = PublicKey.from_bech32(user) @@ -125,25 +123,25 @@ class AdvancedSearch(DVMTaskInterface): userkeys.append(userkey) if not options["users"]: - notes_filter = Filter().kind(Kind(1)).search(options["search"]).since(search_since).until(search_until).limit(options["max_results"]) + notes_filter = Filter().kind(Kind(1)).search(options["search"]).since(search_since).until( + search_until).limit(options["max_results"]) elif options["search"] == "": - notes_filter = Filter().kind(Kind(1)).authors(userkeys).since(search_since).until( - search_until).limit(options["max_results"]) + notes_filter = Filter().kind(Kind(1)).authors(userkeys).since(search_since).until( + search_until).limit(options["max_results"]) else: - notes_filter = Filter().kind(Kind(1)).authors(userkeys).search(options["search"]).since( - search_since).until(search_until).limit(options["max_results"]) + notes_filter = Filter().kind(Kind(1)).authors(userkeys).search(options["search"]).since( + search_since).until(search_until).limit(options["max_results"]) - - events = cli.get_events_of([notes_filter], timedelta(seconds=5)) + events = await cli.get_events_of([notes_filter], timedelta(seconds=5)) result_list = [] if len(events) > 0: for event in events: e_tag = Tag.parse(["e", event.id().to_hex()]) - #print(e_tag.as_vec()) result_list.append(e_tag.as_vec()) + await cli.shutdown() return json.dumps(result_list) def post_process(self, result, event): diff --git a/nostr_dvm/tasks/advanced_search_wine.py b/nostr_dvm/tasks/advanced_search_wine.py index f1089f9..1bb9060 100644 --- a/nostr_dvm/tasks/advanced_search_wine.py +++ b/nostr_dvm/tasks/advanced_search_wine.py @@ -27,11 +27,9 @@ class AdvancedSearchWine(DVMTaskInterface): FIX_COST: float = 0 dvm_config: DVMConfig - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -56,7 +54,6 @@ class AdvancedSearchWine(DVMTaskInterface): search = "" max_results = 100 - for tag in event.tags(): if tag.as_vec()[0] == 'i': input_type = tag.as_vec()[2] @@ -87,7 +84,7 @@ class AdvancedSearchWine(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from nostr_sdk import Filter options = self.set_options(request_form) userkeys = [] @@ -106,13 +103,17 @@ class AdvancedSearchWine(DVMTaskInterface): print("Sending job to nostr.wine API") if options["users"]: - url = ('https://api.nostr.wine/search?query=' + options["search"] + '&kind=1' + '&pubkey=' + options["users"][0][1] + "&limit=100" + "&sort=time" + "&until=" + str(options["until"]) + "&since=" + str(options["since"])) + url = ('https://api.nostr.wine/search?query=' + options["search"] + '&kind=1' + '&pubkey=' + + options["users"][0][1] + "&limit=100" + "&sort=time" + "&until=" + str( + options["until"]) + "&since=" + str(options["since"])) else: - url = ('https://api.nostr.wine/search?query=' + options["search"] + '&kind=1' + "&limit=100" + "&sort=time" + "&until=" + str(options["until"]) + "&since=" + str(options["since"])) + url = ('https://api.nostr.wine/search?query=' + options[ + "search"] + '&kind=1' + "&limit=100" + "&sort=time" + "&until=" + str( + options["until"]) + "&since=" + str(options["since"])) headers = {'Content-type': 'application/x-www-form-urlencoded'} response = requests.get(url, headers=headers) - #print(response.text) + # print(response.text) result_list = [] try: ob = json.loads(response.text) @@ -126,7 +127,6 @@ class AdvancedSearchWine(DVMTaskInterface): except Exception as e: print(e) - return json.dumps(result_list) def post_process(self, result, event): @@ -183,7 +183,7 @@ def build_example(name, identifier, admin_config): nip89config.CONTENT = json.dumps(nip89info) return AdvancedSearchWine(name=name, dvm_config=dvm_config, nip89config=nip89config, - admin_config=admin_config) + admin_config=admin_config) if __name__ == '__main__': diff --git a/nostr_dvm/tasks/audiogeneration_suno_ai.py b/nostr_dvm/tasks/audiogeneration_suno_ai.py index 6f19f83..688dd0a 100644 --- a/nostr_dvm/tasks/audiogeneration_suno_ai.py +++ b/nostr_dvm/tasks/audiogeneration_suno_ai.py @@ -30,11 +30,9 @@ class AudioGenerationSonoAI(DVMTaskInterface): TASK: str = "prompt-to-music" FIX_COST: float = 120 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) self.base_url = 'http://localhost:3000' def is_input_supported(self, tags, client=None, dvm_config=None): @@ -100,7 +98,7 @@ class AudioGenerationSonoAI(DVMTaskInterface): response = requests.post(url, json=payload) return response.json() - def process(self, request_form): + async def process(self, request_form): try: options = self.set_options(request_form) has_quota = False diff --git a/nostr_dvm/tasks/content_discovery_currently_popular.py b/nostr_dvm/tasks/content_discovery_currently_popular.py index 1ff1df7..b849d24 100644 --- a/nostr_dvm/tasks/content_discovery_currently_popular.py +++ b/nostr_dvm/tasks/content_discovery_currently_popular.py @@ -1,3 +1,4 @@ +import asyncio import json import os from datetime import timedelta @@ -27,6 +28,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): TASK: str = "discover-content" FIX_COST: float = 0 dvm_config: DVMConfig + request_form = None last_schedule: int db_since = 3600 db_name = "db/nostr_recent_notes.db" @@ -34,12 +36,10 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): personalized = False result = "" - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): - dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): + dvm_config.SCRIPT = os.path.abspath(__file__) self.request_form = {"jobID": "generic"} opts = { "max_results": 200, @@ -53,16 +53,14 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): if self.options.get("db_since"): self.db_since = int(self.options.get("db_since")) - use_logger = False if use_logger: init_logger(LogLevel.DEBUG) if self.dvm_config.UPDATE_DATABASE: - self.sync_db() - + await self.sync_db() if not self.personalized: - self.result = self.calculate_result(self.request_form) + self.result = await self.calculate_result(self.request_form) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -96,7 +94,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): # if the dvm supports individual results, recalculate it every time for the request if self.personalized: return self.calculate_result(request_form) @@ -104,7 +102,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): else: return self.result - def calculate_result(self, request_form): + async def calculate_result(self, request_form): from nostr_sdk import Filter from types import SimpleNamespace ns = SimpleNamespace() @@ -116,9 +114,9 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().database(database).signer(signer).opts(opts).build() - cli.connect() + await cli.connect() # Negentropy reconciliation # Query events from database @@ -126,7 +124,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): since = Timestamp.from_secs(timestamp_hour_ago) filter1 = Filter().kind(definitions.EventDefinitions.KIND_NOTE).since(since) - events = cli.database().query([filter1]) + events = await cli.database().query([filter1]) print("[" + self.dvm_config.NIP89.NAME + "] Considering " + str(len(events)) + " Events") ns.finallist = {} for event in events: @@ -134,11 +132,11 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): filt = Filter().kinds([definitions.EventDefinitions.KIND_ZAP, definitions.EventDefinitions.KIND_REPOST, definitions.EventDefinitions.KIND_REACTION, definitions.EventDefinitions.KIND_NOTE]).event(event.id()).since(since) - reactions = cli.database().query([filt]) + reactions = await cli.database().query([filt]) if len(reactions) >= self.min_reactions: ns.finallist[event.id().to_hex()] = len(reactions) if len(ns.finallist) == 0: - cli.shutdown() + await cli.shutdown() return self.result result_list = [] @@ -147,7 +145,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): # print(EventId.parse(entry[0]).to_bech32() + "/" + EventId.parse(entry[0]).to_hex() + ": " + str(entry[1])) e_tag = Tag.parse(["e", entry[0]]) result_list.append(e_tag.as_vec()) - cli.shutdown() + await cli.shutdown() print("[" + self.dvm_config.NIP89.NAME + "] Filtered " + str( len(result_list)) + " fitting events.") return json.dumps(result_list) @@ -163,29 +161,29 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): # if not text/plain, don't post-process return result - def schedule(self, dvm_config): + async def schedule(self, dvm_config): if dvm_config.SCHEDULE_UPDATES_SECONDS == 0: return 0 else: if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS: if self.dvm_config.UPDATE_DATABASE: - self.sync_db() + await self.sync_db() self.last_schedule = Timestamp.now().as_secs() - self.result = self.calculate_result(self.request_form) + self.result = await self.calculate_result(self.request_form) return 1 - def sync_db(self): + async def sync_db(self): opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT))) sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY) keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().signer(signer).database(database).opts(opts).build() for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST: - cli.add_relay(relay) + await cli.add_relay(relay) - cli.connect() + await cli.connect() timestamp_since = Timestamp.now().as_secs() - self.db_since since = Timestamp.from_secs(timestamp_since) @@ -197,10 +195,10 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str( self.db_since) + " seconds.. this might take a while..") dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN) - cli.reconcile(filter1, dbopts) - cli.database().delete(Filter().until(Timestamp.from_secs( + await cli.reconcile(filter1, dbopts) + await cli.database().delete(Filter().until(Timestamp.from_secs( Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full. - cli.shutdown() + await cli.shutdown() print( "[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..") @@ -208,7 +206,8 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface): # We build an example here that we can call by either calling this file directly from the main directory, # or by adding it to our playground. You can call the example and adjust it to your needs or redefine it in the # playground or elsewhere -def build_example(name, identifier, admin_config, options, cost=0, update_rate=180, processing_msg=None, update_db=True): +def build_example(name, identifier, admin_config, options, cost=0, update_rate=180, processing_msg=None, + update_db=True): dvm_config = build_default_config(identifier) dvm_config.USE_OWN_VENV = False dvm_config.SHOWLOG = True @@ -247,14 +246,15 @@ def build_example(name, identifier, admin_config, options, cost=0, update_rate= nip89config.DTAG = check_and_set_d_tag(identifier, name, dvm_config.PRIVATE_KEY, nip89info["image"]) nip89config.CONTENT = json.dumps(nip89info) - #admin_config.UPDATE_PROFILE = False - #admin_config.REBROADCAST_NIP89 = False + # admin_config.UPDATE_PROFILE = False + # admin_config.REBROADCAST_NIP89 = False return DicoverContentCurrentlyPopular(name=name, dvm_config=dvm_config, nip89config=nip89config, admin_config=admin_config, options=options) -def build_example_subscription(name, identifier, admin_config, options, update_rate=180, processing_msg=None, update_db=True): +def build_example_subscription(name, identifier, admin_config, options, update_rate=180, processing_msg=None, + update_db=True): dvm_config = build_default_config(identifier) dvm_config.USE_OWN_VENV = False dvm_config.SHOWLOG = True @@ -304,9 +304,9 @@ def build_example_subscription(name, identifier, admin_config, options, update_r nip88config.PERK2DESC = "Support NostrDVM & NostrSDK development" nip88config.PAYMENT_VERIFIER_PUBKEY = "5b5c045ecdf66fb540bdf2049fe0ef7f1a566fa427a4fe50d400a011b65a3a7e" - #admin_config.UPDATE_PROFILE = False - #admin_config.REBROADCAST_NIP89 = False - #admin_config.REBROADCAST_NIP88 = False + # admin_config.UPDATE_PROFILE = False + # admin_config.REBROADCAST_NIP89 = False + # admin_config.REBROADCAST_NIP88 = False # admin_config.FETCH_NIP88 = True # admin_config.EVENTID = "" @@ -318,4 +318,4 @@ def build_example_subscription(name, identifier, admin_config, options, update_r if __name__ == '__main__': - process_venv(DicoverContentCurrentlyPopular) \ No newline at end of file + process_venv(DicoverContentCurrentlyPopular) diff --git a/nostr_dvm/tasks/content_discovery_currently_popular_by_top_zaps.py b/nostr_dvm/tasks/content_discovery_currently_popular_by_top_zaps.py index 5bf1622..dc2d2c6 100644 --- a/nostr_dvm/tasks/content_discovery_currently_popular_by_top_zaps.py +++ b/nostr_dvm/tasks/content_discovery_currently_popular_by_top_zaps.py @@ -1,3 +1,4 @@ +import asyncio import json import os from datetime import timedelta @@ -28,6 +29,7 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): TASK: str = "discover-content" FIX_COST: float = 0 dvm_config: DVMConfig + request_form = None last_schedule: int db_since = 3600 db_name = "db/nostr_recent_notes.db" @@ -36,11 +38,8 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): result = "" logger = False - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): - dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): self.request_form = {"jobID": "generic"} opts = { @@ -60,12 +59,10 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): if self.logger: init_logger(LogLevel.DEBUG) - if self.dvm_config.UPDATE_DATABASE: - self.sync_db() - + await self.sync_db() if not self.personalized: - self.result = self.calculate_result(self.request_form) + self.result = await self.calculate_result(self.request_form) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -98,7 +95,7 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): # if the dvm supports individual results, recalculate it every time for the request if self.personalized: return self.calculate_result(request_form) @@ -106,7 +103,7 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): else: return self.result - def calculate_result(self, request_form): + async def calculate_result(self, request_form): from nostr_sdk import Filter from types import SimpleNamespace ns = SimpleNamespace() @@ -118,10 +115,10 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().database(database).signer(signer).opts(opts).build() - cli.connect() + await cli.connect() # Negentropy reconciliation # Query events from database @@ -129,14 +126,14 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): since = Timestamp.from_secs(timestamp_hour_ago) filter1 = Filter().kind(definitions.EventDefinitions.KIND_NOTE).since(since) - events = cli.database().query([filter1]) + events = await cli.database().query([filter1]) print("[" + self.dvm_config.NIP89.NAME + "] Considering " + str(len(events)) + " Events") ns.finallist = {} for event in events: if event.created_at().as_secs() > timestamp_hour_ago: filt = Filter().kinds([definitions.EventDefinitions.KIND_ZAP]).event(event.id()).since(since) - reactions = cli.database().query([filt]) + reactions = await cli.database().query([filt]) invoice_amount = 0 haspreimage = False if len(reactions) >= self.min_reactions: @@ -162,11 +159,11 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): e_tag = Tag.parse(["e", entry[0]]) result_list.append(e_tag.as_vec()) - print("[" + self.dvm_config.NIP89.NAME+ "] Filtered " + str( + print("[" + self.dvm_config.NIP89.NAME + "] Filtered " + str( len(result_list)) + " fitting events.") - cli.disconnect() - cli.shutdown() + await cli.disconnect() + await cli.shutdown() return json.dumps(result_list) @@ -181,33 +178,33 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): # if not text/plain, don't post-process return result - def schedule(self, dvm_config): + async def schedule(self, dvm_config): if dvm_config.SCHEDULE_UPDATES_SECONDS == 0: return 0 else: if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS: try: if self.dvm_config.UPDATE_DATABASE: - self.sync_db() + await self.sync_db() self.last_schedule = Timestamp.now().as_secs() if not self.personalized: - self.result = self.calculate_result(self.request_form) + self.result = await self.calculate_result(self.request_form) except Exception as e: print(e) return 1 - def sync_db(self): + async def sync_db(self): opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT))) sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY) keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().signer(signer).database(database).opts(opts).build() for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST: - cli.add_relay(relay) + await cli.add_relay(relay) - cli.connect() + await cli.connect() timestamp_since = Timestamp.now().as_secs() - self.db_since since = Timestamp.from_secs(timestamp_since) @@ -219,10 +216,10 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface): print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str( self.db_since) + " seconds.. this might take a while..") dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN) - cli.reconcile(filter1, dbopts) - cli.database().delete(Filter().until(Timestamp.from_secs( + await cli.reconcile(filter1, dbopts) + await cli.database().delete(Filter().until(Timestamp.from_secs( Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full. - cli.shutdown() + await cli.shutdown() print( "[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..") diff --git a/nostr_dvm/tasks/content_discovery_currently_popular_followers.py b/nostr_dvm/tasks/content_discovery_currently_popular_followers.py index 1f0f874..3535d80 100644 --- a/nostr_dvm/tasks/content_discovery_currently_popular_followers.py +++ b/nostr_dvm/tasks/content_discovery_currently_popular_followers.py @@ -1,3 +1,4 @@ +import asyncio import json import os from datetime import timedelta @@ -32,11 +33,8 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): db_name = "db/nostr_recent_notes2.db" min_reactions = 2 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): - dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): self.last_schedule = Timestamp.now().as_secs() @@ -50,7 +48,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): init_logger(LogLevel.DEBUG) if self.dvm_config.UPDATE_DATABASE: - self.sync_db() + await self.sync_db() def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -90,7 +88,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from nostr_sdk import Filter from types import SimpleNamespace ns = SimpleNamespace() @@ -102,20 +100,20 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().database(database).signer(signer).opts(opts).build() - cli.add_relay("wss://relay.damus.io") - cli.add_relay("wss://nostr.oxtr.dev") - cli.add_relay("wss://nostr.mom") + await cli.add_relay("wss://relay.damus.io") + await cli.add_relay("wss://nostr.oxtr.dev") + await cli.add_relay("wss://nostr.mom") #ropts = RelayOptions().ping(False) #cli.add_relay_with_opts("wss://nostr.band", ropts) - cli.connect() + await cli.connect() user = PublicKey.parse(options["user"]) followers_filter = Filter().author(user).kinds([Kind(3)]) - followers = cli.get_events_of([followers_filter], timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)) + followers = await cli.get_events_of([followers_filter], timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)) #print(followers) # Negentropy reconciliation @@ -142,7 +140,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): followings.append(following) filter1 = Filter().kind(definitions.EventDefinitions.KIND_NOTE).authors(followings).since(since) - events = cli.database().query([filter1]) + events = await cli.database().query([filter1]) print("[" + self.dvm_config.NIP89.NAME + "] Considering " + str(len(events)) + " Events") ns.finallist = {} @@ -151,7 +149,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): filt = Filter().kinds( [definitions.EventDefinitions.KIND_ZAP, definitions.EventDefinitions.KIND_REACTION, definitions.EventDefinitions.KIND_REPOST, definitions.EventDefinitions.KIND_NOTE]).event(event.id()).since(since) - reactions = cli.database().query([filt]) + reactions = await cli.database().query([filt]) if len(reactions) >= self.min_reactions: ns.finallist[event.id().to_hex()] = len(reactions) @@ -162,8 +160,8 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): # print(EventId.parse(entry[0]).to_bech32() + "/" + EventId.parse(entry[0]).to_hex() + ": " + str(entry[1])) e_tag = Tag.parse(["e", entry[0]]) result_list.append(e_tag.as_vec()) - cli.connect() - cli.shutdown() + await cli.connect() + await cli.shutdown() print("[" + self.dvm_config.NIP89.NAME + "] Filtered " + str( len(result_list)) + " fitting events.") @@ -180,7 +178,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): # if not text/plain, don't post-process return result - def schedule(self, dvm_config): + async def schedule(self, dvm_config): if dvm_config.SCHEDULE_UPDATES_SECONDS == 0: return 0 # We simply use the db from the other dvm that contains all notes @@ -188,23 +186,23 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): else: if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS: if self.dvm_config.UPDATE_DATABASE: - self.sync_db() + await self.sync_db() self.last_schedule = Timestamp.now().as_secs() return 1 - def sync_db(self): + async def sync_db(self): opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT))) sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY) keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().signer(signer).database(database).opts(opts).build() for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST: - cli.add_relay(relay) + await cli.add_relay(relay) - cli.connect() + await cli.connect() timestamp_since = Timestamp.now().as_secs() - self.db_since since = Timestamp.from_secs(timestamp_since) @@ -217,10 +215,10 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface): print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str( self.db_since) + " seconds.. this might take a while..") dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN) - cli.reconcile(filter1, dbopts) - cli.database().delete(Filter().until(Timestamp.from_secs( + await cli.reconcile(filter1, dbopts) + await cli.database().delete(Filter().until(Timestamp.from_secs( Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full. - cli.shutdown() + await cli.shutdown() print("[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str( self.db_since) + " seconds..") diff --git a/nostr_dvm/tasks/content_discovery_currently_popular_topic.py b/nostr_dvm/tasks/content_discovery_currently_popular_topic.py index 480c124..4d83d0c 100644 --- a/nostr_dvm/tasks/content_discovery_currently_popular_topic.py +++ b/nostr_dvm/tasks/content_discovery_currently_popular_topic.py @@ -1,3 +1,4 @@ +import asyncio import json import os from datetime import timedelta @@ -26,9 +27,10 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): TASK: str = "discover-content" FIX_COST: float = 0 dvm_config: DVMConfig + request_form = None last_schedule: int min_reactions = 2 - db_since = 10*3600 + db_since = 10 * 3600 db_name = "db/nostr_default_recent_notes.db" search_list = [] avoid_list = [] @@ -36,15 +38,9 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): personalized = False result = "" - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): - - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) - - # Generate Generic request form for dvms that provide generic results (e.g only a calculation per update, - # not per call) self.request_form = {"jobID": "generic"} opts = { "max_results": 200, @@ -58,7 +54,7 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): self.last_schedule = Timestamp.now().as_secs() if self.options.get("search_list"): self.search_list = self.options.get("search_list") - #print(self.search_list) + # print(self.search_list) if self.options.get("avoid_list"): self.avoid_list = self.options.get("avoid_list") if self.options.get("must_list"): @@ -68,15 +64,14 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): if self.options.get("db_since"): self.db_since = int(self.options.get("db_since")) - use_logger = False if use_logger: init_logger(LogLevel.DEBUG) if self.dvm_config.UPDATE_DATABASE: - self.sync_db() + await self.sync_db() if not self.personalized: - self.result = self.calculate_result(self.request_form) + self.result = await self.calculate_result(self.request_form) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -110,15 +105,14 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): self.request_form = request_form return request_form - def process(self, request_form): + async def process(self, request_form): # if the dvm supports individual results, recalculate it every time for the request if self.personalized: - return self.calculate_result(request_form) - #else return the result that gets updated once every schenduled update. In this case on database update. + return await self.calculate_result(request_form) + # else return the result that gets updated once every schenduled update. In this case on database update. else: return self.result - def post_process(self, result, event): """Overwrite the interface function to return a social client readable format, if requested""" for tag in event.tags(): @@ -129,7 +123,8 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): # if not text/plain, don't post-process return result - def calculate_result(self, request_form): + + async def calculate_result(self, request_form): from nostr_sdk import Filter from types import SimpleNamespace ns = SimpleNamespace() @@ -141,10 +136,10 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().database(database).signer(signer).opts(opts).build() - cli.connect() + await cli.connect() # Negentropy reconciliation # Query events from database @@ -153,7 +148,7 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): filter1 = Filter().kind(definitions.EventDefinitions.KIND_NOTE).since(since) - events = cli.database().query([filter1]) + events = await cli.database().query([filter1]) print("[" + self.dvm_config.NIP89.NAME + "] Considering " + str(len(events)) + " Events") ns.finallist = {} @@ -166,7 +161,7 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): [definitions.EventDefinitions.KIND_ZAP, definitions.EventDefinitions.KIND_REACTION, definitions.EventDefinitions.KIND_REPOST, definitions.EventDefinitions.KIND_NOTE]).event(event.id()).since(since) - reactions = cli.database().query([filt]) + reactions = await cli.database().query([filt]) if len(reactions) >= self.min_reactions: ns.finallist[event.id().to_hex()] = len(reactions) @@ -179,56 +174,57 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface): print("[" + self.dvm_config.NIP89.NAME + "] Filtered " + str( len(result_list)) + " fitting events.") - cli.shutdown() + await cli.shutdown() return json.dumps(result_list) - - def schedule(self, dvm_config): + async def schedule(self, dvm_config): if dvm_config.SCHEDULE_UPDATES_SECONDS == 0: return 0 else: if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS: if self.dvm_config.UPDATE_DATABASE: - self.sync_db() + await self.sync_db() self.last_schedule = Timestamp.now().as_secs() - self.result = self.calculate_result(self.request_form) - - + self.result = await self.calculate_result(self.request_form) return 1 - def sync_db(self): + async def sync_db(self): opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT))) sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY) keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().signer(signer).database(database).opts(opts).build() for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST: - cli.add_relay(relay) + await cli.add_relay(relay) - cli.connect() + await cli.connect() timestamp_since = Timestamp.now().as_secs() - self.db_since since = Timestamp.from_secs(timestamp_since) - filter1 = Filter().kinds([definitions.EventDefinitions.KIND_NOTE, definitions.EventDefinitions.KIND_REACTION, definitions.EventDefinitions.KIND_ZAP]).since(since) # Notes, reactions, zaps + filter1 = Filter().kinds([definitions.EventDefinitions.KIND_NOTE, definitions.EventDefinitions.KIND_REACTION, + definitions.EventDefinitions.KIND_ZAP]).since(since) # Notes, reactions, zaps # filter = Filter().author(keys.public_key()) - print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str(self.db_since) + " seconds.. this might take a while..") + print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str( + self.db_since) + " seconds.. this might take a while..") dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN) - cli.reconcile(filter1, dbopts) - cli.database().delete(Filter().until(Timestamp.from_secs( + await cli.reconcile(filter1, dbopts) + await cli.database().delete(Filter().until(Timestamp.from_secs( Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full. - cli.shutdown() - print("[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..") + await cli.shutdown() + print( + "[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..") # We build an example here that we can call by either calling this file directly from the main directory, # or by adding it to our playground. You can call the example and adjust it to your needs or redefine it in the # playground or elsewhere -def build_example(name, identifier, admin_config, options, image, description, update_rate=600, cost=0, processing_msg=None, update_db=True): +def build_example(name, identifier, admin_config, options, image, description, update_rate=600, cost=0, + processing_msg=None, update_db=True): dvm_config = build_default_config(identifier) dvm_config.USE_OWN_VENV = False dvm_config.SHOWLOG = True @@ -241,7 +237,6 @@ def build_example(name, identifier, admin_config, options, image, description, u dvm_config.CUSTOM_PROCESSING_MESSAGE = processing_msg admin_config.LUD16 = dvm_config.LN_ADDRESS - # Add NIP89 nip89info = { "name": name, @@ -267,10 +262,11 @@ def build_example(name, identifier, admin_config, options, image, description, u nip89config.CONTENT = json.dumps(nip89info) return DicoverContentCurrentlyPopularbyTopic(name=name, dvm_config=dvm_config, nip89config=nip89config, - admin_config=admin_config, options=options) + admin_config=admin_config, options=options) -def build_example_subscription(name, identifier, admin_config, options, image, description, processing_msg=None, update_db=True): +def build_example_subscription(name, identifier, admin_config, options, image, description, processing_msg=None, + update_db=True): dvm_config = build_default_config(identifier) dvm_config.USE_OWN_VENV = False dvm_config.SHOWLOG = True @@ -281,7 +277,6 @@ def build_example_subscription(name, identifier, admin_config, options, image, d dvm_config.CUSTOM_PROCESSING_MESSAGE = processing_msg admin_config.LUD16 = dvm_config.LN_ADDRESS - # Add NIP89 nip89info = { "name": name, @@ -319,17 +314,15 @@ def build_example_subscription(name, identifier, admin_config, options, image, d nip88config.PERK2DESC = "Support NostrDVM & NostrSDK development" nip88config.PAYMENT_VERIFIER_PUBKEY = "5b5c045ecdf66fb540bdf2049fe0ef7f1a566fa427a4fe50d400a011b65a3a7e" - - # admin_config.FETCH_NIP88 = True - # admin_config.EVENTID = "63a791cdc7bf78c14031616963105fce5793f532bb231687665b14fb6d805fdb" - # admin_config.PRIVKEY = dvm_config.PRIVATE_KEY - + # admin_config.FETCH_NIP88 = True + # admin_config.EVENTID = "63a791cdc7bf78c14031616963105fce5793f532bb231687665b14fb6d805fdb" + # admin_config.PRIVKEY = dvm_config.PRIVATE_KEY return DicoverContentCurrentlyPopularbyTopic(name=name, dvm_config=dvm_config, nip89config=nip89config, - nip88config=nip88config, - admin_config=admin_config, - options=options) + nip88config=nip88config, + admin_config=admin_config, + options=options) if __name__ == '__main__': - process_venv(DicoverContentCurrentlyPopularbyTopic) \ No newline at end of file + process_venv(DicoverContentCurrentlyPopularbyTopic) diff --git a/nostr_dvm/tasks/content_discovery_update_db_only.py b/nostr_dvm/tasks/content_discovery_update_db_only.py index a267abf..9af0d12 100644 --- a/nostr_dvm/tasks/content_discovery_update_db_only.py +++ b/nostr_dvm/tasks/content_discovery_update_db_only.py @@ -1,3 +1,4 @@ +import asyncio import json import os from datetime import timedelta @@ -26,6 +27,7 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface): TASK: str = "update-db-on-schedule" FIX_COST: float = 0 dvm_config: DVMConfig + request_form = None last_schedule: int min_reactions = 2 db_since = 10 * 3600 @@ -36,11 +38,8 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface): personalized = False result = "" - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): - - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): # Generate Generic request form for dvms that provide generic results (e.g only a calculation per update, # not per call) @@ -72,7 +71,7 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface): init_logger(LogLevel.DEBUG) if self.dvm_config.UPDATE_DATABASE: - self.sync_db() + await self.sync_db() def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -106,7 +105,7 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface): self.request_form = request_form return request_form - def process(self, request_form): + async def process(self, request_form): return "I don't return results, I just update the DB." def post_process(self, result, event): @@ -120,28 +119,28 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface): # if not text/plain, don't post-process return result - def schedule(self, dvm_config): + async def schedule(self, dvm_config): if dvm_config.SCHEDULE_UPDATES_SECONDS == 0: return 0 else: if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS: if self.dvm_config.UPDATE_DATABASE: - self.sync_db() + await self.sync_db() self.last_schedule = Timestamp.now().as_secs() return 1 - def sync_db(self): + async def sync_db(self): opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT))) sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY) keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().signer(signer).database(database).opts(opts).build() for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST: - cli.add_relay(relay) + await cli.add_relay(relay) - cli.connect() + await cli.connect() timestamp_since = Timestamp.now().as_secs() - self.db_since since = Timestamp.from_secs(timestamp_since) @@ -153,10 +152,10 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface): print("[" + self.dvm_config.IDENTIFIER + "] Syncing notes of the last " + str( self.db_since) + " seconds.. this might take a while..") dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN) - cli.reconcile(filter1, dbopts) - cli.database().delete(Filter().until(Timestamp.from_secs( + await cli.reconcile(filter1, dbopts) + await cli.database().delete(Filter().until(Timestamp.from_secs( Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full. - cli.shutdown() + await cli.shutdown() print( "[" + self.dvm_config.IDENTIFIER + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..") diff --git a/nostr_dvm/tasks/convert_media.py b/nostr_dvm/tasks/convert_media.py index f4c3b2e..ee2a9ed 100644 --- a/nostr_dvm/tasks/convert_media.py +++ b/nostr_dvm/tasks/convert_media.py @@ -27,11 +27,9 @@ class MediaConverter(DVMTaskInterface): FIX_COST = 20 PER_UNIT_COST = 0.1 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -72,7 +70,7 @@ class MediaConverter(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): options = self.set_options(request_form) url = upload_media_to_hoster(options["filepath"]) diff --git a/nostr_dvm/tasks/discovery_bot_farms.py b/nostr_dvm/tasks/discovery_bot_farms.py index 133e641..716f780 100644 --- a/nostr_dvm/tasks/discovery_bot_farms.py +++ b/nostr_dvm/tasks/discovery_bot_farms.py @@ -27,17 +27,10 @@ class DiscoveryBotFarms(DVMTaskInterface): dvm_config: DVMConfig last_schedule: int = 0 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) - - use_logger = False - if use_logger: - init_logger(LogLevel.DEBUG) - - self.sync_db() + await self.sync_db() def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -55,7 +48,7 @@ class DiscoveryBotFarms(DVMTaskInterface): request_form = {"jobID": event.id().to_hex()} # default values - search = "airdrop;just your average nostr enjoyer" #;@nostrich.house; + search = "airdrop;just your average nostr enjoyer" # ;@nostrich.house; max_results = 500 for tag in event.tags(): @@ -75,7 +68,7 @@ class DiscoveryBotFarms(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from nostr_sdk import Filter options = self.set_options(request_form) @@ -84,19 +77,19 @@ class DiscoveryBotFarms(DVMTaskInterface): keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite("db/nostr_profiles.db") + database = await NostrDatabase.sqlite("db/nostr_profiles.db") cli = ClientBuilder().database(database).signer(signer).opts(opts).build() - cli.add_relay("wss://relay.damus.io") + await cli.add_relay("wss://relay.damus.io") # cli.add_relay("wss://atl.purplerelay.com") - cli.connect() + await cli.connect() # Negentropy reconciliation # Query events from database filter1 = Filter().kind(Kind(0)) - events = cli.database().query([filter1]) + events = await cli.database().query([filter1]) # for event in events: # print(event.as_json()) @@ -121,7 +114,7 @@ class DiscoveryBotFarms(DVMTaskInterface): print(str(exp) + " " + event.author().to_hex()) else: break - + await cli.shutdown() return json.dumps(result_list) def post_process(self, result, event): @@ -135,33 +128,33 @@ class DiscoveryBotFarms(DVMTaskInterface): # if not text/plain, don't post-process return result - def schedule(self, dvm_config): + async def schedule(self, dvm_config): if dvm_config.SCHEDULE_UPDATES_SECONDS == 0: return 0 else: if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS: - self.sync_db() + await self.sync_db() self.last_schedule = Timestamp.now().as_secs() return 1 - def sync_db(self): + async def sync_db(self): opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))) sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY) keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite("db/nostr_profiles.db") + database = await NostrDatabase.sqlite("db/nostr_profiles.db") cli = ClientBuilder().signer(signer).database(database).opts(opts).build() - cli.add_relay("wss://relay.damus.io") - cli.add_relay("wss://nostr21.com") - cli.connect() + await cli.add_relay("wss://relay.damus.io") + await cli.add_relay("wss://nostr21.com") + await cli.connect() filter1 = Filter().kind(Kind(0)) # filter = Filter().author(keys.public_key()) print("Syncing Profile Database.. this might take a while..") dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN) - cli.reconcile(filter1, dbopts) + await cli.reconcile(filter1, dbopts) print("Done Syncing Profile Database.") @@ -197,7 +190,7 @@ def build_example(name, identifier, admin_config): options = {"relay": "wss://relay.damus.io"} return DiscoveryBotFarms(name=name, dvm_config=dvm_config, nip89config=nip89config, - admin_config=admin_config, options=options) + admin_config=admin_config, options=options) if __name__ == '__main__': diff --git a/nostr_dvm/tasks/discovery_censor_wot.py b/nostr_dvm/tasks/discovery_censor_wot.py index d51d78a..326b257 100644 --- a/nostr_dvm/tasks/discovery_censor_wot.py +++ b/nostr_dvm/tasks/discovery_censor_wot.py @@ -30,11 +30,10 @@ class DiscoverReports(DVMTaskInterface): client: Client dvm_config: DVMConfig - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + def is_input_supported(self, tags, client=None, dvm_config=None): return True @@ -68,7 +67,7 @@ class DiscoverReports(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from nostr_sdk import Filter from types import SimpleNamespace ns = SimpleNamespace() @@ -85,9 +84,9 @@ class DiscoverReports(DVMTaskInterface): cli.add_relay(relay) # add nostr band, too. ropts = RelayOptions().ping(False) - cli.add_relay_with_opts("wss://nostr.band", ropts) + await cli.add_relay_with_opts("wss://nostr.band", ropts) - cli.connect() + await cli.connect() options = self.set_options(request_form) step = 20 @@ -99,7 +98,7 @@ class DiscoverReports(DVMTaskInterface): # if we don't add users, e.g. by a wot, we check all our followers. if len(pubkeys) == 0: followers_filter = Filter().author(PublicKey.parse(options["sender"])).kind(Kind(3)) - followers = cli.get_events_of([followers_filter], timedelta(seconds=5)) + followers = await cli.get_events_of([followers_filter], timedelta(seconds=5)) if len(followers) > 0: result_list = [] @@ -119,7 +118,7 @@ class DiscoverReports(DVMTaskInterface): ago = Timestamp.now().as_secs() - 60*60*24*int(options["since_days"]) #TODO make this an option, 180 days for now since = Timestamp.from_secs(ago) kind1984_filter = Filter().authors(pubkeys).kind(Kind(1984)).since(since) - reports = cli.get_events_of([kind1984_filter], timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)) + reports = await cli.get_events_of([kind1984_filter], timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)) bad_actors = [] ns.dic = {} @@ -149,6 +148,7 @@ class DiscoverReports(DVMTaskInterface): bad_actors.append(p_tag.as_vec()) print(json.dumps(bad_actors)) + await cli.shutdown() return json.dumps(bad_actors) def post_process(self, result, event): diff --git a/nostr_dvm/tasks/discovery_inactive_follows.py b/nostr_dvm/tasks/discovery_inactive_follows.py index 93a3293..f2f0d24 100644 --- a/nostr_dvm/tasks/discovery_inactive_follows.py +++ b/nostr_dvm/tasks/discovery_inactive_follows.py @@ -30,11 +30,9 @@ class DiscoverInactiveFollows(DVMTaskInterface): client: Client dvm_config: DVMConfig - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): # no input required @@ -64,7 +62,7 @@ class DiscoverInactiveFollows(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from nostr_sdk import Filter from types import SimpleNamespace ns = SimpleNamespace() @@ -82,17 +80,17 @@ class DiscoverInactiveFollows(DVMTaskInterface): cli = Client.with_opts(signer, opts) for relay in self.dvm_config.RELAY_LIST: - cli.add_relay(relay) + await cli.add_relay(relay) ropts = RelayOptions().ping(False) - cli.add_relay_with_opts("wss://nostr.band", ropts) + await cli.add_relay_with_opts("wss://nostr.band", ropts) - cli.connect() + await cli.connect() options = self.set_options(request_form) step = 20 followers_filter = Filter().author(PublicKey.parse(options["user"])).kind(Kind(3)) - followers = cli.get_events_of([followers_filter], timedelta(seconds=5)) + followers = await cli.get_events_of([followers_filter], timedelta(seconds=5)) if len(followers) > 0: @@ -126,7 +124,7 @@ class DiscoverInactiveFollows(DVMTaskInterface): dif = Timestamp.now().as_secs() - not_active_since_seconds not_active_since = Timestamp.from_secs(dif) - def scanList(users, instance, i, st, notactivesince): + async def scanList(users, instance, i, st, notactivesince): from nostr_sdk import Filter keys = Keys.parse(self.dvm_config.PRIVATE_KEY) @@ -135,8 +133,8 @@ class DiscoverInactiveFollows(DVMTaskInterface): signer = NostrSigner.keys(keys) cli = Client.with_opts(signer, opts) for relay in self.dvm_config.RELAY_LIST: - cli.add_relay(relay) - cli.connect() + await cli.add_relay(relay) + await cli.connect() filters = [] for i in range(i, i + st): diff --git a/nostr_dvm/tasks/discovery_nonfollowers.py b/nostr_dvm/tasks/discovery_nonfollowers.py index a0a4abf..bce3be0 100644 --- a/nostr_dvm/tasks/discovery_nonfollowers.py +++ b/nostr_dvm/tasks/discovery_nonfollowers.py @@ -30,11 +30,9 @@ class DiscoverNonFollowers(DVMTaskInterface): client: Client dvm_config: DVMConfig - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): # no input required @@ -59,7 +57,7 @@ class DiscoverNonFollowers(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from nostr_sdk import Filter from types import SimpleNamespace ns = SimpleNamespace() @@ -71,12 +69,12 @@ class DiscoverNonFollowers(DVMTaskInterface): cli = Client.with_opts(signer, opts) # cli.add_relay("wss://relay.nostr.band") for relay in self.dvm_config.RELAY_LIST: - cli.add_relay(relay) + await cli.add_relay(relay) #add nostr band, too. ropts = RelayOptions().ping(False) - cli.add_relay_with_opts("wss://nostr.band", ropts) + await cli.add_relay_with_opts("wss://nostr.band", ropts) - cli.connect() + await cli.connect() options = self.set_options(request_form) step = 20 @@ -103,7 +101,7 @@ class DiscoverNonFollowers(DVMTaskInterface): ns.dic[following] = "True" print("Followings: " + str(len(followings))) - def scanList(users, instance, i, st): + async def scanList(users, instance, i, st): from nostr_sdk import Filter keys = Keys.parse(self.dvm_config.PRIVATE_KEY) @@ -112,14 +110,14 @@ class DiscoverNonFollowers(DVMTaskInterface): signer = NostrSigner.keys(keys) cli = Client.with_opts(signer, opts) for relay in self.dvm_config.RELAY_LIST: - cli.add_relay(relay) + await cli.add_relay(relay) cli.connect() for i in range(i, i + st): filters = [] filter1 = Filter().author(PublicKey.from_hex(users[i])).kind(Kind(3)) filters.append(filter1) - followers = cli.get_events_of(filters, timedelta(seconds=3)) + followers = await cli.get_events_of(filters, timedelta(seconds=3)) if len(followers) > 0: result_list = [] diff --git a/nostr_dvm/tasks/discovery_trending_notes_nostrband.py b/nostr_dvm/tasks/discovery_trending_notes_nostrband.py index 9c1c85e..e9d4970 100644 --- a/nostr_dvm/tasks/discovery_trending_notes_nostrband.py +++ b/nostr_dvm/tasks/discovery_trending_notes_nostrband.py @@ -25,11 +25,9 @@ class TrendingNotesNostrBand(DVMTaskInterface): dvm_config: DVMConfig logger = False - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) if self.options is not None: if self.options.get("logger"): @@ -67,7 +65,7 @@ class TrendingNotesNostrBand(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): options = self.set_options(request_form) import requests diff --git a/nostr_dvm/tasks/imagegeneration_openai_dalle.py b/nostr_dvm/tasks/imagegeneration_openai_dalle.py index 00f31fd..1f6ad3a 100644 --- a/nostr_dvm/tasks/imagegeneration_openai_dalle.py +++ b/nostr_dvm/tasks/imagegeneration_openai_dalle.py @@ -31,11 +31,9 @@ class ImageGenerationDALLE(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("openai", "openai==1.3.5")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -94,7 +92,7 @@ class ImageGenerationDALLE(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): try: options = self.set_options(request_form) diff --git a/nostr_dvm/tasks/imagegeneration_replicate_sdxl.py b/nostr_dvm/tasks/imagegeneration_replicate_sdxl.py index 41a18be..6199543 100644 --- a/nostr_dvm/tasks/imagegeneration_replicate_sdxl.py +++ b/nostr_dvm/tasks/imagegeneration_replicate_sdxl.py @@ -30,11 +30,9 @@ class ImageGenerationReplicateSDXL(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("replicate", "replicate")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -91,7 +89,7 @@ class ImageGenerationReplicateSDXL(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): try: options = self.set_options(request_form) diff --git a/nostr_dvm/tasks/imagegeneration_sd21_mlx.py b/nostr_dvm/tasks/imagegeneration_sd21_mlx.py index 8c8e160..053e241 100644 --- a/nostr_dvm/tasks/imagegeneration_sd21_mlx.py +++ b/nostr_dvm/tasks/imagegeneration_sd21_mlx.py @@ -34,11 +34,9 @@ class ImageGenerationMLX(DVMTaskInterface): ("tqdm", "tqdm"), ] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -95,7 +93,7 @@ class ImageGenerationMLX(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): try: import mlx.core as mx from nostr_dvm.backends.mlx.modules.stable_diffusion import StableDiffusion diff --git a/nostr_dvm/tasks/imagegeneration_sdxl.py b/nostr_dvm/tasks/imagegeneration_sdxl.py index de64068..069bae1 100644 --- a/nostr_dvm/tasks/imagegeneration_sdxl.py +++ b/nostr_dvm/tasks/imagegeneration_sdxl.py @@ -1,4 +1,5 @@ import json +import os from multiprocessing.pool import ThreadPool from nostr_sdk import Kind @@ -26,10 +27,9 @@ class ImageGenerationSDXL(DVMTaskInterface): TASK: str = "text-to-image" FIX_COST: float = 50 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): + dvm_config.SCRIPT = os.path.abspath(__file__) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -144,7 +144,7 @@ class ImageGenerationSDXL(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): try: # Call the process route of n-server with our request form. response = send_request_to_server(request_form, self.options['server']) diff --git a/nostr_dvm/tasks/imagegeneration_sdxlimg2img.py b/nostr_dvm/tasks/imagegeneration_sdxlimg2img.py index fc656bc..0246848 100644 --- a/nostr_dvm/tasks/imagegeneration_sdxlimg2img.py +++ b/nostr_dvm/tasks/imagegeneration_sdxlimg2img.py @@ -1,4 +1,5 @@ import json +import os from multiprocessing.pool import ThreadPool from nostr_sdk import Kind @@ -26,10 +27,9 @@ class ImageGenerationSDXLIMG2IMG(DVMTaskInterface): TASK: str = "image-to-image" FIX_COST: float = 70 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): + dvm_config.SCRIPT = os.path.abspath(__file__) def is_input_supported(self, tags, client=None, dvm_config=None): hasurl = False @@ -170,7 +170,7 @@ class ImageGenerationSDXLIMG2IMG(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): try: # Call the process route of NOVA-Server with our request form. response = send_request_to_server(request_form, self.options['server']) diff --git a/nostr_dvm/tasks/imageinterrogator.py b/nostr_dvm/tasks/imageinterrogator.py index 92f8ad7..9c3d62b 100644 --- a/nostr_dvm/tasks/imageinterrogator.py +++ b/nostr_dvm/tasks/imageinterrogator.py @@ -1,4 +1,5 @@ import json +import os from multiprocessing.pool import ThreadPool from nostr_sdk import Kind @@ -25,10 +26,9 @@ class ImageInterrogator(DVMTaskInterface): TASK: str = "image-to-text" FIX_COST: float = 80 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): + dvm_config.SCRIPT = os.path.abspath(__file__) def is_input_supported(self, tags, client=None, dvm_config=None): hasurl = False @@ -87,7 +87,7 @@ class ImageInterrogator(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): try: # Call the process route of NOVA-Server with our request form. response = send_request_to_server(request_form, self.options['server']) diff --git a/nostr_dvm/tasks/imageupscale.py b/nostr_dvm/tasks/imageupscale.py index e487695..26d89c5 100644 --- a/nostr_dvm/tasks/imageupscale.py +++ b/nostr_dvm/tasks/imageupscale.py @@ -1,4 +1,5 @@ import json +import os from multiprocessing.pool import ThreadPool from nostr_sdk import Kind @@ -25,10 +26,9 @@ class ImageUpscale(DVMTaskInterface): TASK: str = "image-to-image" FIX_COST: float = 20 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): + dvm_config.SCRIPT = os.path.abspath(__file__) def is_input_supported(self, tags, client=None, dvm_config=None): hasurl = False @@ -89,7 +89,7 @@ class ImageUpscale(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): try: # Call the process route of NOVA-Server with our request form. response = send_request_to_server(request_form, self.options['server']) diff --git a/nostr_dvm/tasks/search_users.py b/nostr_dvm/tasks/search_users.py index b4b39cf..7659e29 100644 --- a/nostr_dvm/tasks/search_users.py +++ b/nostr_dvm/tasks/search_users.py @@ -28,17 +28,10 @@ class SearchUser(DVMTaskInterface): last_schedule: int = 0 db_name = "db/nostr_profiles.db" - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) - - use_logger = False - if use_logger: - init_logger(LogLevel.DEBUG) - - self.sync_db() + await self.sync_db() def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -76,7 +69,7 @@ class SearchUser(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from nostr_sdk import Filter options = self.set_options(request_form) @@ -85,19 +78,19 @@ class SearchUser(DVMTaskInterface): keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().database(database).signer(signer).opts(opts).build() - cli.add_relay("wss://relay.damus.io") + await cli.add_relay("wss://relay.damus.io") # cli.add_relay("wss://atl.purplerelay.com") - cli.connect() + await cli.connect() # Negentropy reconciliation # Query events from database filter1 = Filter().kind(Kind(0)) - events = cli.database().query([filter1]) + events = await cli.database().query([filter1]) # for event in events: # print(event.as_json()) @@ -121,6 +114,7 @@ class SearchUser(DVMTaskInterface): else: break + await cli.disconnect() return json.dumps(result_list) def post_process(self, result, event): @@ -134,33 +128,34 @@ class SearchUser(DVMTaskInterface): # if not text/plain, don't post-process return result - def schedule(self, dvm_config): + async def schedule(self, dvm_config): if dvm_config.SCHEDULE_UPDATES_SECONDS == 0: return 0 else: if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS: - self.sync_db() + await self.sync_db() self.last_schedule = Timestamp.now().as_secs() return 1 - def sync_db(self): + async def sync_db(self): opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))) sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY) keys = Keys.parse(sk.to_hex()) signer = NostrSigner.keys(keys) - database = NostrDatabase.sqlite(self.db_name) + database = await NostrDatabase.sqlite(self.db_name) cli = ClientBuilder().signer(signer).database(database).opts(opts).build() - cli.add_relay("wss://relay.damus.io") - cli.connect() + await cli.add_relay("wss://relay.damus.io") + await cli.connect() filter1 = Filter().kind(Kind(0)) # filter = Filter().author(keys.public_key()) print("Syncing Profile Database.. this might take a while..") dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN) - cli.reconcile(filter1, dbopts) + await cli.reconcile(filter1, dbopts) print("Done Syncing Profile Database.") + await cli.shutdown() # We build an example here that we can call by either calling this file directly from the main directory, diff --git a/nostr_dvm/tasks/summarization_huggingchat.py b/nostr_dvm/tasks/summarization_huggingchat.py index 718cd9e..36101b8 100644 --- a/nostr_dvm/tasks/summarization_huggingchat.py +++ b/nostr_dvm/tasks/summarization_huggingchat.py @@ -26,11 +26,10 @@ class TextSummarizationHuggingChat(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("hugchat", "hugchat")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -92,7 +91,7 @@ class TextSummarizationHuggingChat(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): from hugchat import hugchat from hugchat.login import Login sign = Login(os.getenv("HUGGINGFACE_EMAIL"), os.getenv("HUGGINGFACE_PASSWORD")) diff --git a/nostr_dvm/tasks/summarization_unleashed_chat.py b/nostr_dvm/tasks/summarization_unleashed_chat.py index b6c2861..ed9d0d3 100644 --- a/nostr_dvm/tasks/summarization_unleashed_chat.py +++ b/nostr_dvm/tasks/summarization_unleashed_chat.py @@ -25,11 +25,9 @@ class SummarizationUnleashedChat(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("openai", "openai")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -93,7 +91,7 @@ class SummarizationUnleashedChat(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): from openai import OpenAI temp_open_ai_api_key = os.environ["OPENAI_API_KEY"] os.environ["OPENAI_API_KEY"] = os.getenv("UNLEASHED_API_KEY") diff --git a/nostr_dvm/tasks/textextraction_google.py b/nostr_dvm/tasks/textextraction_google.py index b36f466..b2ebf15 100644 --- a/nostr_dvm/tasks/textextraction_google.py +++ b/nostr_dvm/tasks/textextraction_google.py @@ -29,13 +29,10 @@ class SpeechToTextGoogle(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("speech_recognition", "SpeechRecognition==3.10.0")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) - if options is None: - self.options = {} + def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -106,7 +103,7 @@ class SpeechToTextGoogle(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): import speech_recognition as sr if self.options.get("api_key"): api_key = self.options['api_key'] diff --git a/nostr_dvm/tasks/textextraction_pdf.py b/nostr_dvm/tasks/textextraction_pdf.py index 4ffde17..e1933f9 100644 --- a/nostr_dvm/tasks/textextraction_pdf.py +++ b/nostr_dvm/tasks/textextraction_pdf.py @@ -28,11 +28,9 @@ class TextExtractionPDF(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("pypdf", "pypdf==3.17.1")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -69,7 +67,7 @@ class TextExtractionPDF(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from pypdf import PdfReader from pathlib import Path import requests diff --git a/nostr_dvm/tasks/textextraction_whisperx.py b/nostr_dvm/tasks/textextraction_whisperx.py index 282dbd9..7b1e90f 100644 --- a/nostr_dvm/tasks/textextraction_whisperx.py +++ b/nostr_dvm/tasks/textextraction_whisperx.py @@ -29,11 +29,9 @@ class SpeechToTextWhisperX(DVMTaskInterface): FIX_COST: float = 10 PER_UNIT_COST: float = 0.1 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -132,7 +130,7 @@ class SpeechToTextWhisperX(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): try: # Call the process route of NOVA-Server with our request form. response = send_request_to_server(request_form, self.options['server']) diff --git a/nostr_dvm/tasks/textgeneration_huggingchat.py b/nostr_dvm/tasks/textgeneration_huggingchat.py index c7c5702..ec95c83 100644 --- a/nostr_dvm/tasks/textgeneration_huggingchat.py +++ b/nostr_dvm/tasks/textgeneration_huggingchat.py @@ -25,11 +25,9 @@ class TextGenerationHuggingChat(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("hugchat", "hugchat")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -58,7 +56,7 @@ class TextGenerationHuggingChat(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): from hugchat import hugchat from hugchat.login import Login sign = Login(os.getenv("HUGGINGFACE_EMAIL"), os.getenv("HUGGINGFACE_PASSWORD")) diff --git a/nostr_dvm/tasks/textgeneration_llmlite.py b/nostr_dvm/tasks/textgeneration_llmlite.py index 9f202fb..494012d 100644 --- a/nostr_dvm/tasks/textgeneration_llmlite.py +++ b/nostr_dvm/tasks/textgeneration_llmlite.py @@ -25,11 +25,9 @@ class TextGenerationLLMLite(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("litellm", "litellm==1.12.3")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -68,7 +66,7 @@ class TextGenerationLLMLite(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): from litellm import completion options = self.set_options(request_form) diff --git a/nostr_dvm/tasks/textgeneration_unleashed_chat.py b/nostr_dvm/tasks/textgeneration_unleashed_chat.py index 2d3b958..9d407ac 100644 --- a/nostr_dvm/tasks/textgeneration_unleashed_chat.py +++ b/nostr_dvm/tasks/textgeneration_unleashed_chat.py @@ -25,11 +25,9 @@ class TextGenerationUnleashedChat(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("openai", "openai")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -60,7 +58,7 @@ class TextGenerationUnleashedChat(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): from openai import OpenAI temp_open_ai_api_key = os.environ["OPENAI_API_KEY"] os.environ["OPENAI_API_KEY"] = os.getenv("UNLEASHED_API_KEY") diff --git a/nostr_dvm/tasks/texttospeech.py b/nostr_dvm/tasks/texttospeech.py index 3e5132c..dfaa112 100644 --- a/nostr_dvm/tasks/texttospeech.py +++ b/nostr_dvm/tasks/texttospeech.py @@ -33,11 +33,9 @@ class TextToSpeech(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("TTS", "TTS==0.22.0")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -103,7 +101,7 @@ class TextToSpeech(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): import torch from TTS.api import TTS options = self.set_options(request_form) diff --git a/nostr_dvm/tasks/translation_google.py b/nostr_dvm/tasks/translation_google.py index 1c00fec..f4e6209 100644 --- a/nostr_dvm/tasks/translation_google.py +++ b/nostr_dvm/tasks/translation_google.py @@ -27,11 +27,9 @@ class TranslationGoogle(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("translatepy", "translatepy==2.3")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -77,7 +75,7 @@ class TranslationGoogle(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): from translatepy.translators.google import GoogleTranslate options = self.set_options(request_form) diff --git a/nostr_dvm/tasks/translation_libretranslate.py b/nostr_dvm/tasks/translation_libretranslate.py index 19e8229..8daddba 100644 --- a/nostr_dvm/tasks/translation_libretranslate.py +++ b/nostr_dvm/tasks/translation_libretranslate.py @@ -27,11 +27,9 @@ class TranslationLibre(DVMTaskInterface): TASK: str = "translation" FIX_COST: float = 0 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -77,7 +75,7 @@ class TranslationLibre(DVMTaskInterface): request_form['options'] = json.dumps(options) return request_form - def process(self, request_form): + async def process(self, request_form): options = self.set_options(request_form) request = { "q": options["text"], diff --git a/nostr_dvm/tasks/videogeneration_replicate_svd.py b/nostr_dvm/tasks/videogeneration_replicate_svd.py index fecc8bd..5e3f514 100644 --- a/nostr_dvm/tasks/videogeneration_replicate_svd.py +++ b/nostr_dvm/tasks/videogeneration_replicate_svd.py @@ -31,11 +31,9 @@ class VideoGenerationReplicateSVD(DVMTaskInterface): dependencies = [("nostr-dvm", "nostr-dvm"), ("replicate", "replicate")] - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): dvm_config.SCRIPT = os.path.abspath(__file__) - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -79,7 +77,7 @@ class VideoGenerationReplicateSVD(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): try: options = self.set_options(request_form) print(options["url"]) diff --git a/nostr_dvm/tasks/videogeneration_svd.py b/nostr_dvm/tasks/videogeneration_svd.py index 2e4794c..b668251 100644 --- a/nostr_dvm/tasks/videogeneration_svd.py +++ b/nostr_dvm/tasks/videogeneration_svd.py @@ -26,10 +26,9 @@ class VideoGenerationSVD(DVMTaskInterface): TASK: str = "image-to-video" FIX_COST: float = 120 - def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, - admin_config: AdminConfig = None, options=None): - super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config, - admin_config=admin_config, options=options) + async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None, + admin_config: AdminConfig = None, options=None): + dvm_config.SCRIPT = os.path.abspath(__file__) def is_input_supported(self, tags, client=None, dvm_config=None): for tag in tags: @@ -80,7 +79,7 @@ class VideoGenerationSVD(DVMTaskInterface): return request_form - def process(self, request_form): + async def process(self, request_form): try: # Call the process route of n-server with our request form. response = send_request_to_server(request_form, self.options['server']) diff --git a/nostr_dvm/utils/admin_utils.py b/nostr_dvm/utils/admin_utils.py index 363075c..faeb5f3 100644 --- a/nostr_dvm/utils/admin_utils.py +++ b/nostr_dvm/utils/admin_utils.py @@ -36,7 +36,7 @@ class AdminConfig: PRIVKEY: str = "" -def admin_make_database_updates(adminconfig: AdminConfig = None, dvmconfig: DVMConfig = None, client: Client = None): +async def admin_make_database_updates(adminconfig: AdminConfig = None, dvmconfig: DVMConfig = None, client: Client = None): # This is called on start of Server, Admin function to manually whitelist/blacklist/add balance/delete users if adminconfig is None or dvmconfig is None: return @@ -64,7 +64,7 @@ def admin_make_database_updates(adminconfig: AdminConfig = None, dvmconfig: DVMC publickey = npub if adminconfig.WHITELISTUSER: - user = get_or_add_user(db, publickey, client=client, config=dvmconfig) + user = await get_or_add_user(db, publickey, client=client, config=dvmconfig) update_sql_table(db, user.npub, user.balance, True, False, user.nip05, user.lud16, user.name, user.lastactive, user.subscribed) user = get_from_sql_table(db, publickey) print(str(user.name) + " is whitelisted: " + str(user.iswhitelisted)) diff --git a/nostr_dvm/utils/cashu_utils.py b/nostr_dvm/utils/cashu_utils.py index 747993f..8a473e2 100644 --- a/nostr_dvm/utils/cashu_utils.py +++ b/nostr_dvm/utils/cashu_utils.py @@ -107,7 +107,7 @@ def parse_cashu(cashu_token: str): return None, None, None, "Cashu Parser: " + str(e) -def redeem_cashu(cashu, config, client, required_amount=0, update_self=False) -> (bool, str, int, int): +async def redeem_cashu(cashu, config, client, required_amount=0, update_self=False) -> (bool, str, int, int): proofs, mint, total_amount, message = parse_cashu(cashu) if message is not None: return False, message, 0, 0 @@ -121,7 +121,7 @@ def redeem_cashu(cashu, config, client, required_amount=0, update_self=False) -> invoice, paymenthash = create_bolt11_ln_bits(estimated_redeem_invoice_amount, config) else: - user = get_or_add_user(db=config.DB, npub=config.PUBLIC_KEY, + user = await get_or_add_user(db=config.DB, npub=config.PUBLIC_KEY, client=client, config=config, update=update_self) invoice = create_bolt11_lud16(user.lud16, estimated_redeem_invoice_amount) print(invoice) @@ -148,7 +148,7 @@ def redeem_cashu(cashu, config, client, required_amount=0, update_self=False) -> invoice, paymenthash = create_bolt11_ln_bits(redeem_invoice_amount, config) else: - user = get_or_add_user(db=config.DB, npub=config.PUBLIC_KEY, + user = await get_or_add_user(db=config.DB, npub=config.PUBLIC_KEY, client=client, config=config, update=update_self) invoice = create_bolt11_lud16(user.lud16, redeem_invoice_amount) print(invoice) diff --git a/nostr_dvm/utils/database_utils.py b/nostr_dvm/utils/database_utils.py index a1faa1e..d50e564 100644 --- a/nostr_dvm/utils/database_utils.py +++ b/nostr_dvm/utils/database_utils.py @@ -110,7 +110,6 @@ def get_from_sql_table(db, npub): add_sql_table_column(db) # Migrate
 - user = User user.npub = row[0] user.balance = row[1] @@ -217,8 +216,7 @@ def update_user_subscription(npub, subscribed_until, client, dvm_config): print("Updated user subscription for: " + str(user.name)) - -def get_or_add_user(db, npub, client, config, update=False, skip_meta = False): +async def get_or_add_user(db, npub, client, config, update=False, skip_meta=False): user = get_from_sql_table(db, npub) if user is None: try: @@ -227,7 +225,7 @@ def get_or_add_user(db, npub, client, config, update=False, skip_meta = False): nip05 = "" lud16 = "" else: - name, nip05, lud16 = fetch_user_metadata(npub, client) + name, nip05, lud16 = await fetch_user_metadata(npub, client) print("Adding User: " + npub + " (" + npub + ")") add_to_sql_table(db, npub, config.NEW_USER_BALANCE, False, False, nip05, lud16, name, Timestamp.now().as_secs(), 0) @@ -237,7 +235,7 @@ def get_or_add_user(db, npub, client, config, update=False, skip_meta = False): print("Error Adding User to DB: " + str(e)) elif update: try: - name, nip05, lud16 = fetch_user_metadata(npub, client) + name, nip05, lud16 = await fetch_user_metadata(npub, client) print("Updating User: " + npub + " (" + npub + ")") update_sql_table(db, user.npub, user.balance, user.iswhitelisted, user.isblacklisted, nip05, lud16, name, Timestamp.now().as_secs(), user.subscribed) @@ -249,14 +247,14 @@ def get_or_add_user(db, npub, client, config, update=False, skip_meta = False): return user -def fetch_user_metadata(npub, client): +async def fetch_user_metadata(npub, client): name = "" nip05 = "" lud16 = "" pk = PublicKey.parse(npub) print(f"\nGetting profile metadata for {pk.to_bech32()}...") profile_filter = Filter().kind(Kind(0)).author(pk).limit(1) - events = client.get_events_of([profile_filter], timedelta(seconds=1)) + events = await client.get_events_of([profile_filter], timedelta(seconds=1)) if len(events) > 0: latest_entry = events[0] latest_time = 0 diff --git a/nostr_dvm/utils/nostr_utils.py b/nostr_dvm/utils/nostr_utils.py index 1f7be53..1db4302 100644 --- a/nostr_dvm/utils/nostr_utils.py +++ b/nostr_dvm/utils/nostr_utils.py @@ -1,3 +1,4 @@ +import asyncio import json import os from datetime import timedelta @@ -31,22 +32,30 @@ def get_event_by_id(event_id: str, client: Client, config=None) -> Event | None: event_id = EventId.from_hex(event_id) id_filter = Filter().id(event_id).limit(1) - events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT)) + #events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT)) + events = asyncio.run(get_events_async(client, id_filter, config.RELAY_TIMEOUT)) + if len(events) > 0: return events[0] else: return None +async def get_events_async(client, filter, timeout): + events = client.get_events_of([filter], timedelta(seconds=timeout) + return events + def get_events_by_ids(event_ids, client: Client, config=None) -> List | None: search_ids = [] + events = [] for event_id in event_ids: split = event_id.split(":") if len(split) == 3: pk = PublicKey.from_hex(split[1]) id_filter = Filter().author(pk).custom_tag(SingleLetterTag.lowercase(Alphabet.D), [split[2]]) - events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT)) + events = asyncio.run(get_events_async(client, id_filter, config.RELAY_TIMEOUT)) + #events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT)) else: if str(event_id).startswith('note'): event_id = EventId.from_bech32(event_id) @@ -59,12 +68,13 @@ def get_events_by_ids(event_ids, client: Client, config=None) -> List | None: else: event_id = EventId.from_hex(event_id) - search_ids.append(event_id) + search_ids.append(event_id) - id_filter = Filter().ids(search_ids) - events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT)) + id_filter = Filter().ids(search_ids) + events = asyncio.run(get_events_async(client, id_filter, config.RELAY_TIMEOUT)) + + #events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT)) if len(events) > 0: - return events else: return None @@ -72,7 +82,8 @@ def get_events_by_ids(event_ids, client: Client, config=None) -> List | None: def get_events_by_id(event_ids: list, client: Client, config=None) -> list[Event] | None: id_filter = Filter().ids(event_ids) - events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT)) + events = asyncio.run(get_events_async(client, id_filter, config.RELAY_TIMEOUT)) + #events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT)) if len(events) > 0: return events else: @@ -98,7 +109,8 @@ def get_referenced_event_by_id(event_id, client, dvm_config, kinds) -> Event | N else: job_id_filter = Filter().event(event_id).limit(1) - events = client.get_events_of([job_id_filter], timedelta(seconds=dvm_config.RELAY_TIMEOUT)) + events = asyncio.run(get_events_async(client, job_id_filter, dvm_config.RELAY_TIMEOUT)) + #events = client.get_events_of([job_id_filter], timedelta(seconds=dvm_config.RELAY_TIMEOUT)) if len(events) > 0: return events[0] @@ -106,7 +118,7 @@ def get_referenced_event_by_id(event_id, client, dvm_config, kinds) -> Event | N return None -def get_inbox_relays(event_to_send: Event, client: Client, dvm_config): +async def get_inbox_relays(event_to_send: Event, client: Client, dvm_config): ptags = [] for tag in event_to_send.tags(): if tag.as_vec()[0] == 'p': @@ -114,7 +126,7 @@ def get_inbox_relays(event_to_send: Event, client: Client, dvm_config): ptags.append(ptag) filter = Filter().kinds([EventDefinitions.KIND_RELAY_ANNOUNCEMENT]).authors(ptags) - events = client.get_events_of([filter], timedelta(dvm_config.RELAY_TIMEOUT)) + events = await client.get_events_of([filter], timedelta(dvm_config.RELAY_TIMEOUT)) if len(events) == 0: return [] else: @@ -130,7 +142,7 @@ def get_inbox_relays(event_to_send: Event, client: Client, dvm_config): return relays -def send_event_outbox(event: Event, client, dvm_config) -> EventId: +async def send_event_outbox(event: Event, client, dvm_config) -> EventId: # 1. OK, Let's overcomplicate things. # 2. If our event has a relays tag, we just send the event to these relay in the classical way. @@ -149,12 +161,12 @@ def send_event_outbox(event: Event, client, dvm_config) -> EventId: # 3. If we couldn't find relays, we look in the receivers inbox if len(relays) == 0: - relays = get_inbox_relays(event, client, dvm_config) + relays = await get_inbox_relays(event, client, dvm_config) # 4. If we don't find inbox relays (e.g. because the user didn't announce them, we just send to our default relays if len(relays) == 0: print("[" + dvm_config.NIP89.NAME + "] No Inbox found, replying to generic relays") - eventid = send_event(event, client, dvm_config) + eventid = await send_event(event, client, dvm_config) return eventid # 5. Otherwise, we create a new Outbox client with the inbox relays and send the event there @@ -172,22 +184,22 @@ def send_event_outbox(event: Event, client, dvm_config) -> EventId: for relay in relays: opts = RelayOptions().ping(False) try: - outboxclient.add_relay_with_opts(relay, opts) + await outboxclient.add_relay_with_opts(relay, opts) except: print("[" + dvm_config.NIP89.NAME + "] " + relay + " couldn't be added to outbox relays") - outboxclient.connect() + await outboxclient.connect() try: - event_id = outboxclient.send_event(event) + event_id = await outboxclient.send_event(event) except: - event_id = send_event(event, client, dvm_config) + event_id = await send_event(event, client, dvm_config) - outboxclient.shutdown() + await outboxclient.shutdown() return event_id -def send_event(event: Event, client: Client, dvm_config, blastr=False) -> EventId: +async def send_event(event: Event, client: Client, dvm_config, blastr=False) -> EventId: try: relays = [] for tag in event.tags(): @@ -198,16 +210,16 @@ def send_event(event: Event, client: Client, dvm_config, blastr=False) -> EventI for relay in relays: if relay not in dvm_config.RELAY_LIST: - client.add_relay(relay) + await client.add_relay(relay) #if blastr: # client.add_relay("wss://nostr.mutinywallet.com") - event_id = client.send_event(event) + event_id = await client.send_event(event) for relay in relays: if relay not in dvm_config.RELAY_LIST: - client.remove_relay(relay) + await client.remove_relay(relay) #if blastr: # client.remove_relay("wss://nostr.mutinywallet.com") return event_id