| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108 |
- #!/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
- import sys
- 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"])
- new_articles = 0
- # print(kw)
- if l is not None:
- for a in l.links:
- if not d.article_exists(a.lower()):
- 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(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"])
- return new_articles
- def news_worker(source, kw):
- 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("whut")
- print("#### END EXCEPTION ####")
- finally:
- q.task_done()
- q = Queue()
- d = db()
- d.purge_error()
- n = 0
- for s in d.sources():
- n = n + parseSource(s)
- sys.stdout.flush()
- print("[%s] %d new articles" % (strftime("%H:%M:%S"), n))
- print("###############################################################################")
|