feat(lbac-2431): import france travail to jobs partners

server/src/common/apis/franceTravail/franceTravail.client.ts

@@ -4,6 +4,8 @@ import querystring from "querystring"
 import { internal } from "@hapi/boom"
 import { ObjectId } from "bson"
 import FormData from "form-data"
+import { sleep } from "openai/core.mjs"
+import { IFTJobRaw } from "shared"
 import { IRomeoAPIResponse } from "shared/models/cacheRomeo.model"
 import { IFranceTravailAccess, IFranceTravailAccessType } from "shared/models/franceTravailAccess.model"
 
@@ -15,6 +17,7 @@ import { logger } from "../../logger"
 import { apiRateLimiter } from "../../utils/apiUtils"
 import { getDbCollection } from "../../utils/mongodbUtils"
 import { sentryCaptureException } from "../../utils/sentryUtils"
+import { notifyToSlack } from "../../utils/slackUtils"
 import getApiClient from "../client"
 
 const axiosClient = getApiClient({}, { cache: false })
@@ -222,3 +225,72 @@ export const getRomeoPredictions = async (payload: IRomeoPayload[], options: IRo
     }
   })
 }
+// Documentation https://francetravail.io/produits-partages/catalogue/offres-emploi/documentation#/api-reference/operations/recupererListeOffre
+export const getAllFTJobsByDepartments = async (departement: string) => {
+  const jobLimit = 150
+  let start = 0
+  let total = 1
+
+  let allJobs = [] as Omit<IFTJobRaw, "_id" | "createdAt">[]
+
+  while (start < total) {
+    // Construct the range for this "page"
+    const range = `${start}-${start + jobLimit - 1}`
+
+    // Prepare your query params
+    const params: Parameters<typeof searchForFtJobs>[0] = {
+      natureContrat: "E2,FS", // E2 -> Contrat d'Apprentissage, FS -> contrat de professionalisation
+      range,
+      departement,
+      // publieeDepuis: 7, // Il vaut mieux ne pas mettre de date de publication pour avoir le plus de résultats possible
+      sort: 1, // making sure we get the most recent jobs first
+    }
+
+    try {
+      const response = await searchForFtJobs(params, { throwOnError: true })
+      await sleep(1500)
+      if (!response) {
+        throw new Error("No response from FranceTravail")
+      }
+
+      const { data: jobs, contentRange } = response
+
+      if (!jobs.resultats) {
+        //  logger.info("No resultats from FranceTravail", params)
+        break
+      }
+
+      allJobs = [...allJobs, ...(jobs.resultats as Omit<IFTJobRaw, "_id" | "createdAt">[])]
+
+      // Safely parse out the total
+      // Usually, contentRange might look like "offres 0-149/9981"
+      // We split by "/" and take the second part (9981), converting to Number
+      if (contentRange) {
+        const totalString = contentRange.split("/")[1]
+        if (totalString) {
+          total = parseInt(totalString, 10)
+        }
+      }
+
+      // Move to the next "page"
+      start += jobLimit
+    } catch (error: any) {
+      // handle 3000 limit page reach
+      if (error.response?.data?.message === "La position de début doit être inférieure ou égale à 3000.") {
+        sentryCaptureException(error)
+        await notifyToSlack({
+          subject: "Import Offres France Travail",
+          message: `Limite des 3000 offres par département dépassée! dept: ${departement} total: ${total}`,
+          error: true,
+        })
+        throw error
+      }
+      if (error.response?.data?.message === "Valeur du paramètre « region » incorrecte." || error.response?.data?.message === "Valeur du paramètre « departement » incorrecte.") {
+        // code region or departement not found
+        break
+      }
+      logger.error("Error while fetching jobs", error)
+    }
+  }
+  return allJobs
+}

server/src/jobs/jobs.ts

@@ -25,13 +25,11 @@ import updateDomainesMetiers from "./domainesMetiers/updateDomainesMetiers"
 import { updateDomainesMetiersFile } from "./domainesMetiers/updateDomainesMetiersFile"
 import { importCatalogueFormationJob } from "./formationsCatalogue/formationsCatalogue"
 import { updateParcoursupAndAffelnetInfoOnFormationCatalogue } from "./formationsCatalogue/updateParcoursupAndAffelnetInfoOnFormationCatalogue"
-import { classifyFranceTravailJobs } from "./franceTravail/classifyJobsFranceTravail"
 import { generateFranceTravailAccess } from "./franceTravail/generateFranceTravailAccess"
-import { importFranceTravailJobs } from "./franceTravail/importJobsFranceTravail"
-import { pocRomeo } from "./franceTravail/pocRomeo"
 import { createJobsCollectionForMetabase } from "./metabase/metabaseJobsCollection"
 import { createRoleManagement360 } from "./metabase/metabaseRoleManagement360"
 import { runGarbageCollector } from "./misc/runGarbageCollector"
+import { importFranceTravailRaw } from "./offrePartenaire/france-travail/importJobsFranceTravail"
 import { processJobPartners } from "./offrePartenaire/processJobPartners"
 import { exportLbaJobsToS3 } from "./partenaireExport/exportJobsToS3"
 import { exportJobsToFranceTravail } from "./partenaireExport/exportToFranceTravail"
@@ -217,7 +215,7 @@ export async function setupJobProcessor() {
             handler: () => processApplications(),
           },
           "Génération du token France Travail pour la récupération des offres": {
-            cron_string: "*/5 * * * *",
+            cron_string: "*/20 * * * *", // le token dure 25 minutes et le TTL DB est équivalent
             handler: generateFranceTravailAccess,
           },
           "Mise à jour du référentiel commune": {
@@ -238,7 +236,7 @@ export async function setupJobProcessor() {
           },
           "Import complet des offres France Travail": {
             cron_string: "0 6 * * *",
-            handler: importFranceTravailJobs,
+            handler: importFranceTravailRaw,
           },
           "Emission des intentions des recruteurs": {
             cron_string: "30 20 * * *",
@@ -249,18 +247,6 @@ export async function setupJobProcessor() {
       "remove:duplicates:recruiters": {
         handler: async () => removeDuplicateRecruiters(),
       },
-      "poc:romeo": {
-        handler: async () => pocRomeo(),
-      },
-      "francetravail:token-offre": {
-        handler: async () => generateFranceTravailAccess(),
-      },
-      "francetravail:jobs:import": {
-        handler: async () => importFranceTravailJobs(),
-      },
-      "francetravail:jobs:classify": {
-        handler: async () => classifyFranceTravailJobs(),
-      },
       "recreate:indexes": {
         handler: async (job) => {
           const { drop } = job.payload as any

server/src/jobs/offrePartenaire/france-travail/franceTravailMapper.ts

@@ -0,0 +1,49 @@
+import { ObjectId } from "bson"
+import { IFTJobRaw, parseEnum } from "shared"
+import { TRAINING_CONTRACT_TYPE } from "shared/constants"
+import dayjs from "shared/helpers/dayjs"
+import { JOBPARTNERS_LABEL } from "shared/models/jobsPartners.model"
+import { IComputedJobsPartners } from "shared/models/jobsPartnersComputed.model"
+
+import { blankComputedJobPartner } from "../fillComputedJobsPartners"
+
+export const franceTravailJobsToJobsPartners = (job: IFTJobRaw): IComputedJobsPartners => {
+  const now = new Date()
+  const contractDuration = parseInt(job.typeContratLibelle, 10)
+  const contractType = parseEnum(TRAINING_CONTRACT_TYPE, job.natureContrat)
+  return {
+    ...blankComputedJobPartner,
+    _id: new ObjectId(),
+    created_at: now,
+    updated_at: now,
+    partner_label: JOBPARTNERS_LABEL.FRANCE_TRAVAIL,
+    partner_job_id: job.id,
+    contract_start: null,
+    contract_duration: isNaN(contractDuration) ? null : contractDuration,
+    contract_type: contractType ? [contractType] : ["Apprentissage"],
+    offer_title: job.intitule,
+    offer_rome_codes: [job.romeCode],
+    offer_description: job.description,
+    offer_access_conditions: job.formations ? job.formations?.map((formation) => `${formation.domaineLibelle} - ${formation.niveauLibelle}`) : [],
+    offer_to_be_acquired_skills: job.competences ? job.competences.map((competence) => competence.libelle) : [],
+    offer_creation: new Date(job.dateCreation),
+    offer_expiration: dayjs.tz(job.dateCreation).add(2, "months").toDate(),
+    offer_opening_count: job.nombrePostes,
+    offer_multicast: true,
+    workplace_name: job.entreprise.nom,
+    workplace_description: job.entreprise.description,
+    workplace_address_label: job.lieuTravail.libelle,
+    workplace_geopoint: job.lieuTravail.longitude
+      ? {
+          type: "Point",
+          coordinates: [job.lieuTravail.longitude!, job.lieuTravail.latitude!],
+        }
+      : null,
+    workplace_siret: job.entreprise.siret,
+    workplace_naf_code: job.codeNAF,
+    workplace_naf_label: job.secteurActiviteLibelle,
+    workplace_website: job.entreprise.url,
+
+    apply_url: job.origineOffre.partenaires?.[0]?.url || job.origineOffre.urlOrigine,
+  }
+}

server/src/jobs/offrePartenaire/france-travail/importJobsFranceTravail.ts

@@ -0,0 +1,52 @@
+import { ApiClient } from "api-alternance-sdk"
+import { ObjectId } from "bson"
+import { ZFTJobRaw, type IFTJobRaw } from "shared/models"
+import { JOBPARTNERS_LABEL } from "shared/models/jobsPartners.model"
+import rawFranceTravailModel from "shared/models/rawFranceTravail.model"
+
+import { getAllFTJobsByDepartments } from "@/common/apis/franceTravail/franceTravail.client"
+import { getDbCollection } from "@/common/utils/mongodbUtils"
+import config from "@/config"
+
+import { rawToComputedJobsPartners } from "../rawToComputedJobsPartners"
+
+import { franceTravailJobsToJobsPartners } from "./franceTravailMapper"
+
+export const importFranceTravailRaw = async () => {
+  const apiClient = new ApiClient({
+    key: config.apiApprentissage.apiKey,
+  })
+  const departements = await apiClient.geographie.listDepartements()
+  const codesDepartements = departements.map((d) => d.codeInsee)
+
+  let allJobs = [] as Omit<IFTJobRaw, "_id" | "createdAt">[]
+  for (const codeDepartement of codesDepartements) {
+    const jobs = await getAllFTJobsByDepartments(codeDepartement)
+    allJobs = [...allJobs, ...jobs]
+  }
+
+  await getDbCollection("raw_francetravail").updateMany(
+    { id: { $nin: allJobs.map(({ id }) => id) } as any, unpublishedAt: { $exists: false } },
+    { $set: { unpublishedAt: new Date(), updatedAt: new Date() } }
+  )
+
+  for (const rawFtJob of allJobs) {
+    await getDbCollection("raw_francetravail").findOneAndUpdate(
+      { id: rawFtJob.id as string },
+      {
+        $set: { ...rawFtJob, updatedAt: new Date() },
+        $setOnInsert: { _id: new ObjectId(), createdAt: new Date() },
+      },
+      { upsert: true }
+    )
+  }
+}
+
+export const importFranceTravailToComputed = async () => {
+  await rawToComputedJobsPartners({
+    collectionSource: rawFranceTravailModel.collectionName,
+    partnerLabel: JOBPARTNERS_LABEL.FRANCE_TRAVAIL,
+    zodInput: ZFTJobRaw,
+    mapper: franceTravailJobsToJobsPartners,
+  })
+}