|
@@ -27,20 +27,20 @@ def parseSource(s):
|
|
|
d = db()
|
|
d = db()
|
|
|
kw = d.get_keywords(s["category"])
|
|
kw = d.get_keywords(s["category"])
|
|
|
new_articles = 0
|
|
new_articles = 0
|
|
|
-# print(kw)
|
|
|
|
|
if l is not None:
|
|
if l is not None:
|
|
|
for a in l.links:
|
|
for a in l.links:
|
|
|
- if not d.article_exists(a.lower(),s["_id"]):
|
|
|
|
|
|
|
+ if not d.article_exists(a.lower(), s["_id"]):
|
|
|
q.put(a)
|
|
q.put(a)
|
|
|
- #print("[%s][Source] %s: %d/%d new/total articles" %
|
|
|
|
|
|
|
+ # print("[%s][Source] %s: %d/%d new/total articles" %
|
|
|
# (strftime("%H:%M:%S"), s["name"], q.qsize(), len(l.links)))
|
|
# (strftime("%H:%M:%S"), s["name"], q.qsize(), len(l.links)))
|
|
|
new_articles = q.qsize()
|
|
new_articles = q.qsize()
|
|
|
|
|
|
|
|
if not q.empty():
|
|
if not q.empty():
|
|
|
num_worker_threads = min(5, q.qsize())
|
|
num_worker_threads = min(5, q.qsize())
|
|
|
threads = []
|
|
threads = []
|
|
|
- for i in range(num_worker_threads):
|
|
|
|
|
|
|
+ for _ in range(num_worker_threads):
|
|
|
t = threading.Thread(target=news_worker, args=[s, kw, d, q])
|
|
t = threading.Thread(target=news_worker, args=[s, kw, d, q])
|
|
|
|
|
+ # source, keywords, db, queue
|
|
|
t.daemon = True
|
|
t.daemon = True
|
|
|
t.start()
|
|
t.start()
|
|
|
threads.append(t)
|
|
threads.append(t)
|
|
@@ -49,17 +49,18 @@ def parseSource(s):
|
|
|
q.join() # block until all tasks are done
|
|
q.join() # block until all tasks are done
|
|
|
# print("Finished waiting for the queue")
|
|
# print("Finished waiting for the queue")
|
|
|
|
|
|
|
|
- for i in range(num_worker_threads):
|
|
|
|
|
|
|
+ for _ in range(num_worker_threads):
|
|
|
|
|
+ # put num_worker empty jobs so
|
|
|
|
|
+ # each worker can work once
|
|
|
q.put(None)
|
|
q.put(None)
|
|
|
|
|
+
|
|
|
for t in threads:
|
|
for t in threads:
|
|
|
t.join(10)
|
|
t.join(10)
|
|
|
ex_time = time()-init
|
|
ex_time = time()-init
|
|
|
print("[%s][Source] %s => Finished parsing, %d/%d new articles, in %0.3fsec" %
|
|
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)
|
|
|
|
|
- )
|
|
|
|
|
|
|
+ (strftime("%H:%M:%S"),
|
|
|
|
|
+ s["name"], new_articles, len(l.links), ex_time))
|
|
|
sys.stdout.flush()
|
|
sys.stdout.flush()
|
|
|
- #return new_articles
|
|
|
|
|
return (ex_time, len(l.links), s["name"], new_articles)
|
|
return (ex_time, len(l.links), s["name"], new_articles)
|
|
|
|
|
|
|
|
|
|
|
|
@@ -71,12 +72,12 @@ def news_worker(source, kw, d, q):
|
|
|
if "paid" in source and source["paid"]:
|
|
if "paid" in source and source["paid"]:
|
|
|
paid = True
|
|
paid = True
|
|
|
|
|
|
|
|
- tname = threading.current_thread().name
|
|
|
|
|
|
|
+ # tname = threading.current_thread().name
|
|
|
while True:
|
|
while True:
|
|
|
try:
|
|
try:
|
|
|
item = q.get(timeout=5)
|
|
item = q.get(timeout=5)
|
|
|
except:
|
|
except:
|
|
|
- #print("Failed to get item")
|
|
|
|
|
|
|
+ # print("Failed to get item")
|
|
|
q.task_done()
|
|
q.task_done()
|
|
|
break
|
|
break
|
|
|
|
|
|
|
@@ -84,7 +85,7 @@ def news_worker(source, kw, d, q):
|
|
|
q.task_done()
|
|
q.task_done()
|
|
|
break
|
|
break
|
|
|
|
|
|
|
|
- #print("[Thread %s] %s" % (tname,item))
|
|
|
|
|
|
|
+ # print("[Thread %s] %s" % (tname,item))
|
|
|
try:
|
|
try:
|
|
|
html = None
|
|
html = None
|
|
|
if paid:
|
|
if paid:
|
|
@@ -93,7 +94,7 @@ def news_worker(source, kw, d, q):
|
|
|
|
|
|
|
|
n = News(item, source_id, source_name,
|
|
n = News(item, source_id, source_name,
|
|
|
source_category, kw, html=html)
|
|
source_category, kw, html=html)
|
|
|
- #print("[Thread %s] Finished" % tname)
|
|
|
|
|
|
|
+ # print("[Thread %s] Finished" % tname)
|
|
|
d.insert_article(n.get())
|
|
d.insert_article(n.get())
|
|
|
except Exception as e:
|
|
except Exception as e:
|
|
|
print("#### EXCEPTION ########")
|
|
print("#### EXCEPTION ########")
|
|
@@ -103,16 +104,20 @@ def news_worker(source, kw, d, q):
|
|
|
finally:
|
|
finally:
|
|
|
q.task_done()
|
|
q.task_done()
|
|
|
|
|
|
|
|
|
|
+
|
|
|
|
|
+# TODO influx start
|
|
|
start = time()
|
|
start = time()
|
|
|
sources = db().sources()
|
|
sources = db().sources()
|
|
|
db().purge_error()
|
|
db().purge_error()
|
|
|
-p = Pool(processes=4)
|
|
|
|
|
-ret = p.map(parseSource, sources)
|
|
|
|
|
-n = sum([ r[3] for r in ret ])
|
|
|
|
|
-tot = sum([ r[1] for r in ret ])
|
|
|
|
|
-#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"), n, tot, time()-start))
|
|
|
|
|
-print("###############################################################################")
|
|
|
|
|
|
|
+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])
|
|
|
|
|
+# 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
|