alexis2/parser.py

#!/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
from time import strftime,time

from multiprocessing import Pool
import sys
from analytics import post_data


def parseSource(s):
    init = time()
    q = Queue()
    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)

    d = db()
    kw = d.get_keywords(s["category"])
    new_articles = 0
    if l is not None:
        for a in l.links:
            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)))
    new_articles = q.qsize()

    if not q.empty():
        num_worker_threads = min(5, q.qsize())
        threads = []
        for _ in range(num_worker_threads):
            t = threading.Thread(target=news_worker, args=[s, kw, d, q])
            # source, keywords, db, queue
            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 _ in range(num_worker_threads):
            # put num_worker empty jobs so
            # each worker can work once
            q.put(None)

        for t in threads:
            t.join(10)
    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 (ex_time, len(l.links), s["name"], new_articles)


def news_worker(source, kw, d, q):
    source_id = source["_id"]
    source_name = source["name"]
    source_category = source["category"]
    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,
                     source_category, kw, html=html)
            # print("[Thread %s] Finished" % tname)
            d.insert_article(n.get())
        except Exception as e:
            print("#### EXCEPTION ########")
            print(e)
            print(item)
            print("#### END EXCEPTION ####")
        finally:
            q.task_done()


# TODO influx start

post_data('event', 'start')
start = time()
sources = db().sources()
db().purge_error()
post_data('event', {'type': 'purge_complete'}, 1)
pool = Pool(processes=4)
ret = pool.map(parseSource, sources)
new = sum([r[3] for r in ret])
tot = sum([r[1] for r in ret])

post_data('value', {'type': 'new'}, new)
post_data('value', {'type': 'total'}, tot)
# TODO influx source new
# TODO influx source total
# n = sum(ret)
# sret = sorted(ret, key=lambda tup: tup[0])
# print(sret)

print("[%s] %d/%d new articles, total time %d sec" %
      (strftime("%H:%M:%S"), new, tot, time()-start))
print("#" * 75)
# TODO influx end
post_data('event', 'end')