Jobs APIjobs.lk ↗

"""
TopJobs Sri Lanka Job Search API Client
Search and retrieve job listings from TopJobs.lk

Get your API key from: https://parse.bot/settings
"""

import os
import requests
from typing import Optional, Dict, Any, List


class ParseClient:
    """Client for TopJobs Sri Lanka Job Search API"""
    
    def __init__(self, api_key: Optional[str] = None):
        """Initialize the Parse API client
        
        Args:
            api_key: API key for authentication. If not provided, uses PARSE_API_KEY env var
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "38dde6af-0fbb-4015-b446-25b7b99f9fea"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")
        
        if not self.api_key:
            raise ValueError("API key must be provided or set in PARSE_API_KEY environment variable")
    
    def _call(self, endpoint: str, method: str = "POST", **params) -> Dict[str, Any]:
        """Make an API call to the Parse Bot service
        
        Args:
            endpoint: API endpoint name
            method: HTTP method (GET or POST)
            **params: Query/body parameters
            
        Returns:
            API response as dictionary
            
        Raises:
            requests.RequestException: If API call fails
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {
            "X-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        try:
            if method == "GET":
                response = requests.get(url, headers=headers, params=params)
            else:  # POST
                response = requests.post(url, headers=headers, json=params)
            
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"API Error: {e}")
            raise
    
    def search_jobs(self, keyword: Optional[str] = None, 
                   category: Optional[str] = None, 
                   page: int = 1) -> Dict[str, Any]:
        """Search for open job vacancies
        
        Args:
            keyword: Search keyword (e.g., 'engineer', 'accountant')
            category: Job category code (e.g., 'SDQ' for IT roles)
            page: Page number for pagination (default: 1)
            
        Returns:
            Dictionary containing total_jobs, total_pages, current_page, and jobs list
            
        Raises:
            ValueError: If neither keyword nor category is provided
        """
        if not keyword and not category:
            raise ValueError("At least one of keyword or category must be provided")
        
        params = {}
        if keyword:
            params["keyword"] = keyword
        if category:
            params["category"] = category
        if page > 1:
            params["page"] = page
        
        return self._call("search_jobs", method="POST", **params)
    
    def get_job_details(self, job_ref_no: str) -> Dict[str, Any]:
        """Get detailed information about a specific job vacancy
        
        Args:
            job_ref_no: Job reference number (e.g., '1512058')
            
        Returns:
            Dictionary containing detailed job information
        """
        return self._call("get_job_details", method="GET", job_ref_no=job_ref_no)
    
    def list_categories(self) -> Dict[str, Any]:
        """List all available job categories
        
        Returns:
            Dictionary containing list of categories with code and name
        """
        return self._call("list_categories", method="GET")


def main():
    """Main function demonstrating practical workflow"""
    
    # Initialize the client
    client = ParseClient()
    
    print("=" * 60)
    print("TopJobs Sri Lanka Job Search - Practical Workflow")
    print("=" * 60)
    
    # Step 1: List available categories
    print("\n1. Fetching available job categories...")
    try:
        categories_response = client.list_categories()
        categories = categories_response.get("categories", [])
        print(f"   Found {len(categories)} job categories")
        
        # Display some sample categories
        print("\n   Sample categories:")
        for category in categories[:5]:
            print(f"   - {category['code']}: {category['name']}")
    except Exception as e:
        print(f"   Error fetching categories: {e}")
        return
    
    # Step 2: Search for IT jobs using keyword
    print("\n2. Searching for 'engineer' positions...")
    try:
        search_response = client.search_jobs(keyword="engineer", page=1)
        total_jobs = search_response.get("total_jobs", 0)
        jobs = search_response.get("jobs", [])
        
        print(f"   Found {total_jobs} total jobs matching 'engineer'")
        print(f"   Showing {len(jobs)} jobs on current page")
        
        if jobs:
            print("\n   Top job listings:")
            for i, job in enumerate(jobs[:3], 1):
                print(f"\n   {i}. {job.get('position', 'N/A')}")
                print(f"      Employer: {job.get('employer', 'N/A')}")
                print(f"      Location: {job.get('town', 'N/A')}")
                print(f"      Posted: {job.get('opening_date', 'N/A')}")
                print(f"      Closing: {job.get('closing_date', 'N/A')}")
    except Exception as e:
        print(f"   Error searching jobs: {e}")
        return
    
    # Step 3: Get detailed information for the first job
    if jobs:
        first_job = jobs[0]
        job_ref_no = first_job.get("job_ref_no")
        
        print(f"\n3. Fetching detailed information for job {job_ref_no}...")
        try:
            job_details = client.get_job_details(job_ref_no)
            
            print(f"\n   Job Details:")
            print(f"   Position: {job_details.get('position', 'N/A')}")
            print(f"   Employer: {job_details.get('employer', 'N/A')}")
            print(f"   Location: {job_details.get('location', 'N/A')}")
            print(f"   Job Type: {job_details.get('job_type', 'N/A')}")
            print(f"   Closing Date: {job_details.get('closing_date', 'N/A')}")
            print(f"   Date Posted: {job_details.get('date_posted', 'N/A')}")
            
            if job_details.get('company_email'):
                print(f"   Company Email: {job_details.get('company_email')}")
            
            locations = job_details.get('locations', [])
            if locations:
                print(f"   Locations: {', '.join(locations)}")
            
            apply_links = job_details.get('apply_links', [])
            if apply_links:
                print(f"\n   Application Links:")
                for link in apply_links:
                    print(f"   - {link}")
        except Exception as e:
            print(f"   Error fetching job details: {e}")
    
    # Step 4: Search by category (IT Software)
    print("\n4. Searching for IT Software/Database/QA positions...")
    try:
        category_search = client.search_jobs(category="SDQ", page=1)
        category_jobs = category_search.get("jobs", [])
        total_category_jobs = category_search.get("total_jobs", 0)
        
        print(f"   Found {total_category_jobs} total jobs in IT-Software category")
        print(f"   Showing {len(category_jobs)} jobs on current page")
        
        if category_jobs:
            print("\n   Recent postings in this category:")
            for i, job in enumerate(category_jobs[:3], 1):
                print(f"   {i}. {job.get('position', 'N/A')} at {job.get('employer', 'N/A')}")
    except Exception as e:
        print(f"   Error searching by category: {e}")
    
    print("\n" + "=" * 60)
    print("Workflow completed successfully!")
    print("=" * 60)


if __name__ == "__main__":
    main()