1 from Abe.abe import hash_to_address, decode_check_address
2 from Abe.DataStore import DataStore as Datastore_class
3 from Abe import DataStore, readconf, BCDataStream, deserialize, util, base58
5 import psycopg2, binascii
7 import thread, traceback, sys, urllib, operator
8 from json import dumps, loads
10 dblock = thread.allocate_lock()
12 class MyStore(Datastore_class):
14 def __init__(self,config, address_queue):
15 conf = DataStore.CONFIG_DEFAULTS
16 args, argv = readconf.parse_argv( [], conf)
17 args.dbtype = config.get('database','type')
18 if args.dbtype == 'sqlite3':
19 args.connect_args = { 'database' : config.get('database','database') }
20 elif args.dbtype == 'MySQLdb':
21 args.connect_args = { 'db' : config.get('database','database'), 'user' : config.get('database','username'), 'passwd' : config.get('database','password') }
22 elif args.dbtype == 'psycopg2':
23 args.connect_args = { 'database' : config.get('database','database') }
25 Datastore_class.__init__(self,args)
28 self.mempool_keys = {}
29 self.bitcoind_url = 'http://%s:%s@%s:%s/' % ( config.get('bitcoind','user'), config.get('bitcoind','password'), config.get('bitcoind','host'), config.get('bitcoind','port'))
31 self.address_queue = address_queue
35 def import_block(self, b, chain_ids=frozenset()):
36 block_id = super(MyStore, self).import_block(b, chain_ids)
37 #print "block", block_id
38 for pos in xrange(len(b['transactions'])):
39 tx = b['transactions'][pos]
41 tx['hash'] = util.double_sha256(tx['tx'])
42 tx_id = store.tx_find_id_and_value(tx)
44 self.update_tx_cache(tx_id)
46 print "error: import_block: no tx_id"
50 def update_tx_cache(self, txid):
51 inrows = self.get_tx_inputs(txid, False)
53 _hash = self.binout(row[6])
54 address = hash_to_address(chr(0), _hash)
55 if self.tx_cache.has_key(address):
56 print "cache: invalidating", address
57 self.tx_cache.pop(address)
58 self.address_queue.put(address)
60 outrows = self.get_tx_outputs(txid, False)
62 _hash = self.binout(row[6])
63 address = hash_to_address(chr(0), _hash)
64 if self.tx_cache.has_key(address):
65 print "cache: invalidating", address
66 self.tx_cache.pop(address)
67 self.address_queue.put(address)
69 def safe_sql(self,sql, params=(), lock=True):
71 if lock: dblock.acquire()
72 ret = self.selectall(sql,params)
73 if lock: dblock.release()
76 print "sql error", sql
79 def get_tx_outputs(self, tx_id, lock=True):
80 return self.safe_sql("""SELECT
82 txout.txout_scriptPubKey,
89 LEFT JOIN txin ON (txin.txout_id = txout.txout_id)
90 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
91 LEFT JOIN tx nexttx ON (txin.tx_id = nexttx.tx_id)
92 WHERE txout.tx_id = %d
93 ORDER BY txout.txout_pos
94 """%(tx_id), (), lock)
96 def get_tx_inputs(self, tx_id, lock=True):
97 return self.safe_sql(""" SELECT
101 COALESCE(prevtx.tx_hash, u.txout_tx_hash),
103 COALESCE(txout.txout_pos, u.txout_pos),
106 LEFT JOIN txout ON (txout.txout_id = txin.txout_id)
107 LEFT JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
108 LEFT JOIN tx prevtx ON (txout.tx_id = prevtx.tx_id)
109 LEFT JOIN unlinked_txin u ON (u.txin_id = txin.txin_id)
110 WHERE txin.tx_id = %d
111 ORDER BY txin.txin_pos
112 """%(tx_id,), (), lock)
114 def get_address_out_rows(self, dbhash):
115 return self.safe_sql(""" SELECT
125 FROM chain_candidate cc
126 JOIN block b ON (b.block_id = cc.block_id)
127 JOIN block_tx ON (block_tx.block_id = b.block_id)
128 JOIN tx ON (tx.tx_id = block_tx.tx_id)
129 JOIN txin ON (txin.tx_id = tx.tx_id)
130 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
131 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
132 WHERE pubkey.pubkey_hash = ?
133 AND cc.in_longest = 1""", (dbhash,))
135 def get_address_out_rows_memorypool(self, dbhash):
136 return self.safe_sql(""" SELECT
143 JOIN txin ON (txin.tx_id = tx.tx_id)
144 JOIN txout prevout ON (txin.txout_id = prevout.txout_id)
145 JOIN pubkey ON (pubkey.pubkey_id = prevout.pubkey_id)
146 WHERE pubkey.pubkey_hash = ? """, (dbhash,))
148 def get_address_in_rows(self, dbhash):
149 return self.safe_sql(""" SELECT
159 FROM chain_candidate cc
160 JOIN block b ON (b.block_id = cc.block_id)
161 JOIN block_tx ON (block_tx.block_id = b.block_id)
162 JOIN tx ON (tx.tx_id = block_tx.tx_id)
163 JOIN txout ON (txout.tx_id = tx.tx_id)
164 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
165 WHERE pubkey.pubkey_hash = ?
166 AND cc.in_longest = 1""", (dbhash,))
168 def get_address_in_rows_memorypool(self, dbhash):
169 return self.safe_sql( """ SELECT
176 JOIN txout ON (txout.tx_id = tx.tx_id)
177 JOIN pubkey ON (pubkey.pubkey_id = txout.pubkey_id)
178 WHERE pubkey.pubkey_hash = ? """, (dbhash,))
180 def get_history(self, addr):
182 cached_version = self.tx_cache.get( addr )
183 if cached_version is not None:
184 return cached_version
186 version, binaddr = decode_check_address(addr)
190 dbhash = self.binin(binaddr)
192 rows += self.get_address_out_rows( dbhash )
193 rows += self.get_address_in_rows( dbhash )
200 nTime, chain_id, height, is_in, blk_hash, tx_hash, tx_id, pos, value = row
202 print "cannot unpack row", row
204 tx_hash = self.hashout_hex(tx_hash)
207 "height": int(height),
209 "blk_hash": self.hashout_hex(blk_hash),
216 txpoints.append(txpoint)
217 known_tx.append(self.hashout_hex(tx_hash))
220 # todo: sort them really...
221 txpoints = sorted(txpoints, key=operator.itemgetter("nTime"))
225 rows += self.get_address_in_rows_memorypool( dbhash )
226 rows += self.get_address_out_rows_memorypool( dbhash )
227 address_has_mempool = False
230 is_in, tx_hash, tx_id, pos, value = row
231 tx_hash = self.hashout_hex(tx_hash)
232 if tx_hash in known_tx:
235 # this means that pending transactions were added to the db, even if they are not returned by getmemorypool
236 address_has_mempool = True
238 # this means pending transactions are returned by getmemorypool
239 if tx_hash not in self.mempool_keys:
242 #print "mempool", tx_hash
247 "blk_hash": 'mempool',
253 txpoints.append(txpoint)
256 for txpoint in txpoints:
257 tx_id = txpoint['tx_id']
260 inrows = self.get_tx_inputs(tx_id)
262 _hash = self.binout(row[6])
263 address = hash_to_address(chr(0), _hash)
264 txinputs.append(address)
265 txpoint['inputs'] = txinputs
267 outrows = self.get_tx_outputs(tx_id)
269 _hash = self.binout(row[6])
270 address = hash_to_address(chr(0), _hash)
271 txoutputs.append(address)
272 txpoint['outputs'] = txoutputs
274 # for all unspent inputs, I want their scriptpubkey. (actually I could deduce it from the address)
275 if not txpoint['is_in']:
276 # detect if already redeemed...
278 if row[6] == dbhash: break
281 #row = self.get_tx_output(tx_id,dbhash)
282 # pos, script, value, o_hash, o_id, o_pos, binaddr = row
283 # if not redeemed, we add the script
285 if not row[4]: txpoint['raw_scriptPubKey'] = row[1]
288 if not address_has_mempool:
289 self.tx_cache[addr] = txpoints
295 def memorypool_update(store):
297 ds = BCDataStream.BCDataStream()
298 previous_transactions = store.mempool_keys
299 store.mempool_keys = []
301 postdata = dumps({"method": 'getmemorypool', 'params': [], 'id':'jsonrpc'})
303 respdata = urllib.urlopen(store.bitcoind_url, postdata).read()
305 if r['error'] != None:
308 v = r['result'].get('transactions')
311 ds.write(hextx.decode('hex'))
312 tx = deserialize.parse_Transaction(ds)
313 tx['hash'] = util.double_sha256(tx['tx'])
314 tx_hash = store.hashin(tx['hash'])
316 store.mempool_keys.append(tx_hash)
317 if store.tx_find_id_and_value(tx):
320 tx_id = store.import_tx(tx, False)
321 store.update_tx_cache(tx_id)
326 def main_iteration(store):
330 store.memorypool_update()
332 block_number = store.get_block_number(1)
335 print "IOError: cannot reach bitcoind"
338 traceback.print_exc(file=sys.stdout)