Spaces:
Sleeping
Sleeping
| import requests | |
| import os | |
| from dotenv import load_dotenv | |
| from datetime import datetime | |
| load_dotenv() | |
| def get_job_listings(job_title, location="Worldwide"): | |
| try: | |
| client_id = os.environ.get('JOBSPIKR_CLIENT_ID') | |
| client_auth_key = os.environ.get('JOBSPIKR_AUTH_KEY') | |
| if not client_id or not client_auth_key: | |
| raise ValueError("Client ID and Auth Key are required. Set them in environment variables or hardcode for testing.") | |
| if not job_title.strip(): | |
| raise ValueError("Job title cannot be empty.") | |
| api_url = "https://api.jobspikr.com/v2/data" | |
| payload = { | |
| "size": 50, | |
| "format": "json", | |
| "search_query_json": { | |
| "query_string": { | |
| "default_field": "job_title", | |
| "query": f"(\"{job_title}\") AND (\"{location}\")" | |
| } | |
| } | |
| } | |
| headers = { | |
| "Content-Type": "application/json", | |
| "client_id": client_id, | |
| "client_auth_key": client_auth_key | |
| } | |
| response = requests.post(api_url, json=payload, headers=headers) | |
| response.raise_for_status() | |
| data = response.json() | |
| if data.get("status") != "success": | |
| raise ValueError(f"API Error: {data.get('message', 'Unknown error')}") | |
| job_data = data.get("job_data", []) | |
| if not job_data: | |
| print("No job listings found.") | |
| return [] | |
| # Removing duplicates based on job title and company | |
| unique_jobs = {f"{job.get('job_title', '').lower()}|{job.get('company_name', '').lower()}": job for job in job_data} | |
| job_data = list(unique_jobs.values()) | |
| job_data.sort(key=lambda x: datetime.strptime(x.get('post_date', '2010-01-01'), '%Y-%m-%d'), reverse=True) | |
| job_data = job_data[:8] | |
| for job in job_data: | |
| print(f"Job Title: {job.get('job_title', 'N/A')}") | |
| print(f"Company: {job.get('company_name', 'N/A')}") | |
| print(f"Location: {job.get('city', 'N/A')}, {job.get('country', 'N/A')}") | |
| print(f"Posted on: {job.get('post_date', 'N/A')}") | |
| print(f"URL: {job.get('url', 'N/A')}") | |
| print(f"Apply URL: {job.get('apply_url', 'N/A')}") | |
| print(f"Has Expired: {job.get('has_expired', False)}") | |
| print(f"Remote: {job.get('is_remote', 'N/A')}") | |
| print(f"Job Type: {job.get('job_type', 'N/A')}") | |
| print(f"Inferred Skills: {', '.join(job.get('inferred_skills', [])) or 'N/A'}") | |
| #print(f"Seniority Level: {job.get('inferred_seniority_level', 'N/A')}") | |
| # Displaying offered salary if it exists | |
| offered_salary = job.get('salary_offered') | |
| if offered_salary: | |
| print(f"Offered Salary: {offered_salary}") | |
| print("\n") | |
| return job_data | |
| except requests.exceptions.RequestException as req_err: | |
| print(f"Request Error: {req_err}") | |
| return False | |
| except Exception as e: | |
| print(f"Error occurred: {e}") | |
| return False | |
| if __name__ == "__main__": | |
| job_title = input("Enter the job title: ").strip() | |
| location = input("Enter the location: ").strip() | |
| get_job_listings(job_title, location) | |
| # import requests | |
| # import os | |
| # from dotenv import load_dotenv | |
| # from datetime import datetime | |
| # load_dotenv() | |
| # def get_job_listings(job_title, location="Worldwide"): | |
| # try: | |
| # # Fetch the API credentials from environment variables | |
| # client_id = os.environ.get('JOBSPIKR_CLIENT_ID') | |
| # client_auth_key = os.environ.get('JOBSPIKR_AUTH_KEY') | |
| # if not client_id or not client_auth_key: | |
| # raise ValueError("Client ID and Auth Key are required. Set them in environment variables or hardcode for testing.") | |
| # # Validate inputs | |
| # if not job_title.strip(): | |
| # raise ValueError("Job title cannot be empty.") | |
| # # Construct API endpoint | |
| # api_url = "https://api.jobspikr.com/v2/data" | |
| # # Construct the POST request payload | |
| # payload = { | |
| # "size": 50, | |
| # "format": "json", | |
| # "search_query_json": { | |
| # "query_string": { | |
| # "default_field": "job_title", | |
| # "query": f"(\"{job_title}\") AND (\"{location}\")" | |
| # } | |
| # } | |
| # } | |
| # # Add headers | |
| # headers = { | |
| # "Content-Type": "application/json", | |
| # "client_id": client_id, | |
| # "client_auth_key": client_auth_key | |
| # } | |
| # # Make the API request | |
| # response = requests.post(api_url, json=payload, headers=headers) | |
| # response.raise_for_status() # Raise an error for HTTP response codes 4xx/5xx | |
| # # Parse the response | |
| # data = response.json() | |
| # if data.get("status") != "success": | |
| # raise ValueError(f"API Error: {data.get('message', 'Unknown error')}") | |
| # # Extract job data | |
| # job_data = data.get("job_data", []) | |
| # if not job_data: | |
| # print("No job listings found.") | |
| # return [] | |
| # # Filter jobs by location | |
| # filtered_jobs = [ | |
| # job for job in job_data | |
| # if location.lower() in (job.get('city', '').lower() or '') or | |
| # location.lower() in (job.get('country', '').lower() or '') | |
| # ] | |
| # if not filtered_jobs: | |
| # print(f"No job listings found for location: {location}.") | |
| # return [] | |
| # # Remove duplicates based on job title and company | |
| # unique_jobs = { | |
| # f"{job.get('job_title', '').lower()}|{job.get('company_name', '').lower()}": job | |
| # for job in filtered_jobs | |
| # } | |
| # filtered_jobs = list(unique_jobs.values()) | |
| # # Sort jobs by post date (newest to oldest) | |
| # filtered_jobs.sort( | |
| # key=lambda x: datetime.strptime(x.get('post_date', '2010-01-01'), '%Y-%m-%d'), | |
| # reverse=True | |
| # ) | |
| # filtered_jobs = filtered_jobs[:8] | |
| # # Print job listings with all fields | |
| # for job in filtered_jobs: | |
| # print(f"Job Title: {job.get('job_title', 'N/A')}") | |
| # print(f"Company: {job.get('company_name', 'N/A')}") | |
| # print(f"Location: {job.get('city', 'N/A')}, {job.get('country', 'N/A')}") | |
| # print(f"Posted on: {job.get('post_date', 'N/A')}") | |
| # print(f"URL: {job.get('url', 'N/A')}") | |
| # print(f"Apply URL: {job.get('apply_url', 'N/A')}") | |
| # print(f"Has Expired: {job.get('has_expired', False)}") # Default to False | |
| # print(f"Remote: {job.get('is_remote', 'N/A')}") | |
| # print(f"Job Type: {job.get('job_type', 'N/A')}") | |
| # print(f"Inferred Skills: {', '.join(job.get('inferred_skills', [])) or 'N/A'}") | |
| # offered_salary = job.get('salary_offered') or job.get('offered_salary') | |
| # if offered_salary: | |
| # print(f"Offered Salary: {offered_salary}") | |
| # print("\n") | |
| # return filtered_jobs | |
| # except requests.exceptions.RequestException as req_err: | |
| # print(f"Request Error: {req_err}") | |
| # return False | |
| # except Exception as e: | |
| # print(f"Error occurred: {e}") | |
| # return False | |
| # if __name__ == "__main__": | |
| # # Prompt user for input | |
| # job_title = input("Enter the job title: ").strip() | |
| # location = input("Enter the location (leave blank for Worldwide): ").strip() | |
| # # Call the function with user input | |
| # if job_title: | |
| # location = location if location else "Worldwide" | |
| # get_job_listings(job_title, location) | |
| # else: | |
| # print("Job title is required.") | |