scripts/merchant.py

   1 #!/usr/bin/env python
   2 #
   3 # Electrum - lightweight Bitcoin client
   4 # Copyright (C) 2011 thomasv@gitorious
   5 #
   6 # This program is free software: you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation, either version 3 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14 # GNU General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
  18
  19
  20 import time, thread, sys, socket, os
  21 import urllib2,json
  22 import MySQLdb as mdb
  23 import Queue
  24 from electrum import Wallet, Interface, WalletVerifier
  25
  26 import ConfigParser
  27 config = ConfigParser.ConfigParser()
  28 config.read("merchant.conf")
  29
  30 db_instance = config.get('db','instance')
  31 db_user = config.get('db','user')
  32 db_password = config.get('db','password')
  33 db_name = config.get('db','name')
  34
  35 electrum_server = config.get('electrum','server')
  36
  37 my_password = config.get('main','password')
  38 my_host = config.get('main','host')
  39 my_port = config.getint('main','port')
  40
  41 cb_received = config.get('callback','received')
  42 cb_expired = config.get('callback','expired')
  43 cb_password = config.get('callback','password')
  44
  45 wallet = Wallet()
  46 wallet.master_public_key = config.get('electrum','mpk')
  47
  48
  49
  50 omg_addresses = {}
  51
  52 def electrum_input_thread(in_queue):
  53     while True:
  54         addr, amount, confirmations = in_queue.get(True,1000000000)
  55         if addr in omg_addresses:
  56             continue
  57         else:
  58             print "subscribing to ", addr
  59             omg_addresses[addr] = {'requested':float(amount), 'confirmations':int(confirmations)}
  60             interface.send([('blockchain.address.subscribe',[addr])])
  61
  62
  63 def electrum_output_thread(out_queue):
  64     while True:
  65         r = interface.get_response()
  66         print r
  67         method = r.get('method')
  68
  69         if method == 'blockchain.address.subscribe':
  70             addr = r.get('params')[0]
  71             interface.send([('blockchain.address.get_history',[addr])])
  72
  73         elif method == 'blockchain.address.get_history':
  74             addr = r.get('params')[0]
  75             h = r.get('result')
  76             if h is None: continue
  77             omg_addresses[addr]['history'] = h
  78             for item in h:
  79                 tx_hash = item.get('tx_hash')
  80                 verifier.add(tx_hash)
  81
  82         elif method == 'blockchain.numblocks.subscribe':
  83             for addr in omg_addresses:
  84                 h = omg_addresses[addr].get('history',[])
  85                 amount = omg_addresses[addr].get('requested')
  86                 confs = omg_addresses[addr].get('confirmations')
  87                 val = 0
  88
  89                 for item in h:
  90                     tx_hash = item.get('tx_hash')
  91                     v = item['value']
  92                     if v<0: continue
  93                     if verifier.get_confirmations(tx_hash) >= conf:
  94                         val += v
  95
  96                 s = (val)/1.e8
  97                 print "balance for %s:"%addr, s
  98                 if s>=amount:
  99                     out_queue.put( ('payment',addr) )
 100
 101
 102 stopping = False
 103
 104 def do_stop():
 105     global stopping
 106     stopping = True
 107
 108 def do_create(conn):
 109     # creation
 110     cur = conn.cursor()
 111     cur.execute("CREATE TABLE electrum_payments (id INT PRIMARY KEY, address VARCHAR(40), amount FLOAT, confirmations INT(8), received_at TIMESTAMP, expires_at TIMESTAMP, paid INT(1), processed INT(1));")
 112     conn.commit()
 113
 114 def process_request(i, amount, confirmations, expires_in, password):
 115     print "process_request", i, amount, confirmations, expires_in
 116     if password!=my_password:
 117         print "wrong password ", password
 118         return
 119     addr = wallet.get_new_address(i, 0)
 120     out_queue.put( ('request',(i,addr,amount,expires_in) ))
 121     return addr
 122
 123 def get_mpk():
 124     return wallet.master_public_key
 125
 126
 127 def server_thread(conn):
 128     from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer
 129     server = SimpleJSONRPCServer(( my_host, my_port))
 130     server.register_function(process_request, 'request')
 131     server.register_function(get_mpk, 'mpk')
 132     server.register_function(do_stop, 'stop')
 133     server.serve_forever()
 134
 135
 136 def handle_command(cmd):
 137     import jsonrpclib
 138     server = jsonrpclib.Server('http://%s:%d'%(my_host, my_port))
 139     try:
 140         if cmd == 'mpk':
 141             out = server.mpk()
 142         elif cmd == 'stop':
 143             out = server.stop()
 144         elif cmd == 'create':
 145             conn = mdb.connect(db_instance, db_user, db_password, db_name);
 146             do_create(conn)
 147             out = "ok"
 148         else:
 149             out = "unknown command"
 150     except socket.error:
 151         print "Server not running"
 152         return 1
 153
 154     print out
 155     return 0
 156
 157
 158 if __name__ == '__main__':
 159
 160     if len(sys.argv) > 1:
 161         ret = handle_command(sys.argv[1])
 162         sys.exit(ret)
 163
 164     print "using database", db_name
 165     conn = mdb.connect(db_instance, db_user, db_password, db_name);
 166
 167     interface = Interface({'server':"%s:%d:t"%(electrum_server, 50001)})
 168     interface.start()
 169     interface.send([('blockchain.numblocks.subscribe',[])])
 170
 171     verifier = WalletVerifier(interface, {})
 172     verifier.start()
 173
 174
 175     # this process detects when addresses have paid
 176     in_queue = Queue.Queue()
 177     out_queue = Queue.Queue()
 178     thread.start_new_thread(electrum_input_thread, (in_queue,))
 179     thread.start_new_thread(electrum_output_thread, (out_queue,))
 180
 181     thread.start_new_thread(server_thread, (conn,))
 182
 183
 184     while not stopping:
 185         cur = conn.cursor()
 186
 187         # get a list of addresses to watch
 188         cur.execute("SELECT address, amount, confirmations FROM electrum_payments WHERE paid IS NULL;")
 189         data = cur.fetchall()
 190         for item in data:
 191             in_queue.put(item)
 192
 193         try:
 194             cmd, params = out_queue.get(True, 10)
 195         except Queue.Empty:
 196             cmd = ''
 197
 198         if cmd == 'payment':
 199             # set paid=1 for received payments
 200             print "received payment from", addr
 201             cur.execute("select id from electrum_payments where address='%s';"%addr)
 202             id = cur.fetchone()[0]
 203             cur.execute("update electrum_payments set paid=1 where id=%d;"%(id))
 204         elif cmd == 'request':
 205             i, addr, amount, confs, hours = params
 206             sql = "INSERT INTO electrum_payments (id, address, amount, confirmations, received_at, expires_at, paid, processed)"\
 207                 + " VALUES (%d, '%s', %f, %d, CURRENT_TIMESTAMP, ADDTIME(CURRENT_TIMESTAMP, '0 %d:0:0'), NULL, NULL);"%(i, addr, amount, confs, hours)
 208             cur.execute(sql)
 209
 210
 211         # set paid=0 for expired requests
 212         cur.execute("""UPDATE electrum_payments set paid=0 WHERE expires_at < CURRENT_TIMESTAMP and paid is NULL;""")
 213
 214         # do callback for addresses that received payment
 215         cur.execute("""SELECT id, address, paid from electrum_payments WHERE paid is not NULL and processed is NULL;""")
 216         data = cur.fetchall()
 217         for item in data:
 218             print "callback:", item
 219             id = int(item[0])
 220             address = item[1]
 221             paid = int(item[2])
 222             headers = {'content-type':'application/json'}
 223             data_json = { 'id':id, 'address':address, 'btc_auth':cb_password }
 224             data_json = json.dumps(data_json)
 225             url = cb_received if paid else cb_expired
 226             req = urllib2.Request(url, data_json, headers)
 227             try:
 228                 response_stream = urllib2.urlopen(req)
 229                 cur.execute("UPDATE electrum_payments SET processed=1 WHERE id=%d;"%(id))
 230             except urllib2.HTTPError:
 231                 print "cannot do callback", data_json
 232
 233         conn.commit()
 234
 235
 236     conn.close()
 237
 238