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, SimpleConfig, WalletSynchronizer
  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
  46 wallet_config = SimpleConfig()
  47 master_public_key = config.get('electrum','mpk')
  48 wallet_config.set_key('master_public_key',master_public_key)
  49 wallet = Wallet(wallet_config)
  50 wallet.synchronize = lambda: None # prevent address creation by the wallet
  51
  52
  53 omg_addresses = {}
  54
  55 def input_reader_thread(request_queue):
  56     while True:
  57         addr, amount, confirmations = request_queue.get(True,1000000000)
  58         if addr in omg_addresses:
  59             continue
  60         else:
  61             print "subscribing to ", addr
  62             omg_addresses[addr] = {'requested':float(amount), 'confirmations':int(confirmations)}
  63
  64         if addr not in wallet.addresses:
  65             with wallet.lock:
  66                 print "adding %s to wallet"%addr
  67                 wallet.addresses.append(addr)
  68                 wallet.history[addr] = []
  69                 synchronizer.subscribe_to_addresses([addr])
  70                 wallet.up_to_date = False
  71
  72
  73
  74 def on_wallet_update():
  75     print "updated_callback"
  76     for addr in omg_addresses:
  77         h = wallet.history.get(addr)
  78
  79         requested_amount = omg_addresses[addr].get('requested')
  80         requested_confs  = omg_addresses[addr].get('confirmations')
  81
  82         value = 0
  83         for tx_hash, tx_height in h:
  84
  85             tx = wallet.transactions.get(tx_hash)
  86             if not tx: continue
  87             if verifier.get_confirmations(tx_hash) < requested_confs: continue
  88             for o in tx.get('outputs'):
  89                 if o.get('address') == addr:
  90                     value += o.get('value')
  91
  92         s = (value)/1.e8
  93         print "balance for %s:"%addr, s, requested_amount
  94         if s>= requested_amount:
  95             print "payment accepted", addr
  96             out_queue.put( ('payment', addr))
  97
  98
  99 stopping = False
 100
 101 def do_stop():
 102     global stopping
 103     stopping = True
 104
 105 def do_create(conn):
 106     # creation
 107     cur = conn.cursor()
 108     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));")
 109     conn.commit()
 110
 111 def process_request(i, amount, confirmations, expires_in, password):
 112     print "process_request", i, amount, confirmations, expires_in
 113     if password!=my_password:
 114         print "wrong password ", password
 115         return
 116     addr = wallet.get_new_address(0, i, 0)
 117     out_queue.put( ('request', (i, addr, amount, confirmations, expires_in) ))
 118     return addr
 119
 120 def get_mpk():
 121     return wallet.master_public_key
 122
 123
 124 def server_thread(conn):
 125     from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer
 126     server = SimpleJSONRPCServer(( my_host, my_port))
 127     server.register_function(process_request, 'request')
 128     server.register_function(get_mpk, 'mpk')
 129     server.register_function(do_stop, 'stop')
 130     server.serve_forever()
 131
 132
 133 def handle_command(cmd):
 134     import jsonrpclib
 135     server = jsonrpclib.Server('http://%s:%d'%(my_host, my_port))
 136     try:
 137         if cmd == 'mpk':
 138             out = server.mpk()
 139         elif cmd == 'stop':
 140             out = server.stop()
 141         elif cmd == 'create':
 142             conn = mdb.connect(db_instance, db_user, db_password, db_name);
 143             do_create(conn)
 144             out = "ok"
 145         else:
 146             out = "unknown command"
 147     except socket.error:
 148         print "Server not running"
 149         return 1
 150
 151     print out
 152     return 0
 153
 154
 155 if __name__ == '__main__':
 156
 157     if len(sys.argv) > 1:
 158         ret = handle_command(sys.argv[1])
 159         sys.exit(ret)
 160
 161     print "using database", db_name
 162     conn = mdb.connect(db_instance, db_user, db_password, db_name);
 163
 164     interface = Interface({'server':"%s:%d:t"%(electrum_server, 50001)})
 165     interface.start()
 166     interface.send([('blockchain.numblocks.subscribe',[])])
 167
 168     wallet.interface = interface
 169     interface.register_callback('updated', on_wallet_update)
 170
 171     verifier = WalletVerifier(interface, wallet_config)
 172     wallet.set_verifier(verifier)
 173
 174     synchronizer = WalletSynchronizer(wallet, wallet_config)
 175     synchronizer.start()
 176
 177     verifier.start()
 178
 179
 180     # this process detects when addresses have paid
 181     request_queue = Queue.Queue()
 182     out_queue = Queue.Queue()
 183     thread.start_new_thread(input_reader_thread, (request_queue,))
 184     thread.start_new_thread(server_thread, (conn,))
 185
 186     while not stopping:
 187         cur = conn.cursor()
 188
 189         # read pending requests from table
 190         cur.execute("SELECT address, amount, confirmations FROM electrum_payments WHERE paid IS NULL;")
 191         data = cur.fetchall()
 192         for item in data:
 193             request_queue.put(item)
 194
 195         try:
 196             cmd, params = out_queue.get(True, 10)
 197         except Queue.Empty:
 198             cmd = ''
 199
 200         if cmd == 'payment':
 201             addr = params
 202             # set paid=1 for received payments
 203             print "received payment from", addr
 204             cur.execute("select id from electrum_payments where address='%s';"%addr)
 205             id = cur.fetchone()[0]
 206             cur.execute("update electrum_payments set paid=1 where id=%d;"%(id))
 207
 208         elif cmd == 'request':
 209             # add a new request to the table.
 210             i, addr, amount, confs, hours = params
 211             sql = "INSERT INTO electrum_payments (id, address, amount, confirmations, received_at, expires_at, paid, processed)"\
 212                 + " VALUES (%d, '%s', %f, %d, CURRENT_TIMESTAMP, ADDTIME(CURRENT_TIMESTAMP, '0 %d:0:0'), NULL, NULL);"%(i, addr, amount, confs, hours)
 213             cur.execute(sql)
 214
 215
 216         # set paid=0 for expired requests
 217         cur.execute("""UPDATE electrum_payments set paid=0 WHERE expires_at < CURRENT_TIMESTAMP and paid is NULL;""")
 218
 219         # do callback for addresses that received payment
 220         cur.execute("""SELECT id, address, paid from electrum_payments WHERE paid is not NULL and processed is NULL;""")
 221         data = cur.fetchall()
 222         for item in data:
 223             print "callback:", item
 224             id = int(item[0])
 225             address = item[1]
 226             paid = int(item[2])
 227             headers = {'content-type':'application/json'}
 228             data_json = { 'id':id, 'address':address, 'btc_auth':cb_password }
 229             data_json = json.dumps(data_json)
 230             url = cb_received if paid else cb_expired
 231             req = urllib2.Request(url, data_json, headers)
 232             try:
 233                 response_stream = urllib2.urlopen(req)
 234                 cur.execute("UPDATE electrum_payments SET processed=1 WHERE id=%d;"%(id))
 235             except urllib2.HTTPError:
 236                 print "cannot do callback", data_json
 237
 238         conn.commit()
 239
 240
 241     conn.close()
 242     print "terminated"
 243
 244
 245