Merge pull request #17 from pct-org/improvements

refactor: Change from pMap to pTimes

Author: TriPSs

apps/scraper/src/scraper.module.ts

@@ -2,7 +2,7 @@ import { Inject, Logger, Module, OnApplicationBootstrap } from '@nestjs/common'
 import { ScheduleModule, SchedulerRegistry } from '@nestjs/schedule'
 import { MongooseModule } from '@nestjs/mongoose'
 import { CronJob } from 'cron'
-import * as pMap from 'p-map'
+import * as pLimit from 'p-limit'
 
 import { ModelsModule } from './shared/models.module'
 import { ConfigModule } from './shared/config/config.module'
@@ -60,21 +60,19 @@ export class ScraperModule implements OnApplicationBootstrap {
 
     this.logger.log(`Enabled cron on '${this.configService.get(ConfigService.CRON_TIME)}'`)
 
-    if (this.configService.get(ConfigService.SCRAPE_ON_START)) {
+    if (this.configService.get(ConfigService.SCRAPE_ON_START) || true) {
       this.scrapeConfigs()
     }
   }
 
   private async scrapeConfigs(): Promise<void> {
     this.logger.log('Start scraping')
 
-    await pMap(
-      this.providersService.getProviders(),
-      (provider) => provider.scrapeConfigs(),
-      {
-        concurrency: 1
-      }
-    )
+    const limit = pLimit(1)
+
+    await Promise.all(this.providersService.getProviders().map((provider) => (
+      limit(() => provider.scrapeConfigs())
+    )))
   }
 
 }

libs/scraper/providers/base/src/base.provider.ts

@@ -7,6 +7,7 @@ import { formatTorrents } from '@pct-org/torrent/utils'
 import { BaseHelper } from '@pct-org/scraper/helpers/base'
 import { MOVIE_TYPE } from '@pct-org/types/movie'
 import { SHOW_TYPE } from '@pct-org/types/show'
+import * as pLimit from 'p-limit'
 
 import {
   ScrapedItem,
@@ -62,24 +63,27 @@ export abstract class BaseProvider {
   /**
    * Starts scraping the provided configs
    */
-  async scrapeConfigs(): Promise<void> {
+  public async scrapeConfigs(): Promise<void> {
     this.logger.log(`Started scraping...`)
 
-    await pMap(
-      this.configs,
-      (config) => this.scrapeConfig(config),
-      {
-        concurrency: 1
-      }
-    )
+    const limit = pLimit(1)
+
+    await Promise.all(this.configs.map((config) => (
+      limit(() => this.scrapeConfig(config))
+    )))
 
     this.logger.log(`Done scraping`)
   }
 
   /**
    * Set the configuration to scrape with.
    */
-  protected setConfig({ query, contentType, regexps = [], language = 'en' }: ScraperProviderConfig): void {
+  protected setConfig({
+    query,
+    contentType,
+    regexps = [],
+    language = 'en'
+  }: ScraperProviderConfig): void {
     this.contentType = contentType
     this.query = query
     this.language = language
@@ -105,36 +109,32 @@ export abstract class BaseProvider {
       this.logger.log(`Total pages ${totalPages}`)
 
       const torrents = await this.getAllTorrents(totalPages)
-      const allContent = await this.getAllContent(torrents)
+      const contents = await this.getAllContent(torrents)
 
-      this.logger.log(`Total content ${allContent.length}`)
+      this.logger.log(`Total content ${contents.length}`)
 
-      await pMap(
-        allContent,
-        async (content) => {
-          const isInBlacklist = await this.isItemBlackListed(content)
+      const limit = pLimit(this.maxWebRequests)
 
-          // Only get data for this item if it's not in the blacklist
-          if (!isInBlacklist) {
-            try {
-              await this.enhanceAndImport(content)
+      await Promise.all(contents.map((content) => limit(async() => {
+        const isInBlacklist = await this.isItemBlackListed(content)
 
-            } catch (err) {
-              const errorMessage = err.message || err
+        // Only get data for this item if it's not in the blacklist
+        if (!isInBlacklist) {
+          try {
+            await this.enhanceAndImport(content)
 
-              this.logger.error(`BaseProvider.scrapeConfig: ${errorMessage}`, err.stack)
+          } catch (err) {
+            const errorMessage = err.message || err
 
-              // Log the content so it can be better debugged from logs
-              if (errorMessage.includes('Could not find any data with slug')) {
-                this.logger.error(JSON.stringify(content))
-              }
+            this.logger.error(`BaseProvider.scrapeConfig: ${errorMessage}`, err.stack)
+
+            // Log the content so it can be better debugged from logs
+            if (errorMessage.includes('Could not find any data with slug')) {
+              this.logger.error(JSON.stringify(content))
             }
           }
-        },
-        {
-          concurrency: this.maxWebRequests
         }
-      )
+      })))
     } catch (err) {
       this.logger.error(`Catch BaseProvider.scrapeConfig: ${err.message || err}`, err.stack)
     }
@@ -146,7 +146,10 @@ export abstract class BaseProvider {
    * @returns {Promise<Boolean|Error>}
    */
   protected async isItemBlackListed(content: ScrapedItem): Promise<boolean | Error> {
-    const { slug, imdb } = content
+    const {
+      slug,
+      imdb
+    } = content
 
     const blacklistedItem = await this.blackListModel.findOne({
       $or: [
@@ -253,7 +256,11 @@ export abstract class BaseProvider {
   /**
    * Extract content information based on a regex.
    */
-  abstract extractContent({ torrent, regex, lang }): ScrapedItem | undefined
+  abstract extractContent({
+    torrent,
+    regex,
+    lang
+  }): ScrapedItem | undefined
 
   /**
    * Get content info from a given torrent.
@@ -307,7 +314,8 @@ export abstract class BaseProvider {
       return items.set(slug, item)
     }, {
       concurrency: 1
-    }).then(() => Array.from(items.values()))
+    })
+      .then(() => Array.from(items.values()))
   }
 
   /**
@@ -352,7 +360,7 @@ export abstract class BaseProvider {
   getAllTorrents(totalPages: number): Promise<Array<any>> {
     let torrents = []
 
-    return pTimes(totalPages, async (page) => {
+    return pTimes(totalPages, async(page) => {
       this.logger.debug(`Started searching ${this.name} on page ${page + 1} out of ${totalPages}`)
 
       // Get the page
@@ -364,17 +372,19 @@ export abstract class BaseProvider {
     }, {
       concurrency: 1,
       stopOnError: false
-    }).then(() => {
-      this.logger.log(`Found ${torrents.length} torrents.`)
+    })
+      .then(() => {
+        this.logger.log(`Found ${torrents.length} torrents.`)
 
-      return Promise.resolve(torrents)
-    }).catch((err) => {
-      // Only log the errors
-      this.logger.error(`Catch BaseProvider.getAllTorrents: ${err.message || err}`)
+        return Promise.resolve(torrents)
+      })
+      .catch((err) => {
+        // Only log the errors
+        this.logger.error(`Catch BaseProvider.getAllTorrents: ${err.message || err}`)
 
-      // We still want to resolve all the pages that did go well
-      return Promise.resolve(torrents)
-    })
+        // We still want to resolve all the pages that did go well
+        return Promise.resolve(torrents)
+      })
   }
 
   /**

libs/scraper/providers/eztv/src/eztv-provider.service.ts

@@ -1,7 +1,7 @@
 import { Inject, Injectable, Logger } from '@nestjs/common'
 import { BaseProvider, ScrapedItem, ScraperProviderConfig } from '@pct-org/scraper/providers/base'
 import { EztvApi } from '@pct-org/eztv-api'
-import * as pMap from 'p-map'
+import * as pLimit from 'p-limit'
 import { ShowHelperService } from '@pct-org/scraper/helpers/show'
 import { SHOW_TYPE } from '@pct-org/types/show'
 
@@ -36,36 +36,32 @@ export class EztvProviderService extends BaseProvider {
 
     this.logger.log(`${this.name}: Found ${contents.length} ${this.contentType}s.`)
 
-    await pMap(
-      contents,
-      async (content) => {
-        const isInBlacklist = await this.isItemBlackListed(content)
+    const limit = pLimit(this.maxWebRequests)
 
-        // Only get data for this item if it's not in the blacklist
-        if (!isInBlacklist) {
-          try {
-            // Get full show data
-            const show: ScrapedItem = await this.api.getShowData(content)
+    await Promise.all(contents.map((content) => limit(async() => {
+      const isInBlacklist = await this.isItemBlackListed(content)
 
-            // Enhance and import the show
-            await this.enhanceAndImport(show)
+      // Only get data for this item if it's not in the blacklist
+      if (!isInBlacklist) {
+        try {
+          // Get full show data
+          const show: ScrapedItem = await this.api.getShowData(content)
 
-          } catch (err) {
-            const errorMessage = err.message || err
+          // Enhance and import the show
+          await this.enhanceAndImport(show)
 
-            this.logger.error(`EztvProviderService.scrapeConfig: ${errorMessage}`, err.stack)
+        } catch (err) {
+          const errorMessage = err.message || err
 
-            // Log the content so it can be better debugged from logs
-            if (errorMessage.includes('Could not find any data with slug')) {
-              this.logger.error(JSON.stringify(content))
-            }
+          this.logger.error(`EztvProviderService.scrapeConfig: ${errorMessage}`, err.stack)
+
+          // Log the content so it can be better debugged from logs
+          if (errorMessage.includes('Could not find any data with slug')) {
+            this.logger.error(JSON.stringify(content))
           }
         }
-      },
-      {
-        concurrency: this.maxWebRequests
       }
-    )
+    })))
   }
 
   extractContent({ torrent, regex, lang }: { torrent: any; regex: any; lang: any }): undefined {

package.json

@@ -72,6 +72,7 @@
     "node-tvdb": "^4.1.0",
     "omdb-api-pt": "^2.0.1",
     "opensubtitles-api": "^5.1.2",
+    "p-limit": "^3.1.0",
     "p-map": "^4.0.0",
     "p-times": "^3.0.0",
     "pm2": "^5.1.1",

yarn.lock

@@ -9767,6 +9767,13 @@ p-limit@^3.0.1, p-limit@^3.0.2:
   dependencies:
     p-try "^2.0.0"
 
+p-limit@^3.1.0:
+  version "3.1.0"
+  resolved "https://registry.yarnpkg.com/p-limit/-/p-limit-3.1.0.tgz#e1daccbe78d0d1388ca18c64fea38e3e57e3706b"
+  integrity sha512-TYOanM3wGwNGsZN2cVTYPArw454xnXj5qmWF1bEoAc4+cU/ol7GVh7odevjp1FNHduHc3KZMcFduxU5Xc6uJRQ==
+  dependencies:
+    yocto-queue "^0.1.0"
+
 p-locate@^3.0.0:
   version "3.0.0"
   resolved "https://registry.yarnpkg.com/p-locate/-/p-locate-3.0.0.tgz#322d69a05c0264b25997d9f40cd8a891ab0064a4"
@@ -13792,6 +13799,11 @@ [email protected]:
   resolved "https://registry.yarnpkg.com/yn/-/yn-3.1.1.tgz#1e87401a09d767c1d5eab26a6e4c185182d2eb50"
   integrity sha512-Ux4ygGWsu2c7isFWe8Yu1YluJmqVhxqK2cLXNQA5AcC3QfbGNpM7fu0Y8b/z16pXLnFxZYvWhd3fhBY9DLmC6Q==
 
+yocto-queue@^0.1.0:
+  version "0.1.0"
+  resolved "https://registry.yarnpkg.com/yocto-queue/-/yocto-queue-0.1.0.tgz#0294eb3dee05028d31ee1a5fa2c556a6aaf10a1b"
+  integrity sha512-rVksvsnNCdJ/ohGc6xgPwyN8eheCxsiLM8mxuE/t/mOVqJewPuO1miLpTHQiRgTKCLexL4MeAFVagts7HmNZ2Q==
+
 yts-api-pt@^2.0.0:
   version "2.0.0"
   resolved "https://registry.yarnpkg.com/yts-api-pt/-/yts-api-pt-2.0.0.tgz#f1048d258e3ab5b1e0862e9732ea59fab1ead64a"