Compare commits
No commits in common. "master" and "home_page" have entirely different histories.
|
|
@ -1,5 +0,0 @@
|
|||
log
|
||||
nohup.out
|
||||
*.swp
|
||||
*.log
|
||||
*.out
|
||||
66
consumer.py
66
consumer.py
|
|
@ -1,66 +0,0 @@
|
|||
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()
|
||||
|
||||
|
|
@ -0,0 +1,63 @@
|
|||
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()
|
||||
|
|
@ -1,79 +0,0 @@
|
|||
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()
|
||||
48
producer.py
48
producer.py
|
|
@ -1,48 +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='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()
|
||||
|
||||
|
|
@ -0,0 +1,44 @@
|
|||
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()
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
MySQL-python==1.2.5
|
||||
redis==2.10.5
|
||||
|
|
@ -1,46 +0,0 @@
|
|||
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()
|
||||
32
text.py
32
text.py
|
|
@ -1,32 +0,0 @@
|
|||
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
|
||||
18
token.py
18
token.py
|
|
@ -1,18 +0,0 @@
|
|||
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())
|
||||
|
||||
13
tokens
13
tokens
|
|
@ -1,13 +0,0 @@
|
|||
c9b850151b3aeb6d750619653bdd25ad27e79e76
|
||||
a36b692adab5f4bfa8d03fba34669b78fe52d134
|
||||
d1e7a9d2eb0f54ec0e96ca06a64d68897d122513
|
||||
cd1c3cda5c269c30ca3527968ea1a4e41038d8f1
|
||||
25826285cf58b8bea2712298f2bc1dc598c8200b
|
||||
277152ee058361c684ee253b881141bba722dfc5
|
||||
7933edb46487203f2752a93c702fd630cd552bae
|
||||
9377749a42d8492428c52329624a8388b620e926
|
||||
0e862a968c0f7ba84acc9b909c3fccdf3e9cd15a
|
||||
877738b0ede13b627605e301dd4f00725697ca0d
|
||||
8368357f10e6318309b7e278b900e375f73421bd
|
||||
6ac08b18aa04a36b602957be808a813900487173
|
||||
e94b02919d147be8df7dfc5ede4de92c3a75b1b4
|
||||
Loading…
Reference in New Issue