#!/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): sel=None rss=False if "selector" in s: sel=s["selector"] if "rss" in s: rss=bool(["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())) 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 (0 if num_workers > new_articles)" % q.qsize()) q.join() #block until all tasks are done print("Finished waiting for the queue") #stop workers for i in range(num_worker_threads): q.put(None) print("Waiting for %d threads" % len(threads)) for t in threads: t.join() print("[Source] %s => Finished parsing" %s["name"]) def news_worker(source_id,kw): while True: item = q.get() if item is None: break print("[Thread %s] %s" % (threading.current_thread().name,item)) n=News(item,source_id,kw) print("[Thread %s] Finished" % threading.current_thread().name) d.insert_article(n.get()) q.task_done() q=Queue() d=db() d.purge_error() num_worker_threads=10 for s in d.sources(): parseSource(s)