Merge branch 'master' into develop

api/search/fetch_data.py

@@ -23,16 +23,19 @@ This file is part of Blockstack.
     along with Blockstack. If not, see <http://www.gnu.org/licenses/>.
 """
 
-import sys
+import sys, os, time
+import tempfile
 import json
+from datetime import datetime
 
-from api.config import SEARCH_BLOCKCHAIN_DATA_FILE as BLOCKCHAIN_DATA_FILE, \
-    SEARCH_PROFILE_DATA_FILE as PROFILE_DATA_FILE
+from api.config import (
+    SEARCH_BLOCKCHAIN_DATA_FILE, SEARCH_PROFILE_DATA_FILE,
+    SEARCH_LAST_INDEX_DATA_FILE, SEARCH_LOCKFILE)
 
 from .utils import validUsername
 from .utils import get_json, config_log
 
-from blockstack_client.proxy import get_all_names
+from blockstack_client  import proxy
 from blockstack_client.profile import get_profile
 from api.utils import profile_log
 import logging
@@ -44,14 +47,10 @@ def fetch_namespace():
         Fetch all names in a namespace that should be indexed.
         Data is saved in data/ directory
     """
+    resp = proxy.get_all_names()
 
-    resp = get_all_names()
-
-    fout = open(BLOCKCHAIN_DATA_FILE, 'w')
-    fout.write(json.dumps(resp))
-    fout.close()
-
-    return
+    with open(SEARCH_BLOCKCHAIN_DATA_FILE, 'w') as fout:
+        fout.write(json.dumps(resp))
 
 def print_status_bar(filled, total):
     pct = float(filled) / total
@@ -60,6 +59,75 @@ def print_status_bar(filled, total):
     sys.stdout.write(out)
     sys.stdout.flush()
 
+def update_profiles():
+    if not os.path.exists(SEARCH_LAST_INDEX_DATA_FILE):
+        return {'error' : 'No last index, you need to rebuild the whole index.'}
+    with open(SEARCH_LAST_INDEX_DATA_FILE, 'r') as fin:
+        search_indexer_info = json.load(fin)
+
+    last_block_processed = search_indexer_info['last_block_height']
+    last_full_index = search_indexer_info['last_full_index']
+
+    info_resp = proxy.getinfo()
+    try:
+        new_block_height = info_resp['last_block_processed']
+    except:
+        print info_resp
+        raise
+
+    if last_block_processed - 1 > new_block_height:
+        return {'status' : True, 'message' : 'No new blocks since last indexing'}
+
+    # aaron: note, sometimes it may take a little while for
+    #  new zonefiles to have propagated to the network, so
+    #  we over-fetch a little bit
+    zonefiles_resp = proxy.get_zonefiles_by_block(
+        last_block_processed - 1, new_block_height)
+    zonefiles_updated = zonefiles_resp['zonefile_info']
+    names_updated = set(
+        [ zf_info['name'] for zf_info in zonefiles_updated
+          if 'name' in zf_info ])
+    updated_profiles = {}
+    actually_updated_names = set()
+    print "Updating {} entries...".format(len(names_updated))
+    for ix, name in enumerate(names_updated):
+        print_status_bar(ix+1, len(names_updated))
+        profile_entry = {}
+        profile_entry['fqu'] = name
+
+        try:
+            profile_entry['profile'] = get_profile(name, use_legacy = True)['profile']
+            updated_profiles[name] = (profile_entry)
+            actually_updated_names.add(name)
+        except KeyboardInterrupt as e:
+            raise e
+        except:
+            import traceback as tb; tb.print_exc()
+
+    names_updated = actually_updated_names
+    with open(SEARCH_PROFILE_DATA_FILE, 'r') as fin:
+        all_profiles = json.load(fin)
+    to_remove = []
+    for ix, profile in enumerate(all_profiles):
+        if profile['fqu'] in names_updated:
+            all_profiles[ix] = updated_profiles[profile['fqu']]
+
+
+    if not obtain_lockfile():
+        return {'error' : 'Could not obtain lockfile, abandoning my update.'}
+    with open(SEARCH_LAST_INDEX_DATA_FILE, 'r') as fin:
+        search_indexer_info = json.load(fin)
+    if search_indexer_info['last_full_index'] != last_full_index:
+        return {'error' : 'Full re-index written during our update. Abandoning'}
+
+    with open(SEARCH_PROFILE_DATA_FILE, 'w') as fout:
+        json.dump(all_profiles, fout)
+    with open(SEARCH_LAST_INDEX_DATA_FILE, 'w') as fout:
+        search_indexer_info['last_block_height'] = new_block_height
+        json.dump(search_indexer_info, fout)
+
+    return {'status' : True, 'message' : 'Indexed {} profiles'.format(len(names_updated))}
+
 def fetch_profiles(max_to_fetch = None, just_test_set = False):
     """
         Fetch profile data using Blockstack Core and save the data.
@@ -69,14 +137,14 @@ def fetch_profiles(max_to_fetch = None, just_test_set = False):
         * profile: json profile data
     """
 
-    fin = open(BLOCKCHAIN_DATA_FILE, 'r')
-    file = fin.read()
-    fin.close()
+    with open(SEARCH_BLOCKCHAIN_DATA_FILE, 'r') as fin:
+        all_names = json.load(file)
 
-    all_names = json.loads(file)
+    info_resp = proxy.getinfo()
+    last_block_processed = info_resp['last_block_processed']
 
     all_profiles = []
-    
+
     if max_to_fetch == None:
         max_to_fetch = len(all_names)
 
@@ -101,13 +169,52 @@ def fetch_profiles(max_to_fetch = None, just_test_set = False):
         except:
             pass
 
-    fout = open(PROFILE_DATA_FILE, 'w')
-    fout.write(json.dumps(all_profiles))
-    fout.close()
+    attempts = 0
+    while not obtain_lockfile():
+        attempts += 1
+        time.sleep(5)
+        if attempts > 10:
+            print "ERROR! Could not obtain lockfile"
+            return
 
-    return
+    with open(SEARCH_PROFILE_DATA_FILE, 'w') as fout:
+        json.dump(all_profiles, fout)
+    with open(SEARCH_LAST_INDEX_DATA_FILE, 'w') as fout:
+        search_index_data = {
+            'last_block_height' : last_block_processed,
+            'last_full_index' : datetime.now().isoformat()
+        }
+        json.dump(search_index_data, fout)
 
 
+def obtain_lockfile():
+    if os.path.exists(SEARCH_LOCKFILE):
+        with open(SEARCH_LOCKFILE, 'r') as fin:
+            pid = json.load(fin)
+        try:
+            os.kill(pid, 0)
+            return False # lockfile exists, pid still running.
+        except:
+            pass
+        # lockfile stale. unlink it
+        os.unlink(SEARCH_LOCKFILE)
+    fd, path = tempfile.mkstemp(prefix=".indexer.lock.", dir=os.path.dirname(SEARCH_LOCKFILE))
+    try:
+        with os.fdopen(fd, 'w') as fout:
+            json.dump(os.getpid(), fout)
+        os.link( path, SEARCH_LOCKFILE )
+        os.unlink( path )
+    except:
+        import traceback as tb; tb.print_exc()
+        return False
+    # make sure we got it
+    with open(SEARCH_LOCKFILE, 'r') as fin:
+        pid = json.load(fin)
+    if pid == os.getpid():
+        return True
+    print "Wrong pid : {} != {}".format(pid, os.getpid())
+    return False
+
 if __name__ == "__main__":
 
     if(len(sys.argv) < 2):
@@ -129,6 +236,8 @@ if __name__ == "__main__":
             else:
                 args['max_to_fetch'] = int(sys.argv[2])
         fetch_profiles(**args)
-
+    elif(option == '--update_profiles'):
+        print json.dumps(update_profiles(),
+                         indent = 2)
     else:
         print "Usage error"