#!/usr/bin/python3 import threading from queue import Queue from db import db from source import Source from linklist import LinkList from news import News def parseSource(s): num_worker_threads=10 sel=None rss=False if "selector" in s: sel=s["selector"] if "rss" in s: rss=bool(s["rss"]) l=LinkList(s["link"],sel,rss=rss) kw=d.get_keywords(s["category"]) 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): q.put(a) print("[Source] %s: %d new articles" % ( s["name"], q.qsize())) num_worker_threads=min(10,q.qsize()) if not q.empty(): threads=[] for i in range(num_worker_threads): t = threading.Thread(target=news_worker,args=[s["_id"],kw]) t.daemon=True t.start() threads.append(t) print("Waiting for %d queue elements" % q.qsize()) q.join() #block until all tasks are done print("Finished waiting for the queue") for i in range(num_worker_threads): q.put(None) print("Waiting for %d threads" % len(threads)) for t in threads: t.join(10) print("[Source] %s => Finished parsing" %s["name"]) def news_worker(source_id,kw): tname=threading.current_thread().name while True: try: item = q.get(timeout=5) except: print("[%s] Empty queue" % tname) break if item is None: print("[%s] Empty queue (none)" % tname) q.task_done() break print("[Thread %s] %s" % (tname,item)) try: n=News(item,source_id,kw) print("[Thread %s] Finished" % tname) d.insert_article(n.get()) except: print("whut") finally: q.task_done() q=Queue() d=db() d.purge_error() for s in d.sources(): parseSource(s)