Final tweaks to member enrichment

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

@@ -65,7 +65,7 @@ export async function getEnrichmentData(
   input: IEnrichmentSourceInput,
 ): Promise<IMemberEnrichmentData | null> {
   const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log)
-  if ((await service.isEnrichableBySource(input)) && (await hasRemainingCredits(source))) {
+  if (await service.isEnrichableBySource(input)) {
     return service.getData(input)
   }
   return null
@@ -228,7 +228,7 @@ export async function updateMemberUsingSquashedPayload(
 
     // process identities
     if (squashedPayload.identities.length > 0) {
-      svc.log.info({ memberId }, 'Adding to member identities!')
+      svc.log.debug({ memberId }, 'Adding to member identities!')
       for (const i of squashedPayload.identities) {
         updated = true
         promises.push(
@@ -247,7 +247,7 @@ export async function updateMemberUsingSquashedPayload(
     // process contributions
     // if squashed payload has data from progai, we should fetch contributions here
     // it's ommited from the payload because it takes a lot of space
-    svc.log.info('Processing contributions! ', { memberId, hasContributions })
+    svc.log.debug('Processing contributions! ', { memberId, hasContributions })
     if (hasContributions) {
       promises.push(
         findMemberEnrichmentCache([MemberEnrichmentSource.PROGAI], memberId)
@@ -265,7 +265,6 @@ export async function updateMemberUsingSquashedPayload(
           .then((normalized) => {
             if (normalized) {
               const typed = normalized as IMemberEnrichmentDataNormalized
-              svc.log.info('Normalized contributions: ', { contributions: typed.contributions })
 
               if (typed.contributions) {
                 updated = true
@@ -280,7 +279,7 @@ export async function updateMemberUsingSquashedPayload(
     let attributes = existingMemberData.attributes as Record<string, unknown>
 
     if (squashedPayload.attributes) {
-      svc.log.info({ memberId }, 'Updating member attributes!')
+      svc.log.debug({ memberId }, 'Updating member attributes!')
 
       attributes = _.merge({}, attributes, squashedPayload.attributes)
 
@@ -298,7 +297,7 @@ export async function updateMemberUsingSquashedPayload(
 
     // process reach
     if (squashedPayload.reach && Object.keys(squashedPayload.reach).length > 0) {
-      svc.log.info({ memberId }, 'Updating member reach!')
+      svc.log.debug({ memberId }, 'Updating member reach!')
       let reach: IMemberReach
 
       if (existingMemberData.reach && existingMemberData.reach.total) {
@@ -608,7 +607,7 @@ export async function findWhichLinkedinProfileToUseAmongScraperResult(
     }
   }
 
-  if (!categorized.selected && profilesFromUnverfiedIdentities.length > 0) {
+  if (profilesFromUnverfiedIdentities.length > 0) {
     const result = await findRelatedLinkedinProfilesWithLLM(
       memberId,
       memberData,
@@ -617,7 +616,9 @@ export async function findWhichLinkedinProfileToUseAmongScraperResult(
 
     // check if empty object
     if (result.profileIndex !== null) {
-      categorized.selected = profilesFromUnverfiedIdentities[result.profileIndex]
+      if (!categorized.selected) {
+        categorized.selected = profilesFromUnverfiedIdentities[result.profileIndex]
+      }
       // add profiles not selected to discarded
       for (let i = 0; i < profilesFromUnverfiedIdentities.length; i++) {
         if (i !== result.profileIndex) {
@@ -796,46 +797,35 @@ function prepareWorkExperiences(
   // we iterate through the existing version experiences to see if update is needed
   for (const current of orderedCurrentVersion) {
     // try and find a matching experience in the new versions by title
-    let match = orderedNewVersion.find(
+    const match = orderedNewVersion.find(
       (e) =>
         e.title === current.jobTitle &&
         e.identities &&
         e.identities.some((e) => e.organizationId === current.orgId),
     )
-    if (!match) {
-      // if we didn't find a match by title we should check dates
-      match = orderedNewVersion.find(
-        (e) =>
-          dateIntersects(current.dateStart, current.dateEnd, e.startDate, e.endDate) &&
-          e.identities &&
-          e.identities.some((e) => e.organizationId === current.orgId),
-      )
-    }
 
     // if we found a match we can check if we need something to update
-    if (match) {
-      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    if (
+      match &&
+      current.dateStart === match.startDate &&
+      current.dateEnd === null &&
+      match.endDate !== null
+    ) {
       const toUpdateInner: Record<string, any> = {}
 
-      // lets check if the dates and title are the same otherwise we need to update them
-      if (current.dateStart !== match.startDate) {
-        toUpdateInner.dateStart = match.startDate
-      }
-
-      if (current.dateEnd !== match.endDate) {
-        toUpdateInner.dateEnd = match.endDate
-      }
-
-      if (current.jobTitle !== match.title) {
-        toUpdateInner.title = match.title
-      }
-
-      if (Object.keys(toUpdateInner).length > 0) {
-        toUpdate.set(current, toUpdateInner)
-      }
+      toUpdateInner.dateEnd = match.endDate
+      toUpdate.set(current, toUpdateInner)
 
       // remove the match from the new version array so we later don't process it again
       orderedNewVersion = orderedNewVersion.filter((e) => e.id !== match.id)
+    } else if (
+      match &&
+      (current.dateStart !== match.startDate || current.dateEnd !== null || match.endDate === null)
+    ) {
+      // there's an incoming work experiences, but it's conflicting with the existing manually updated data
+      // we shouldn't add or update anything when this happens
+      // we can only update dateEnd of existing manually changed data, when it has a null dateEnd
+      orderedNewVersion = orderedNewVersion.filter((e) => e.id !== match.id)
     }
     // if we didn't find a match we should just leave it as it is in the database since it was manual input
   }
@@ -850,28 +840,6 @@ function prepareWorkExperiences(
   }
 }
 
-function dateIntersects(
-  d1Start?: string | null,
-  d1End?: string | null,
-  d2Start?: string | null,
-  d2End?: string | null,
-): boolean {
-  // If both periods have no dates at all, we can't determine intersection
-  if ((!d1Start && !d1End) || (!d2Start && !d2End)) {
-    return false
-  }
-
-  // Convert strings to timestamps, using fallbacks for missing dates
-  const start1 = d1Start ? new Date(d1Start).getTime() : -Infinity
-  const end1 = d1End ? new Date(d1End).getTime() : Infinity
-  const start2 = d2Start ? new Date(d2Start).getTime() : -Infinity
-  const end2 = d2End ? new Date(d2End).getTime() : Infinity
-
-  // Periods intersect if one period's start is before other period's end
-  // and that same period's end is after the other period's start
-  return start1 <= end2 && end1 >= start2
-}
-
 export async function cleanAttributeValue(
   attributeValue: string | string[] | Record<string, any>,
 ): Promise<string | string[] | Record<string, any>> {

services/apps/members_enrichment_worker/src/bin/onboarding.ts

@@ -27,7 +27,7 @@ const tenantId = processArguments[0]
 
 const minMemberActivities = 100
 const maxConcurrentProcessing = 5
-const maxMembersToProcess = 1000
+const maxMembersToProcess = Infinity
 
 async function getEnrichableMembers(limit: number): Promise<string[]> {
   const query = `

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

@@ -1,5 +1,6 @@
 import axios from 'axios'
 
+import { replaceDoubleQuotes } from '@crowd/common'
 import { Logger, LoggerBase } from '@crowd/logging'
 import {
   MemberAttributeName,
@@ -216,10 +217,10 @@ export default class EnrichmentServiceClearbit extends LoggerBase implements IEn
       }
 
       normalized.memberOrganizations.push({
-        name: data.employment.name,
+        name: replaceDoubleQuotes(data.employment.name),
         source: OrganizationSource.ENRICHMENT_CLEARBIT,
         identities: orgIdentities,
-        title: data.employment.title,
+        title: replaceDoubleQuotes(data.employment.title),
         startDate: null,
         endDate: null,
       })

services/apps/members_enrichment_worker/src/sources/crustdata/service.ts

@@ -1,6 +1,6 @@
 import axios from 'axios'
 
-import { isEmail } from '@crowd/common'
+import { isEmail, replaceDoubleQuotes } from '@crowd/common'
 import { Logger, LoggerBase } from '@crowd/logging'
 import {
   IMemberEnrichmentCache,
@@ -352,13 +352,15 @@ export default class EnrichmentServiceCrustdata extends LoggerBase implements IE
         }
 
         normalized.memberOrganizations.push({
-          name: workExperience.employer_name,
+          name: replaceDoubleQuotes(workExperience.employer_name),
           source: OrganizationSource.ENRICHMENT_CRUSTDATA,
           identities,
-          title: workExperience.employee_title,
+          title: replaceDoubleQuotes(workExperience.employee_title),
           startDate: workExperience?.start_date ?? null,
           endDate: workExperience?.end_date ?? null,
-          organizationDescription: workExperience.employer_linkedin_description,
+          organizationDescription: replaceDoubleQuotes(
+            workExperience.employer_linkedin_description,
+          ),
         })
       }
     }

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

@@ -1,7 +1,7 @@
 import axios from 'axios'
 import lodash from 'lodash'
 
-import { websiteNormalizer } from '@crowd/common'
+import { replaceDoubleQuotes, websiteNormalizer } from '@crowd/common'
 import { Logger, LoggerBase } from '@crowd/logging'
 import {
   MemberAttributeName,
@@ -261,51 +261,80 @@ export default class EnrichmentServiceProgAI extends LoggerBase implements IEnri
   ): IMemberEnrichmentDataNormalized {
     if (data.work_experiences) {
       for (const workExperience of data.work_experiences) {
-        const identities = []
-
-        if (workExperience.companyUrl) {
-          const normalizedDomain = websiteNormalizer(workExperience.companyUrl, false)
+        if (
+          workExperience.company !== null ||
+          workExperience.companyUrl !== null ||
+          workExperience.companyLinkedInUrl !== null
+        ) {
+          const identities = []
+          let hasPrimaryDomainIdentity = false
+
+          if (workExperience.companyUrl) {
+            const normalizedDomain = websiteNormalizer(workExperience.companyUrl, false)
+
+            // sometimes companyUrl is a github link, we don't want to add it as a primary domain
+            if (
+              normalizedDomain &&
+              !workExperience.companyUrl.toLowerCase().includes('github') &&
+              !(workExperience.company || '').toLowerCase().includes('github')
+            ) {
+              identities.push({
+                platform: PlatformType.LINKEDIN,
+                value: normalizedDomain,
+                type: OrganizationIdentityType.PRIMARY_DOMAIN,
+                verified: true,
+              })
+              hasPrimaryDomainIdentity = true
+            }
+          }
 
-          // sometimes companyUrl is a github link, we don't want to add it as a primary domain
           if (
-            normalizedDomain &&
-            !workExperience.companyUrl.toLowerCase().includes('github') &&
-            !workExperience.company.toLowerCase().includes('github')
+            workExperience.companyLinkedInUrl &&
+            this.getLinkedInProfileHandle(workExperience.companyLinkedInUrl)
           ) {
             identities.push({
               platform: PlatformType.LINKEDIN,
-              value: normalizedDomain,
-              type: OrganizationIdentityType.PRIMARY_DOMAIN,
-              verified: true,
+              value: this.getLinkedInProfileHandle(workExperience.companyLinkedInUrl),
+              type: OrganizationIdentityType.USERNAME,
+              verified: !hasPrimaryDomainIdentity,
             })
           }
-        }
 
-        if (workExperience.companyLinkedInUrl) {
-          identities.push({
-            platform: PlatformType.LINKEDIN,
-            value: `company:${workExperience.companyLinkedInUrl.split('/').pop()}`,
-            type: OrganizationIdentityType.USERNAME,
-            verified: true,
+          normalized.memberOrganizations.push({
+            name: replaceDoubleQuotes(workExperience.company),
+            source: OrganizationSource.ENRICHMENT_PROGAI,
+            identities,
+            title: replaceDoubleQuotes(workExperience.title),
+            startDate: workExperience.startDate
+              ? workExperience.startDate.replace('Z', '+00:00')
+              : null,
+            endDate: workExperience.endDate ? workExperience.endDate.replace('Z', '+00:00') : null,
           })
         }
-
-        normalized.memberOrganizations.push({
-          name: workExperience.company,
-          source: OrganizationSource.ENRICHMENT_PROGAI,
-          identities,
-          title: workExperience.title,
-          startDate: workExperience.startDate
-            ? workExperience.startDate.replace('Z', '+00:00')
-            : null,
-          endDate: workExperience.endDate ? workExperience.endDate.replace('Z', '+00:00') : null,
-        })
       }
     }
 
     return normalized
   }
 
+  private getLinkedInProfileHandle(url: string): string | null {
+    let regex = /company\/([^/]+)/
+    let match = url.match(regex)
+
+    if (match) {
+      return `company:${match[1]}`
+    }
+
+    regex = /school\/([^/]+)/
+    match = url.match(regex)
+
+    if (match) {
+      return `school:${match[1]}`
+    }
+
+    return null
+  }
+
   async getDataUsingGitHubHandle(githubUsername: string): Promise<IMemberEnrichmentDataProgAI> {
     const url = `${process.env['CROWD_ENRICHMENT_PROGAI_URL']}/get_profile`
     const config = {

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

@@ -128,13 +128,19 @@ export default class EnrichmentServiceSerpApi extends LoggerBase implements IEnr
           platform: PlatformType.LINKEDIN,
           type: MemberIdentityType.USERNAME,
           verified: false,
-          value: this.normalizeLinkedUrl(data.linkedinUrl).split('/').pop(),
+          value: this.getLinkedInProfileHandle(this.normalizeLinkedUrl(data.linkedinUrl)),
         },
       ],
     }
     return normalized
   }
 
+  private getLinkedInProfileHandle(url: string): string | null {
+    const regex = /in\/([^/]+)/
+    const match = url.match(regex)
+    return match ? match[1] : null
+  }
+
   private normalizeLinkedUrl(url: string): string {
     try {
       const parsedUrl = new URL(url)

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

@@ -8,6 +8,7 @@ import {
 import { IEnrichableMember, MemberEnrichmentSource } from '@crowd/types'
 
 import * as activities from '../activities'
+import { hasRemainingCredits } from '../activities/enrichment'
 import { IEnrichmentSourceInput } from '../types'
 import { sourceHasDifferentDataComparedToCache } from '../utils/common'
 
@@ -46,6 +47,12 @@ export async function enrichMember(
     if (await isCacheObsolete(source, cache)) {
       const enrichmentInput: IEnrichmentSourceInput = await getEnrichmentInput(input)
 
+      if (!(await hasRemainingCredits(source))) {
+        // no credits remaining, only update cache.updatedAt and keep the old data
+        await touchMemberEnrichmentCacheUpdatedAt(source, input.id)
+        continue
+      }
+
       const data = await getEnrichmentData(source, enrichmentInput)
 
       if (!cache) {