lever_slack_update_driver.py

import os
from dotenv import load_dotenv
import requests
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from datetime import date
from collections import defaultdict
import modal

load_dotenv()
stub = modal.Stub()

API_KEY = os.getenv("LEVER_API_KEY")
BASE_URL = os.getenv("LEVER_ROOT_URL")
SLACK_BOT_TOKEN = os.getenv("SLACK_API_BOT_KEY")
SLACK_CHANNEL = "#ops-hiring"
ROLES = [
    "Product Manager",
    "Founding Backend Engineer",
    "Machine Learning Engineer Summer Intern",
    "Founding Machine Learning Engineer",
    "Frontend Engineer",
]


class SlackBot:
    def __init__(self):
        self.client = WebClient(token=SLACK_BOT_TOKEN)

    def post_in_blocks(self, blocks):
        try:
            response = self.client.chat_postMessage(channel=SLACK_CHANNEL, blocks=blocks)
            return response["ts"]  # return timestamp of message for future reference
        except SlackApiError as e:
            print(f"Error sending message: {e}")

    def create_url(self, text, url):
        return f"<{url}|{text}>"

    def create_block(self, text):
        return {"type": "section", "text": {"type": "mrkdwn", "text": text}}

    def create_divider(
        self,
    ):
        return {"type": "divider"}

    def post_header(self, stats_dict):
        date_str = create_date_str()
        self.post_in_blocks(
            [
                self.create_block(f"*{date_str} ATS Update*\n_{format_counts(stats_dict)}_"),
                self.create_divider(),
            ],
        )

    def post_candidate_blocks(self, candidate_map):
        MAX_BLOCKS = 50  # Slack limits to 50 blocks per message
        for role in candidate_map.keys():
            role_text = role
            if "engineer" in role.lower():
                role_text = "🛠 " + role + " 🛠"
            if "product" in role.lower():
                role_text = f"📦 {role} 📦"

            blocks = []
            text = f"*{role_text}* \n \n "
            blocks.append(self.create_block(text))
            candidates = candidate_map[role]
            for candidate in candidates:
                linkedin_url = candidate["linkedin_url"]
                twitter_url = candidate["twitter_url"]
                github_url = candidate["github_url"]

                entry = f"👤 {candidate['name']} \n "
                if candidate["headline"]:
                    entry += f"📖 {candidate['headline']} \n "
                if linkedin_url:
                    entry += "🔗 " + self.create_url("LinkedIn", linkedin_url)
                    if twitter_url:
                        entry += ", " + self.create_url("Twitter", twitter_url)
                    if github_url:
                        entry += ", " + self.create_url("Github", github_url)
                    entry += "\n"

                entry += f"{self.create_url('📂 Lever', candidate['lever_url'])} \n \n "
                blocks.append(self.create_block(entry))
                blocks.append(self.create_block("\n"))
                if len(blocks) == MAX_BLOCKS - 3:
                    break

            blocks.append(self.create_divider())
            self.post_in_blocks(blocks)


class LeverAPI:
    def __init__(self):
        self.base_url = BASE_URL
        self.stats = None

    def _make_request(self, endpoint, params=None):
        response = requests.get(BASE_URL + endpoint, auth=(API_KEY, ""), params=params)
        response.raise_for_status()  # raise an exception if the request was unsuccessful
        return response.json()

    def _get_unprocessed_opportunities(self):
        data = self._make_request("/opportunities", params={"limit": 100, "archived": False, "stage_id":["lead-new", "applicant-new"]})
        return data["data"]

    def _extract_candidate_data(self, d):
        # Grab relevant fields
        lever_url = d["urls"]["show"]
        name = d["name"]
        links = d["links"]

        # Get links
        linkedin_url = None
        twitter_url = None
        github_url = None
        for link in links:
            if "linkedin" in link.lower():
                linkedin_url = link
            elif "twitter" in link.lower():
                twitter_url = link
            elif "github" in link.lower():
                github_url = link

        headline = d["headline"]
        tags = d["tags"]

        candidate = {
            "lever_url": lever_url,
            "name": name,
            "linkedin_url": linkedin_url,
            "links": links,
            "headline": headline,
            "tags": tags,
            "github_url": github_url,
            "twitter_url": twitter_url,
        }
        return candidate

    def get_candidates_by_role(self):
        data = self._get_unprocessed_opportunities()
        candidate_map = defaultdict(list)
        for d in data:
            candidate = self._extract_candidate_data(d)
            for role in ROLES:
                if role in d["tags"]:
                    candidate_map[role].append(candidate)
        self.stats = {role: len(candidate) for role, candidate in candidate_map.items()}
        return candidate_map


def create_date_str():
    return date.today().strftime("%B") + " " + date.today().strftime("%d")


def format_counts(counts):
    return ", ".join(
        f"{count} {key}" if count == 1 else f"{count} {key}s" for key, count in counts.items()
    )


@stub.function(
    schedule=modal.Cron("0 12 * * *"),  # every monday at 8am ET
    image=modal.Image.debian_slim().pip_install(["slack-sdk", "python-dotenv", "requests"]),
    secret=modal.Secret.from_name("hearth-operations-secrets"),
)
def hearth_ops_lever_driver():
    # 1. Fetch candidates by role
    lever_api = LeverAPI()
    candidate_map = lever_api.get_candidates_by_role()
    stats_dict = lever_api.stats  # {role: cnt ...}

    # 2. Create slack messgae
    slack_api = SlackBot()
    slack_api.post_header(stats_dict)

    slack_api.post_candidate_blocks(candidate_map)


if __name__ == "__main__":
    hearth_ops_lever_driver()