12 changed files with 310 additions and 107 deletions
--- a/.gitignore
+++ b/.gitignore
@ -0,0 +1,5 @@
+log
+nohup.out
+*.swp
+*.log
+*.out
--- a/consumer.py
+++ b/consumer.py
@ -0,0 +1,66 @@
+import redis
+import urllib2
+import logging
+import MySQLdb
+import time
+import sys
+
+r = redis.Redis(host="192.168.80.55",port=6379,db=0)
+
+logger = logging.getLogger()
+hdlr = logging.FileHandler("consumer-%s.log"%sys.argv[1])
+formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
+hdlr.setFormatter(formatter)
+logger.addHandler(hdlr)
+logger.setLevel(logging.NOTSET)
+send_headers = {"Content-Type":"application/json","Authorization":""}
+conn = MySQLdb.connect(host="192.168.80.104",user="influx",passwd="influx1234",db="pages",charset='utf8' )
+cursor = conn.cursor()
+sql_add_repo = "insert into github_html_detail(url,detail,time) values(%s,%s,%s)"
+sql_update_state = "update repo_urls set state=1 where id=%s"
+sql_error = "insert into errors(url,message,time) values(%s,%s,%s)"
+def start():
+    logger.info("consumer begains to work")
+    task = r.blpop("repo_urls")[1]
+    while True:
+	logger.info("task : %s",task)
+	id_to_set_state = task[0:task.index("-")]
+	url_download = task[task.index("-")+1:]
+	logger.info(">>id:%s - url:%s"%(id_to_set_state,url_download))
+	
+	token = r.blpop("tokens")[1]
+	r.rpush("tokens",token)
+	logger.info(">>token:%s will be used"%token)
+	
+	send_headers["Authorization"] = "token %s"%token
+	req = urllib2.Request(url_download,headers = send_headers)
+	try:	
+		response = urllib2.urlopen(req,timeout=20)
+		repo = response.read()
+		logger.info(">>done downloading")
+		cursor.execute(sql_add_repo,(url_download,repo,time.strftime('%Y-%m-%d %H:%M:%S')))
+		cursor.execute(sql_update_state,(id_to_set_state,))
+		conn.commit()
+		logger.info(">>done storing")
+	except urllib2.URLError,e:
+        	if hasattr(e,"reason"):
+            		logger.error("Failed to reach the server")
+            		logger.error("The reason: "%e.reason)
+            		cursor.execute(sql_error,(url_download,e.reason,time.strftime('%Y-%m-%d %H:%M:%S')))
+       		elif hasattr(e,"code"):
+            		logger.error("The server couldn't fulfill the request")
+            		logger.error("Error code: %s"%str(e.code))
+            		logger.error("Return content: %s"%e.read())
+            		cursor.execute(sql_error,(url_download,e.read(),time.strftime('%Y-%m-%d %H:%M:%S')))
+        	else:
+           		logger.error(e.msg)     
+        	conn.commit()
+	except Exception,e:
+		logger.error(e)
+		cursor.execute(sql_error,(url_download,e,time.strftime('%Y-%m-%d %H:%M:%S')))
+		conn.commit()
+	task = r.blpop("repo_urls")[1] 
+
+if __name__ == "__main__":
+    start()
+
--- a/consumer_home_page.py
+++ b/consumer_home_page.py
@ -1,63 +0,0 @@
-import json
-import redis
-import urllib2
-import logging
-import MySQLdb
-import time
-import sys
-r = redis.Redis(host="192.168.80.55",port=6379,db=0)
-logger = logging.getLogger()
-hdlr = logging.FileHandler("consumer-%s.log"%sys.argv[1])
-formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
-hdlr.setFormatter(formatter)
-logger.addHandler(hdlr)
-logger.setLevel(logging.NOTSET)
-send_headers = {"Content-Type":"application/json","Authorization":""}
-conn = MySQLdb.connect(host="192.168.80.104",user="influx",passwd="influx1234",db="pages",charset='utf8' )
-cursor = conn.cursor()
-sql_add_repo = "insert into github_home_detail(url,detail,time,id) values(%s,%s,%s,%s)"
-sql_update_state = "update repo_urls set home_state=1 where id=%s"
-sql_update_error="update repo_urls set home_errors=%s where id=%s"
-sql_error = "insert into errors(url,message,time) values(%s,%s,%s)"
-retry_time=5
-def start():
-    logger.info("consumer_home_page begains to work")
-    task = r.blpop("home_urls")[1]
-    while True:
-        logger.info("task : %s",task)
-        id_to_set_state = task[0:task.index("-")]
-        error_num= task[task.index("-")+1:task.index("-",task.index("-")+1)]
-        url_download = task[task.index("-",task.index("-")+1)+1:]
-        logger.info(">>id:%s - url:%s"%(id_to_set_state,url_download))
-        req = urllib2.Request(url_download)
-        try:
-                response = urllib2.urlopen(req,timeout=20)
-                repo = response.read()
-                logger.info(">>done downloading")
-                cursor.execute(sql_add_repo,(url_download,repo,time.strftime('%Y-%m-%d %H:%M:%S'),id_to_set_state))
-                cursor.execute(sql_update_state,(id_to_set_state,))
-                conn.commit()
-                logger.info(">>done storing")
-        except urllib2.URLError,e:
-                if hasattr(e,"reason"):
-                            logger.error("Failed to reach the server")
-                            logger.error("The reason: %s"%e.reason)
-                            cursor.execute(sql_error,(url_download,e.reason,time.strftime('%Y-%m-%d %H:%M:%S')))
-                elif hasattr(e,"code"):
-                            logger.error("The server couldn't fulfill the request")
-                            logger.error("Error code: %s"%str(e.code))
-                            logger.error("Return content: %s"%e.read())
-                            cursor.execute(sql_error,(url_download,e.read(),time.strftime('%Y-%m-%d %H:%M:%S')))
-                else:
-                           logger.error(e.msg)
-                cursor.execute(sql_update_error,(int(error_num)+1,id_to_set_state))
-                if(error_num<retry_time):
-                    r.rpush("home_urls","%d-%d-%s"%(id_to_set_state,int(error_num)+1,url_download))
-                conn.commit()
-        except Exception,e:
-                logger.error(e)
-                cursor.execute(sql_error,(url_download,e,time.strftime('%Y-%m-%d %H:%M:%S')))
-                conn.commit()
-        task = r.blpop("home_urls")[1]
-if __name__ == "__main__":
-    start()
--- a/list_repos.py
+++ b/list_repos.py
@ -0,0 +1,79 @@
+import urllib2
+import json
+import MySQLdb
+import time
+import logging
+
+logger = logging.getLogger()
+hdlr = logging.FileHandler("log")
+formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
+hdlr.setFormatter(formatter)
+logger.addHandler(hdlr)
+logger.setLevel(logging.NOTSET)
+
+conn = MySQLdb.connect(host="192.168.80.104",user="influx",passwd="influx1234",db="pages",charset='utf8' )
+
+cursor = conn.cursor()
+sql_insert = "insert into repo_urls(link,url,full_name,state) values(%s,%s,%s,%s)"
+sql_error = "insert into errors(url,message,time) values(%s,%s,%s)"
+
+send_headers = {"Content-Type":"application/json","Authorization":"token e41a69a15db215bd7959320360e77a988c4e19a3"}
+url = "https://api.github.com/repositories?since=444679"
+
+while True:  
+    try:
+        req = urllib2.Request(url,headers = send_headers) 
+        logger.info("process link :%s"%url)
+        result = urllib2.urlopen(req,timeout=10)
+        prjs = json.loads(result.read())    
+	logger.info("done downloading and parsing json")
+        for prj in prjs:
+            try:
+                cursor.execute(sql_insert,(url,prj.get("url"),prj.get("full_name"),0))
+            except Exception,e:
+                logger.error(e.message)
+	logger.info("done storing to db")
+        conn.commit()
+        links = result.info().get("Link")
+        if "next" in links:
+            url = links[1:links.index(">")]
+            logger.info("next url:%s"%url)
+        else:
+            logger.info("no next link any more")
+            num_repos = len(prjs)
+            while True:
+                time.sleep(10*60)
+                logger.info("check it again")
+                result = urllib2.urlopen(req)
+                prjs = json.loads(result.read())
+                new_num_repos = len(prjs)
+                if new_num_repos > num_repos:
+                    for prj in prjs[num_repos:new_num_repos]:
+                        cursor.execute(sql_insert,(url,prj.get("url"),prj.get("full_name"),0))
+                    conn.commit()
+                    num_repos = new_num_repos
+                links = result.info().get("Link")
+                if "next" in links:
+                    url = links[0:links.index(">")]
+                    logger.info("yeah, next link is here!!")
+                    break
+            
+    except urllib2.URLError,e:
+        if hasattr(e,"reason"):
+            logger.error("Failed to reach the server")
+            logger.error("The reason: "%e.reason)
+            cursor.execute(sql_error,(url,e.reason,time.strftime('%Y-%m-%d %H:%M:%S')))
+        elif hasattr(e,"code"):
+            logger.error("The server couldn't fulfill the request")
+            logger.error("Error code: %s"%str(e.code))
+            logger.error("Return content: %s"%e.read())
+            cursor.execute(sql_error,(url,e.read(),time.strftime('%Y-%m-%d %H:%M:%S')))
+        else:
+            pass     
+        conn.commit()
+    except Exception,e:
+        logger.error(e.message)
+        pass
+
+cursor.close()
+conn.close()
--- a/producer.py
+++ b/producer.py
@ -0,0 +1,48 @@
+import redis
+import logging
+import MySQLdb
+import time
+
+r = redis.Redis(host="192.168.80.55",port=6379,db=0)
+
+logger = logging.getLogger()
+hdlr = logging.FileHandler("produce.log")
+formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
+hdlr.setFormatter(formatter)
+logger.addHandler(hdlr)
+logger.setLevel(logging.NOTSET)
+
+conn = MySQLdb.connect(host="192.168.80.104",user="influx",passwd="influx1234",db="pages",charset='utf8' )
+cursor = conn.cursor()
+sql_select_pointer = "select pointer from pointers where table_name='repo_urls'"
+sql_update_pointer = "update pointers set pointer=%s where table_name='repo_urls'"
+sql_add_repo = "insert into github_html_detail(url,detail,time) values(%s,%s,%s)"
+sql_update_pushtime = "update repo_urls set push_time=%s where id=%s"
+sql_select_url = "select id,url from repo_urls where id>%s limit %s"
+patch_size = 1000 #how many urls push to redis every time
+def start():
+    logger.info("producer begains to work")
+    while True:
+	cursor.execute(sql_select_pointer)
+	last_id = cursor.fetchone()[0]
+	logger.info(">>id after last process: %d",last_id)
+	
+	num_redis = r.llen("repo_urls")
+	logger.info(">>%d urls in redis"%num_redis)
+	if num_redis < 1000:
+	    logger.info(">>add urls to redis")
+	    cursor.execute(sql_select_url,(last_id,patch_size))
+	    urls_added = cursor.fetchall()
+	    for url in urls_added:
+		r.rpush("repo_urls","%d-%s"%(url[0],url[1]))
+		cursor.execute(sql_update_pushtime,(time.time(),url[0]))
+	    cursor.execute(sql_update_pointer,(last_id + len(urls_added),))
+	    conn.commit()
+	else:
+	    logger.info(">>enough urls")
+
+	time.sleep(10)
+
+if __name__ == "__main__":
+    start()
+
--- a/producer_home_page.py
+++ b/producer_home_page.py
@ -1,44 +0,0 @@
-import redis
-import logging
-import MySQLdb
-import time
-r = redis.Redis(host="192.168.80.55",port=6379,db=0)
-logger = logging.getLogger()
-hdlr = logging.FileHandler("produce.log")
-formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
-hdlr.setFormatter(formatter)
-logger.addHandler(hdlr)
-logger.setLevel(logging.NOTSET)
-conn = MySQLdb.connect(host="192.168.80.104",user="influx",passwd="influx1234",db="pages",charset='utf8' )
-cursor = conn.cursor()
-sql_select_pointer = "select pointer from pointers where table_name='home_urls'"
-sql_update_pointer = "update pointers set pointer=%s where table_name='home_urls'"
-sql_update_pushtime = "update repo_urls set home_push_time=%s where id=%s"
-sql_select_url = "select id,home_errors,url from repo_urls where id>%s and home_state=0 limit %s"
-patch_size = 1000 #how many urls push to redis every time
-html_mode="https://github.com/%s/%s"
-# r.flushdb()
-def start():
-    logger.info("producer_home_page begains to work")
-    while True:
-        cursor.execute(sql_select_pointer)
-        last_id = cursor.fetchone()[0]
-        logger.info(">>id after last process: %d",last_id)
-        num_redis = r.llen("home_urls")
-        print num_redis
-        logger.info(">>%d urls in redis"%num_redis)
-        if num_redis < patch_size:
-            logger.info(">>add home_urls to redis")
-            cursor.execute(sql_select_url,(last_id,patch_size))
-            urls_added = cursor.fetchall()
-            for url in urls_added:
-                html_way=html_mode%(url[2].split("/")[4],url[2].split("/")[5])
-                r.rpush("home_urls","%d-%d-%s"%(url[0],url[1],html_way))
-                cursor.execute(sql_update_pushtime,(time.time(),url[0]))
-            cursor.execute(sql_update_pointer,(last_id + len(urls_added),))
-            conn.commit()
-        else:
-            logger.info(">>enough urls")
-        time.sleep(10)
-if __name__ == "__main__":
-    start()
--- a/requirements.txt
+++ b/requirements.txt
@ -0,0 +1,2 @@
+MySQL-python==1.2.5
+redis==2.10.5
--- a/start.sh
+++ b/start.sh
@ -0,0 +1 @@
+nohup python consumer.py $1 2>&1 &
--- a/state_monitor.py
+++ b/state_monitor.py
@ -0,0 +1,46 @@
+import MySQLdb
+import time
+import redis
+
+import logging
+r = redis.Redis(host="192.168.80.55",port=6379,db=0)
+logger = logging.getLogger()
+hdlr = logging.FileHandler("state_monitor.log")
+formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
+hdlr.setFormatter(formatter)
+logger.addHandler(hdlr)
+logger.setLevel(logging.NOTSET)
+
+conn = MySQLdb.connect(host="192.168.80.104",user="influx",passwd="influx1234",db="pages",charset='utf8' )
+
+cursor = conn.cursor()
+sql_update_pushtime = "update repo_urls set push_time=%s where id=%s"
+sql_scan_state = "select id,url,push_time from repo_urls where id>%s and state=0 and push_time<%s limit 500"
+sql_select_monitor_pointer = "select pointer from pointers where table_name='state_monitor'"
+sql_update_monitor_pointer = "update pointers set pointer=%s where table_name='state_monitor'"
+#url whose state hasn't been set to 1 'time_limit' after  pushed will be repushed
+time_limit = 60 * 10
+scan_period = time_limit
+def start():
+	logger.info("state_monitor begains to work")
+	while True:
+		# scan the database, select top state-unset urls
+		cursor.execute(sql_select_monitor_pointer)
+		last_id = cursor.fetchone()[0]
+		logger.info("last scan pointer:%d"%last_id)
+		cursor.execute(sql_scan_state,(last_id,time.time() - time_limit))
+		unstated_urls = cursor.fetchall()	
+		logger.info("%d unstated urls will be repushed"%len(unstated_urls))
+		for url in unstated_urls:
+			r.rpush("repo_urls","%d-%s"%(url[0],url[1]))
+			cursor.execute(sql_update_pushtime,(time.time(),url[0]))
+			logger.info( "%d-%d-%s"%(url[0],url[2],time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(url[2]))))	
+		last_id = unstated_urls[len(unstated_urls)-1][0]
+		cursor.execute(sql_update_monitor_pointer,(last_id,))
+		conn.commit()
+		time.sleep(time_limit)
+
+
+
+if __name__ == "__main__":
+	start()
--- a/text.py
+++ b/text.py
@ -0,0 +1,32 @@
+import urllib2
+import json
+import commands
+import time
+url = "https://api.github.com/repositories"
+send_headers = {"Content-Type":"application/json"}
+
+def get_ip_address(ifname):
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    return socket.inet_ntoa(fcntl.ioctl(
+        s.fileno(),
+        0x8915,  # SIOCGIFADDR
+        struct.pack('256s', ifname[:15])
+    )[20:24])
+
+while True:
+    req = urllib2.Request(url,headers = send_headers)
+    try :
+	result = urllib2.urlopen(req)
+	ratelimit_remaining = result.info().get("X-RateLimit-Remaining")
+        print "ratelimit_remaining: %s"%ratelimit_remaining
+        if ratelimit_remaining == 0:
+	    print get_ip_address("eth0")
+            commands.getoutput("pppoe-stop")
+	    time.sleep(1)
+	    commands.getoutput("pppoe-start")
+	    print get_ip_address("eth0")
+	next_url = result.info().get("Link")
+	print next_url.index(">")
+	url = next_url[1,next_url.index(">")]
+    except Exception, e:
+	print e
--- a/token.py
+++ b/token.py
@ -0,0 +1,18 @@
+import redis
+import logging
+
+
+logger = logging.getLogger()
+hdlr = logging.FileHandler("token.log")
+formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
+hdlr.setFormatter(formatter)
+logger.addHandler(hdlr)
+logger.setLevel(logging.NOTSET)
+
+r = redis.Redis(host="192.168.80.55",port=6379,db=0)
+
+with open("tokens","r") as file:
+	for line in file.readlines():
+		logger.info("add token: %s",line.strip())
+		r.rpush("tokens",line.strip())	
+
--- a/13
+++ b/13
@ -0,0 +1,13 @@
+c9b850151b3aeb6d750619653bdd25ad27e79e76
+a36b692adab5f4bfa8d03fba34669b78fe52d134
+d1e7a9d2eb0f54ec0e96ca06a64d68897d122513
+cd1c3cda5c269c30ca3527968ea1a4e41038d8f1
+25826285cf58b8bea2712298f2bc1dc598c8200b
+277152ee058361c684ee253b881141bba722dfc5
+7933edb46487203f2752a93c702fd630cd552bae
+9377749a42d8492428c52329624a8388b620e926
+0e862a968c0f7ba84acc9b909c3fccdf3e9cd15a
+877738b0ede13b627605e301dd4f00725697ca0d
+8368357f10e6318309b7e278b900e375f73421bd
+6ac08b18aa04a36b602957be808a813900487173
+e94b02919d147be8df7dfc5ede4de92c3a75b1b4