#!/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 from paid_parser import PaidParser def parseSource(s): global q q=Queue() 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"]) # print(kw) if l is not None: for a in l.links: if not d.article_exists(a): q.put(a) print("[Source] %s: %d/%d new/total articles" % ( s["name"], q.qsize(), len(l.links))) if not q.empty(): num_worker_threads=min(10,q.qsize()) threads=[] for i in range(num_worker_threads): t = threading.Thread(target=news_worker,args=[s,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) for t in threads: t.join(10) # print("[Source] %s => Finished parsing" %s["name"]) def news_worker(source,kw): source_id=source["_id"] source_name=source["name"] paid=False if "paid" in source and source["paid"]: paid=True tname=threading.current_thread().name while True: try: item = q.get(timeout=5) except: print("Failed to get item") q.task_done() break if item is None: q.task_done() break print("[Thread %s] %s" % (tname,item)) try: html=None if paid: p=PaidParser(item) html=p.html n=News(item,source_id,source_name,kw,html=html) print("[Thread %s] Finished" % tname) d.insert_article(n.get()) except Exception as e: print(e) print("whut") finally: q.task_done() q=Queue() d=db() d.purge_error() for s in d.sources(): parseSource(s)