import { Injectable, HttpException, HttpStatus, OnModuleInit } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; import axios from 'axios'; const moment = require('moment'); import { env } from 'process'; import { uploadFileToS3 } from 'src/common/config/upload-s3-file-using-link'; import { Constants } from 'src/common/constants'; import { ApifyScraper } from 'src/entities/apify-screper.entity'; import { FacebookPages } from 'src/entities/facebook-pages.entity'; import { FacebookPosts } from 'src/entities/faceboook-posts.entity'; import { InstagramMedias } from 'src/entities/instagram-medias.entity'; import { InstagramProfiles } from 'src/entities/instagram-profile.entity'; import { KeyType, Settings } from 'src/entities/settings.entity'; import { SocialMediaPosts } from 'src/entities/social-media-posts.entity'; import { SocialMedias, SocialType } from 'src/entities/social-medias.entity'; import { TikTokProfiles } from 'src/entities/tiktok-profile.entity'; import { TikTokVideos } from 'src/entities/tiktok-videos.entity'; import { Users } from 'src/entities/users.entity'; import { YoutubeChannels } from 'src/entities/youtube-channel.entity'; import { YoutubeVideos } from 'src/entities/youtube-videos.entity'; import { responseMessages } from 'src/messages/response-messages'; import { In, Not, Repository } from 'typeorm'; import { EncryptionService } from '../encryption/encryption.service'; import { DataSource } from 'typeorm'; import { ApifyClient } from 'apify-client'; @Injectable() export class ApifyServices implements OnModuleInit { public APIFY_ACCESS_TOKEN = ""; public s3Config = {}; public client: any; constructor( private readonly encryptionService: EncryptionService, private readonly dataSource: DataSource, @InjectRepository(Users) private Users: Repository, @InjectRepository(SocialMedias) private SocialMedias: Repository, @InjectRepository(TikTokProfiles) private TikTokProfiles: Repository, @InjectRepository(TikTokVideos) private TikTokVideos: Repository, @InjectRepository(InstagramProfiles) private InstagramProfiles: Repository, @InjectRepository(FacebookPages) private FacebookPages: Repository, @InjectRepository(YoutubeChannels) private YoutubeChannels: Repository, @InjectRepository(InstagramMedias) private InstagramMedias: Repository, @InjectRepository(SocialMediaPosts) private SocialMediaPosts: Repository, @InjectRepository(YoutubeVideos) private YoutubeVideos: Repository, @InjectRepository(FacebookPosts) private FacebookPosts: Repository, ) { } async onModuleInit() { this.APIFY_ACCESS_TOKEN = await this.encryptionService.decrypt(process.env.APIFY_ACCESS_TOKEN); this.s3Config = { accessKeyId: await this.encryptionService.decrypt(process.env.AWS_ACCESS_KEY_ID), secretAccessKey: await this.encryptionService.decrypt(process.env.AWS_SECRET_ACCESS_KEY), region: process.env.AWS_DEFAULT_REGION, } this.client = new ApifyClient({ token: this.APIFY_ACCESS_TOKEN, }); } async sleep(ms) { return new Promise(resolve => setTimeout(resolve, ms)); } /** * @description * This function is used to scrape Instagram accounts' data using Apify's instagram-profile-scraper actor. * It takes two parameters: page and pageSize. page is the current page number of the pagination, and pageSize is the number of records per page. * The function first makes a query to the database to get the list of Instagram accounts' usernames. Then it sends a POST request to Apify's instagram-profile-scraper actor * with the list of usernames. After the actor finishes scraping the data, the function makes a GET request to the dataset of the actor to get the scraped data. * Finally, it calls instagramProfileUpdate function to update the database with the scraped data. If the total number of records is greater than the offset of the next page, the function calls itself * with the next page number and pageSize until all records are scraped. */ async instagramScraper(pageSize, social_user_name = null) { const daysAgo = moment().subtract(Constants.SOCIAL_MEDIA_CRON_DAY, 'days').toDate(); let query = await this.SocialMedias.createQueryBuilder('sm') .select(['sm.id', 'sm.social_user_name']) .where('social_type = :social_type', { social_type: SocialType.INSTAGRAM }) .andWhere('sm.social_media_not_found = false') .andWhere('sm.apify_data_updated_attempts < 3') .andWhere('(sm.updated_by_cron <= :dateLimit OR sm.updated_by_cron IS NULL)', { dateLimit: daysAgo }) .andWhere('is_verified = true').andWhere('is_token_expire = true'); // UPDATE SINGLE RECORD if (social_user_name) { query.andWhere('sm.social_user_name = :social_user_name', { social_user_name }) } const [result, total] = await query.orderBy('created_at', 'ASC').take(pageSize).getManyAndCount(); const instagramIds = result.map(item => item.id); if (instagramIds.length) { await this.SocialMedias .createQueryBuilder() .update() .set({ apify_data_updated_attempts: () => 'apify_data_updated_attempts + 1' }) .where('id IN (:...ids)', { ids: instagramIds }) .execute(); } let userNames = result.map(item => item.social_user_name) console.log(userNames) if (userNames.length) { await this.instagramApify(userNames, pageSize); if (total) { await this.instagramScraper(pageSize, social_user_name); } } } /************* ✨ Codeium Command ⭐ *************/ /** * @description * @param country the country of the Instagram accounts to be scraped * @param condition the condition of the followers count, can be "BETWEEN", "GREATER_THAN", "LESS_THAN", or "EQUAL_TO" * @param days the number of days ago that the records were last updated * @param from the from value of the followers count range * @param to the to value of the followers count range */ /****** 1b9ad9e9-590c-4c21-a236-e4c2744e8cce *******/ async instagramScraperCountryFilter(country: string, condition: string, days: number, from: number, to: number = 0) { const pageSize = 100; const daysAgo = moment().subtract(days, 'days').toDate(); let query = await this.SocialMedias.createQueryBuilder('sm') .select(['sm.id', 'sm.social_user_name', 'sm.follower_count']) .where('social_type = :social_type', { social_type: SocialType.INSTAGRAM }) .andWhere('sm.social_media_not_found = false') .andWhere('sm.apify_data_updated_attempts < 3') .andWhere('(sm.updated_by_cron <= :dateLimit OR sm.updated_by_cron IS NULL)', { dateLimit: daysAgo }) .andWhere('country = :country', { country }) .andWhere('is_verified = false'); switch (condition) { case "BETWEEN": query.andWhere('follower_count BETWEEN :from AND :to', { from, to }); break; case "GREATER_THAN": query.andWhere('follower_count > :from', { from }); break; case "LESS_THAN": query.andWhere('follower_count < :from', { from }); break; case "EQUAL_TO": query.andWhere('follower_count = :from', { from }); break; } const [result, total] = await query.orderBy('created_at', 'ASC').take(pageSize).getManyAndCount(); const instagramCountryIds = result.map(item => item.id); if (instagramCountryIds.length) { await this.SocialMedias .createQueryBuilder() .update() .set({ apify_data_updated_attempts: () => 'apify_data_updated_attempts + 1' }) .where('id IN (:...ids)', { ids: instagramCountryIds }) .execute(); } let userNames = result.map(item => item.social_user_name) console.log(userNames) if (userNames.length) { await this.instagramApify(userNames, pageSize); if (total) { await this.instagramScraperCountryFilter(country, condition, days, from, to); } } } async instagramApify(userNames, pageSize) { const headers = {}; const params = {}; const instagramScraperUrl = process.env.APIFY_BASE_URL + `acts/apify~instagram-profile-scraper/runs?token=${this.APIFY_ACCESS_TOKEN}`; const data = { "usernames": userNames } // const datasetResponse = await this.postAxios(instagramScraperUrl, headers, data, params); // const datasetId = datasetResponse.data.defaultDatasetId; const { defaultDatasetId } = await this.client.actor('apify~instagram-profile-scraper').call(data); //await this.sleep(60000); const instagramDatasetUrl = process.env.APIFY_BASE_URL + `datasets/${defaultDatasetId}/items?limit=${pageSize}&offset=0&token=${this.APIFY_ACCESS_TOKEN}` // const response = await this.getAxios(instagramDatasetUrl, headers, params); const { items } = await this.client.dataset(defaultDatasetId).listItems(); console.log(items) await this.instagramProfileUpdate(items); } /** * @description * This function is used to update Instagram profiles in the database with the scraped data. * It takes one parameter: response, which is an array of objects returned from the Apify's instagram-profile-scraper actor's dataset. * The function loops through the array and for each object, it updates the SocialMedias table with the scraped data, then updates the InstagramProfiles table * with the scraped data. If the record does not exist in the InstagramProfiles table, it will not be created. */ async instagramProfileUpdate(response) { for (const row of response) { if (row?.error) { const userName = row.url.split("/").pop(); console.log(userName) await this.SocialMedias .createQueryBuilder() .update() .set({ source_posts_from: 'apify', social_media_not_found: true, is_updated: true, updated_by_cron: moment().format('YYYY-MM-DD HH:mm:ss'), }) .where('LOWER(social_user_name) = LOWER(:username)', { username: userName }) .andWhere('social_type = :type', { type: SocialType.INSTAGRAM }) .execute(); // this.SocialMedias.update({ social_user_name: userName, social_type: SocialType.INSTAGRAM }, { social_media_not_found: true, is_updated: true, updated_by_cron: moment().format('YYYY-MM-DD HH:mm:ss') }); continue }; // const socialMedia = await this.SocialMedias.findOne({ relations: ['users'], where: { social_user_name: row.username, social_type: SocialType.INSTAGRAM } }); const socialMedia = await this.SocialMedias .createQueryBuilder('social') .leftJoinAndSelect('social.users', 'user') .where('LOWER(social.social_user_name) = LOWER(:username)', { username: row.username }) .andWhere('social.social_type = :type', { type: SocialType.INSTAGRAM }) .getOne(); if (socialMedia) { const s3Response = await uploadFileToS3(row.profilePicUrl, `images/profile/`, this.s3Config); if (!s3Response) { this.SocialMedias.update({ id: socialMedia.id }, { apify_data_updated_attempts: socialMedia.apify_data_updated_attempts + 1 }); if (socialMedia.apify_data_updated_attempts >= 3) { this.SocialMedias.update({ id: socialMedia.id }, { source_posts_from: 'apify', social_media_not_found: true, is_updated: true, updated_by_cron: moment().format('YYYY-MM-DD HH:mm:ss') }); } continue } const todayDate = moment().format('YYYY-MM-DD HH:mm:ss'); this.Users.update({ id: socialMedia.users.id }, { image: s3Response.path, updated_by_cron: todayDate }); // if (socialMedia.users.is_updated_by_cron) { socialMedia['follower_count'] = row.followersCount; socialMedia['following_count'] = row.followsCount; socialMedia['post_count'] = row.postsCount; socialMedia['apify_data_updated_attempts'] = 0; socialMedia['source_posts_from'] = 'apify'; // } socialMedia['is_updated'] = true; socialMedia['updated_by_cron'] = todayDate; await this.SocialMedias.save(socialMedia); const instagramProfiles = await this.InstagramProfiles.findOne({ where: { socialMedia: { id: socialMedia.id } } }); if (instagramProfiles) { instagramProfiles['user_name'] = row.username; instagramProfiles['name'] = row.fullName; instagramProfiles['profile_picture_url'] = row.profilePicUrl; // if (socialMedia.users.is_updated_by_cron) { instagramProfiles['follows_count'] = row.followsCount; instagramProfiles['media_count'] = row.postsCount; instagramProfiles['followers_count'] = row.followersCount; // } instagramProfiles['biography'] = row.biography; await this.InstagramProfiles.save(instagramProfiles); console.log(row.username) // await this.ApifyScraper.delete({ socialMedia: socialMedia }); // this.ApifyScraper.save({ socialMedia: socialMedia, platform: SocialType.INSTAGRAM, response_json: row }); console.log(row?.latestPosts?.length) if (row?.latestPosts?.length) { await this.instagramPosts(instagramProfiles, socialMedia, socialMedia.users, row.latestPosts); } } } } } async instagramPosts(instagramProfile, socialMedia, authUser, posts) { await this.dataSource .createQueryBuilder() .delete() .from('instagram_medias') .where('instagramProfiles = :id', { id: instagramProfile.id }) .execute(); //await this.InstagramMedias.delete({ instagramProfiles: { id: instagramProfile.id } }); await this.dataSource .createQueryBuilder() .delete() .from('social_media_posts') .where('socialMedia = :id', { id: socialMedia.id }) // column name in DB .execute(); // await this.SocialMediaPosts.delete({ socialMedia: { id: socialMedia.id } }); let totalLikes: bigint = BigInt(0); let totalComments: bigint = BigInt(0); let totalViews: bigint = BigInt(0); let totalReels = 0; const instagramMedias = []; const socialMediaPosts = []; for (const media of posts) { totalLikes += BigInt(media.likesCount || 0); totalComments += BigInt(media.commentsCount || 0); totalViews += BigInt(media?.videoViewCount || 0); totalReels += media.type == 'Video' ? 1 : 0; instagramMedias.push({ comments_count: media.commentsCount, like_count: media.likesCount, username: media.username, permalink: media.url, media_url: media.url, media_type: media.type, shortcode: media.shortCode, instagram_timestamp: moment(media.timestamp).format('YYYY-MM-DD HH:mm:ss'), instagram_media_id: media.id, caption: media.caption, thumbnail_url: media.displayUrl, reel_views_count: media?.videoViewCount || 0, users: { id: authUser.id }, socialMedia: { id: socialMedia.id }, instagramProfiles: { id: instagramProfile.id }, }) socialMediaPosts.push({ users: { id: authUser.id }, socialMedia: { id: socialMedia.id }, title: media.caption || "", description: "", like_count: media.likesCount || 0, comment_count: media.commentsCount || 0, post_id: media.id, post_url: media.url, image_url: media.displayUrl || "", social_type: SocialType.INSTAGRAM, created_time: moment(media.timestamp).format('YYYY-MM-DD HH:mm:ss') }); } await this.InstagramMedias.insert(instagramMedias); await this.SocialMediaPosts.insert(socialMediaPosts); if (posts.length) { const engagementRate = parseFloat((((Number(totalLikes) + Number(totalComments)) / posts.length) / socialMedia.follower_count * 100).toFixed(4)); await this.updateSocialMediaEngagementRate(socialMedia.id, engagementRate); } console.log(totalViews, totalReels) if (totalReels) { const reelAvg = parseFloat((Number(totalViews) / Number(totalReels)).toFixed(4)); console.log(reelAvg) await this.SocialMedias.update({ id: socialMedia.id }, { reel_average: reelAvg }); } // await this.SocialMedias.update({ id: socialMedia.id }, { engagement_rate: engagementRate }); } /** * @description * This function is used to scrape Facebook pages. It takes three parameters: page, pageSize, and social_id. * The function first queries the SocialMedias table to get all the records with the specified social media type. * Then it loops through the records and for each record, it creates an array of start URLs for the Apify's facebook-pages-scraper actor. * The start URLs are the Facebook page URLs of the records. The function then calls the Apify's facebook-pages-scraper actor and gets the dataset ID. * After waiting for 60 seconds, it calls the Apify's dataset API to get the scraped data. Finally it calls the facebookPageUpdate function to update the FacebookPages table with the scraped data. * If the total number of records is more than the next offset, it will call itself with the next page number and page size. */ async facebookScraper(pageSize, social_user_name = null) { const daysAgo = moment().subtract(Constants.SOCIAL_MEDIA_CRON_DAY, 'days').toDate(); let query = await this.SocialMedias.createQueryBuilder('sm') .select(['sm.id', 'sm.social_user_name']) .where('social_type = :social_type', { social_type: SocialType.FACEBOOK }) .andWhere('sm.social_media_not_found = false') .andWhere('sm.apify_data_updated_attempts < 3') .andWhere('(sm.updated_by_cron <= :dateLimit OR sm.updated_by_cron IS NULL)', { dateLimit: daysAgo }) .andWhere('is_verified = true').andWhere('is_token_expire = true'); // UPDATE SINGLE RECORD if (social_user_name) { query.andWhere('sm.social_user_name = :social_user_name', { social_user_name }) } const [result, total] = await query.orderBy('created_at', 'ASC').take(pageSize).getManyAndCount(); let ids = result.map(item => item.id) if (ids.length) { await this.SocialMedias .createQueryBuilder() .update() .set({ apify_data_updated_attempts: () => 'apify_data_updated_attempts + 1' }) .where('id IN (:...ids)', { ids }) .execute(); } console.log(ids) if (ids.length) { const facebookPages = await this.FacebookPages.find({ select: ['id', 'user_name'], where: { socialMedia: { id: In(ids) }, user_name: Not("") } }); if (facebookPages) { await this.facebookApify(facebookPages, pageSize); if (total) { await this.facebookScraper(pageSize, social_user_name); } } } } async updateSocialMediaEngagementRate(id, engagementRate) { const socialMedia = await this.SocialMedias.findOne({ where: { id: id } }); const oldEngagementRate = socialMedia['engagement_rate'] || 0; socialMedia['old_engagement_rate'] = oldEngagementRate; socialMedia['engagement_rate'] = engagementRate; if (oldEngagementRate == 0.0000) { socialMedia['engagement_rate_percentage'] = 100; // or 100, depending on logic. } else { socialMedia['engagement_rate_percentage'] = Math.abs((engagementRate - oldEngagementRate) / oldEngagementRate * 100); } await this.SocialMedias.save(socialMedia); } async facebookApify(facebookPages, pageSize) { const headers = { 'Content-Type': 'application/json' }; const params = {}; const facebookScraperUrl = process.env.APIFY_BASE_URL + `acts/apify~facebook-pages-scraper/runs?token=${this.APIFY_ACCESS_TOKEN}`; let data = { "startUrls": [] }; let isExistsUserName = new Set(); // ONLY ALLOW UNIUE START URL LINKS for (const item of facebookPages) { if (!isExistsUserName.has(item.user_name)) { isExistsUserName.add(item.user_name); // Add username to the Set data.startUrls.push({ url: "https://www.facebook.com/" + item.user_name, method: "GET" }); } } // const datasetResponse = await this.postAxios(facebookScraperUrl, headers, data, params); // const datasetId = datasetResponse.data.defaultDatasetId; // console.log(datasetId) // await this.sleep(60000); // const facebookDatasetUrl = process.env.APIFY_BASE_URL + `datasets/${datasetId}/items?limit=${pageSize}&offset=0&token=${this.APIFY_ACCESS_TOKEN}` // const response = await this.getAxios(facebookDatasetUrl, headers, params); // console.log(response) const { defaultDatasetId } = await this.client.actor('apify~facebook-pages-scraper').call(data); const { items } = await this.client.dataset(defaultDatasetId).listItems(); console.log(items) await this.facebookPageUpdate(items); } async facebookScraperCountryFilter(country: string, condition: string, days: number, from: number, to: number = 0) { const pageSize = 100; const daysAgo = moment().subtract(days, 'days').toDate(); let query = await this.SocialMedias.createQueryBuilder('sm') .select(['sm.id', 'sm.social_user_name', 'sm.follower_count']) .where('social_type = :social_type', { social_type: SocialType.FACEBOOK }) .andWhere('sm.social_media_not_found = false') .andWhere('sm.apify_data_updated_attempts < 3') .andWhere('(sm.updated_by_cron <= :dateLimit OR sm.updated_by_cron IS NULL)', { dateLimit: daysAgo }) .andWhere('is_verified = false') .andWhere('country = :country', { country }); switch (condition) { case "BETWEEN": query.andWhere('follower_count BETWEEN :from AND :to', { from, to }); break; case "GREATER_THAN": query.andWhere('follower_count > :from', { from }); break; case "LESS_THAN": query.andWhere('follower_count < :from', { from }); break; case "EQUAL_TO": query.andWhere('follower_count = :from', { from }); break; } const [result, total] = await query.orderBy('created_at', 'ASC').take(pageSize).getManyAndCount(); let ids = result.map(item => item.id) if (ids.length) { await this.SocialMedias .createQueryBuilder() .update() .set({ apify_data_updated_attempts: () => 'apify_data_updated_attempts + 1' }) .where('id IN (:...ids)', { ids }) .execute(); } console.log(ids) if (ids.length) { const facebookPages = await this.FacebookPages.find({ select: ['id', 'user_name'], where: { socialMedia: { id: In(ids) }, user_name: Not("") } }); if (facebookPages) { await this.facebookApify(facebookPages, pageSize); if (total) { await this.facebookScraperCountryFilter(country, condition, days, from, to); } } } } /** * Updates the FacebookPages table with the scraped data from Apify's facebook-pages-scraper actor. */ async facebookPageUpdate(response) { for (const row of response) { if (row?.error) { const userName = row.url.split("/").pop(); this.SocialMedias.update({ social_user_name: userName, social_type: SocialType.FACEBOOK }, { source_posts_from: 'apify', social_media_not_found: true, is_updated: true, updated_by_cron: moment().format('YYYY-MM-DD HH:mm:ss') }); continue }; const facebookPage = await this.FacebookPages.findOne({ relations: ['users', 'socialMedia'], where: { user_name: row.pageName } }); console.log("facebookPage", facebookPage) if (facebookPage) { const s3Response = await uploadFileToS3(row.profilePictureUrl, `images/profile/`, this.s3Config); if (!s3Response) { this.SocialMedias.update({ id: facebookPage.socialMedia.id }, { source_posts_from: 'apify', apify_data_updated_attempts: facebookPage.socialMedia.apify_data_updated_attempts + 1 }); if (facebookPage.socialMedia.apify_data_updated_attempts >= 3) { this.SocialMedias.update({ id: facebookPage.socialMedia.id }, { source_posts_from: 'apify', social_media_not_found: true, is_updated: true, updated_by_cron: moment().format('YYYY-MM-DD HH:mm:ss') }); } continue; } await this.Users.update({ id: facebookPage.users.id }, { image: s3Response.path }); let socialMedia = facebookPage.socialMedia; if (socialMedia) { const fCount = row.followers - Number(socialMedia?.follower_count); socialMedia['follower_count'] = String(fCount); socialMedia['apify_data_updated_attempts'] = 0; socialMedia['is_updated'] = true; socialMedia['source_posts_from'] = 'apify'; socialMedia['updated_by_cron'] = moment().format('YYYY-MM-DD HH:mm:ss'); await this.SocialMedias.save(socialMedia); } facebookPage['description'] = row?.about_me?.text || ""; facebookPage['name'] = row.title || ""; facebookPage['page_logo'] = row.profilePictureUrl || ""; facebookPage['cover_image'] = row.coverPhotoUrl || ""; facebookPage['phone'] = row.phone || ""; facebookPage['website'] = row.website || ""; facebookPage['category'] = row.categories.join(', ') || ""; // if (facebookPage.users.is_updated_by_cron) { facebookPage['followers_count'] = row.followers || 0; // } await this.FacebookPages.save(facebookPage); } } } /** * @description * This function scrapes TikTok profiles' data using Apify's tiktok-profile-scraper actor. * It accepts three parameters: page, pageSize, and an optional social_id. * The function queries the database for TikTok usernames, optionally filtering by social_id. * It sends a POST request to the tiktok-profile-scraper actor with the usernames and retrieves the dataset ID. * After a delay, it makes a GET request to Apify to fetch the scraped data. * The function then updates the database with the TikTok profile data. * If there are more records to be processed, the function calls itself recursively with the next page number. */ async tiktokScraper(pageSize, social_user_name = null) { const daysAgo = moment().subtract(Constants.SOCIAL_MEDIA_CRON_DAY, 'days').toDate(); let query = await this.SocialMedias.createQueryBuilder('sm') .select(['sm.id', 'sm.social_user_name']) .where('social_type = :social_type', { social_type: SocialType.TIKTOK }) .andWhere('sm.social_media_not_found = false') .andWhere('sm.apify_data_updated_attempts < 3') .andWhere('(sm.updated_by_cron <= :dateLimit OR sm.updated_by_cron IS NULL)', { dateLimit: daysAgo }) .andWhere('is_verified = true').andWhere('is_token_expire = true'); // UPDATE SINGLE RECORD if (social_user_name) { query.andWhere('sm.social_user_name = :social_user_name', { social_user_name }) } const [result, total] = await query.orderBy('created_at', 'ASC').take(pageSize).getManyAndCount(); console.log("result", result) let ids = result.map(item => item.id) if (ids.length) { await this.SocialMedias .createQueryBuilder() .update() .set({ apify_data_updated_attempts: () => 'apify_data_updated_attempts + 1' }) .where('id IN (:...ids)', { ids }) .execute(); } let userNames = result.map(item => item.social_user_name) if (userNames.length) { //{"input":{"body":"{\"excludePinnedPosts\":false,\"profileSorting\":\"latest\",\"profiles\":[\"khaby.lame\"],\"resultsPerPage\":10,\"shouldDownloadAvatars\":false,\"shouldDownloadCovers\":false,\"shouldDownloadSlideshowImages\":false,\"shouldDownloadSubtitles\":false,\"shouldDownloadVideos\":false}","contentType":"application/json; charset=utf-8"},"options":{}} await this.tiktokApify(userNames, pageSize); if (total) { await this.tiktokScraper(pageSize, social_user_name); } } } async tiktokApify(userNames, pageSize) { console.log(userNames) const headers = {}; const params = {}; const tiktokScraperUrl = process.env.APIFY_BASE_URL + `acts/clockworks~tiktok-profile-scraper/runs?token=${this.APIFY_ACCESS_TOKEN}`; const data = { "profiles": userNames, resultsPerPage: 12 } // const datasetResponse = await this.postAxios(tiktokScraperUrl, headers, data, params); // const datasetId = datasetResponse.data.defaultDatasetId; // console.log(datasetId) // await this.sleep(60000); // const tiktokDatasetUrl = process.env.APIFY_BASE_URL + `datasets/${datasetId}/items?offset=0&token=${this.APIFY_ACCESS_TOKEN}` // const response = await this.getAxios(tiktokDatasetUrl, headers, params); const { defaultDatasetId } = await this.client.actor('clockworks~tiktok-profile-scraper').call(data); const { items } = await this.client.dataset(defaultDatasetId).listItems(); // TIKTOK SCRAPER PROVIDES POSTS await this.saveTikTokPosts(items); } async tikTokScraperCountryFilter(country: string, condition: string, days: number, from: number, to: number = 0) { const pageSize = 100; const daysAgo = moment().subtract(days, 'days').toDate(); let query = await this.SocialMedias.createQueryBuilder('sm') .select(['sm.id', 'sm.social_user_name', 'sm.follower_count']) .where('social_type = :social_type', { social_type: SocialType.TIKTOK }) .andWhere('sm.social_media_not_found = false') .andWhere('sm.apify_data_updated_attempts < 3') .andWhere('(sm.updated_by_cron <= :dateLimit OR sm.updated_by_cron IS NULL)', { dateLimit: daysAgo }) .andWhere('is_verified = false') .andWhere('country = :country', { country }); switch (condition) { case "BETWEEN": query.andWhere('follower_count BETWEEN :from AND :to', { from, to }); break; case "GREATER_THAN": query.andWhere('follower_count > :from', { from }); break; case "LESS_THAN": query.andWhere('follower_count < :from', { from }); break; case "EQUAL_TO": query.andWhere('follower_count = :from', { from }); break; } const [result, total] = await query.orderBy('created_at', 'ASC').take(pageSize).getManyAndCount(); const tikTokCountryIds = result.map(item => item.id); if (tikTokCountryIds.length) { await this.SocialMedias .createQueryBuilder() .update() .set({ apify_data_updated_attempts: () => 'apify_data_updated_attempts + 1' }) .where('id IN (:...ids)', { ids: tikTokCountryIds }) .execute(); } let userNames = result.map(item => item.social_user_name) console.log(userNames) if (userNames.length) { await this.tiktokApify(userNames, pageSize); if (total) { await this.tikTokScraperCountryFilter(country, condition, days, from, to); } } } /** * Updates TikTok profile information in the database based on the scraped data. */ async tiktokProfileUpdate(row) { const { authorMeta, text, musicMeta } = row; const socialMedia = await this.SocialMedias.findOne({ relations: ['users'], where: { social_user_name: authorMeta.name, social_type: SocialType.TIKTOK } }); if (socialMedia) { const s3Response = await uploadFileToS3(authorMeta.originalAvatarUrl, `images/profile/`, this.s3Config); if (!s3Response) { this.SocialMedias.update({ id: socialMedia.id }, { apify_data_updated_attempts: socialMedia.apify_data_updated_attempts + 1 }); if (socialMedia.apify_data_updated_attempts >= 3) { this.SocialMedias.update({ id: socialMedia.id }, { social_media_not_found: true, is_updated: true, updated_by_cron: moment().format('YYYY-MM-DD HH:mm:ss') }); } return false; } // if (socialMedia.users.is_updated_by_cron) { socialMedia['follower_count'] = authorMeta.fans; socialMedia['following_count'] = authorMeta.following; socialMedia['apify_data_updated_attempts'] = 0; socialMedia['post_count'] = authorMeta.video; socialMedia['source_posts_from'] = 'apify'; // } socialMedia['is_updated'] = true; socialMedia['updated_by_cron'] = moment().format('YYYY-MM-DD HH:mm:ss'); await this.SocialMedias.save(socialMedia); const tiktokProfiles = await this.TikTokProfiles.findOne({ where: { socialMedia: { id: socialMedia.id } } }); if (tiktokProfiles) { tiktokProfiles['avatar_url'] = authorMeta.originalAvatarUrl || ""; tiktokProfiles['bio_description'] = text || ""; tiktokProfiles['display_name'] = authorMeta.nickName || ""; tiktokProfiles['profile_deep_link'] = authorMeta.bioLink || ""; // if (socialMedia.users.is_updated_by_cron) { tiktokProfiles['likes_count'] = authorMeta.heart || 0; tiktokProfiles['video_count'] = authorMeta.video || 0; tiktokProfiles['following_count'] = authorMeta.following || 0; tiktokProfiles['follower_count'] = authorMeta.fans || 0; // } this.TikTokProfiles.save(tiktokProfiles); await this.Users.update({ id: socialMedia.users.id }, { image: s3Response.path }); } } return true; } async saveTikTokPosts(posts) { let userName = null; let totalLikes: bigint = BigInt(0); let totalComments: bigint = BigInt(0); let totalViews: bigint = BigInt(0); let followers: bigint = BigInt(0); let socialMedia = null; for (const post of posts) { if (post?.invalidUrls) { await this.SocialMedias.update( { social_user_name: In(post.invalidUrls), social_type: SocialType.TIKTOK }, { source_posts_from: 'apify', social_media_not_found: true, is_updated: true, updated_by_cron: moment().format('YYYY-MM-DD HH:mm:ss'), }, ); continue; } if (post?.authorMeta?.name == undefined) { this.SocialMedias.update({ social_user_name: post.input, social_type: SocialType.TIKTOK }, { source_posts_from: 'apify', social_media_not_found: true }); continue; } console.log("post", post) const tiktokProfile = await this.TikTokProfiles.findOne({ relations: ['socialMedia', 'users'], where: { username: post?.authorMeta?.name } }); if (tiktokProfile) { socialMedia = tiktokProfile.socialMedia; totalLikes += BigInt(post.diggCount || 0); totalComments += BigInt(post.commentCount || 0); totalViews += BigInt(post.playCount || 0); followers = BigInt(post.authorMeta.fans || 0); if (post.authorMeta.name != userName) { await this.TikTokVideos.delete({ tikTokProfiles: { id: tiktokProfile.id } }); await this.SocialMediaPosts.delete({ socialMedia: { id: tiktokProfile.socialMedia.id } }); const isUpdated = await this.tiktokProfileUpdate(post); if (!isUpdated) { continue; } userName = post.authorMeta.name } const newVideo = { users: tiktokProfile.users, socialMedia: tiktokProfile.socialMedia, tikTokProfiles: tiktokProfile, video_id: post.id || "", create_time: moment.unix(post.createTime).format('YYYY-MM-DD HH:mm:ss') || "", title: post.text || "", cover_image_url: post.videoMeta.coverUrl || "", embed_html: "", share_url: post.webVideoUrl || "", embed_link: "", height: post.videoMeta.height || 0, width: post.videoMeta.width || 0, like_count: post.diggCount || 0, share_count: post.shareCount || 0, comment_count: post.commentCount || 0, duration: post.videoMeta.duration || 0, view_count: post.playCount || 0 } const socialPosts = { users: tiktokProfile.users, socialMedia: tiktokProfile.socialMedia, title: post.text || "", description: "", like_count: post.diggCount || 0, comment_count: post.commentCount || 0, post_id: post.id, post_url: post.webVideoUrl || "", image_url: post?.videoMeta?.coverUrl || "", social_type: SocialType.TIKTOK, created_time: moment.unix(post.createTime).format('YYYY-MM-DD HH:mm:ss') || "" } await this.TikTokVideos.save(newVideo); await this.SocialMediaPosts.save(socialPosts) } } if (posts.length && followers) { const engagementRate = parseFloat( (((Number(totalLikes) + Number(totalComments) + Number(totalViews)) / Number(posts.length)) / Number(followers) * 100).toFixed(4) ); await this.updateSocialMediaEngagementRate(socialMedia.id, engagementRate); } // await this.SocialMedias.update({ id: socialMediaId }, { engagement_rate: engagementRate }); } async uploadPostImageOnS3Bucket(perPage) { const [posts, total] = await this.SocialMediaPosts.createQueryBuilder('posts') .innerJoinAndSelect('posts.socialMedia', 'socialMedia') .select(['posts.id', 'posts.image_url', 'posts.created_at', 'posts.post_id', 'posts.social_type', 'socialMedia.id', 'socialMedia.apify_data_updated_attempts']) .where('posts.is_uploaded_on_s3 IS NULL') .orderBy('posts.created_at', 'ASC') .take(perPage) .getManyAndCount(); if (posts.length == 0) return; for (const post of posts) { //await this.sleep(500); const s3Response = await uploadFileToS3(post.image_url, `files/campaign/posts/${post.socialMedia.id}/${post.social_type}`, this.s3Config); if (s3Response) { console.log(s3Response.path) await this.SocialMediaPosts.update({ id: post.id }, { image_url: s3Response.path, is_uploaded_on_s3: true }); switch (post.social_type) { case SocialType.INSTAGRAM: this.InstagramMedias.update({ instagram_media_id: post.post_id }, { thumbnail_url: s3Response.path }); break; case SocialType.TIKTOK: this.TikTokVideos.update({ video_id: post.post_id }, { cover_image_url: s3Response.path }); break; case SocialType.YOUTUBE: this.YoutubeVideos.update({ video_id: post.post_id }, { thumbnails: s3Response.path }); break; case SocialType.FACEBOOK: this.FacebookPosts.update({ post_id: post.post_id }, { thumbnails: s3Response.path }); break; } } else { await this.SocialMedias.update({ id: post.socialMedia.id }, { apify_data_updated_attempts: post.socialMedia.apify_data_updated_attempts + 1, updated_by_cron: null }); await this.SocialMediaPosts.update({ id: post.id }, { is_uploaded_on_s3: false }); } } if (total) { await this.uploadPostImageOnS3Bucket(perPage); } } async getAxios(url, headers, params) { try { const response = await axios.get(url, { headers, params }); // Handle the response data return response.data; } catch (error) { throw new HttpException(error.message, HttpStatus.BAD_REQUEST); } } async postAxios(url, headers, data, params = null) { try { const response = await axios.post(url, data, { headers, params }); // Handle the response data return response.data; } catch (error) { console.log(error.message) throw new HttpException(error.message, HttpStatus.BAD_REQUEST); } } }