fix: lbac-1777: refactor import referentiel opco Constructys (#848)

* fix: refactor import referentiel opco Constructys

* Update server/src/jobs/lba_recruteur/opco/constructys/constructysImporter.ts

Co-authored-by: Kevin Barnoin <[email protected]>

* fix: remove accents from emails

---------

Co-authored-by: Kevin Barnoin <[email protected]>

server/src/commands.ts

@@ -530,6 +530,12 @@ program
   .option("-q, --queued", "Run job asynchronously", false)
   .action(createJobAction("user-recruters:data-validation:fix"))
 
+program
+  .command("import-referentiel-opco-constructys")
+  .description("Importe les emails pour la collection ReferentielOpco depuis l'opco Constructys")
+  .option("-q, --queued", "Run job asynchronously", false)
+  .action(createJobAction("referentiel-opco:constructys:import"))
+
 export async function startCLI() {
   await program.parseAsync(process.argv)
 }

server/src/common/model/schema/referentielOpco/referentielOpco.schema.ts

@@ -1,6 +1,6 @@
+import { OPCOS } from "shared/constants/recruteur"
 import { IReferentielOpco } from "shared/models"
 
-import { OPCOS } from "../../../../services/constant.service"
 import { model, Schema } from "../../../mongodb"
 
 export const referentielOpcoSchema = new Schema<IReferentielOpco>(

server/src/common/utils/fileUtils.ts

@@ -1,7 +1,7 @@
 import path from "path"
 
 import csvToJson from "convert-csv-to-json"
-import { parse } from "csv-parse"
+import { Options as CsvParseOptions, parse } from "csv-parse"
 import { isEmpty, pickBy } from "lodash-es"
 import XLSX from "xlsx"
 
@@ -48,7 +48,7 @@ export const prepareMessageForMail = (data) => {
   return result ? result.replace(/\r\n|\r|\n/gi, "<br />") : result
 }
 
-export const parseCsv = (options = {}) => {
+export const parseCsv = (options: CsvParseOptions = {}) => {
   return parse({
     trim: true,
     delimiter: ";",

server/src/jobs/jobs.ts

@@ -33,6 +33,7 @@ import { updateAddressDetailOnRecruitersCollection } from "./lba_recruteur/formu
 import { updateMissingStartDate } from "./lba_recruteur/formulaire/misc/updateMissingStartDate"
 import { relanceFormulaire } from "./lba_recruteur/formulaire/relanceFormulaire"
 import { generateIndexes } from "./lba_recruteur/indexes/generateIndexes"
+import { importReferentielOpcoFromConstructys } from "./lba_recruteur/opco/constructys/constructysImporter"
 import { relanceOpco } from "./lba_recruteur/opco/relanceOpco"
 import { createOffreCollection } from "./lba_recruteur/seed/createOffre"
 import { fillRecruiterRaisonSociale } from "./lba_recruteur/user/misc/fillRecruiterRaisonSociale"
@@ -329,6 +330,8 @@ export async function runJob(job: IInternalJobsCronTask | IInternalJobsSimple):
         return fixRecruiterDataValidation()
       case "user-recruters:data-validation:fix":
         return fixUserRecruiterDataValidation()
+      case "referentiel-opco:constructys:import":
+        return importReferentielOpcoFromConstructys()
       ///////
       case "mongodb:indexes:create":
         return createMongoDBIndexes()

server/src/jobs/lba_recruteur/opco/constructys/constructysImporter.ts

@@ -1,76 +1,80 @@
 import { createReadStream } from "fs"
 import path from "path"
 
-import Joi from "joi"
-import { filterData, oleoduc, transformData, writeData } from "oleoduc"
+import { oleoduc, transformData, writeData } from "oleoduc"
+import { removeAccents } from "shared"
+import { OPCOS } from "shared/constants/recruteur"
+
+import { notifyToSlack } from "@/common/utils/slackUtils"
+import { prepareReferentielOpcoForInsert } from "@/services/opco.service"
 
 import __dirname from "../../../../common/dirname"
 import { logger } from "../../../../common/logger"
 import { ReferentielOpco } from "../../../../common/model/index"
 import { fileDownloader, parseCsv } from "../../../../common/utils/fileUtils"
 import config from "../../../../config"
-import { runScript } from "../../../scriptWrapper"
-
-const importer = async (filePath, remoteFileName, opco_label) => {
-  logger.info("Downloading file...")
-  await fileDownloader(filePath, remoteFileName, config.ftp.constructys)
 
+const importer = async (filePath: string, opco_label: OPCOS) => {
   logger.info(`Deleting collection entries for ${opco_label}...`)
   await ReferentielOpco.deleteMany({ opco_label })
 
   logger.info("Importing Data...")
 
-  const stat = {
+  const stats = {
     error: 0,
     total: 0,
     imported: 0,
   }
 
   await oleoduc(
     createReadStream(filePath),
-    parseCsv(),
-    filterData((e) => e.Mails),
+    parseCsv({ delimiter: ";", encoding: "latin1" /* identique à ISO-8859-1 */ }),
     transformData((e) => {
-      const emails: string[] = []
-      const { Siret, Mails } = e
-
-      const emailsArray = Mails.split(/,|;| /).filter((x) => x)
-      const emailsArrayDuplicateFree = [...new Set(emailsArray)]
-
-      for (const email of emailsArrayDuplicateFree) {
-        stat.total++
-        const { error, value } = Joi.string().email().validate(email, { abortEarly: false })
-
-        if (error) {
-          stat.error++
-          return
-        }
-
-        stat.imported++
-        emails.push(value)
+      const { Siret } = e
+      stats.total++
+      const csvEmailStr = e["Email du contact"]
+      const emailsArray = removeAccents(csvEmailStr)
+        .split(/,|;| /)
+        .filter((x) => x)
+      const referentielOpt = prepareReferentielOpcoForInsert({ opco_label, siret_code: Siret, emails: emailsArray })
+      if (referentielOpt) {
+        stats.imported++
+        return referentielOpt
+      } else {
+        logger.error("could not import", { siret: Siret, emails: csvEmailStr })
+        stats.error++
+        return
       }
-
-      return { siret_code: Siret, emails: [...new Set(emails)] }
     }),
     writeData(
-      async ({ siret_code, emails }) => {
-        await ReferentielOpco.create({ opco_label, siret_code, emails })
+      async (referentiel) => {
+        const { siret_code } = referentiel
+        await ReferentielOpco.findOneAndUpdate({ siret_code }, { $set: referentiel }, { upsert: true }).lean()
       },
       { parallel: 500 }
     )
   )
 
   logger.info("Data import done.")
-  return stat
+  await notifyToSlack({
+    subject: "import referentiel opco Constructys",
+    message: `${stats.total} documents. ${stats.error} erreurs. ${stats.imported} mises à jour`,
+    error: stats.error > 0,
+  })
+  return stats
 }
 
-runScript(async () => {
+export const importReferentielOpcoFromConstructys = async () => {
   logger.info("Constructys data import starting...")
+
+  logger.info("Downloading file...")
   const dirname = __dirname(import.meta.url)
   const filePath = path.resolve(dirname, "./constructys-data.csv")
   const remoteFileName = "CTYS_MATCHA.csv"
-  const opco_label = "Constructys"
+  await fileDownloader(filePath, remoteFileName, config.ftp.constructys)
 
-  const result = await importer(filePath, remoteFileName, opco_label)
+  logger.info("Importing file...")
+  const opco_label = OPCOS.CONSTRUCTYS
+  const result = await importer(filePath, opco_label)
   return result
-})
+}

server/src/services/constant.service.ts

@@ -34,19 +34,6 @@ export const REGEX = {
   GEO: /^(-?\d+(\.\d+)?),\s*(-?\d+(\.\d+)?)$/,
   TELEPHONE: /^[0-9]{10}$/,
 }
-export const OPCOS = {
-  AFDAS: "AFDAS",
-  AKTO: "AKTO / Opco entreprises et salariés des services à forte intensité de main d'oeuvre",
-  ATLAS: "ATLAS",
-  CONSTRUCTYS: "Constructys",
-  OPCOMMERCE: "L'Opcommerce",
-  OCAPIAT: "OCAPIAT",
-  OPCO2I: "OPCO 2i",
-  EP: "Opco entreprises de proximité",
-  MOBILITE: "Opco Mobilités",
-  SANTE: "Opco Santé",
-  UNIFORMATION: "Uniformation, l'Opco de la Cohésion sociale",
-}
 
 export const NIVEAUX_POUR_LBA = {
   INDIFFERENT: "Indifférent",

server/src/services/etablissement.service.ts

@@ -324,6 +324,7 @@ export const getGeoCoordinates = async (adresse: string): Promise<GeoCoord> => {
     throw newError
   }
 }
+
 /**
  * @description Get matching records from the ReferentielOpco collection for a given siret & email
  * @param {IReferentielOpco["siret_code"]} siretCode

server/src/services/opco.service.ts

@@ -1,10 +1,11 @@
 import memoize from "memoizee"
+import { OPCOS } from "shared/constants/recruteur"
+import { IReferentielOpco, ZReferentielOpcoInsert } from "shared/models"
 
 import { Opco } from "../common/model/index"
 import { IOpco } from "../common/model/schema/opco/opco.types"
 
 import { CFADOCK_FILTER_LIMIT, fetchOpcosFromCFADock } from "./cfadock.service"
-import { OPCOS } from "./constant.service"
 
 /**
  * @description get opco from database collection OPCOS
@@ -123,3 +124,13 @@ export const filterJobsByOpco = async ({ jobs, opco, opcoUrl }: { jobs: any[]; o
 
   return results
 }
+
+export const prepareReferentielOpcoForInsert = (referentiel: Omit<IReferentielOpco, "_id">) => {
+  if (ZReferentielOpcoInsert.safeParse(referentiel).success && referentiel.emails.length) {
+    const deduplicatedEmails = [...new Set(referentiel.emails)]
+    referentiel.emails = deduplicatedEmails
+    return referentiel
+  } else {
+    return false
+  }
+}

shared/constants/recruteur.ts

@@ -34,18 +34,18 @@ export const REGEX = {
   GEO: /^(-?\d+(\.\d+)?),\s*(-?\d+(\.\d+)?)$/,
   TELEPHONE: /^[0-9]{10}$/,
 }
-export const OPCOS = {
-  AFDAS: "AFDAS",
-  AKTO: "AKTO / Opco entreprises et salariés des services à forte intensité de main d'oeuvre",
-  ATLAS: "ATLAS",
-  CONSTRUCTYS: "Constructys",
-  OPCOMMERCE: "L'Opcommerce",
-  OCAPIAT: "OCAPIAT",
-  OPCO2I: "OPCO 2i",
-  EP: "Opco entreprises de proximité",
-  MOBILITE: "Opco Mobilités",
-  SANTE: "Opco Santé",
-  UNIFORMATION: "Uniformation, l'Opco de la Cohésion sociale",
+export enum OPCOS {
+  AFDAS = "AFDAS",
+  AKTO = "AKTO / Opco entreprises et salariés des services à forte intensité de main d'oeuvre",
+  ATLAS = "ATLAS",
+  CONSTRUCTYS = "Constructys",
+  OPCOMMERCE = "L'Opcommerce",
+  OCAPIAT = "OCAPIAT",
+  OPCO2I = "OPCO 2i",
+  EP = "Opco entreprises de proximité",
+  MOBILITE = "Opco Mobilités",
+  SANTE = "Opco Santé",
+  UNIFORMATION = "Uniformation, l'Opco de la Cohésion sociale",
 }
 
 export const NIVEAUX_POUR_LBA = {

shared/models/referentielOpco.model.ts

@@ -13,4 +13,10 @@ export const ZReferentielOpco = z
   })
   .strict()
 
+export const ZReferentielOpcoInsert = ZReferentielOpco.pick({
+  opco_label: true,
+  siret_code: true,
+  emails: true,
+})
+
 export type IReferentielOpco = z.output<typeof ZReferentielOpco>

shared/utils/index.ts

@@ -1 +1,2 @@
 export * from "./assertUnreachable"
+export * from "./stringUtils"

shared/utils/stringUtils.test.ts

@@ -0,0 +1,26 @@
+import { describe, it, expect } from "vitest"
+
+import { removeAccents } from "./stringUtils"
+
+describe("stringUtils", () => {
+  describe("removeAccents", () => {
+    it("should remove standard accents", () => {
+      expect(removeAccents("àâä")).toBe("aaa")
+      expect(removeAccents("éêëè")).toBe("eeee")
+      expect(removeAccents("ïî")).toBe("ii")
+      expect(removeAccents("ôö")).toBe("oo")
+      expect(removeAccents("üùû")).toBe("uuu")
+      expect(removeAccents("ÿŷ")).toBe("yy")
+    })
+    it("should remove ç => c", () => {
+      expect(removeAccents("ç")).toBe("c")
+    })
+    it("should remove accents from capital letters", () => {
+      expect(removeAccents("ÄÂÊËÏÎÔÖÛÜŸŶ")).toBe("AAEEIIOOUUYY")
+    })
+    it("should not change standard characters", () => {
+      const unchanged = `&"'(-_)=$*µ$£%!§:/;.,?~#{}[]|^@\``
+      expect(removeAccents(unchanged)).toBe(unchanged)
+    })
+  })
+})

shared/utils/stringUtils.ts

@@ -0,0 +1 @@
+export const removeAccents = (str: string) => str.normalize("NFD").replace(/[\u0300-\u036f]/g, "")