state_monitor"

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()