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}")
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
@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 >>>
Le Tout est souvent plus grand que la somme de ses parties.
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
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é.
Le Tout est souvent plus grand que la somme de ses parties.