Setup custom listing processors

The custom processors will allow customers to append extra information to the listing or transform listing information to better match their expected result

index.js

@@ -35,7 +35,14 @@ setInterval(
             .forEach(async (prov) => {
               const pro = fetchedProvider.find((fp) => fp.metaInformation.id === prov.id);
               pro.init(prov, job.blacklist);
-              await new FredyRuntime(pro.config, job.notificationAdapter, prov.id, job.id, similarityCache).execute();
+              await new FredyRuntime(
+                pro.config,
+                job.notificationAdapter,
+                prov.id,
+                job.id,
+                similarityCache,
+                job.listingProcessors
+              ).execute();
               setLastJobExecution(job.id);
             });
         });

lib/FredyRuntime.js

@@ -1,6 +1,7 @@
 import { NoNewListingsWarning } from './errors.js';
 import { setKnownListings, getKnownListings } from './services/storage/listingsStorage.js';
 import * as notify from './notification/notify.js';
+import * as process from './processors/process.js';
 import xray from './services/scraper.js';
 import * as scrapingAnt from './services/scrapingAnt.js';
 import urlModifier from './services/queryStringMutator.js';
@@ -13,12 +14,13 @@ class FredyRuntime {
    * @param jobKey key of the job that is currently running (from within the config)
    * @param similarityCache cache instance holding values to check for similarity of entries
    */
-  constructor(providerConfig, notificationConfig, providerId, jobKey, similarityCache) {
+  constructor(providerConfig, notificationConfig, providerId, jobKey, similarityCache, listingProcessors) {
     this._providerConfig = providerConfig;
     this._notificationConfig = notificationConfig;
     this._providerId = providerId;
     this._jobKey = jobKey;
     this._similarityCache = similarityCache;
+    this._listingProcessors = listingProcessors;
   }
   execute() {
     return (
@@ -36,6 +38,7 @@ class FredyRuntime {
         .then(this._save.bind(this))
         //check for similar listings. if found, remove them before notifying
         .then(this._filterBySimilarListings.bind(this))
+        .then(this._processListings.bind(this))
         //notify the user using the configured notification adapter
         .then(this._notify.bind(this))
         //if an error occurred on the way, handle it here.
@@ -124,6 +127,11 @@ class FredyRuntime {
     filteredList.forEach((filter) => this._similarityCache.addCacheEntry(this._jobKey, filter.title));
     return filteredList;
   }
+
+  _processListings(listings) {
+    return process.processListings(listings, this._listingProcessors);
+  }
+
   _handleError(err) {
     if (err.name !== 'NoNewListingsWarning') console.error(err);
   }

lib/api/routes/jobRouter.js

@@ -42,7 +42,7 @@ jobRouter.get('/processingTimes', async (req, res) => {
   res.send();
 });
 jobRouter.post('/', async (req, res) => {
-  const { provider, notificationAdapter, name, blacklist = [], jobId, enabled } = req.body;
+  const { provider, notificationAdapter, name, blacklist = [], jobId, enabled, listingProcessors } = req.body;
   if (
     provider.find((p) => p.id === immoscoutProvider.metaInformation.id) != null &&
     (config.scrapingAnt.apiKey == null || config.scrapingAnt.apiKey.length === 0)
@@ -61,6 +61,7 @@ jobRouter.post('/', async (req, res) => {
       blacklist,
       provider,
       notificationAdapter,
+      listingProcessors,
     });
   } catch (error) {
     res.send(new Error(error));

lib/processors/process.js

@@ -0,0 +1,23 @@
+import fs from 'fs';
+const path = '.';
+
+const registeredProcessors = await Promise.all(
+  fs
+    .readdirSync('./lib/processors')
+    .filter((file) => file.endsWith('.js') && file !== 'process.js')
+    .map(async (integPath) => await import(`${path}/${integPath}`))
+);
+
+const findProcessor = (processor) => {
+  return registeredProcessors.find((a) => a.config.id === processor.id);
+};
+export function processListings(listings, listingProcessors) {
+  const processors = listingProcessors.map(findProcessor);
+  const processedListingsPromises = listings.map(async (listing) => {
+    return processors.reduce(
+      async (listingAcc, processor) => await processor.processListing({ listing: await listingAcc }),
+      listing
+    );
+  });
+  return Promise.resolve(Promise.all(processedListingsPromises));
+}

lib/processors/staticProcessor.js

@@ -0,0 +1,9 @@
+export async function processListing({ listing }) {
+  return { ...listing, processed: true };
+}
+export const config = {
+  id: 'static',
+  name: 'Static',
+  description: 'This processor adds an extra `processed: true` property to the listing',
+  config: {},
+};

lib/services/storage/jobStorage.js

@@ -11,8 +11,16 @@ const db = new LowdashAdapter(adapter, { jobs: [] });
 
 db.read();
 
-
-export const upsertJob = ({ jobId, name, blacklist = [], enabled = true, provider, notificationAdapter, userId }) => {
+export const upsertJob = ({
+  jobId,
+  name,
+  blacklist = [],
+  enabled = true,
+  provider,
+  notificationAdapter,
+  userId,
+  listingProcessors,
+}) => {
   const currentJob =
     jobId == null
       ? null
@@ -33,6 +41,7 @@ export const upsertJob = ({ jobId, name, blacklist = [], enabled = true, provide
     blacklist,
     provider,
     notificationAdapter,
+    listingProcessors,
   });
   db.chain.set('jobs', jobs).value();
   db.write();