import asyncio
from urllib.parse import urljoin
import csv
import os
import time
import json
from pathlib import Path
import configparser
from datetime import datetime
from urllib.parse import urlsplit
import re
from bs4 import BeautifulSoup
import requests
from telegram import Bot
from openai import OpenAI
#========================
# SETTINGS
CONFIG_DIRECTORY_NAME = "config"
DATA_DIRECTORY_NAME = "data"
CONFIG_NAME = "config.ini"
PROJECT_DIRECTORY = Path(__file__).parent.parent
CONFIG_DIRECTORY = os.path.join(PROJECT_DIRECTORY, CONFIG_DIRECTORY_NAME)
DATA_DIRECTORY = os.path.join(PROJECT_DIRECTORY, DATA_DIRECTORY_NAME)
CONFIG_PATH = os.path.join(CONFIG_DIRECTORY, CONFIG_NAME)
#========================
# READ CONFIG FILE
config = configparser.ConfigParser()
config_path = Path(PROJECT_DIRECTORY) / "config" / CONFIG_NAME
config.read(config_path)
config_default = config["DEFAULT"]
#========================
# BUILD PATHS
JOB_LISTINGS_PATH = os.path.join(DATA_DIRECTORY, config_default["ListingsFile"])
MY_CV_PATH = os.path.join(DATA_DIRECTORY, config_default["CVFile"])
PROMPT_PATH = os.path.join(DATA_DIRECTORY, config_default["PromptFile"])
#========================
def build_headers(url: str):
# Split the URL
split_url = urlsplit(url)
# Reconstruct the base URL and the rest of the URL
base_url = f"{split_url.scheme}://{split_url.netloc}/"
rest_of_url = f"{split_url.path}?{split_url.query}"
headers = {
"authority": base_url,
"method": "GET",
"path": rest_of_url,
"scheme": "https",
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
"accept-encoding": "gzip, zstd",
"accept-language": "sk,en-US;q=0.9,en;q=0.8",
"cache-control": "max-age=0",
"cookie": "RecommendationUserId=s-055qnsudi8m9; browser=detected; PHPSESSID=2rro055qnsudi8m98vf4028caa; uid_tracker=%7BFADD842A-4834-D955-4DAD-CE19CC4BA5B1%7D; lmc_ccm=%7B%22level%22%3A%5B%22necessary%22%5D%2C%22revision%22%3A0%2C%22data%22%3A%7B%22serviceName%22%3A%22profesia.sk%22%2C%22uid%22%3A%22VMvErwXnpZmS0d7GIE2EF%22%7D%2C%22rfc_cookie%22%3Atrue%7D; g_state={\"i_p\":1723806189587,\"i_l\":3}; LastSearchItems=%7B%226de10d0f1051e6167066f2c5cb95834a%22%3A%7B%22criteria%22%3A%7B%22count_days%22%3A%221%22%2C%22region_id%22%3A%221%22%2C%22salary%22%3A%223500%22%2C%22salary_period%22%3A%22m%22%2C%22category_id%22%3A%225%22%2C%22jobtype_id%22%3A%221%22%2C%22remote_work%22%3A%222%22%7D%2C%22lang%22%3A%22sk%22%2C%22created%22%3A%22Friday%2C%2009-Aug-2024%2013%3A23%3A27%20CEST%22%7D%2C%2250253be697a1454c2e18a7719b54cf1c%22%3A%7B%22criteria%22%3A%7B%22region_id%22%3A%221%22%2C%22salary%22%3A%223500%22%2C%22salary_period%22%3A%22m%22%2C%22category_id%22%3A%225%22%2C%22jobtype_id%22%3A%221%22%2C%22remote_work%22%3A%222%22%7D%2C%22lang%22%3A%22sk%22%2C%22created%22%3A%22Friday%2C%2009-Aug-2024%2007%3A38%3A23%20CEST%22%7D%2C%227d6155338fb3d2cdc7e04dcdee325fa8%22%3A%7B%22criteria%22%3A%7B%22region_id%22%3A%221%22%2C%22salary%22%3A%223500%22%2C%22salary_period%22%3A%22m%22%2C%22jobtype_id%22%3A%221%22%2C%22remote_work%22%3A%222%22%7D%2C%22lang%22%3A%22sk%22%2C%22created%22%3A%22Friday%2C%2009-Aug-2024%2007%3A38%3A14%20CEST%22%7D%7D; LastSearch=%7B%22params%22%3A%7B%22region_id%22%3A%221%22%2C%22count_days%22%3A%221%22%2C%22jobtype_id%22%3A%221%22%2C%22category_id%22%3A%225%22%2C%22salary_period%22%3A%22m%22%2C%22salary%22%3A%223500%22%2C%22remote_work%22%3A%222%22%7D%2C%22lang%22%3A%22sk%22%2C%22offer_criteria_title%22%3A%22%22%7D",
"priority": "u=0, i",
"sec-ch-ua": "\"Not)A;Brand\";v=\"99\", \"Microsoft Edge\";v=\"127\", \"Chromium\";v=\"127\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "\"Windows\"",
"sec-fetch-dest": "document",
"sec-fetch-mode": "navigate",
"sec-fetch-site": "none",
"sec-fetch-user": "?1",
"upgrade-insecure-requests": "1",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36 Edg/127.0.0.0"
}
return headers
def read_existing_jobs(file_path):
existing_jobs = set()
if os.path.exists(file_path):
with open(file_path, mode='r', newline='', encoding='utf-8') as file:
reader = csv.DictReader(file, delimiter=';')
for row in reader:
existing_jobs.add((row['Title'], row['Employer'], row['Salary']))
return existing_jobs
def write_new_jobs(file_path, jobs):
with open(file_path, mode='a', newline='\n', encoding='utf-8') as file:
writer = csv.DictWriter(file, fieldnames=['Title', 'Employer', 'Salary', 'Link', 'Description', 'DateScraped', 'AIExpectedSalaryRange', 'AICompatibility', 'AISummary', 'AISkills', 'AICustomers', 'AIPromptTokens', 'AICompletionTokens', 'AITotalTokens'], delimiter=';')
if os.stat(file_path).st_size == 0:
writer.writeheader()
writer.writerows(jobs)
def scrape_job_description(job_url):
"""Scrape the job description from the job listing page."""
headers = build_headers(job_url)
try:
# Send a GET request to the job listing page
response = requests.get(job_url, headers=headers)
# Check if the request was successful
if response.status_code == 200:
# Parse the HTML content using BeautifulSoup
soup = BeautifulSoup(response.content, 'html.parser')
# Extract the job description text
# description_div = soup.find('div', class_='maintextearea') # Adjust this selector based on the actual HTML structure
# description_div = soup.find('main', id='detail', class_='col-sm-8')
description_div = soup.find('div', class_='card card-content')
if description_div:
job_description = description_div.get_text(separator="\n").strip()
return job_description
else:
return "Description not found."
else:
print(f"Failed to retrieve the job description. Status code: {response.status_code}")
return "Failed to retrieve the job description."
except requests.RequestException as e:
print(f"Request exception occurred: {e}")
return "Failed to retrieve the job description."
def scrape_jobs(url):
jobs = []
while url:
# Send a GET request with the specified User-Agent
response = requests.get(url, headers=headers)
# Check if the request was successful
if response.status_code == 200:
# Parse the HTML content using BeautifulSoup
soup = BeautifulSoup(response.content, 'html.parser')
# Find all job postings
# job_postings = soup.find_all('li', class_='list-row')
# job_postings = [element for element in soup.find_all('li', class_='list-row') if 'native-agent' not in element.get('class', [])]
job_postings = soup.find_all(lambda tag: tag.name == 'li' and tag.get('class') == ['list-row'])
# Extract job details for each posting
for job in job_postings:
try:
title = job.find('span', class_='title').text.strip()
employer = job.find('span', class_='employer').text.strip()
salary = job.find('span', class_='label label-bordered green half-margin-on-top').text.strip()
link = job.find('h2').find('a')['href']
link = urljoin(config_default["ProfesiaUrl"], link)
job_details = {
'Title': title,
'Employer': employer,
'Salary': salary,
'Link': link,
'Description': None,
'DateScraped': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
'AIExpectedSalaryRange': None,
'AICompatibility': None,
'AISummary': None,
'AISkills': None,
'AICustomers': None,
'AIPromptTokens': None,
'AICompletionTokens': None,
'AITotalTokens': None
}
jobs.append(job_details)
except AttributeError as e:
print(f"An error occurred while scraping job details: {e}")
# Skip if any of the expected fields are not found in the job posting
continue
except Exception as e:
print(f"An error occurred while scraping job details: {e}")
# Check for the "next" button and get the next page URL
next_button = soup.find('a', class_='next')
if next_button and 'href' in next_button.attrs:
next_url = next_button['href']
url = urljoin(config_default["ProfesiaUrl"], next_url)
else:
url = None
else:
print(f"Failed to retrieve the webpage. Status code: {response.status_code}")
break
time.sleep(3)
return jobs
def filter_new_jobs(scraped_jobs, existing_jobs):
new_jobs = [job for job in scraped_jobs if (job['Title'], job['Employer'], job['Salary']) not in existing_jobs]
return new_jobs
async def send_new_job_postings(token, chat_id, new_jobs):
bot = Bot(token=token)
if new_jobs:
message = "New job postings:\n----------\n"
for job in new_jobs:
message = f"Title: {job['Title']}\nEmployer: {job['Employer']}\nSalary: {job['Salary']}\nExpected Salary Range: {job['AIExpectedSalaryRange']}\nCompatibility: {job['AICompatibility']}\n\nSummary: {job['AISummary']}\n\nSkills: {job['AISkills']}\n\nCustomers: {job['AICustomers']}\n\nJob Posting: View Details"
# Send the message
await bot.send_message(chat_id=chat_id, text=message, parse_mode='HTML')
print("New job postings sent to Telegram.")
else:
print("No new job postings to send.")
headers = build_headers(config_default["ProfesiaUrl"])
# Read existing jobs from the CSV file
existing_jobs = read_existing_jobs(JOB_LISTINGS_PATH)
# Scrape all jobs
scraped_jobs = scrape_jobs(config_default["ProfesiaUrl"])
# Filter new jobs
new_jobs = filter_new_jobs(scraped_jobs, existing_jobs)
# Scrape descriptions for new jobs only
for job in new_jobs:
job['Description'] = scrape_job_description(job['Link'])
time.sleep(3) # Optional: Sleep to avoid overloading the server
# Read my CV
with open(MY_CV_PATH, 'r', encoding='utf-8') as file:
my_cv = file.read()
# Read prompt file
with open(PROMPT_PATH, 'r', encoding='utf-8') as file:
prompt = file.read()
# Check compatibility for new jobs only
#========================
for job in new_jobs:
job_listing = job['Description']
client = OpenAI(api_key=config_default["OpenAIKey"],)
chat_completion = client.chat.completions.create(
messages=[
{
"role": "system",
"content": "You are a career advisor specializing in assessing job compatibility based on provided resumes and job descriptions."
},
{
"role": "user",
"content": f"Job listing: {job_listing}"
},
{
"role": "user",
"content": f"CV: {my_cv}"
},
{
"role": "user",
"content": prompt
}
],
model=config_default["Model"],
temperature=0.0
)
chat_reply = chat_completion.choices[0].message.content
# Regular expression to extract the JSON part
json_match = re.search(r'{.*}', chat_reply)
if json_match:
chat_reply = json_match.group()
# Parse the string to a Python dictionary
json_data = json.loads(chat_reply)
print(chat_reply)
job['AIExpectedSalaryRange'] = json_data['Expected Salary Range']
job['AICompatibility'] = json_data['Compatibility']
job['AISummary'] = json_data['Summary']
job['AISkills'] = json_data['Skills']
job['AICustomers'] = json_data['Customers']
job['AIPromptTokens'] = chat_completion.usage.prompt_tokens
job['AICompletionTokens'] = chat_completion.usage.completion_tokens
job['AITotalTokens'] = chat_completion.usage.total_tokens
# #========================
# Print the new job details
if new_jobs:
print("New job postings:")
for job in new_jobs:
print(f"Title: {job['Title']}")
print(f"Employer: {job['Employer']}")
print(f"Salary: {job['Salary']}")
print(f"Link: {job['Link']}")
print(f"AICompatibility: {job['AICompatibility']}")
print(f"AISummary: {job['AISummary']}")
print(f'AISkills: {job["AISkills"]}')
print(f'AICustomers: {job["AICustomers"]}')
print('---')
# Append new jobs to the CSV file
write_new_jobs(JOB_LISTINGS_PATH, new_jobs)
loop = asyncio.get_event_loop()
loop.run_until_complete(send_new_job_postings(config_default["TelegramToken"], config_default["TelegramChatID"], new_jobs))
else:
print("No new job postings found.")