Simple OBS script that shows a notification upon pending transactions.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

duniter-alert.py 7.0KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169
  1. import urllib.request as http
  2. import urllib.error
  3. import json
  4. def query_duniter(addr, node="https://g1.duniter.org", tx_type="pending"):
  5. net = "g1"
  6. url = node + "/tx/history/" + addr
  7. data = None
  8. # https://g1.duniter.org/tx/history/<address>
  9. with http.urlopen(url) as response:
  10. data = response.read()
  11. table = json.loads(data)
  12. return table["history"][tx_type]
  13. # this function might not properly attribute issuers
  14. # if the blockchain allows three participants in one transaction.
  15. def parse_tx(tx, address):
  16. # only display balance going into streamer's address
  17. issuer = tx["issuers"][0]
  18. for output in tx["outputs"]:
  19. if address in output and address != issuer:
  20. data = output.split(":", 1) # [balance, 0, SIG(address)]
  21. return issuer[:5], int(data[0])/100
  22. return None, None
  23. def make_text(transactions,
  24. address,
  25. f,
  26. prepend="Pending transaction(s) from ",
  27. append=". Thank you! "):
  28. text = prepend
  29. for tx in reversed(transactions):
  30. sender, amount = parse_tx(tx, address)
  31. if sender != None and amount != None:
  32. text += f % (sender, amount)
  33. text += append + (len(text) * ' ')
  34. return text
  35. # invoke from standard terminal - easy testing
  36. if __name__ == "__main__":
  37. import sys
  38. def main():
  39. print ("I loaded.")
  40. address = sys.argv[1]
  41. print(address)
  42. txs = query_duniter(address, tx_type="received")
  43. print(make_text(txs, address, "%s (%.2f Ğ1) "))
  44. main()
  45. else: # invoke from OBS
  46. import obspython as obs
  47. def script_load(settings):
  48. obs.script_log(obs.LOG_INFO,
  49. "duniter alert active.")
  50. def script_unload():
  51. obs.script_log(obs.LOG_INFO,
  52. "duniter alert unloaded.")
  53. obs.timer_remove(update_text)
  54. def test_btn(props, prop):
  55. update_text("received")
  56. def stop_btn(props, prop):
  57. obs.script_log(obs.LOG_INFO,
  58. "Stop button pressed. Hit Reload to resume contacting nodes.")
  59. obs.timer_remove(update_text)
  60. def script_description():
  61. return "Creates a notification upon pending Duniter transactions.\nTo use, place a text source onto your scene and select.\nSee Script Log for additional information."
  62. def update_text(tx_type="pending"):
  63. global address
  64. global peer
  65. global interval
  66. global src
  67. global pre
  68. global fmt
  69. global app
  70. source = obs.obs_get_source_by_name(src)
  71. text = " "
  72. if source != None:
  73. try:
  74. txs = query_duniter(address, peer, tx_type)
  75. if txs != []:
  76. # empty string = default; space = blank
  77. if pre == "" and app == "":
  78. text = make_text(txs, address, fmt)
  79. elif pre != "" and app == "":
  80. text = make_text(txs, address, fmt, prepend=pre)
  81. elif pre == "" and app != "":
  82. text = make_text(txs, address, fmt, append=app)
  83. elif pre != "" and app != "":
  84. text = make_text(txs, address, fmt, prepend=pre, append=app)
  85. # if text is empty string, then the text source won't be visible
  86. settings = obs.obs_data_create()
  87. obs.obs_data_set_string(settings, "text", text)
  88. obs.obs_source_update(source, settings)
  89. obs.obs_data_release(settings)
  90. except urllib.error.URLError as err:
  91. obs.script_log(obs.LOG_WARNING,
  92. "Error: Could not connect to URL.")
  93. # putting err.reason in script_log will cause another exception
  94. print(err.reason)
  95. obs.remove_current_callback()
  96. except Exception as err:
  97. obs.script_log(obs.LOG_WARNING,
  98. "Error: generic exception")
  99. print(err)
  100. obs.remove_current_callback()
  101. except:
  102. obs.script_log(obs.LOG_WARNING,
  103. "Error: Bad programming. Suppress tick-spam.")
  104. obs.remove_current_callback()
  105. def script_update(settings):
  106. global address
  107. address = obs.obs_data_get_string(settings, "address")
  108. global peer
  109. peer = obs.obs_data_get_string(settings, "peer")
  110. global interval
  111. interval = obs.obs_data_get_int(settings, "interval")
  112. global src
  113. src = obs.obs_data_get_string(settings, "source")
  114. global pre
  115. pre = obs.obs_data_get_string(settings, "prepend")
  116. global fmt
  117. fmt = obs.obs_data_get_string(settings, "format")
  118. global app
  119. app = obs.obs_data_get_string(settings, "append")
  120. obs.timer_remove(update_text)
  121. if address != "" and src != "":
  122. if len(address) == 43 or len(address) == 44:
  123. obs.timer_add(update_text, interval * 1000)
  124. else:
  125. obs.script_log(obs.LOG_WARNING, "Warning: Given address is either too short or too long to be a Duniter address. No timer instantiated.")
  126. def script_defaults(settings):
  127. obs.obs_data_set_default_string(settings, "peer", "https://g1.duniter.org")
  128. obs.obs_data_set_default_string(settings, "format", "%s (%.2f Ğ1)")
  129. obs.obs_data_set_default_int(settings, "interval", 60)
  130. def script_properties():
  131. props = obs.obs_properties_create()
  132. obs.obs_properties_add_text(props, "address", "Address", obs.OBS_TEXT_DEFAULT)
  133. obs.obs_properties_add_text(props, "peer", "Peer", obs.OBS_TEXT_DEFAULT)
  134. obs.obs_properties_add_int(props, "interval", "Interval (seconds)", 15, 3600, 1)
  135. p = obs.obs_properties_add_list(props,
  136. "source",
  137. "Text Source",
  138. obs.OBS_COMBO_TYPE_EDITABLE,
  139. obs.OBS_COMBO_FORMAT_STRING)
  140. sources = obs.obs_enum_sources()
  141. if sources is not None:
  142. for source in sources:
  143. source_id = obs.obs_source_get_unversioned_id(source)
  144. if source_id == "text_gdiplus" or source_id == "text_ft2_source":
  145. name = obs.obs_source_get_name(source)
  146. obs.obs_property_list_add_string(p, name, name)
  147. obs.source_list_release(sources)
  148. obs.obs_properties_add_text(props, "prepend", "Custom Prepend (optional)", obs.OBS_TEXT_DEFAULT)
  149. obs.obs_properties_add_text(props, "format", "Format (sender = s, amount = f)", obs.OBS_TEXT_DEFAULT)
  150. obs.obs_properties_add_text(props, "append", "Custom Append (optional)", obs.OBS_TEXT_DEFAULT)
  151. obs.obs_properties_add_button(props, "button1", "Test", test_btn)
  152. obs.obs_properties_add_button(props, "button2", "Stop", stop_btn)
  153. return props