multithread v1.1

parser.py

@@ -19,6 +19,8 @@ class LinkList():
         self.links=[]
         if rss:
             feed = feedparser.parse(url)
+            if feed["bozo"]:
+                print("RSS ERROR. PANIC")
             #print(feed["bozo"]) FIXME: If bozo==1 => error
             self.links=[i["link"] for i in feed["items"]]
             print(self.links)
@@ -33,12 +35,12 @@ class LinkList():
                 s.parse_categories()
                 s.generate_articles()
     
-                self.links=list(set([a.url for a in s.articles])) #avoid dupes
+                self.links=[a.url for a in s.articles] 
             else:
                 f=FrontPage(url,selector)
                 self.links=f.links
-
         self.links=self.purgeLinks(self.links)
+        self.links=list(set(self.links)) #avoid dupes
 
     def purgeLinks(self,l):
         return [ sanitizeUrl(link) for link in l if not "presslist" in link.lower() and not "videolist" in link.lower() ]
@@ -64,7 +66,12 @@ class News():
 
         a = Article(url, language=lang,keep_article_html=True)
         a.download()
-        a.parse()
+        try:
+            a.parse()
+        except newspaper.article.ArticleException:
+            print("ERROR parsing article")
+            ERROR=True
+            return
         a.nlp()
         
         a.authors=self.dedup([self.fix_author(value) for value in a.authors if not self.isComment(value) and not self.isNoise(value)])
@@ -129,28 +136,27 @@ def parseSource(s):
     l=LinkList(s["link"],sel,rss=rss)
 
     if l is not None:
+        print("[Source] %s: %d total articles" % ( s["name"], len(l.links)))
         for a in l.links:
             if not d.article_exists(a):
-                print(a)
                 q.put(a)
-        print(len(l.links))
-
-    num_worker_threads=8
-    threads=[]
-    for i in range(num_worker_threads):
-        t = threading.Thread(target=news_worker,args=[s["_id"]])
-        t.daemon=True
-        t.start()
-        threads.append(t)
-
-    q.join()
-    for i in range(num_worker_threads):
-        q.put(None)
-    for t in threads:
-        t.join()
-    print("Finished parsing %s" %s)
+        print("[Source] %s: %d new articles" % ( s["name"], q.qsize()))
 
 
+    if not q.empty():
+        threads=[]
+        for i in range(num_worker_threads):
+            t = threading.Thread(target=news_worker,args=[s["_id"]])
+            t.daemon=True
+            t.start()
+            threads.append(t)
+    
+        q.join()
+        for i in range(num_worker_threads):
+            q.put(None)
+        for t in threads:
+            t.join()
+    print("[Source] %s => Finished parsing" %s["name"])
 
 def news_worker(source_id):
     while True:
@@ -158,13 +164,12 @@ def news_worker(source_id):
         if item is None:
             break
         n=News(item,source_id)
-        print("[Thread %s] %s" % (threading.current_thread(),n.URL))
+        print("[Thread %s] %s" % (threading.current_thread().name,n.URL))
         d.insert_article(n.get())
         q.task_done()
 
 q=Queue()
 d=db()
+num_worker_threads=20
 for s in d.sources():
-    print(s["name"])
     parseSource(s)
-    break