PULL_DATA.js 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602
  1. const CMD = {}
  2. const time_count = 1000;
  3. const redis_help = require('../src/use_redis');
  4. const origin_data_controllers = require('../src/data_manager/Controllers/origin_data_controllers');
  5. const axios = require('axios')
  6. const config = require('../etc/config.json');
  7. const DailyRecordManager = require('./daily_records');
  8. const schedule = require('node-schedule');
  9. const manager = new DailyRecordManager();
  10. const rabbitMq = require('../src/mq/rabbit-mq');
  11. const helper = require('../src/helper');
  12. const PullDataService = require('../src/PullDataService');
  13. const pull_data_task_item = require('./pull_data_task_item');
  14. var _24HourRanges = []
  15. var jobs = []
  16. var cur_day_data = new Map()
  17. var pull_data_task_map = []
  18. var cur_timeRange = null
  19. var pullDataService = new PullDataService(redis_help)
  20. var detail_task = []
  21. var task_params = []
  22. var cur_page = 0
  23. const messageHandler = async (msg) => {
  24. let PullDataConfig = await redis_help.getKeyValue("PullDataConfig")
  25. PullDataConfig = JSON.parse(PullDataConfig)
  26. _24HourRanges = PullDataConfig
  27. CMD.startScheduler(taskCallback)
  28. };
  29. const get_pull_data_task = function(){
  30. for (let index = 0; index < pull_data_task_map.length; index++) {
  31. const element = pull_data_task_map[index];
  32. if(!element.isRunTask()){
  33. return element
  34. }
  35. }
  36. let item = new pull_data_task_item(redis_help)
  37. pull_data_task_map.push(item)
  38. return item
  39. }
  40. const taskCallback = async (context) => {
  41. var { timeRange, executionTime } = context;
  42. console.log('执行任务:', {
  43. 时间区间: timeRange.name,
  44. 开始时间: timeRange.start,
  45. 执行时间: executionTime
  46. });
  47. const today = helper.getLocalDate();
  48. cur_day_data.forEach((v,k)=>{
  49. if(k==today){
  50. }else{
  51. cur_day_data.set(k, {})
  52. }
  53. })
  54. if(!cur_day_data.has(today)){
  55. cur_day_data.set(today,{})
  56. }
  57. if(!cur_day_data.get(today)[timeRange.start]){
  58. cur_day_data.get(today)[timeRange.start] = {name:timeRange.name,interval_minute:timeRange.interval_minute,finish_count:0}
  59. }
  60. console.log("cur_day_data[today]:",cur_day_data.get(today),today)
  61. const existingRecord = await manager.getRecord(today);
  62. if (existingRecord) {
  63. // 如果存在今天的记录,则更新
  64. const result = await manager.updateRecord(today, cur_day_data.get(today));
  65. console.log("更新记录结果:", result);
  66. } else {
  67. // 如果不存在今天的记录,则创建新记录
  68. const result = await manager.createRecord(today, cur_day_data.get(today));
  69. console.log("创建新记录结果:", result);
  70. }
  71. cur_timeRange = timeRange
  72. cur_timeRange.finish_count = 0
  73. cur_page = 0
  74. await processTask();
  75. };
  76. // 启动消费者
  77. async function startConsumer() {
  78. try {
  79. await rabbitMq.consumerDirectMsg(messageHandler,"exchange_update_pull_config","updatePullConig");
  80. } catch (error) {
  81. console.error('启动消费者失败:', error);
  82. }
  83. }
  84. async function processDetailTask(task_item) {
  85. try{
  86. console.log(" start processDetailTask")
  87. let materialId_list = task_item.materialId_list
  88. let materialId_data_list = task_item.materialId_data_list
  89. let response = await pullDataService.get_detail(materialId_list)
  90. if(!response.success){
  91. console.log("get_detail:",response)
  92. throw {msg:response,timeRange:cur_timeRange,fun:"get_detail",materialId_list:materialId_list}
  93. }
  94. if(cur_timeRange!=null){
  95. cur_timeRange.finish_count+=response.data.list.length
  96. const today = helper.getLocalDate();
  97. cur_day_data.get(today)[cur_timeRange.start].finish_count = cur_timeRange.finish_count
  98. await manager.updateRecord(today,cur_day_data.get(today));
  99. }
  100. for (let index = 0; index < response.data.list.length; index++) {
  101. const element = response.data.list[index];
  102. const info = materialId_data_list[element.materialId]
  103. if(element.hasAnchorInfo){
  104. await origin_data_controllers.createOriginData(
  105. {
  106. video_id:element.awemeId,
  107. materialId:element.materialId,
  108. video_link:element.awemeUrl,
  109. title:CMD.subTitle(element.title),
  110. publish_time:new Date(element.publishTime),
  111. kepp_num:info.kepp_num,
  112. comment_num:info.comment_num,
  113. like_num:info.like_num,
  114. shared_num:info.shared_num,
  115. is_guajian:1,
  116. guajian_link:element.landingUrl,
  117. status:0,
  118. createTime:info.createTime,
  119. updateTime:info.updateTime
  120. }
  121. )
  122. }
  123. }
  124. console.log(" finish processDetailTask")
  125. }catch (error) {
  126. console.error("processDetailTask:",error)
  127. }finally{
  128. // global.setTimeout(processDetailTask,500)
  129. }
  130. }
  131. async function process_material_list(response,page) {
  132. try{
  133. let FilterConfig = await redis_help.getKeyValue("FilterConfig")
  134. FilterConfig = JSON.parse(FilterConfig)
  135. let materialId_list = []
  136. let materialId_data_list = []
  137. if(!response.success){
  138. throw {msg:response,timeRange:cur_timeRange,fun:"get_novel_material_list"}
  139. }
  140. if(response.data == undefined||response.data == null){
  141. throw {msg:response,timeRange:cur_timeRange,fun:"get_novel_material_list"}
  142. }
  143. if(response.data.list == undefined||response.data.list == null){
  144. throw "没有数据了"
  145. }
  146. if(response.data.list.length<=0){
  147. throw "没有数据了"
  148. }
  149. if(cur_timeRange!=null){
  150. const today = helper.getLocalDate();
  151. // console.log("response.data:",response.data,today)
  152. if(cur_day_data.get(today)[cur_timeRange.start]){
  153. console.log("response.data.count:",response.data.count)
  154. console.log("(cur_day_data[today][timeRange.start]:",cur_day_data.get(today)[cur_timeRange.start])
  155. cur_timeRange.count = cur_timeRange.count + response.data.list.length
  156. cur_timeRange.cur_page = page
  157. cur_day_data.get(today)[cur_timeRange.start].count = response.data.count
  158. cur_day_data.get(today)[cur_timeRange.start].cur_page = page
  159. }else{
  160. console.log("cur_day_data空")
  161. }
  162. manager.updateRecord(today,cur_day_data.get(today));
  163. }
  164. console.log("materialId_list:",response.data.list.length)
  165. for (let index = 0; index < response.data.list.length; index++) {
  166. const origin_element = response.data.list[index];
  167. const result = await origin_data_controllers.getOriginData({
  168. materialId:origin_element.materialId})
  169. if(result.success){
  170. // await origin_data_controllers.updateOriginData({id:result.data.id},
  171. // {
  172. // kepp_num:origin_element.favoriteCount.count,
  173. // comment_num:origin_element.commentCount.count,
  174. // like_num:origin_element.likeCount.count,
  175. // shared_num:origin_element.shareCount.count
  176. // })
  177. }else{
  178. if(CMD.isRight(FilterConfig,origin_element)){
  179. materialId_list.push(origin_element.materialId)
  180. // materialId_data_list[origin_element.materialId] = {
  181. // kepp_num:origin_element.favoriteCount.count,
  182. // comment_num:origin_element.commentCount.count,
  183. // like_num:origin_element.likeCount.count,
  184. // shared_num:origin_element.shareCount.count
  185. // }
  186. console.log("materialId_list:",index,origin_element.materialId)
  187. materialId_data_list[origin_element.materialId] = {
  188. kepp_num:0,
  189. comment_num:0,
  190. like_num:0,
  191. shared_num:0
  192. }
  193. }
  194. }
  195. }
  196. console.log("fuck!:",materialId_list.length)
  197. return {materialId_list:materialId_list,materialId_data_list:materialId_data_list,count:response.data.list.length}
  198. }catch(e){
  199. console.log("materialId_list:",e)
  200. return null
  201. }
  202. }
  203. async function finish_material_list(response,page) {
  204. if(!response.success){
  205. console.error("finish_material_list:",response)
  206. }else{
  207. let detail_item = await process_material_list(response,page,null)
  208. detail_task.push(detail_item)
  209. }
  210. }
  211. async function process_update_token() {
  212. try{
  213. let response = await pullDataService.getToken()
  214. console.log("token:",response.data.token)
  215. }catch(e){
  216. console.error("process_update_token:",e)
  217. }finally{
  218. // setTimeout(process_update_token,200)
  219. }
  220. }
  221. async function process_novel_material_list() {
  222. try{
  223. if(task_params.length>0){
  224. let element = task_params.shift()
  225. const pull_task_item = get_pull_data_task();
  226. await pull_task_item.runTask(
  227. finish_material_list,
  228. cur_timeRange,
  229. element.page,
  230. element.limit
  231. );
  232. await new Promise(resolve => setTimeout(resolve, 200));
  233. }
  234. }catch(e){
  235. }finally{
  236. if(task_params.length>0){
  237. setTimeout(process_novel_material_list,200)
  238. }
  239. }
  240. }
  241. async function processTask(){
  242. try{
  243. let FilterConfig = await redis_help.getKeyValue("FilterConfig")
  244. FilterConfig = JSON.parse(FilterConfig)
  245. let response = await pullDataService.get_novel_material_list(cur_timeRange,cur_page,500)
  246. if(response.success){
  247. if(response.data.list.length<=0){
  248. cur_page = -1;
  249. throw response
  250. }
  251. }else{
  252. cur_page = -1;
  253. throw response
  254. }
  255. // console.log("get_novel_material_list:",response)
  256. let detail_item = await process_material_list(response,cur_page,cur_timeRange)
  257. // { page: 1, offset: 0, limit: 500 },
  258. // task_params = helper.getPaginationParams(detail_item.count,500)
  259. // task_params.shift()
  260. // detail_task.push(detail_item)
  261. console.log("detail_item:",detail_item)
  262. await processDetailTask(detail_item)
  263. console.log("processTask over!")
  264. }catch(e){
  265. console.error("PULL_DATA: error:",e)
  266. }finally{
  267. if(cur_page!=-1){
  268. cur_page++;
  269. global.setTimeout(processTask,1000)
  270. }
  271. }
  272. }
  273. CMD.subTitle = function(title){
  274. if(title.length>299){
  275. return title.substring(0, 299);
  276. }
  277. return title
  278. }
  279. CMD.isRight = function(FilterConfig,origin_element){
  280. if(origin_element.likeCount.count < FilterConfig.like_num){// console.log("点赞数正确")
  281. return false
  282. }
  283. if(origin_element.favoriteCount.count < FilterConfig.kepp_num){ // console.log("收藏数正确")
  284. return false
  285. }
  286. if(origin_element.shareCount.count < FilterConfig.shared_num){ // console.log("分享数正确")
  287. return false
  288. }
  289. if(origin_element.commentCount.count < FilterConfig.comment_num){ // console.log("评论")
  290. return false
  291. }
  292. return true
  293. }
  294. CMD.init = async function(){
  295. // await startConsumer();
  296. redis_help.connect(async ()=>{
  297. startConsumer()
  298. let PullDataConfig = await redis_help.getKeyValue("PullDataConfig")
  299. PullDataConfig = JSON.parse(PullDataConfig)
  300. let response = await pullDataService.getToken()
  301. console.log("token:",response)
  302. // processTask()
  303. _24HourRanges = PullDataConfig
  304. // 定义任务回调函数
  305. let key = helper.getLocalDate();
  306. let record_res = await manager.getRecord(key);
  307. if(record_res!=null){
  308. if(!cur_day_data.has(key)){
  309. cur_day_data.set(key,record_res.content)
  310. }
  311. }
  312. await pullDataService.getToken()
  313. // 启动调度器
  314. CMD.startScheduler(taskCallback);
  315. // process_update_token()
  316. })
  317. }
  318. // 生成24小时的时间区间
  319. CMD.generate24HourRanges = function() {
  320. // const ranges = [];
  321. // const today = new Date();
  322. // today.setMinutes(0);
  323. // today.setSeconds(0);
  324. // today.setMilliseconds(0);
  325. // for (let hour = 0; hour < 24; hour++) {
  326. // const timeString = `${hour.toString().padStart(2, '0')}:00`;
  327. // let periodName = '';
  328. // // 设置当天的小时
  329. // today.setHours(hour);
  330. // const timestamp = today.getTime();
  331. // // 根据时间划分时段
  332. // if (hour >= 0 && hour < 6) {
  333. // periodName = '凌晨区间';
  334. // } else if (hour >= 6 && hour < 9) {
  335. // periodName = '早晨区间';
  336. // } else if (hour >= 9 && hour < 12) {
  337. // periodName = '上午区间';
  338. // } else if (hour >= 12 && hour < 14) {
  339. // periodName = '中午区间';
  340. // } else if (hour >= 14 && hour < 18) {
  341. // periodName = '下午区间';
  342. // } else if (hour >= 18 && hour < 22) {
  343. // periodName = '晚上区间';
  344. // } else {
  345. // periodName = '深夜区间';
  346. // }
  347. // ranges.push({
  348. // name: periodName,
  349. // start: timeString,
  350. // timestamp: timestamp,
  351. // hour:hour,
  352. // timeFormat: new Date(timestamp).toLocaleString() // 可读的时间格式
  353. // });
  354. // }
  355. let ranges = [
  356. {
  357. name: '凌晨区间',
  358. start: '00:00',
  359. timestamp: 1733760000000,
  360. hour: 0,
  361. timeFormat: '12/10/2024, 12:00:00 AM'
  362. },
  363. {
  364. name: '凌晨区间',
  365. start: '01:00',
  366. timestamp: 1733763600000,
  367. hour: 1,
  368. timeFormat: '12/10/2024, 1:00:00 AM'
  369. },
  370. {
  371. name: '凌晨区间',
  372. start: '02:00',
  373. timestamp: 1733767200000,
  374. hour: 2,
  375. timeFormat: '12/10/2024, 2:00:00 AM'
  376. },
  377. {
  378. name: '凌晨区间',
  379. start: '03:00',
  380. timestamp: 1733770800000,
  381. hour: 3,
  382. timeFormat: '12/10/2024, 3:00:00 AM'
  383. },
  384. {
  385. name: '凌晨区间',
  386. start: '04:00',
  387. timestamp: 1733774400000,
  388. hour: 4,
  389. timeFormat: '12/10/2024, 4:00:00 AM'
  390. },
  391. {
  392. name: '凌晨区间',
  393. start: '05:00',
  394. timestamp: 1733778000000,
  395. hour: 5,
  396. timeFormat: '12/10/2024, 5:00:00 AM'
  397. },
  398. {
  399. name: '早晨区间',
  400. start: '06:00',
  401. timestamp: 1733781600000,
  402. hour: 6,
  403. timeFormat: '12/10/2024, 6:00:00 AM'
  404. },
  405. {
  406. name: '早晨区间',
  407. start: '07:00',
  408. timestamp: 1733785200000,
  409. hour: 7,
  410. timeFormat: '12/10/2024, 7:00:00 AM'
  411. },
  412. {
  413. name: '早晨区间',
  414. start: '08:00',
  415. timestamp: 1733788800000,
  416. hour: 8,
  417. timeFormat: '12/10/2024, 8:00:00 AM'
  418. },
  419. {
  420. name: '上午区间',
  421. start: '09:00',
  422. timestamp: 1733792400000,
  423. hour: 9,
  424. timeFormat: '12/10/2024, 9:00:00 AM'
  425. },
  426. {
  427. name: '上午区间',
  428. start: '10:00',
  429. timestamp: 1733796000000,
  430. hour: 10,
  431. timeFormat: '12/10/2024, 10:00:00 AM'
  432. },
  433. {
  434. name: '上午区间',
  435. start: '11:00',
  436. timestamp: 1733799600000,
  437. hour: 11,
  438. timeFormat: '12/10/2024, 11:00:00 AM'
  439. },
  440. {
  441. name: '中午区间',
  442. start: '12:00',
  443. timestamp: 1733803200000,
  444. hour: 12,
  445. timeFormat: '12/10/2024, 12:00:00 PM'
  446. },
  447. {
  448. name: '中午区间',
  449. start: '13:00',
  450. timestamp: 1733806800000,
  451. hour: 13,
  452. timeFormat: '12/10/2024, 1:00:00 PM'
  453. },
  454. {
  455. name: '下午区间',
  456. start: '14:00',
  457. timestamp: 1733810400000,
  458. hour: 14,
  459. timeFormat: '12/10/2024, 2:00:00 PM'
  460. },
  461. {
  462. name: '下午区间',
  463. start: '15:00',
  464. timestamp: 1733814000000,
  465. hour: 15,
  466. timeFormat: '12/10/2024, 3:00:00 PM'
  467. },
  468. {
  469. name: '下午区间',
  470. start: '16:00',
  471. timestamp: 1733817600000,
  472. hour: 16,
  473. timeFormat: '12/10/2024, 4:00:00 PM'
  474. },
  475. {
  476. name: '下午区间',
  477. start: '17:00',
  478. timestamp: 1733821200000,
  479. hour: 17,
  480. timeFormat: '12/10/2024, 5:00:00 PM'
  481. },
  482. {
  483. name: '晚上区间',
  484. start: '18:00',
  485. timestamp: 1733824800000,
  486. hour: 18,
  487. timeFormat: '12/10/2024, 6:00:00 PM'
  488. },
  489. {
  490. name: '晚上区间',
  491. start: '19:00',
  492. timestamp: 1733828400000,
  493. hour: 19,
  494. timeFormat: '12/10/2024, 7:00:00 PM'
  495. },
  496. {
  497. name: '晚上区间',
  498. start: '20:45',
  499. timestamp: 1733832000000,
  500. hour: 20,
  501. timeFormat: '12/10/2024, 8:00:00 PM'
  502. },
  503. {
  504. name: '晚上区间',
  505. start: '21:00',
  506. timestamp: 1733835600000,
  507. hour: 21,
  508. timeFormat: '12/10/2024, 9:00:00 PM'
  509. },
  510. {
  511. name: '深夜区间',
  512. start: '22:00',
  513. timestamp: 1733839200000,
  514. hour: 22,
  515. timeFormat: '12/10/2024, 10:00:00 PM'
  516. },
  517. {
  518. name: '深夜区间',
  519. start: '23:00',
  520. timestamp: 1733842800000,
  521. hour: 23,
  522. timeFormat: '12/10/2024, 11:00:00 PM'
  523. }
  524. ]
  525. return ranges;
  526. }
  527. CMD.stopScheduler = function() {
  528. jobs.forEach(({ job }) => job.cancel());
  529. jobs = [];
  530. console.log('调度器已停止,所有任务已清除');
  531. }
  532. CMD.startScheduler = function(taskCallback) {
  533. CMD.stopScheduler()
  534. // 为每个时间点创建定时任务
  535. _24HourRanges.forEach(timeRange => {
  536. // 解析小时
  537. const hour = parseInt(timeRange.start.split(':')[0]);
  538. const minute = parseInt(timeRange.start.split(':')[1]);
  539. // 创建定时任务 - 在每天的指定小时整点执行
  540. const job = schedule.scheduleJob(`${minute} ${hour} * * *`, async () => {
  541. try {
  542. console.log(`开始执行任务: ${timeRange.name} ${timeRange.start}`);
  543. // 执行回调函数
  544. await taskCallback({
  545. timeRange,
  546. executionTime: new Date()
  547. });
  548. } catch (error) {
  549. // 记录失败
  550. console.error('任务执行错误:', error);
  551. }
  552. });
  553. jobs.push({
  554. job,
  555. timeRange
  556. });
  557. console.log(`已安排任务: ${timeRange.name} ${timeRange.start}`);
  558. });
  559. console.log(`调度器已启动,共设置 ${jobs.length} 个定时任务`);
  560. }
  561. if(!config.isDebug){
  562. CMD.init()
  563. }