misc+pep8

parser.py

@@ -7,14 +7,15 @@ from source import Source
 from linklist import LinkList
 from news import News
 from paid_parser import PaidParser
-from time import strftime
+from time import strftime,time
+
+from multiprocessing import Pool
 import sys
 
 
 def parseSource(s):
-    global q
+    init = time()
     q = Queue()
-    num_worker_threads = 10
     sel = None
     rss = False
     if "selector" in s:
@@ -23,22 +24,23 @@ def parseSource(s):
         rss = bool(s["rss"])
     l = LinkList(s["link"], sel, rss=rss)
 
+    d = db()
     kw = d.get_keywords(s["category"])
     new_articles = 0
 #    print(kw)
     if l is not None:
         for a in l.links:
-            if not d.article_exists(a.lower()):
+            if not d.article_exists(a.lower(),s["_id"]):
                 q.put(a)
-        print("[%s][Source] %s: %d/%d new/total articles" %
-              (strftime("%H:%M:%S"), s["name"], q.qsize(), len(l.links)))
+        #print("[%s][Source] %s: %d/%d new/total articles" %
+        #      (strftime("%H:%M:%S"), s["name"], q.qsize(), len(l.links)))
     new_articles = q.qsize()
 
     if not q.empty():
-        num_worker_threads = min(10, q.qsize())
+        num_worker_threads = min(5, q.qsize())
         threads = []
         for i in range(num_worker_threads):
-            t = threading.Thread(target=news_worker, args=[s, kw])
+            t = threading.Thread(target=news_worker, args=[s, kw, d, q])
             t.daemon = True
             t.start()
             threads.append(t)
@@ -51,11 +53,17 @@ def parseSource(s):
             q.put(None)
         for t in threads:
             t.join(10)
-#    print("[Source] %s => Finished parsing" %s["name"])
-    return new_articles
+    ex_time = time()-init
+    print("[%s][Source] %s => Finished parsing, %d/%d new articles, in %0.3fsec" % 
+		(strftime("%H:%M:%S"),
+		s["name"], new_articles, len(l.links), ex_time)
+	)
+    sys.stdout.flush()
+    #return new_articles
+    return (ex_time, len(l.links), s["name"], new_articles)
 
 
-def news_worker(source, kw):
+def news_worker(source, kw, d, q):
     source_id = source["_id"]
     source_name = source["name"]
     source_category = source["category"]
@@ -91,18 +99,20 @@ def news_worker(source, kw):
             print("#### EXCEPTION ########")
             print(e)
             print(item)
-            print("whut")
             print("#### END EXCEPTION ####")
         finally:
             q.task_done()
 
-q = Queue()
-d = db()
-d.purge_error()
-n = 0
-for s in d.sources():
-    n = n + parseSource(s)
-    sys.stdout.flush()
+start = time()
+sources = db().sources()
+db().purge_error()
+p = Pool(processes=4)
+ret = p.map(parseSource, sources)
+n = sum([ r[3] for r in ret ])
+tot = sum([ r[1] for r in ret ])
+#n = sum(ret)
+#sret = sorted(ret, key=lambda tup: tup[0])
+#print(sret)
 
-print("[%s] %d new articles" % (strftime("%H:%M:%S"), n))
+print("[%s] %d/%d new articles, total time %d sec" % (strftime("%H:%M:%S"), n, tot, time()-start))
 print("###############################################################################")