Partage
  • Partager sur Facebook
  • Partager sur Twitter

ERREUR: 400: 400: 'str' object is not callable

    16 février 2025 à 22:13:09

    Bonjour à tous,
    Dans mon projet je n'arrive pas à trouver le probleme sur 'str'

    Vous trouverez ci-joint le code de mon app.

    Merci d'avance de votre aide,

    import uuid
    import json
    from fastapi import HTTPException
    from IA.database import execute_query, fetch_one
    from datetime import datetime
    
    def ensure_skill_exists(skill_id, skill_name):
        """Vérifie si la compétence existe, sinon l'ajoute."""
        print(f"🔍 Vérification de l'existence de la compétence {skill_name} ({skill_id})")
        existing_skill = fetch_one("SELECT id FROM skill WHERE id = ?", (skill_id,))
        
        if not existing_skill:
            print(f"🆕 Ajout de la compétence {skill_name} ({skill_id})")
            execute_query("INSERT INTO skill (id, name) VALUES (?, ?)", (skill_id, skill_name))
        else:
            print(f"✅ La compétence {skill_name} existe déjà.")
    
    def patch_person(cv_id: str, person_update: dict):
        """Met à jour les informations personnelles d'un candidat."""
        
        existing_person = fetch_one("SELECT profileTitle FROM person WHERE id = ?", (cv_id,))
        if not existing_person:
            raise HTTPException(status_code=404, detail=f"CV avec l'ID {cv_id} introuvable.")
    
        # Mise à jour des champs valides
        update_fields = ["lastName", "firstName", "email", "phone", "profileTitle"]
        update_values = {field: person_update[field] for field in update_fields if field in person_update}
    
        if not update_values:
            return {"message": "Aucune mise à jour nécessaire."}
    
        # Générer une requête SQL dynamique
        set_clause = ", ".join([f"{key} = ?" for key in update_values.keys()])
        values = list(update_values.values()) + [datetime.now(), cv_id]
    
        execute_query(f"UPDATE person SET {set_clause}, updatedAt = ? WHERE id = ?", values)
    
        return {"message": f"Informations personnelles mises à jour pour {cv_id}."}
    
    def patch_location(cv_id: str, location_update: dict):
        """Met à jour la localisation d'un candidat."""
        
        existing_cv = fetch_one("SELECT locationId FROM person WHERE id = ?", (cv_id,))
        if not existing_cv or existing_cv.get("locationId") is None:
            raise HTTPException(status_code=404, detail=f"CV avec l'ID {cv_id} introuvable ou sans localisation.")
    
        location_id = existing_cv["locationId"]
        update_values = {key: location_update[key] for key in ["postalCodeId", "cityId"] if key in location_update}
    
        if update_values:
            set_clause = ", ".join([f"{key} = ?" for key in update_values.keys()])
            values = list(update_values.values()) + [location_id]
            execute_query(f"UPDATE location SET {set_clause} WHERE id = ?", values)
    
        return {"message": f"Localisation mise à jour pour le CV {cv_id}."}
    
    def patch_skills(cv_id: str, skills_update: list):
        """Met à jour les compétences d'un candidat."""
    
        print(f"📌 Debug - Données reçues pour mise à jour des skills : {skills_update}")
    
        if not isinstance(skills_update, list) or not all(isinstance(skill, dict) for skill in skills_update):
            print(f"❌ Erreur de type : {type(skills_update)} reçu")
            raise HTTPException(status_code=400, detail="Le format des compétences est incorrect.")
    
        # Supprime les anciennes compétences associées
        execute_query("DELETE FROM candidate_skill WHERE candidate_id = ?", (cv_id,))
    
        for skill in skills_update:
            try:
                print(f"🔍 Vérification de la compétence : {skill}")
                
                # Vérifie les clés requises
                required_keys = {"candidate_skill_id", "skill_id", "name", "level"}
                if not required_keys.issubset(skill.keys()):
                    raise ValueError(f"Clés manquantes : {skill}")
    
                # Vérifie si `skill_id` est bien une string et non une fonction
                if not isinstance(skill["skill_id"], str):
                    raise TypeError(f"🔴 skill_id n'est pas une string : {skill['skill_id']}")
    
                ensure_skill_exists(skill["skill_id"], skill["name"])
    
                # Ajout des compétences dans la base SQLite
                execute_query(
                    "INSERT INTO candidate_skill (id, candidate_id, skill_id, level) VALUES (?, ?, ?, ?)",
                    (skill["candidate_skill_id"], cv_id, skill["skill_id"], skill["level"])
                )
    
            except Exception as e:
                print(f"❌ ERREUR DÉTECTÉE DANS patch_skills : {e}")
                print(f"🔍 Type d'erreur : {type(e).__name__}")
                raise HTTPException(status_code=400, detail=f"Erreur dans patch_skills : {str(e)}")
    
        print(f"✅ Mise à jour réussie des compétences pour {cv_id}")
        return {"message": f"Compétences mises à jour pour {cv_id}."}
    
    
    
    def patch_cv(cv_data):
        """Ajoute un CV dans la base et met à jour ses compétences."""
        try:
            cv_id = cv_data["id"]
    
            existing_cv = fetch_one("SELECT id FROM person WHERE id = ?", (cv_id,))
            if existing_cv:
                raise HTTPException(status_code=400, detail=f"Le CV avec l'ID '{cv_id}' existe déjà.")
    
            execute_query(
                '''INSERT INTO person (id, lastName, firstName, email, phone, profileTitle, locationId, statusId, role, type, createdAt, updatedAt)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'))''',
                (
                    cv_id,
                    cv_data["lastName"],
                    cv_data["firstName"],
                    cv_data["email"],
                    cv_data["phone"],
                    cv_data["profileTitle"],
                    cv_data["location"]["id"],
                    cv_data["status"]["id"],
                    cv_data["role"],
                    cv_data["type"]
                )
            )
    
            patch_skills(cv_id, cv_data["skills"])
    
            return {"message": f"CV ajouté et compétences mises à jour pour {cv_id}"}
    
        except Exception as e:
            print(f"❌ ERREUR DÉTECTÉE DANS patch_cv : {e}")
            print(f"🔍 Type d'erreur : {type(e).__name__}")
            raise HTTPException(status_code=500, detail=f"Erreur inattendue : {e}")
    

     


    from fastapi import FastAPI, HTTPException
    from pydantic import BaseModel
    import sys
    import os
    import importlib
    import IA.services.cv_service as cv_service
    import IA.services.job_service as job_service
    from typing import List, Optional
    from datetime import datetime
    from fastapi.routing import APIRoute
    import logging
    from IA.services.patch import patch_person, patch_location, patch_skills
    from typing import List
    
    
    # Configuration des logs
    sys.path.append(os.path.abspath(os.path.dirname(__file__)))
    
    app = FastAPI()
    
    # 📌 Ajout de logs pour vérifier les routes
    @app.on_event("startup")
    async def startup_event():
        logging.basicConfig(level=logging.DEBUG)
        logging.debug("🚀 API démarrée - Vérification des routes")
        for route in app.routes:
            logging.debug(f"✅ Route chargée: {route.path} [{', '.join(route.methods)}]")
    
    @app.get("/")
    def read_root():
        return {"message": "API Interim-Online is running"}
    
    @app.get("/")
    def read_root():
        return {"message": "API Interim-Online is running"}
    
    class Activity(BaseModel):
        id: str
        name: str
    
    class SubActivity(BaseModel):
        id: str
        name: str
        activity: Activity
    
    class Job(BaseModel):
        id: str
        name: str
        sub_activity: Optional[str] = None
        activity: Optional[str] = None
    
    class Experience(BaseModel):
        id: str
        companyName: str
        startDate: str
        endDate: str
        description: str
        level: Optional[str] = "Basique"
    
    class Education(BaseModel):
        id: str
        title: str
        institution: str
        startDate: str
        endDate: str
        description: Optional[str] = "Basique"
    
    class Skill(BaseModel):
        id: str
        name: str
        level: Optional[str] = "Basique"
    
    class SkillUpdate(BaseModel):
        candidate_skill_id: str
        skill_id: str
        name: str
        level: str 
    
    class SkillsPayload(BaseModel):
        skills: List[SkillUpdate] 
    
    class Language(BaseModel):
        id: str
        name: str
        level: Optional[str] = "Basique"
    
    class Location(BaseModel):
        id: str
        cityId: str
        departmentId: str
        regionId: str
        countryId: str
        city: str
        department: str
        region: str
        country: str
        postalCode: str
    
    class Status(BaseModel):
        id: str
        name: str
    
    class CVRequest(BaseModel):
        id: str
        lastName: str
        firstName: str
        email: str
        phone: str
        profileTitle: str
        location: Location
        status: Status
        role: str = "candidate"
        type: str = "candidate"
        jobs: List[Job]
        skills: List[Skill]
        experience: List[Experience]
        education: List[Education]
        languages: List[Language]
    
    
    
    @app.post("/cv")
    def create_cv(cv: CVRequest):
        try:
            print(f"📌 Debug: Données reçues pour le CV: {cv.dict()}")
    
            # 🔹 Vérifier et récupérer `statusId`
            if isinstance(cv.status, dict):
                status_id = cv.status.get("id") or cv_service.ensure_status_exists(cv.status.get("name"))
            else:
                status_id = cv.status.id if hasattr(cv.status, "id") else cv_service.ensure_status_exists(cv.status.name)
    
            print(f"✅ StatusId déterminé: {status_id}")
    
            # 🔹 Vérifier et insérer la localisation
            location_id = cv_service.ensure_location_exists(cv.location.dict())
            print(f"✅ LocationId déterminé: {location_id}")
    
            # 🔹 Convertir les objets `Job` en dictionnaires avant validation
            jobs_list = [job.dict() if hasattr(job, "dict") else job for job in cv.jobs]
            print(f"✅ Jobs reçus: {jobs_list}")
    
            # 🔹 Vérifier et insérer les métiers
            job_ids = cv_service.ensure_jobs_exist(jobs_list)
    
            # 🔹 Ajouter le CV
            cv_service.add_cv(
                cv.id, cv.lastName, cv.firstName, cv.email, cv.phone, cv.profileTitle,
                location_id, status_id, cv.role, cv.type, job_ids, cv.skills,
                cv.experience, cv.education, cv.languages, updated_at=datetime.utcnow()
            )
    
            return {"message": "CV ajouté avec succès"}
    
        except Exception as e:
            print(f"❌ ERREUR: {str(e)}")
            raise HTTPException(status_code=500, detail=str(e))
    
    
    @app.get("/cvs")
    def list_cvs():
        return cv_service.get_all_cvs()
    
    @app.patch("/cv/{cv_id}/person", summary="Patch Person Data", tags=["CV"])
    def patch_cv_person(cv_id: str, person_update: dict):
        """Met à jour partiellement les informations personnelles du candidat."""
        return patch_person(cv_id, person_update)
    
    @app.patch("/cv/{cv_id}/location", summary="Patch Location Data", tags=["CV"])
    def patch_cv_location(cv_id: str, location_update: dict):
        """Met à jour partiellement la localisation du candidat."""
        return patch_location(cv_id, location_update)
    
    @app.patch("/cv/{cv_id}/skills", summary="Patch Skills Api", tags=["CV"])
    def patch_cv_skills(cv_id: str, skills_update: SkillsPayload):  
        print("✅ Données prêtes pour `patch_skills`:", skills_update.skills)  
        return patch_skills(cv_id, skills_update.skills)
    
    
    @app.put("/cv/{cv_id}/status")
    def update_status_route(cv_id: str, status_data: dict):
        """🔹 Met à jour le statut du CV"""
        return cv_service.update_status(cv_id, status_data)
    
    @app.put("/cv/{cv_id}/location")
    def update_location_route(cv_id: str, location_data: dict):
        """🔹 Met à jour la localisation du CV"""
        return cv_service.update_location(cv_id, location_data)
    
    @app.put("/cv/{cv_id}/skills")
    def update_skills_route(cv_id: str, skills_data: dict):
        """🔹 Met à jour les compétences du CV"""
        return cv_service.update_skills(cv_id, skills_data.get("skills", []))
    
    @app.put("/cv/{cv_id}/experience")
    def update_experience_route(cv_id: str, experience_data: dict):
        """🔹 Met à jour les expériences du CV"""
        return cv_service.update_experience(cv_id, experience_data.get("experience", []))
    
    @app.put("/cv/{cv_id}/education")
    def update_education_route(cv_id: str, education_data: dict):
        """🔹 Met à jour les formations du CV"""
        return cv_service.update_education(cv_id, education_data.get("education", []))
    
    @app.put("/cv/{cv_id}/languages")
    def update_languages_route(cv_id: str, languages_data: dict):
        """🔹 Met à jour les langues du CV"""
        return cv_service.update_languages(cv_id, languages_data.get("languages", []))
    
    



    import torch
    import uuid
    import json
    from transformers import AutoTokenizer, AutoModel
    from IA.database import execute_query, fetch_one, fetch_query
    from fastapi import HTTPException
    from IA.services.patch import patch_person, patch_location, patch_skills
    from IA.services import patch
    
    
    # Charger le modèle CamemBERT pour générer les embeddings
    tokenizer = AutoTokenizer.from_pretrained("camembert-base")
    model = AutoModel.from_pretrained("camembert-base")
    
    def get_embedding(text):
        """Génère un embedding à partir d'un texte en utilisant CamemBERT."""
        if not text.strip():
            return torch.zeros(768).tolist()
        inputs = tokenizer(text, return_tensors='pt', truncation=True, padding=True, max_length=512)
        with torch.no_grad():
            outputs = model(**inputs)
        return outputs.last_hidden_state[:, 0, :].squeeze().tolist()
    
    def ensure_status_exists(status_name):
        """Vérifie si le statut existe, sinon l'insère et retourne son ID."""
        status = fetch_one("SELECT id FROM status WHERE UPPER(name) = UPPER(?)", (status_name,))
        if not status:
            status_id = str(uuid.uuid4())
            execute_query("INSERT INTO status (id, name) VALUES (?, ?)", (status_id, status_name))
            return status_id
        return status["id"]
    
    def ensure_location_exists(city_name):
        """Vérifie si une ville existe et retourne son ID."""
        city = fetch_one("SELECT id FROM city WHERE UPPER(name) = UPPER(?)", (city_name,))
        if not city:
            raise HTTPException(status_code=400, detail=f"La ville '{city_name}' est inconnue.")
        return city["id"]
    
    def ensure_job_exists(job_data):
        """Vérifie et insère les métiers, sous-activités et activités, et retourne leurs IDs."""
        job_name = job_data["name"]
        sub_activity_name = job_data.get("sub_activity")
        activity_name = job_data.get("activity")
    
        # Vérifier si le métier existe déjà
        job_info = fetch_one("""
            SELECT j.id, j.name, sa.id AS sub_activity_id, sa.name AS sub_activity, 
                   a.id AS activity_id, a.name AS activity
            FROM job j
            LEFT JOIN sub_activity sa ON j.subActivityId = sa.id
            LEFT JOIN activity a ON sa.activityId = a.id
            WHERE UPPER(j.name) = UPPER(?)
        """, (job_name,))
    
        if job_info:
            return {
                "id": job_info["id"],
                "name": job_info["name"],
                "sub_activity": {
                    "id": job_info["sub_activity_id"],
                    "name": job_info["sub_activity"],
                    "activity": {
                        "id": job_info["activity_id"],
                        "name": job_info["activity"]
                    }
                }
            }
    
        # Gérer la sous-activité et l'activité
        if not sub_activity_name:
            raise HTTPException(status_code=400, detail=f"Le métier '{job_name}' est inconnu, veuillez spécifier une sous-activité.")
    
        sub_activity = fetch_one("SELECT id FROM sub_activity WHERE UPPER(name) = UPPER(?)", (sub_activity_name,))
        if not sub_activity:
            raise HTTPException(status_code=400, detail=f"La sous-activité '{sub_activity_name}' est inconnue.")
    
        job_id = str(uuid.uuid4())
        execute_query("INSERT INTO job (id, name, subActivityId) VALUES (?, ?, ?)", 
                      (job_id, job_name, sub_activity["id"]))
    
        return {
            "id": job_id,
            "name": job_name,
            "sub_activity": {
                "id": sub_activity["id"],
                "name": sub_activity_name,
                "activity": {
                    "id": sub_activity["id"],  # Correction pour assurer la relation
                    "name": activity_name
                }
            }
        }
    
    def add_job(job_id, title, description, company_id, status, city, job_data, languages, skills_required):
        """Ajoute une offre d'emploi à la base de données avec embedding AI."""
        status_id = ensure_status_exists(status)
        city_id = ensure_location_exists(city)
        job_info = ensure_job_exists(job_data)
    
        embedding = get_embedding(f"{title} {job_info['id']} {job_info['sub_activity']['id']} {job_info['sub_activity']['activity']['id']}")
        
        query = '''INSERT INTO job_offer (id, title, description, companyId, statusId, cityId, jobId, ai_embedding, createdAt)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))'''
        execute_query(query, (job_id, title, description, company_id, status_id, city_id, job_info['id'], str(embedding)))
        
        # 🔹 Gestion des compétences requises
        for skill in skills_required:
            skill_id = fetch_one("SELECT id FROM skill WHERE UPPER(name) = UPPER(?)", (skill,))
            if not skill_id:
                skill_id = str(uuid.uuid4())  # Générer un ID unique
                execute_query("INSERT INTO skill (id, name) VALUES (?, ?)", (skill_id, skill))
            else:
                skill_id = skill_id["id"]
            execute_query("INSERT INTO job_offer_skills_skill (jobOfferId, skillId) VALUES (?, ?)", (job_id, skill_id))
        
        # 🔹 Gestion des langues requises
        for lang in languages:
            lang_id = fetch_one("SELECT id FROM language WHERE UPPER(name) = UPPER(?)", (lang["name"],))
            if not lang_id:
                lang_id = str(uuid.uuid4())  # Générer un ID unique
                execute_query("INSERT INTO language (id, name) VALUES (?, ?)", (lang_id, lang["name"]))
            else:
                lang_id = lang_id["id"]
            execute_query("INSERT INTO job_offer_language (job_offer_id, language_id, level) VALUES (?, ?, ?)", (job_id, lang_id, lang["level"]))
    
    def update_job(job_id, job_update):
        """Met à jour une offre d'emploi existante avec les nouvelles valeurs fournies."""
        existing_job = fetch_one("SELECT * FROM job_offer WHERE id = ?", (job_id,))
        if not existing_job:
            raise HTTPException(status_code=404, detail=f"Aucune offre d'emploi trouvée avec l'ID {job_id}")
        
        update_fields = []
        update_values = []
        for key, value in job_update.items():
            update_fields.append(f"{key} = ?")
            update_values.append(value)
        
        if update_fields:
            update_query = f"UPDATE job_offer SET {', '.join(update_fields)} WHERE id = ?"
            update_values.append(job_id)
            execute_query(update_query, tuple(update_values))
        
        return {"message": "Offre d'emploi mise à jour avec succès"}
    
    def get_all_jobs():
        """Récupère toutes les offres d'emploi enregistrées."""
        return fetch_query("SELECT * FROM job_offer")
    
    def get_job_by_id(job_id):
        """Récupère une offre d'emploi spécifique par son ID avec ses compétences et langues requises."""
        job = fetch_one("SELECT * FROM job_offer WHERE id = ?", (job_id,))
        if not job:
            raise HTTPException(status_code=404, detail=f"Aucune offre d'emploi trouvée avec l'ID {job_id}")
        
        job["skills_required"] = fetch_query(
            "SELECT name FROM skill "
            "INNER JOIN job_offer_skills_skill ON skill.id = job_offer_skills_skill.skillId "
            "WHERE job_offer_skills_skill.jobOfferId = ?", (job_id,)
        )
        
        job["languages_required"] = fetch_query(
            "SELECT name, level FROM language "
            "INNER JOIN job_offer_language ON language.id = job_offer_language.language_id "
            "WHERE job_offer_language.job_offer_id = ?", (job_id,)
        )
        return job
    
    def patch_person(cv_id: str, person_update: dict):
        """Met à jour uniquement les champs fournis du CV."""
    
        existing_person = fetch_one("SELECT id FROM person WHERE id = ?", (cv_id,))
        if not existing_person:
            raise HTTPException(status_code=404, detail=f"Aucun CV trouvé avec l'ID {cv_id}")
    
        updates = []
        values = []
    
        for key, value in person_update.items():
            updates.append(f"{key} = ?")
            values.append(value)
    
        values.append(cv_id)
    
        query = f"UPDATE person SET {', '.join(updates)} WHERE id = ?"
        execute_query(query, values)
    
        return {"message": f"CV {cv_id} mis à jour avec succès"}
    
    def patch_location(cv_id: str, location_update: dict):
        """Met à jour la localisation d'un candidat"""
        print(f"📌 Mise à jour de la localisation pour {cv_id}")
    



    • Partager sur Facebook
    • Partager sur Twitter
      17 février 2025 à 9:50:45

      Salut, serait t'il possible d'avoir l'erreur complète ? Merci
      • Partager sur Facebook
      • Partager sur Twitter
        17 février 2025 à 12:58:52

        dans tous les cas l'erreur dit que tu essais de faire un appel sur une variable qui contient une chaine., et avec le message d'erreur complet, ça sera plus simple de savoir où se passe cette erreur
        • Partager sur Facebook
        • Partager sur Twitter
          17 février 2025 à 14:05:49

          @umfred: la variable n'a pas besoin de contenir une chaîne.
          >>> a=3                                                                                                                 
          >>> n=a(3)                                                                                                              
          Traceback (most recent call last):                                                                                      
            File "<stdin>", line 1, in <module>                                                                                   
          TypeError: 'int' object is not callable                                                                                 
          >>>
          • Partager sur Facebook
          • Partager sur Twitter

          Le Tout est souvent plus grand que la somme de ses parties.

            17 février 2025 à 18:34:43

            vu que l'erreur fait ici mention de 'str' object is not callable, (d'après le titre de son sujet) il est bien dans ce cas, où sa variable contient une chaine
            • Partager sur Facebook
            • Partager sur Twitter

            ERREUR: 400: 400: 'str' object is not callable

            × Après avoir cliqué sur "Répondre" vous serez invité à vous connecter pour que votre message soit publié.
            • Editeur
            • Markdown