newContentIdTask.py 31 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827
  1. """
  2. @author: luojunhui
  3. """
  4. import json
  5. import time
  6. import asyncio
  7. from applications.config import Config
  8. from applications.log import logging
  9. from applications.functions.pqFunctions import publish_to_pq, get_pq_video_detail
  10. from applications.functions.common import shuffle_list
  11. from applications.functions.kimi import KimiServer
  12. from applications.spider import search_videos_from_web
  13. from applications.etl_function import *
  14. from applications.feishu import bot
  15. from applications.functions.aigc import record_trace_id
  16. class NewContentIdTask(object):
  17. """
  18. 不存在历史已经发布的文章的匹配流程
  19. """
  20. TASK_INIT_STATUS = 0
  21. TASK_KIMI_FINISHED_STATUS = 1
  22. TASK_SPIDER_FINISHED_STATUS = 2
  23. TASK_ETL_FINISHED_STATUS = 3
  24. TASK_PUBLISHED_STATUS = 4
  25. TASK_PROCESSING_STATUS = 101
  26. TASK_FAIL_STATUS = 99
  27. KIMI_ILLEGAL_STATUS = 95
  28. ARTICLE_TEXT_TABLE_ERROR = 98
  29. TASK_MAX_PROCESS_TIMES = 3
  30. RECORD_SUCCESS_TRACE_ID_CODE = 2
  31. def __init__(self, mysql_client):
  32. self.mysql_client = mysql_client
  33. self.config = Config()
  34. self.article_match_video_table = self.config.article_match_video_table
  35. self.article_text_table = self.config.article_text_table
  36. self.article_crawler_video_table = self.config.article_crawler_video_table
  37. self.gh_id_dict = json.loads(self.config.get_config_value("testAccountLevel2"))
  38. self.account_map = json.loads(self.config.get_config_value("accountMap"))
  39. self.spider_coroutines = self.config.get_config_value("spiderCoroutines")
  40. async def get_tasks(self):
  41. """
  42. 获取 task
  43. :return:
  44. """
  45. # 获取 content_status 为 处理中 的任务,判断时间, 如果超过 1h 则,则将它改为 0, process_times + 1
  46. select_processing_sql = f"""
  47. SELECT
  48. trace_id, content_status_update_time, process_times
  49. FROM
  50. {self.article_match_video_table}
  51. WHERE
  52. content_status = {self.TASK_PROCESSING_STATUS}
  53. and process_times <= {self.TASK_MAX_PROCESS_TIMES};
  54. """
  55. processing_articles = await self.mysql_client.async_select(select_processing_sql)
  56. if processing_articles:
  57. processing_list = [
  58. {
  59. "trace_id": item[0],
  60. "content_status_update_time": item[1],
  61. "process_times": item[2]
  62. }
  63. for item in processing_articles
  64. ]
  65. for obj in processing_list:
  66. if int(time.time()) - obj['content_status_update_time'] >= 3600:
  67. # 认为该任务失败
  68. await self.roll_back_content_status_when_fails(
  69. process_times=obj['process_times'] + 1,
  70. trace_id=obj['trace_id']
  71. )
  72. # 将 process_times > 3 且状态不为 4 的任务的状态修改为失败,
  73. update_status_sql = f"""
  74. UPDATE
  75. {self.article_match_video_table}
  76. SET
  77. content_status = %s
  78. WHERE
  79. process_times > %s and content_status != %s;
  80. """
  81. await self.mysql_client.async_insert(
  82. update_status_sql,
  83. params=(
  84. self.TASK_FAIL_STATUS,
  85. self.TASK_MAX_PROCESS_TIMES,
  86. self.TASK_PUBLISHED_STATUS
  87. )
  88. )
  89. # 获取 process_times <= 3 且 content_status = 0 的任务
  90. select_sql = f"""
  91. SELECT
  92. trace_id, content_id, flow_pool_level, gh_id, process_times
  93. FROM
  94. {self.article_match_video_table}
  95. WHERE
  96. content_status = {self.TASK_INIT_STATUS}
  97. and process_times <= {self.TASK_MAX_PROCESS_TIMES}
  98. ORDER BY flow_pool_level, request_timestamp
  99. LIMIT {self.spider_coroutines};
  100. """
  101. tasks = await self.mysql_client.async_select(select_sql)
  102. if tasks:
  103. return [
  104. {
  105. "trace_id": i[0],
  106. "content_id": i[1],
  107. "flow_pool_level": i[2],
  108. "gh_id": i[3],
  109. "process_times": i[4]
  110. }
  111. for i in tasks
  112. ]
  113. else:
  114. return []
  115. async def get_video_list(self, content_id):
  116. """
  117. 判断该文章是否存在历史匹配视频
  118. :param content_id
  119. :return:
  120. """
  121. sql = f"""
  122. SELECT id
  123. FROM {self.article_crawler_video_table}
  124. WHERE content_id = '{content_id}' and download_status = 2;
  125. """
  126. res_tuple = await self.mysql_client.async_select(sql)
  127. if len(res_tuple) >= 3:
  128. return True
  129. else:
  130. return False
  131. async def update_content_status(self, new_content_status, trace_id, ori_content_status):
  132. """
  133. :param new_content_status:
  134. :param trace_id:
  135. :param ori_content_status:
  136. :return:
  137. """
  138. update_sql = f"""
  139. UPDATE {self.article_match_video_table}
  140. SET content_status = %s, content_status_update_time = %s
  141. WHERE trace_id = %s and content_status = %s;
  142. """
  143. row_counts = await self.mysql_client.async_insert(
  144. sql=update_sql,
  145. params=(
  146. new_content_status,
  147. int(time.time()),
  148. trace_id,
  149. ori_content_status
  150. )
  151. )
  152. return row_counts
  153. async def roll_back_content_status_when_fails(self, process_times, trace_id):
  154. """
  155. 处理失败,回滚至初始状态,处理次数加 1
  156. :param process_times:
  157. :param trace_id:
  158. :return:
  159. """
  160. update_article_sql = f"""
  161. UPDATE {self.article_match_video_table}
  162. SET
  163. content_status = %s,
  164. content_status_update_time = %s,
  165. process_times = %s
  166. WHERE trace_id = %s and content_status = %s;
  167. """
  168. await self.mysql_client.async_insert(
  169. sql=update_article_sql,
  170. params=(
  171. self.TASK_INIT_STATUS,
  172. int(time.time()),
  173. process_times + 1,
  174. trace_id,
  175. self.TASK_PROCESSING_STATUS
  176. )
  177. )
  178. async def judge_whether_same_content_id_is_processing(self, content_id):
  179. """
  180. 同一个 content_id 只需要处理一次
  181. :param content_id:
  182. :return:
  183. success: 4
  184. init: 0
  185. fail: 99
  186. """
  187. select_sql = f"""
  188. SELECT distinct content_status
  189. FROM {self.article_match_video_table}
  190. WHERE content_id = '{content_id}';
  191. """
  192. result = await self.mysql_client.async_select(select_sql)
  193. if result:
  194. for item in result:
  195. content_status = item[0]
  196. # if content_status not in {self.TASK_INIT_STATUS, self.TASK_PUBLISHED_STATUS} :
  197. if content_status in {
  198. self.TASK_KIMI_FINISHED_STATUS,
  199. self.TASK_SPIDER_FINISHED_STATUS,
  200. self.TASK_ETL_FINISHED_STATUS,
  201. self.TASK_PROCESSING_STATUS,
  202. self.TASK_PUBLISHED_STATUS
  203. }:
  204. return True
  205. return False
  206. else:
  207. return False
  208. async def get_downloaded_videos(self, content_id):
  209. """
  210. 获取已下载的视频
  211. :return:
  212. """
  213. sql = f"""
  214. SELECT platform, play_count, like_count, video_oss_path, cover_oss_path, user_id
  215. FROM {self.article_crawler_video_table}
  216. WHERE content_id = '{content_id}' and download_status = 2
  217. ORDER BY score DESC;
  218. """
  219. res_tuple = await self.mysql_client.async_select(sql)
  220. return [
  221. {
  222. "platform": i[0],
  223. "play_count": i[1],
  224. "like_count": i[2],
  225. "video_oss_path": i[3],
  226. "cover_oss_path": i[4],
  227. "uid": i[5]
  228. }
  229. for i in res_tuple
  230. ]
  231. async def get_kimi_status(self, content_id):
  232. """
  233. 通过 content_id 获取kimi info
  234. :return:
  235. """
  236. select_sql = f"""
  237. select kimi_status
  238. from {self.article_text_table}
  239. where content_id = '{content_id}';
  240. """
  241. response = await self.mysql_client.async_select(select_sql)
  242. if response:
  243. kimi_status = response[0][0]
  244. return kimi_status
  245. else:
  246. return self.ARTICLE_TEXT_TABLE_ERROR
  247. async def kimi_task(self, params):
  248. """
  249. 执行 kimi 任务
  250. :return:
  251. """
  252. KIMI_SUCCESS_STATUS = 1
  253. KIMI_FAIL_STATUS = 2
  254. content_id = params['content_id']
  255. trace_id = params['trace_id']
  256. process_times = params['process_times']
  257. kimi_status_code = await self.get_kimi_status(content_id=content_id)
  258. if kimi_status_code == KIMI_SUCCESS_STATUS:
  259. affected_rows = await self.update_content_status(
  260. new_content_status=self.TASK_KIMI_FINISHED_STATUS,
  261. trace_id=trace_id,
  262. ori_content_status=self.TASK_INIT_STATUS
  263. )
  264. if affected_rows == 0:
  265. logging(
  266. code="6000",
  267. info="多个进程抢占同一个任务的执行状态锁,抢占失败,return"
  268. )
  269. return
  270. get_kimi_sql = f"""
  271. SELECT article_title, kimi_title, kimi_summary, kimi_keys
  272. FROM {self.article_text_table}
  273. WHERE content_id = '{content_id}';
  274. """
  275. kimi_info = await self.mysql_client.async_select(get_kimi_sql)
  276. return {
  277. "kimi_title": kimi_info[0][1],
  278. "ori_title": kimi_info[0][0],
  279. "kimi_summary": kimi_info[0][2],
  280. "kimi_keys": json.loads(kimi_info[0][3])
  281. }
  282. elif kimi_status_code == self.ARTICLE_TEXT_TABLE_ERROR:
  283. logging(
  284. code="4000",
  285. info="long_articles_text表中未找到 content_id"
  286. )
  287. else:
  288. # 开始处理,讲 content_status 从 0 改为 101
  289. affected_rows = await self.update_content_status(
  290. new_content_status=self.TASK_PROCESSING_STATUS,
  291. trace_id=trace_id,
  292. ori_content_status=self.TASK_INIT_STATUS
  293. )
  294. if affected_rows == 0:
  295. logging(
  296. code="6000",
  297. info="多个进程抢占同一个任务的执行状态锁,抢占失败,return"
  298. )
  299. return
  300. K = KimiServer()
  301. try:
  302. select_sql = f"""
  303. select article_title, article_text
  304. from {self.article_text_table}
  305. where content_id = '{content_id}'
  306. """
  307. res = await self.mysql_client.async_select(select_sql)
  308. article_obj = {
  309. "article_title": res[0][0],
  310. "article_text": res[0][1],
  311. "content_id": content_id
  312. }
  313. kimi_info = await K.search_kimi_schedule(params=article_obj)
  314. kimi_title = kimi_info['k_title']
  315. content_title = kimi_info['content_title'].replace("'", "").replace('"', "")
  316. content_keys = json.dumps(kimi_info['content_keys'], ensure_ascii=False)
  317. update_kimi_sql = f"""
  318. UPDATE {self.article_text_table}
  319. SET
  320. kimi_title = %s,
  321. kimi_summary = %s,
  322. kimi_keys = %s,
  323. kimi_status = %s
  324. WHERE content_id = %s;"""
  325. await self.mysql_client.async_insert(
  326. sql=update_kimi_sql,
  327. params=(kimi_title, content_title, content_keys, KIMI_SUCCESS_STATUS, params['content_id'])
  328. )
  329. await self.update_content_status(
  330. new_content_status=self.TASK_KIMI_FINISHED_STATUS,
  331. trace_id=trace_id,
  332. ori_content_status=self.TASK_PROCESSING_STATUS
  333. )
  334. return {
  335. "kimi_title": kimi_title,
  336. "ori_title": article_obj['article_title'],
  337. "kimi_summary": content_title,
  338. "kimi_keys": kimi_info['content_keys']
  339. }
  340. except Exception as e:
  341. # kimi 任务处理失败
  342. update_kimi_sql = f"""
  343. UPDATE {self.article_text_table}
  344. SET
  345. kimi_status = %s
  346. WHERE content_id = %s
  347. """
  348. await self.mysql_client.async_insert(
  349. sql=update_kimi_sql,
  350. params=(
  351. KIMI_FAIL_STATUS,
  352. content_id
  353. )
  354. )
  355. # 将状态由 101 回退为 0
  356. await self.roll_back_content_status_when_fails(
  357. process_times=process_times,
  358. trace_id=trace_id
  359. )
  360. return {}
  361. async def spider_task(self, params, kimi_result):
  362. """
  363. 爬虫任务
  364. :return:
  365. """
  366. SPIDER_INIT_STATUS = 1
  367. DOWNLOAD_SUCCESS_STATUS = 2
  368. trace_id = params['trace_id']
  369. content_id = params['content_id']
  370. process_times = params['process_times']
  371. gh_id = params['gh_id']
  372. select_sql = f"""
  373. select count(id)
  374. from {self.article_crawler_video_table}
  375. where content_id = '{content_id}'
  376. and download_status = {DOWNLOAD_SUCCESS_STATUS};
  377. """
  378. count_tuple = await self.mysql_client.async_select(select_sql)
  379. counts = count_tuple[0][0]
  380. if counts >= 3:
  381. await self.update_content_status(
  382. new_content_status=self.TASK_SPIDER_FINISHED_STATUS,
  383. trace_id=trace_id,
  384. ori_content_status=SPIDER_INIT_STATUS
  385. )
  386. return True
  387. # 开始处理,将状态由 1 改成 101
  388. affected_rows = await self.update_content_status(
  389. new_content_status=self.TASK_PROCESSING_STATUS,
  390. ori_content_status=SPIDER_INIT_STATUS,
  391. trace_id=trace_id
  392. )
  393. if affected_rows == 0:
  394. logging(
  395. code="6000",
  396. info="多个进程抢占同一个任务的执行状态锁,抢占失败,return"
  397. )
  398. return False
  399. try:
  400. search_videos_count = await search_videos_from_web(
  401. info={
  402. "ori_title": kimi_result['ori_title'],
  403. "kimi_summary": kimi_result['kimi_summary'],
  404. "kimi_keys": kimi_result['kimi_keys'],
  405. "trace_id": trace_id,
  406. "gh_id": gh_id,
  407. "content_id": content_id,
  408. "crawler_video_table": self.article_crawler_video_table
  409. },
  410. gh_id_map=self.account_map,
  411. db_client=self.mysql_client
  412. )
  413. if search_videos_count >= 3:
  414. # 表示爬虫任务执行成功, 将状态从 101 改为 2
  415. await self.update_content_status(
  416. new_content_status=self.TASK_SPIDER_FINISHED_STATUS,
  417. trace_id=trace_id,
  418. ori_content_status=self.TASK_PROCESSING_STATUS
  419. )
  420. return True
  421. else:
  422. await self.roll_back_content_status_when_fails(
  423. process_times=process_times + 1,
  424. trace_id=trace_id
  425. )
  426. return False
  427. except Exception as e:
  428. await self.roll_back_content_status_when_fails(
  429. process_times=process_times + 1,
  430. trace_id=trace_id
  431. )
  432. print("爬虫处理失败: {}".format(e))
  433. return False
  434. async def etl_task(self, params):
  435. """
  436. download && upload videos
  437. :param params:
  438. :return:
  439. """
  440. VIDEO_DOWNLOAD_SUCCESS_STATUS = 2
  441. VIDEO_DOWNLOAD_FAIL_STATUS = 3
  442. ETL_TASK_INIT_STATUS = 2
  443. trace_id = params['trace_id']
  444. content_id = params['content_id']
  445. process_times = params['process_times']
  446. # 判断是否有三条已经下载完成的视频
  447. select_sql = f"""
  448. select count(id)
  449. from {self.article_crawler_video_table}
  450. where content_id = '{content_id}' and download_status = {VIDEO_DOWNLOAD_SUCCESS_STATUS};
  451. """
  452. video_count_tuple = await self.mysql_client.async_select(select_sql)
  453. video_count = video_count_tuple[0][0]
  454. if video_count >= 3:
  455. affect_rows = await self.update_content_status(
  456. ori_content_status=ETL_TASK_INIT_STATUS,
  457. trace_id=trace_id,
  458. new_content_status=self.TASK_ETL_FINISHED_STATUS
  459. )
  460. if affect_rows == 0:
  461. logging(
  462. code="6000",
  463. info="多个进程抢占同一个任务的执行状态锁,抢占失败,return"
  464. )
  465. return False
  466. return True
  467. else:
  468. # 开始处理, 将文章状态修改为处理状态
  469. affected_rows = await self.update_content_status(
  470. ori_content_status=ETL_TASK_INIT_STATUS,
  471. trace_id=trace_id,
  472. new_content_status=self.TASK_PROCESSING_STATUS
  473. )
  474. if affected_rows == 0:
  475. logging(
  476. code="6000",
  477. info="多个进程抢占同一个任务的执行状态锁,抢占失败,return"
  478. )
  479. return False
  480. select_sql = f"""
  481. SELECT id, out_video_id, platform, video_title, video_url, cover_url, user_id, trace_id
  482. FROM {self.article_crawler_video_table}
  483. WHERE content_id = '{content_id}' and download_status != {VIDEO_DOWNLOAD_SUCCESS_STATUS}
  484. ORDER BY score DESC;
  485. """
  486. videos_need_to_download_tuple = await self.mysql_client.async_select(select_sql)
  487. downloaded_count = 0
  488. for line in videos_need_to_download_tuple:
  489. params = {
  490. "id": line[0],
  491. "video_id": line[1],
  492. "platform": line[2],
  493. "video_title": line[3],
  494. "video_url": line[4],
  495. "cover_url": line[5],
  496. "user_id": line[6],
  497. "trace_id": line[7]
  498. }
  499. try:
  500. local_video_path, local_cover_path = generate_video_path(params['platform'], params['video_id'])
  501. # download videos
  502. file_path = await download_video(
  503. file_path=local_video_path,
  504. platform=params['platform'],
  505. video_url=params['video_url']
  506. )
  507. # download cover
  508. cover_path = await download_cover(
  509. file_path=local_cover_path,
  510. platform=params['platform'],
  511. cover_url=params['cover_url']
  512. )
  513. oss_video = await upload_to_oss(
  514. local_video_path=file_path,
  515. download_type="video"
  516. )
  517. if cover_path:
  518. oss_cover = await upload_to_oss(
  519. local_video_path=cover_path,
  520. download_type="image"
  521. )
  522. else:
  523. oss_cover = None
  524. update_sql = f"""
  525. UPDATE {self.article_crawler_video_table}
  526. SET video_oss_path = %s, cover_oss_path = %s, download_status = %s
  527. WHERE id = %s;
  528. """
  529. await self.mysql_client.async_insert(
  530. sql=update_sql,
  531. params=(
  532. oss_video,
  533. oss_cover,
  534. VIDEO_DOWNLOAD_SUCCESS_STATUS,
  535. params['id']
  536. )
  537. )
  538. downloaded_count += 1
  539. if downloaded_count > 3:
  540. await self.update_content_status(
  541. ori_content_status=self.TASK_PROCESSING_STATUS,
  542. trace_id=trace_id,
  543. new_content_status=self.TASK_ETL_FINISHED_STATUS
  544. )
  545. return True
  546. except Exception as e:
  547. update_sql = f"""
  548. UPDATE {self.article_crawler_video_table}
  549. SET download_status = %s
  550. WHERE id = %s;
  551. """
  552. await self.mysql_client.async_insert(
  553. sql=update_sql,
  554. params=(VIDEO_DOWNLOAD_FAIL_STATUS, params['id'])
  555. )
  556. if downloaded_count >= 3:
  557. await self.update_content_status(
  558. ori_content_status=self.TASK_PROCESSING_STATUS,
  559. trace_id=trace_id,
  560. new_content_status=self.TASK_ETL_FINISHED_STATUS
  561. )
  562. return True
  563. else:
  564. await self.roll_back_content_status_when_fails(
  565. process_times=process_times + 1,
  566. trace_id=trace_id
  567. )
  568. return False
  569. async def publish_task(self, params, kimi_title):
  570. """
  571. 发布任务
  572. :param kimi_title:
  573. :param params:
  574. :return:
  575. """
  576. PUBLISH_DEFAULT_STATUS = 3
  577. gh_id = params['gh_id']
  578. flow_pool_level = params['flow_pool_level']
  579. content_id = params['content_id']
  580. trace_id = params['trace_id']
  581. process_times = params['process_times']
  582. # 开始处理,将状态修改为操作状态
  583. affected_rows = await self.update_content_status(
  584. ori_content_status=PUBLISH_DEFAULT_STATUS,
  585. trace_id=trace_id,
  586. new_content_status=self.TASK_PROCESSING_STATUS
  587. )
  588. if affected_rows == 0:
  589. logging(
  590. code="6000",
  591. info="多个进程抢占同一个任务的执行状态锁,抢占失败,return"
  592. )
  593. return False
  594. try:
  595. download_videos = await self.get_downloaded_videos(content_id)
  596. match flow_pool_level:
  597. case "autoArticlePoolLevel4":
  598. # 冷启层, 全量做
  599. video_list = shuffle_list(download_videos)[:3]
  600. case "autoArticlePoolLevel3":
  601. if self.gh_id_dict.get(gh_id):
  602. video_list = shuffle_list(download_videos)[:3]
  603. else:
  604. video_list = download_videos[:3]
  605. case "autoArticlePoolLevel2":
  606. # 次条,只针对具体账号做
  607. video_list = []
  608. case "autoArticlePoolLevel1":
  609. # 头条,先不做
  610. video_list = download_videos[:3]
  611. case _:
  612. video_list = download_videos[:3]
  613. L = []
  614. for video_obj in video_list:
  615. params = {
  616. "videoPath": video_obj['video_oss_path'],
  617. "uid": video_obj['uid'],
  618. "title": kimi_title
  619. }
  620. publish_response = await publish_to_pq(params)
  621. video_id = publish_response['data']['id']
  622. response = await get_pq_video_detail(video_id)
  623. obj = {
  624. "uid": video_obj['uid'],
  625. "source": video_obj['platform'],
  626. "kimiTitle": kimi_title,
  627. "videoId": response['data'][0]['id'],
  628. "videoCover": response['data'][0]['shareImgPath'],
  629. "videoPath": response['data'][0]['videoPath'],
  630. "videoOss": video_obj['video_oss_path']
  631. }
  632. L.append(obj)
  633. update_sql = f"""
  634. UPDATE {self.article_match_video_table}
  635. SET content_status = %s, response = %s, process_times = %s
  636. WHERE trace_id = %s and content_status = %s;
  637. """
  638. # 从操作中状态修改为已发布状态
  639. await self.mysql_client.async_insert(
  640. sql=update_sql,
  641. params=(
  642. self.TASK_PUBLISHED_STATUS,
  643. json.dumps(L, ensure_ascii=False),
  644. process_times + 1,
  645. trace_id,
  646. self.TASK_PROCESSING_STATUS
  647. )
  648. )
  649. except Exception as e:
  650. await self.roll_back_content_status_when_fails(
  651. process_times=params['process_times'] + 1,
  652. trace_id=params['trace_id']
  653. )
  654. print(e)
  655. async def start_process(self, params):
  656. """
  657. 处理单篇文章
  658. :param params:
  659. :return:
  660. """
  661. # step1: 执行 kimi 操作
  662. # time.sleep(5) # 测试多个进程操作同一个 task 的等待时间
  663. kimi_result = await self.kimi_task(params)
  664. trace_id = params['trace_id']
  665. process_times = params['process_times']
  666. content_id = params['content_id']
  667. print(kimi_result)
  668. if kimi_result:
  669. # 等待 kimi 操作执行完成之后,开始执行 spider_task
  670. print("kimi success")
  671. logging(
  672. code=3001,
  673. info="kimi success",
  674. trace_id=trace_id
  675. )
  676. spider_flag = await self.spider_task(params=params, kimi_result=kimi_result)
  677. if spider_flag:
  678. # 等待爬虫执行完成后,开始执行 etl_task
  679. print("spider success")
  680. logging(
  681. code=3002,
  682. info="spider_success",
  683. trace_id=trace_id
  684. )
  685. etl_flag = await self.etl_task(params)
  686. if etl_flag:
  687. # 等待下载上传完成,执行发布任务
  688. print("etl success")
  689. logging(
  690. code="3003",
  691. info="etl_success",
  692. trace_id=trace_id
  693. )
  694. try:
  695. await self.publish_task(params, kimi_result['kimi_title'])
  696. logging(
  697. code="3004",
  698. info="publish_success",
  699. trace_id=trace_id
  700. )
  701. await record_trace_id(
  702. trace_id=trace_id,
  703. status=self.RECORD_SUCCESS_TRACE_ID_CODE
  704. )
  705. except Exception as e:
  706. logging(
  707. code="6004",
  708. info="publish 失败--{}".format(e),
  709. trace_id=params['trace_id']
  710. )
  711. else:
  712. logging(
  713. code="6003",
  714. info="ETL 处理失败",
  715. trace_id=params['trace_id']
  716. )
  717. else:
  718. logging(
  719. code="6002",
  720. info="爬虫处理失败",
  721. trace_id=params['trace_id']
  722. )
  723. else:
  724. logging(
  725. code="6001",
  726. info="kimi 处理失败",
  727. trace_id=trace_id
  728. )
  729. if process_times >= self.TASK_MAX_PROCESS_TIMES:
  730. logging(
  731. code="6011",
  732. info="kimi处理次数达到上限, 放弃处理",
  733. trace_id=trace_id
  734. )
  735. # 将相同的content_id && content_status = 0的状态修改为kimi 失败状态
  736. update_sql = f"""
  737. UPDATE {self.article_match_video_table}
  738. SET content_status = %s
  739. WHERE content_id = %s and content_status = %s;
  740. """
  741. affected_rows = await self.mysql_client.async_insert(
  742. sql=update_sql,
  743. params=(
  744. self.KIMI_ILLEGAL_STATUS,
  745. content_id,
  746. self.TASK_INIT_STATUS
  747. )
  748. )
  749. bot(
  750. title="KIMI 处理失败",
  751. detail={
  752. "content_id": content_id,
  753. "affected_rows": affected_rows
  754. }
  755. )
  756. async def process_task(self, params):
  757. """
  758. 处理任务
  759. :return:
  760. """
  761. content_id = params['content_id']
  762. download_videos = await self.get_video_list(content_id)
  763. if not download_videos:
  764. # 开始处理, 判断是否有相同的文章 id 正在处理
  765. processing_flag = await self.judge_whether_same_content_id_is_processing(content_id)
  766. if processing_flag:
  767. logging(
  768. code="9001",
  769. info="该 content id 正在处理中, 跳过此任务--{}".format(content_id)
  770. )
  771. else:
  772. await self.start_process(params=params)
  773. else:
  774. print("存在已下载视频")
  775. async def deal(self):
  776. """
  777. function
  778. :return:
  779. """
  780. task_list = await self.get_tasks()
  781. task_dict = {}
  782. # 对 content_id去重
  783. for task in task_list:
  784. key = task['content_id']
  785. task_dict[key] = task
  786. process_list = []
  787. for item in task_dict:
  788. process_list.append(task_dict[item])
  789. logging(
  790. code="5001",
  791. info="Match Task Got {} this time".format(len(process_list)),
  792. function="Publish Task"
  793. )
  794. if task_list:
  795. total_task = len(process_list)
  796. print(process_list)
  797. a = time.time()
  798. print("开始处理,一共{}个任务".format(total_task))
  799. tasks = [self.process_task(params) for params in process_list]
  800. await asyncio.gather(*tasks)
  801. b = time.time()
  802. print("处理时间: {} s".format(b - a))
  803. else:
  804. logging(
  805. code="9008",
  806. info="没有要处理的请求"
  807. )