item_buffer.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426
  1. # -*- coding: utf-8 -*-
  2. """
  3. Created on 2018-06-19 17:17
  4. ---------
  5. @summary: item 管理器, 负责缓冲添加到数据库中的item, 由该manager统一添加。防止多线程同时访问数据库
  6. ---------
  7. @author: Boris
  8. @email: boris_liu@foxmail.com
  9. """
  10. import importlib
  11. import threading
  12. from queue import Queue
  13. import feapder.setting as setting
  14. import feapder.utils.tools as tools
  15. from feapder.db.redisdb import RedisDB
  16. from feapder.dedup import Dedup
  17. from feapder.network.item import Item, UpdateItem
  18. from feapder.pipelines import BasePipeline
  19. from feapder.pipelines.mysql_pipeline import MysqlPipeline
  20. from feapder.utils import metrics
  21. from feapder.utils.log import log
  22. MAX_ITEM_COUNT = 5000 # 缓存中最大item数
  23. UPLOAD_BATCH_MAX_SIZE = 1000
  24. MYSQL_PIPELINE_PATH = "feapder.pipelines.mysql_pipeline.MysqlPipeline"
  25. class ItemBuffer(threading.Thread):
  26. dedup = None
  27. __redis_db = None
  28. def __init__(self, redis_key, task_table=None):
  29. if not hasattr(self, "_table_item"):
  30. super(ItemBuffer, self).__init__()
  31. self._thread_stop = False
  32. self._is_adding_to_db = False
  33. self._redis_key = redis_key
  34. self._task_table = task_table
  35. self._items_queue = Queue(maxsize=MAX_ITEM_COUNT)
  36. self._table_request = setting.TAB_REQUESTS.format(redis_key=redis_key)
  37. self._table_failed_items = setting.TAB_FAILED_ITEMS.format(
  38. redis_key=redis_key
  39. )
  40. self._item_tables = {
  41. # 'item_name': 'table_name' # 缓存item名与表名对应关系
  42. }
  43. self._item_update_keys = {
  44. # 'table_name': ['id', 'name'...] # 缓存table_name与__update_key__的关系
  45. }
  46. self._pipelines = self.load_pipelines()
  47. self._have_mysql_pipeline = MYSQL_PIPELINE_PATH in setting.ITEM_PIPELINES
  48. self._mysql_pipeline = None
  49. if setting.ITEM_FILTER_ENABLE and not self.__class__.dedup:
  50. self.__class__.dedup = Dedup(
  51. to_md5=False, **setting.ITEM_FILTER_SETTING
  52. )
  53. # 导出重试的次数
  54. self.export_retry_times = 0
  55. # 导出失败的次数 TODO 非air爬虫使用redis统计
  56. self.export_falied_times = 0
  57. @property
  58. def redis_db(self):
  59. if self.__class__.__redis_db is None:
  60. self.__class__.__redis_db = RedisDB()
  61. return self.__class__.__redis_db
  62. def load_pipelines(self):
  63. pipelines = []
  64. for pipeline_path in setting.ITEM_PIPELINES:
  65. module, class_name = pipeline_path.rsplit(".", 1)
  66. pipeline_cls = importlib.import_module(module).__getattribute__(class_name)
  67. pipeline = pipeline_cls()
  68. if not isinstance(pipeline, BasePipeline):
  69. raise ValueError(f"{pipeline_path} 需继承 feapder.pipelines.BasePipeline")
  70. pipelines.append(pipeline)
  71. return pipelines
  72. @property
  73. def mysql_pipeline(self):
  74. if not self._mysql_pipeline:
  75. module, class_name = MYSQL_PIPELINE_PATH.rsplit(".", 1)
  76. pipeline_cls = importlib.import_module(module).__getattribute__(class_name)
  77. self._mysql_pipeline = pipeline_cls()
  78. return self._mysql_pipeline
  79. def run(self):
  80. self._thread_stop = False
  81. while not self._thread_stop:
  82. self.flush()
  83. tools.delay_time(1)
  84. self.close()
  85. def stop(self):
  86. self._thread_stop = True
  87. self._started.clear()
  88. def put_item(self, item):
  89. if isinstance(item, Item):
  90. # 入库前的回调
  91. item.pre_to_db()
  92. self._items_queue.put(item)
  93. def flush(self):
  94. try:
  95. items = []
  96. update_items = []
  97. requests = []
  98. callbacks = []
  99. items_fingerprints = []
  100. data_count = 0
  101. while not self._items_queue.empty():
  102. data = self._items_queue.get_nowait()
  103. data_count += 1
  104. # data 分类
  105. if callable(data):
  106. callbacks.append(data)
  107. elif isinstance(data, UpdateItem):
  108. update_items.append(data)
  109. elif isinstance(data, Item):
  110. items.append(data)
  111. if setting.ITEM_FILTER_ENABLE:
  112. items_fingerprints.append(data.fingerprint)
  113. else: # request-redis
  114. requests.append(data)
  115. if data_count >= UPLOAD_BATCH_MAX_SIZE:
  116. self.__add_item_to_db(
  117. items, update_items, requests, callbacks, items_fingerprints
  118. )
  119. items = []
  120. update_items = []
  121. requests = []
  122. callbacks = []
  123. items_fingerprints = []
  124. data_count = 0
  125. if data_count:
  126. self.__add_item_to_db(
  127. items, update_items, requests, callbacks, items_fingerprints
  128. )
  129. except Exception as e:
  130. log.exception(e)
  131. def get_items_count(self):
  132. return self._items_queue.qsize()
  133. def is_adding_to_db(self):
  134. return self._is_adding_to_db
  135. def __dedup_items(self, items, items_fingerprints):
  136. """
  137. 去重
  138. @param items:
  139. @param items_fingerprints:
  140. @return: 返回去重后的items, items_fingerprints
  141. """
  142. if not items:
  143. return items, items_fingerprints
  144. is_exists = self.__class__.dedup.get(items_fingerprints)
  145. is_exists = is_exists if isinstance(is_exists, list) else [is_exists]
  146. dedup_items = []
  147. dedup_items_fingerprints = []
  148. items_count = dedup_items_count = dup_items_count = 0
  149. while is_exists:
  150. item = items.pop(0)
  151. items_fingerprint = items_fingerprints.pop(0)
  152. is_exist = is_exists.pop(0)
  153. items_count += 1
  154. if not is_exist:
  155. dedup_items.append(item)
  156. dedup_items_fingerprints.append(items_fingerprint)
  157. dedup_items_count += 1
  158. else:
  159. dup_items_count += 1
  160. log.info(
  161. "待入库数据 {} 条, 重复 {} 条,实际待入库数据 {} 条".format(
  162. items_count, dup_items_count, dedup_items_count
  163. )
  164. )
  165. return dedup_items, dedup_items_fingerprints
  166. def __pick_items(self, items, is_update_item=False):
  167. """
  168. 将每个表之间的数据分开 拆分后 原items为空
  169. @param items:
  170. @param is_update_item:
  171. @return:
  172. """
  173. datas_dict = {
  174. # 'table_name': [{}, {}]
  175. }
  176. while items:
  177. item = items.pop(0)
  178. # 取item下划线格式的名
  179. # 下划线类的名先从dict中取,没有则现取,然后存入dict。加快下次取的速度
  180. item_name = item.item_name
  181. table_name = self._item_tables.get(item_name)
  182. if not table_name:
  183. table_name = item.table_name
  184. self._item_tables[item_name] = table_name
  185. if table_name not in datas_dict:
  186. datas_dict[table_name] = []
  187. datas_dict[table_name].append(item.to_dict)
  188. if is_update_item and table_name not in self._item_update_keys:
  189. self._item_update_keys[table_name] = item.update_key
  190. return datas_dict
  191. def __export_to_db(self, table, datas, is_update=False, update_keys=()):
  192. # 打点 校验
  193. self.check_datas(table=table, datas=datas)
  194. for pipeline in self._pipelines:
  195. if is_update:
  196. if table == self._task_table and not isinstance(
  197. pipeline, MysqlPipeline
  198. ):
  199. continue
  200. if not pipeline.update_items(table, datas, update_keys=update_keys):
  201. log.error(
  202. f"{pipeline.__class__.__name__} 更新数据失败. table: {table} items: {datas}"
  203. )
  204. return False
  205. else:
  206. if not pipeline.save_items(table, datas):
  207. log.error(
  208. f"{pipeline.__class__.__name__} 保存数据失败. table: {table} items: {datas}"
  209. )
  210. return False
  211. # 若是任务表, 且上面的pipeline里没mysql,则需调用mysql更新任务
  212. if not self._have_mysql_pipeline and is_update and table == self._task_table:
  213. if not self.mysql_pipeline.update_items(
  214. table, datas, update_keys=update_keys
  215. ):
  216. log.error(
  217. f"{self.mysql_pipeline.__class__.__name__} 更新数据失败. table: {table} items: {datas}"
  218. )
  219. return False
  220. return True
  221. def __add_item_to_db(
  222. self, items, update_items, requests, callbacks, items_fingerprints
  223. ):
  224. export_success = True
  225. self._is_adding_to_db = True
  226. # 去重
  227. if setting.ITEM_FILTER_ENABLE:
  228. items, items_fingerprints = self.__dedup_items(items, items_fingerprints)
  229. # 分捡
  230. items_dict = self.__pick_items(items)
  231. update_items_dict = self.__pick_items(update_items, is_update_item=True)
  232. # item批量入库
  233. failed_items = {"add": [], "update": [], "requests": []}
  234. while items_dict:
  235. table, datas = items_dict.popitem()
  236. log.debug(
  237. """
  238. -------------- item 批量入库 --------------
  239. 表名: %s
  240. datas: %s
  241. """
  242. % (table, tools.dumps_json(datas, indent=16))
  243. )
  244. if not self.__export_to_db(table, datas):
  245. export_success = False
  246. failed_items["add"].append({"table": table, "datas": datas})
  247. # 执行批量update
  248. while update_items_dict:
  249. table, datas = update_items_dict.popitem()
  250. log.debug(
  251. """
  252. -------------- item 批量更新 --------------
  253. 表名: %s
  254. datas: %s
  255. """
  256. % (table, tools.dumps_json(datas, indent=16))
  257. )
  258. update_keys = self._item_update_keys.get(table)
  259. if not self.__export_to_db(
  260. table, datas, is_update=True, update_keys=update_keys
  261. ):
  262. export_success = False
  263. failed_items["update"].append({"table": table, "datas": datas})
  264. if export_success:
  265. # 执行回调
  266. while callbacks:
  267. try:
  268. callback = callbacks.pop(0)
  269. callback()
  270. except Exception as e:
  271. log.exception(e)
  272. # 删除做过的request
  273. if requests:
  274. self.redis_db.zrem(self._table_request, requests)
  275. # 去重入库
  276. if setting.ITEM_FILTER_ENABLE:
  277. if items_fingerprints:
  278. self.__class__.dedup.add(items_fingerprints, skip_check=True)
  279. else:
  280. failed_items["requests"] = requests
  281. if self.export_retry_times > setting.EXPORT_DATA_MAX_RETRY_TIMES:
  282. if self._redis_key != "air_spider":
  283. # 失败的item记录到redis
  284. self.redis_db.sadd(self._table_failed_items, failed_items)
  285. # 删除做过的request
  286. if requests:
  287. self.redis_db.zrem(self._table_request, requests)
  288. log.error(
  289. "入库超过最大重试次数,不再重试,数据记录到redis,items:\n {}".format(
  290. tools.dumps_json(failed_items)
  291. )
  292. )
  293. self.export_retry_times = 0
  294. else:
  295. tip = ["入库不成功"]
  296. if callbacks:
  297. tip.append("不执行回调")
  298. if requests:
  299. tip.append("不删除任务")
  300. exists = self.redis_db.zexists(self._table_request, requests)
  301. for exist, request in zip(exists, requests):
  302. if exist:
  303. self.redis_db.zadd(self._table_request, requests, 300)
  304. if setting.ITEM_FILTER_ENABLE:
  305. tip.append("数据不入去重库")
  306. if self._redis_key != "air_spider":
  307. tip.append("将自动重试")
  308. tip.append("失败items:\n {}".format(tools.dumps_json(failed_items)))
  309. log.error(",".join(tip))
  310. self.export_falied_times += 1
  311. if self._redis_key != "air_spider":
  312. self.export_retry_times += 1
  313. if self.export_falied_times > setting.EXPORT_DATA_MAX_FAILED_TIMES:
  314. # 报警
  315. msg = "《{}》爬虫导出数据失败,失败次数:{},请检查爬虫是否正常".format(
  316. self._redis_key, self.export_falied_times
  317. )
  318. log.error(msg)
  319. tools.send_msg(
  320. msg=msg,
  321. level="error",
  322. message_prefix="《%s》爬虫导出数据失败" % (self._redis_key),
  323. )
  324. self._is_adding_to_db = False
  325. def check_datas(self, table, datas):
  326. """
  327. 打点 记录总条数及每个key情况
  328. @param table: 表名
  329. @param datas: 数据 列表
  330. @return:
  331. """
  332. metrics.emit_counter("total count", len(datas), classify=table)
  333. for data in datas:
  334. for k, v in data.items():
  335. metrics.emit_counter(k, int(bool(v)), classify=table)
  336. def close(self):
  337. # 调用pipeline的close方法
  338. for pipeline in self._pipelines:
  339. try:
  340. pipeline.close()
  341. except:
  342. pass