parser.py 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108
  1. #!/usr/bin/python3
  2. import threading
  3. from queue import Queue
  4. from db import db
  5. from source import Source
  6. from linklist import LinkList
  7. from news import News
  8. from paid_parser import PaidParser
  9. from time import strftime
  10. import sys
  11. def parseSource(s):
  12. global q
  13. q = Queue()
  14. num_worker_threads = 10
  15. sel = None
  16. rss = False
  17. if "selector" in s:
  18. sel = s["selector"]
  19. if "rss" in s:
  20. rss = bool(s["rss"])
  21. l = LinkList(s["link"], sel, rss=rss)
  22. kw = d.get_keywords(s["category"])
  23. new_articles = 0
  24. # print(kw)
  25. if l is not None:
  26. for a in l.links:
  27. if not d.article_exists(a.lower()):
  28. q.put(a)
  29. print("[%s][Source] %s: %d/%d new/total articles" %
  30. (strftime("%H:%M:%S"), s["name"], q.qsize(), len(l.links)))
  31. new_articles = q.qsize()
  32. if not q.empty():
  33. num_worker_threads = min(10, q.qsize())
  34. threads = []
  35. for i in range(num_worker_threads):
  36. t = threading.Thread(target=news_worker, args=[s, kw])
  37. t.daemon = True
  38. t.start()
  39. threads.append(t)
  40. # print("Waiting for %d queue elements" % q.qsize())
  41. q.join() # block until all tasks are done
  42. # print("Finished waiting for the queue")
  43. for i in range(num_worker_threads):
  44. q.put(None)
  45. for t in threads:
  46. t.join(10)
  47. # print("[Source] %s => Finished parsing" %s["name"])
  48. return new_articles
  49. def news_worker(source, kw):
  50. source_id = source["_id"]
  51. source_name = source["name"]
  52. source_category = source["category"]
  53. paid = False
  54. if "paid" in source and source["paid"]:
  55. paid = True
  56. tname = threading.current_thread().name
  57. while True:
  58. try:
  59. item = q.get(timeout=5)
  60. except:
  61. #print("Failed to get item")
  62. q.task_done()
  63. break
  64. if item is None:
  65. q.task_done()
  66. break
  67. #print("[Thread %s] %s" % (tname,item))
  68. try:
  69. html = None
  70. if paid:
  71. p = PaidParser(item)
  72. html = p.html
  73. n = News(item, source_id, source_name,
  74. source_category, kw, html=html)
  75. #print("[Thread %s] Finished" % tname)
  76. d.insert_article(n.get())
  77. except Exception as e:
  78. print("#### EXCEPTION ########")
  79. print(e)
  80. print(item)
  81. print("whut")
  82. print("#### END EXCEPTION ####")
  83. finally:
  84. q.task_done()
  85. q = Queue()
  86. d = db()
  87. d.purge_error()
  88. n = 0
  89. for s in d.sources():
  90. n = n + parseSource(s)
  91. sys.stdout.flush()
  92. print("[%s] %d new articles" % (strftime("%H:%M:%S"), n))
  93. print("###############################################################################")