|
import time |
|
import logging |
|
from fastapi import FastAPI |
|
from fastapi.responses import HTMLResponse |
|
from fastapi.staticfiles import StaticFiles |
|
from selenium import webdriver |
|
from selenium.webdriver.common.by import By |
|
from selenium.webdriver.chrome.service import Service |
|
from selenium.webdriver.chrome.options import Options |
|
from webdriver_manager.chrome import ChromeDriverManager |
|
|
|
app = FastAPI() |
|
|
|
|
|
logging.basicConfig(level=logging.INFO) |
|
|
|
|
|
app.mount("/static", StaticFiles(directory="static"), name="static") |
|
|
|
def scrape_upwork_data(search_query, num_jobs, page): |
|
|
|
options = webdriver.ChromeOptions() |
|
options.add_argument('--headless') |
|
options.add_argument('--no-sandbox') |
|
options.add_argument('--disable-dev-shm-usage') |
|
|
|
|
|
driver=webdriver.Chrome(options=options) |
|
job_listings = [] |
|
try: |
|
url = f'https://www.upwork.com/nx/search/jobs?amount=500-&hourly_rate=25-&location=Americas,Europe,Australia%20and%20New%20Zealand,Canada,India,Switzerland,United%20States&per_page=50&q={search_query}&sort=recency&t=0,1&page={page}' |
|
driver.get(url) |
|
driver.set_window_size(1080, 720) |
|
time.sleep(5) |
|
page_title = driver.title |
|
jobs = driver.find_elements(By.CSS_SELECTOR, 'article[data-test="JobTile"]') |
|
|
|
for job in jobs: |
|
try: |
|
posted_date = job.find_element(By.CSS_SELECTOR, 'small[data-test="job-pubilshed-date"]').text.strip() |
|
title_element = job.find_element(By.CSS_SELECTOR, 'h2.job-tile-title > a') |
|
title = title_element.text.strip() |
|
link = title_element.get_attribute('href') |
|
description = job.find_element(By.CSS_SELECTOR, 'div[data-test="JobTileDetails"] > div > div > p').text.strip() |
|
|
|
job_info = job.find_element(By.CSS_SELECTOR, 'ul.job-tile-info-list') |
|
job_type = job_info.find_element(By.CSS_SELECTOR, 'li[data-test="job-type-label"]').text.strip() |
|
experience_level = job_info.find_element(By.CSS_SELECTOR, 'li[data-test="experience-level"]').text.strip() |
|
|
|
|
|
try: |
|
budget = job_info.find_element(By.CSS_SELECTOR, 'li[data-test="is-fixed-price"]').text.strip() |
|
except Exception as e: |
|
logging.error(f'Error finding budget: {e}') |
|
budget = job_info.find_element(By.CSS_SELECTOR, 'li[data-test="duration-label"]').text.strip() |
|
|
|
job_listings.append({ |
|
'title': title, |
|
'date': posted_date, |
|
'link': link, |
|
'description': description, |
|
'job_type': job_type, |
|
'experience_level': experience_level, |
|
'budget': budget |
|
}) |
|
|
|
except Exception as e: |
|
logging.error(f'Error parsing job listing: {e}') |
|
|
|
finally: |
|
driver.quit() |
|
|
|
return job_listings |
|
|
|
@app.get("/", response_class=HTMLResponse) |
|
async def read_root(): |
|
return """ |
|
<html> |
|
<head> |
|
<title>Upwork Job Listings</title> |
|
</head> |
|
<body> |
|
<h1>Welcome to Upwork Job Scraper</h1> |
|
<form action="/jobs" method="get"> |
|
<input type="text" name="query" placeholder="Search Query" required> |
|
<input type="number" name="num_jobs" value="50" min="1" max="100" required> |
|
<button type="submit">Search Jobs</button> |
|
</form> |
|
</body> |
|
</html> |
|
""" |
|
|
|
@app.get("/jobs", response_class=HTMLResponse) |
|
async def get_jobs(query: str, num_jobs: int = 50): |
|
jobs = [] |
|
for page in range(1, (num_jobs // 50) + 1): |
|
job_listings = scrape_upwork_data(query, num_jobs, page) |
|
jobs.extend(job_listings) |
|
|
|
|
|
html_output = "<h2>Job Listings</h2>" |
|
for job in jobs: |
|
html_output += f""" |
|
<div> |
|
<h3><a href="{job['link']}">{job['title']}</a></h3> |
|
<p>Posted Date: {job['date']}</p> |
|
<p>Type: {job['job_type']}</p> |
|
<p>Experience Level: {job['experience_level']}</p> |
|
<p>Budget: {job['budget']}</p> |
|
<p>Description: {job['description']}</p> |
|
</div> |
|
<hr> |
|
""" |
|
return HTMLResponse(content=html_output) |
|
|
|
if __name__ == "__main__": |
|
import uvicorn |
|
uvicorn.run(app, host="0.0.0.0", port=7860) |
|
|
