Member enrichment ProgAI Linkedin scraper source

services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts

@@ -1,6 +1,7 @@
 import { findMemberIdentityWithTheMostActivityInPlatform as findMemberIdentityWithTheMostActivityInPlatformQuestDb } from '@crowd/data-access-layer/src/activities'
 import {
   findMemberEnrichmentCacheDb,
+  findMemberEnrichmentCacheForAllSourcesDb,
   insertMemberEnrichmentCacheDb,
   touchMemberEnrichmentCacheUpdatedAtDb,
   updateMemberEnrichmentCacheDb,
@@ -42,7 +43,7 @@ export async function getEnrichmentData(
 export async function normalizeEnrichmentData(
   source: MemberEnrichmentSource,
   data: IMemberEnrichmentData,
-): Promise<IMemberEnrichmentDataNormalized> {
+): Promise<IMemberEnrichmentDataNormalized | IMemberEnrichmentDataNormalized[]> {
   const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log)
   return service.normalize(data)
 }
@@ -100,6 +101,12 @@ export async function findMemberEnrichmentCache(
   return findMemberEnrichmentCacheDb(svc.postgres.reader.connection(), memberId, source)
 }
 
+export async function findMemberEnrichmentCacheForAllSources(
+  memberId: string,
+): Promise<IMemberEnrichmentCache<IMemberEnrichmentData>[]> {
+  return findMemberEnrichmentCacheForAllSourcesDb(svc.postgres.reader.connection(), memberId)
+}
+
 export async function insertMemberEnrichmentCache(
   source: MemberEnrichmentSource,
   memberId: string,

services/apps/premium/members_enrichment_worker/src/factory.ts

@@ -3,6 +3,7 @@ import { Logger } from '@crowd/logging'
 import { MemberEnrichmentSource } from '@crowd/types'
 
 import EnrichmentServiceClearbit from './sources/clearbit/service'
+import EnrichmentServiceProgAILinkedinScraper from './sources/progai-linkedin-scraper/service'
 import EnrichmentServiceProgAI from './sources/progai/service'
 import EnrichmentServiceSerpApi from './sources/serp/service'
 import { IEnrichmentService } from './types'
@@ -24,6 +25,8 @@ export class EnrichmentSourceServiceFactory {
         return new EnrichmentServiceClearbit(log)
       case MemberEnrichmentSource.SERP:
         return new EnrichmentServiceSerpApi(log)
+      case MemberEnrichmentSource.PROGAI_LINKEDIN_SCRAPER:
+        return new EnrichmentServiceProgAILinkedinScraper(log)
       default:
         throw new Error(`Enrichment service for ${source} is not found!`)
     }

services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts

@@ -56,7 +56,7 @@ export default class EnrichmentServiceClearbit extends LoggerBase implements IEn
     super(log)
   }
 
-  isEnrichableBySource(input: IEnrichmentSourceInput): boolean {
+  async isEnrichableBySource(input: IEnrichmentSourceInput): Promise<boolean> {
     return !!input.email?.value && input.email?.verified
   }
 

services/apps/premium/members_enrichment_worker/src/sources/progai-linkedin-scraper/service.ts

@@ -0,0 +1,158 @@
+import axios from 'axios'
+
+import { Logger, LoggerBase } from '@crowd/logging'
+import { IMemberIdentity, MemberEnrichmentSource, PlatformType } from '@crowd/types'
+
+import { findMemberEnrichmentCacheForAllSources } from '../../activities/enrichment'
+import { EnrichmentSourceServiceFactory } from '../../factory'
+import {
+  IEnrichmentService,
+  IEnrichmentSourceInput,
+  IMemberEnrichmentDataNormalized,
+} from '../../types'
+import { IMemberEnrichmentDataProgAI, IMemberEnrichmentDataProgAIResponse } from '../progai/types'
+
+export default class EnrichmentServiceProgAILinkedinScraper
+  extends LoggerBase
+  implements IEnrichmentService
+{
+  public source: MemberEnrichmentSource = MemberEnrichmentSource.PROGAI_LINKEDIN_SCRAPER
+  public platform = `enrichment-${this.source}`
+
+  public alsoFindInputsInSourceCaches: MemberEnrichmentSource[] = [
+    MemberEnrichmentSource.CLEARBIT,
+    MemberEnrichmentSource.SERP,
+  ]
+
+  public enrichableBySql = `(mi.verified AND mi.type = 'username' and mi.platform = 'linkedin')`
+
+  // bust cache after 120 days
+  public cacheObsoleteAfterSeconds = 60 * 60 * 24 * 120
+
+  constructor(public readonly log: Logger) {
+    super(log)
+  }
+
+  // in addition to members with linkedin identity
+  // we'll also use existing cache rows from other sources (serp and clearbit)
+  // if there are linkedin urls there as well, we'll enrich using these also
+  async isEnrichableBySource(input: IEnrichmentSourceInput): Promise<boolean> {
+    const caches = await findMemberEnrichmentCacheForAllSources(input.memberId)
+
+    let hasEnrichableLinkedinInCache = false
+    for (const cache of caches) {
+      if (this.alsoFindInputsInSourceCaches.includes(cache.source)) {
+        const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(
+          cache.source,
+          this.log,
+        )
+        const normalized = service.normalize(cache.data) as IMemberEnrichmentDataNormalized
+        if (normalized.identities.some((i) => i.platform === PlatformType.LINKEDIN)) {
+          hasEnrichableLinkedinInCache = true
+          break
+        }
+
+        break
+      }
+    }
+
+    return (
+      hasEnrichableLinkedinInCache ||
+      (input.linkedin && input.linkedin.value && input.linkedin.verified)
+    )
+  }
+
+  async hasRemainingCredits(): Promise<boolean> {
+    return true
+  }
+
+  private async findConsumableLinkedinIdentities(
+    input: IEnrichmentSourceInput,
+  ): Promise<IMemberIdentity[]> {
+    const consumableIdentities: IMemberIdentity[] = []
+    const caches = await findMemberEnrichmentCacheForAllSources(input.memberId)
+    const linkedinUrlHashmap = new Map<string, boolean>()
+
+    for (const cache of caches) {
+      if (this.alsoFindInputsInSourceCaches.includes(cache.source)) {
+        const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(
+          cache.source,
+          this.log,
+        )
+        const normalized = service.normalize(cache.data) as IMemberEnrichmentDataNormalized
+        if (normalized.identities.some((i) => i.platform === PlatformType.LINKEDIN)) {
+          const identity = normalized.identities.find((i) => i.platform === PlatformType.LINKEDIN)
+          if (!linkedinUrlHashmap.get(identity.value)) {
+            consumableIdentities.push(identity)
+            linkedinUrlHashmap.set(identity.value, true)
+          }
+        }
+      }
+    }
+
+    // also add the linkedin identity from the input
+    if (
+      input.linkedin &&
+      input.linkedin.value &&
+      input.linkedin.verified &&
+      !linkedinUrlHashmap.get(input.linkedin.value)
+    ) {
+      consumableIdentities.push(input.linkedin)
+    }
+
+    return consumableIdentities
+  }
+
+  async getData(input: IEnrichmentSourceInput): Promise<IMemberEnrichmentDataProgAI[] | null> {
+    const profiles: IMemberEnrichmentDataProgAI[] = []
+    const consumableIdentities = await this.findConsumableLinkedinIdentities(input)
+
+    for (const identity of consumableIdentities) {
+      const data = await this.getDataUsingLinkedinHandle(identity.value)
+      if (data) {
+        profiles.push(data)
+      }
+    }
+
+    return profiles.length > 0 ? profiles : null
+  }
+
+  private async getDataUsingLinkedinHandle(handle: string): Promise<IMemberEnrichmentDataProgAI> {
+    let response: IMemberEnrichmentDataProgAIResponse
+
+    try {
+      const url = `${process.env['CROWD_ENRICHMENT_PROGAI_URL']}/get_profile`
+      const config = {
+        method: 'get',
+        url,
+        params: {
+          linkedin_url: `https://linkedin.com/in/${handle}`,
+          with_emails: true,
+          api_key: process.env['CROWD_ENRICHMENT_PROGAI_API_KEY'],
+        },
+        headers: {},
+      }
+
+      response = (await axios(config)).data
+    } catch (err) {
+      throw new Error(err)
+    }
+
+    return response.profile
+  }
+
+  normalize(profiles: IMemberEnrichmentDataProgAI[]): IMemberEnrichmentDataNormalized[] {
+    const normalizedProfiles: IMemberEnrichmentDataNormalized[] = []
+    const progaiService = EnrichmentSourceServiceFactory.getEnrichmentSourceService(
+      MemberEnrichmentSource.PROGAI,
+      this.log,
+    )
+
+    for (const profile of profiles) {
+      const normalized = progaiService.normalize(profile) as IMemberEnrichmentDataNormalized
+      normalizedProfiles.push(normalized)
+    }
+
+    return normalizedProfiles.length > 0 ? normalizedProfiles : null
+  }
+}

services/apps/premium/members_enrichment_worker/src/sources/progai/service.ts

@@ -118,7 +118,7 @@ export default class EnrichmentServiceProgAI extends LoggerBase implements IEnri
     super(log)
   }
 
-  isEnrichableBySource(input: IEnrichmentSourceInput): boolean {
+  async isEnrichableBySource(input: IEnrichmentSourceInput): Promise<boolean> {
     const enrichableUsingGithubHandle = !!input.github?.value
     const enrichableUsingEmail = this.alsoUseEmailIdentitiesForEnrichment && !!input.email?.value
     return enrichableUsingGithubHandle || enrichableUsingEmail

services/apps/premium/members_enrichment_worker/src/sources/serp/service.ts

@@ -36,7 +36,7 @@ export default class EnrichmentServiceSerpApi extends LoggerBase implements IEnr
     super(log)
   }
 
-  isEnrichableBySource(input: IEnrichmentSourceInput): boolean {
+  async isEnrichableBySource(input: IEnrichmentSourceInput): Promise<boolean> {
     const displayNameSplit = input.displayName?.split(' ')
     return (
       displayNameSplit?.length > 1 &&
@@ -126,7 +126,7 @@ export default class EnrichmentServiceSerpApi extends LoggerBase implements IEnr
           platform: PlatformType.LINKEDIN,
           type: MemberIdentityType.USERNAME,
           verified: false,
-          value: this.normalizeLinkedUrl(data.linkedinUrl),
+          value: this.normalizeLinkedUrl(data.linkedinUrl).split('/').pop(),
         },
       ],
     }

services/apps/premium/members_enrichment_worker/src/types.ts

@@ -14,6 +14,7 @@ import { IMemberEnrichmentDataProgAI } from './sources/progai/types'
 import { IMemberEnrichmentDataSerp } from './sources/serp/types'
 
 export interface IEnrichmentSourceInput {
+  memberId: string
   github?: IMemberIdentity
   linkedin?: IMemberIdentity
   email?: IMemberIdentity
@@ -25,6 +26,7 @@ export interface IEnrichmentSourceInput {
 
 export type IMemberEnrichmentData =
   | IMemberEnrichmentDataProgAI
+  | IMemberEnrichmentDataProgAI[]
   | IMemberEnrichmentDataClearbit
   | IMemberEnrichmentDataSerp
 
@@ -35,7 +37,7 @@ export interface IEnrichmentService {
   cacheObsoleteAfterSeconds: number
 
   // can the source enrich using this input
-  isEnrichableBySource(input: IEnrichmentSourceInput): boolean
+  isEnrichableBySource(input: IEnrichmentSourceInput): Promise<boolean>
 
   // does the source have credits to enrich members, if returned false the source will be skipped
   // response will be saved to redis for 60 seconds and will be used for subsequent calls
@@ -49,7 +51,9 @@ export interface IEnrichmentService {
 
   // should either return the data or null if it's a miss
   getData(input: IEnrichmentSourceInput): Promise<IMemberEnrichmentData | null>
-  normalize(data: IMemberEnrichmentData): IMemberEnrichmentDataNormalized
+  normalize(
+    data: IMemberEnrichmentData,
+  ): IMemberEnrichmentDataNormalized | IMemberEnrichmentDataNormalized[]
 }
 
 export interface IMemberEnrichmentDataNormalized {

services/apps/premium/members_enrichment_worker/src/workflows/enrichMember.ts

@@ -43,6 +43,7 @@ export async function enrichMember(
     // cache is obsolete when it's not found or cache.updatedAt is older than cacheObsoleteAfterSeconds
     if (await isCacheObsolete(source, cache)) {
       const enrichmentInput: IEnrichmentSourceInput = {
+        memberId: input.id,
         email: input.identities.find((i) => i.verified && i.type === MemberIdentityType.EMAIL),
         linkedin: input.identities.find(
           (i) =>

services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts

@@ -24,6 +24,7 @@ export async function getMembersToEnrich(args: IGetMembersForEnrichmentArgs): Pr
     MemberEnrichmentSource.PROGAI,
     MemberEnrichmentSource.CLEARBIT,
     MemberEnrichmentSource.SERP,
+    MemberEnrichmentSource.PROGAI_LINKEDIN_SCRAPER,
   ]
 
   const members = await getEnrichableMembers(MEMBER_ENRICHMENT_PER_RUN, sources, afterCursor)

services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts

@@ -531,3 +531,20 @@ export async function findMemberEnrichmentCacheDb<T>(
 
   return result ?? null
 }
+
+export async function findMemberEnrichmentCacheForAllSourcesDb<T>(
+  tx: DbConnOrTx,
+  memberId: string,
+): Promise<IMemberEnrichmentCache<T>[]> {
+  const result = await tx.manyOrNone(
+    `
+    select *
+    from "memberEnrichmentCache"
+    where 
+      "memberId" = $(memberId) and data is not null;
+    `,
+    { memberId },
+  )
+
+  return result ?? null
+}

services/libs/types/src/enums/enrichment.ts

@@ -2,4 +2,5 @@ export enum MemberEnrichmentSource {
   PROGAI = 'progai',
   CLEARBIT = 'clearbit',
   SERP = 'serp',
+  PROGAI_LINKEDIN_SCRAPER = 'progai-linkedin-scraper',
 }