406 lines
12 KiB
Python
406 lines
12 KiB
Python
# import os
|
|
# import fitz # PyMuPDF
|
|
# import spacy
|
|
# import requests
|
|
from recruitment import models
|
|
from django.conf import settings
|
|
|
|
# nlp = spacy.load("en_core_web_sm")
|
|
|
|
# def extract_text_from_pdf(pdf_path):
|
|
# text = ""
|
|
# with fitz.open(pdf_path) as doc:
|
|
# for page in doc:
|
|
# text += page.get_text()
|
|
# return text
|
|
|
|
# def extract_summary_from_pdf(pdf_path):
|
|
# if not os.path.exists(pdf_path):
|
|
# return {'error': 'File not found'}
|
|
|
|
# text = extract_text_from_pdf(pdf_path)
|
|
# doc = nlp(text)
|
|
# summary = {
|
|
# 'name': doc.ents[0].text if doc.ents else '',
|
|
# 'skills': [chunk.text for chunk in doc.noun_chunks if len(chunk.text.split()) > 1],
|
|
# 'summary': text[:500]
|
|
# }
|
|
# return summary
|
|
|
|
import requests
|
|
from PyPDF2 import PdfReader
|
|
import os
|
|
import json
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
OPENROUTER_API_KEY ='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'
|
|
OPENROUTER_MODEL = 'qwen/qwen-2.5-72b-instruct:free'
|
|
|
|
if not OPENROUTER_API_KEY:
|
|
logger.warning("OPENROUTER_API_KEY not set. Resume scoring will be skipped.")
|
|
|
|
def extract_text_from_pdf(file_path):
|
|
print("text extraction")
|
|
text = ""
|
|
try:
|
|
with open(file_path, "rb") as f:
|
|
reader = PdfReader(f)
|
|
for page in reader.pages:
|
|
text += (page.extract_text() or "")
|
|
except Exception as e:
|
|
logger.error(f"PDF extraction failed: {e}")
|
|
raise
|
|
return text.strip()
|
|
|
|
def score_resume_with_openrouter(resume_text):
|
|
prompt = f"""
|
|
You are an expert technical recruiter. Your task is to score the following candidate for the role of a Senior Data Analyst based on the provided job criteria.
|
|
|
|
**Job Criteria:**
|
|
- Must-Have Skills: Python, SQL, 5+ years of experience.
|
|
- Nice-to-Have Skills: Tableau, AWS.
|
|
- Experience: Must have led at least one project.
|
|
|
|
**Candidate's Extracted Resume Text:**
|
|
\"\"\"
|
|
{resume_text}
|
|
\"\"\"
|
|
|
|
**Your Task:**
|
|
Provide a response in strict JSON format with the following keys:
|
|
1. 'match_score': A score from 0 to 100 representing how well the candidate fits the role.
|
|
2. 'strengths': A brief summary of why the candidate is a strong fit, referencing specific criteria.
|
|
3. 'weaknesses': A brief summary of where the candidate falls short or what criteria are missing.
|
|
4. 'criteria_checklist': An object where you rate the candidate's match for each specific criterion (e.g., {{'Python': 'Met', 'AWS': 'Not Mentioned'}}).
|
|
|
|
Only output valid JSON. Do not include any other text.
|
|
"""
|
|
print("model call")
|
|
response = requests.post(
|
|
url="https://openrouter.ai/api/v1/chat/completions",
|
|
headers={
|
|
"Authorization": f"Bearer {OPENROUTER_API_KEY}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
data=json.dumps({
|
|
"model": OPENROUTER_MODEL,
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
},
|
|
)
|
|
)
|
|
# print(response.status_code)
|
|
# print(response.json())
|
|
res = {}
|
|
print(response.status_code)
|
|
if response.status_code == 200:
|
|
res = response.json()
|
|
content = res["choices"][0]['message']['content']
|
|
try:
|
|
print(content)
|
|
content = content.replace("```json","").replace("```","")
|
|
print(content)
|
|
res = json.loads(content)
|
|
print(res)
|
|
except Exception as e:
|
|
print(e)
|
|
|
|
# res = raw_output["choices"][0]["message"]["content"]
|
|
else:
|
|
print("error response")
|
|
return res
|
|
# print(f"rawraw_output)
|
|
# print(response)
|
|
|
|
|
|
|
|
# def match_resume_with_job_description(resume, job_description,prompt=""):
|
|
# resume_doc = nlp(resume)
|
|
# job_doc = nlp(job_description)
|
|
# similarity = resume_doc.similarity(job_doc)
|
|
# return similarity
|
|
|
|
def dashboard_callback(request, context):
|
|
total_jobs = models.Job.objects.count()
|
|
total_candidates = models.Candidate.objects.count()
|
|
jobs = models.Job.objects.all()
|
|
job_titles = [job.title for job in jobs]
|
|
job_app_counts = [job.candidates.count() for job in jobs]
|
|
|
|
context.update({
|
|
"total_jobs": total_jobs,
|
|
"total_candidates": total_candidates,
|
|
"job_titles": job_titles,
|
|
"job_app_counts": job_app_counts,
|
|
})
|
|
return context
|
|
|
|
|
|
|
|
|
|
def get_access_token():
|
|
"""Obtain an access token using server-to-server OAuth."""
|
|
client_id = settings.ZOOM_CLIENT_ID
|
|
client_secret = settings.ZOOM_CLIENT_SECRET
|
|
|
|
auth_url = "https://zoom.us/oauth/token"
|
|
headers = {
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
}
|
|
data = {
|
|
"grant_type": "account_credentials",
|
|
"account_id": settings.ZOOM_ACCOUNT_ID,
|
|
}
|
|
|
|
auth = (client_id, client_secret)
|
|
|
|
response = requests.post(auth_url, headers=headers, data=data, auth=auth)
|
|
|
|
if response.status_code == 200:
|
|
return response.json().get("access_token")
|
|
else:
|
|
raise Exception(f"Failed to obtain access token: {response.json()}")
|
|
|
|
def create_zoom_meeting(topic, start_time, duration):
|
|
"""
|
|
Create a Zoom meeting using the Zoom API.
|
|
|
|
Args:
|
|
topic (str): The topic of the meeting.
|
|
start_time (str): The start time of the meeting in ISO 8601 format (e.g., "2023-10-01T10:00:00Z").
|
|
duration (int): The duration of the meeting in minutes.
|
|
|
|
Returns:
|
|
dict: A dictionary containing the meeting details if successful, or an error message if failed.
|
|
"""
|
|
try:
|
|
access_token = get_access_token()
|
|
|
|
meeting_details = {
|
|
"topic": topic,
|
|
"type": 2,
|
|
"start_time": start_time,
|
|
"duration": duration,
|
|
"timezone": "UTC",
|
|
"settings": {
|
|
"host_video": True,
|
|
"participant_video": True,
|
|
"join_before_host": True,
|
|
"mute_upon_entry": False,
|
|
"approval_type": 2,
|
|
"audio": "both",
|
|
"auto_recording": "none"
|
|
}
|
|
}
|
|
|
|
# Make API request to Zoom to create the meeting
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
response = requests.post(
|
|
"https://api.zoom.us/v2/users/me/meetings",
|
|
headers=headers,
|
|
json=meeting_details
|
|
)
|
|
|
|
# Check response status
|
|
if response.status_code == 201:
|
|
meeting_data = response.json()
|
|
return {
|
|
"status": "success",
|
|
"message": "Meeting created successfully.",
|
|
"meeting_details": {
|
|
"join_url": meeting_data['join_url'],
|
|
"meeting_id": meeting_data['id'],
|
|
"password": meeting_data['password'],
|
|
"host_email": meeting_data['host_email']
|
|
},
|
|
"zoom_gateway_response": meeting_data
|
|
}
|
|
else:
|
|
return {
|
|
"status": "error",
|
|
"message": "Failed to create meeting.",
|
|
"details": response.json()
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"status": "error",
|
|
"message": str(e)
|
|
}
|
|
|
|
|
|
def list_zoom_meetings(next_page_token=None):
|
|
"""
|
|
List all meetings for a user using the Zoom API.
|
|
|
|
Args:
|
|
next_page_token (str, optional): The token for paginated results. Defaults to None.
|
|
|
|
Returns:
|
|
dict: A dictionary containing the list of meetings or an error message.
|
|
"""
|
|
try:
|
|
access_token = get_access_token()
|
|
user_id = 'me'
|
|
|
|
params = {}
|
|
if next_page_token:
|
|
params['next_page_token'] = next_page_token
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
response = requests.get(
|
|
f"https://api.zoom.us/v2/users/{user_id}/meetings",
|
|
headers=headers,
|
|
params=params
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
meetings_data = response.json()
|
|
return {
|
|
"status": "success",
|
|
"message": "Meetings retrieved successfully.",
|
|
"meetings": meetings_data.get("meetings", []),
|
|
"next_page_token": meetings_data.get("next_page_token")
|
|
}
|
|
else:
|
|
return {
|
|
"status": "error",
|
|
"message": "Failed to retrieve meetings.",
|
|
"details": response.json()
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"status": "error",
|
|
"message": str(e)
|
|
}
|
|
|
|
|
|
def get_zoom_meeting_details(meeting_id):
|
|
"""
|
|
Retrieve details of a specific meeting using the Zoom API.
|
|
|
|
Args:
|
|
meeting_id (str): The ID of the meeting to retrieve.
|
|
|
|
Returns:
|
|
dict: A dictionary containing the meeting details or an error message.
|
|
"""
|
|
try:
|
|
access_token = get_access_token()
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
response = requests.get(
|
|
f"https://api.zoom.us/v2/meetings/{meeting_id}",
|
|
headers=headers
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
meeting_data = response.json()
|
|
return {
|
|
"status": "success",
|
|
"message": "Meeting details retrieved successfully.",
|
|
"meeting_details": meeting_data
|
|
}
|
|
else:
|
|
return {
|
|
"status": "error",
|
|
"message": "Failed to retrieve meeting details.",
|
|
"details": response.json()
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"status": "error",
|
|
"message": str(e)
|
|
}
|
|
|
|
|
|
def update_zoom_meeting(meeting_id, updated_data):
|
|
"""
|
|
Update a Zoom meeting using the Zoom API.
|
|
|
|
Args:
|
|
meeting_id (str): The ID of the meeting to update.
|
|
updated_data (dict): A dictionary containing the fields to update (e.g., topic, start_time, duration).
|
|
|
|
Returns:
|
|
dict: A dictionary containing the updated meeting details or an error message.
|
|
"""
|
|
try:
|
|
access_token = get_access_token()
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
response = requests.patch(
|
|
f"https://api.zoom.us/v2/meetings/{meeting_id}",
|
|
headers=headers,
|
|
json=updated_data
|
|
)
|
|
|
|
if response.status_code == 204:
|
|
return {
|
|
"status": "success",
|
|
"message": "Meeting updated successfully."
|
|
}
|
|
else:
|
|
print(response.json())
|
|
return {
|
|
"status": "error",
|
|
"message": "Failed to update meeting.",
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"status": "error",
|
|
"message": str(e)
|
|
}
|
|
|
|
|
|
def delete_zoom_meeting(meeting_id):
|
|
"""
|
|
Delete a Zoom meeting using the Zoom API.
|
|
|
|
Args:
|
|
meeting_id (str): The ID of the meeting to delete.
|
|
|
|
Returns:
|
|
dict: A dictionary indicating success or failure.
|
|
"""
|
|
try:
|
|
access_token = get_access_token()
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}"
|
|
}
|
|
response = requests.delete(
|
|
f"https://api.zoom.us/v2/meetings/{meeting_id}",
|
|
headers=headers
|
|
)
|
|
|
|
if response.status_code == 204:
|
|
return {
|
|
"status": "success",
|
|
"message": "Meeting deleted successfully."
|
|
}
|
|
else:
|
|
return {
|
|
"status": "error",
|
|
"message": "Failed to delete meeting.",
|
|
"details": response.json()
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"status": "error",
|
|
"message": str(e)
|
|
} |