Deploy a Cron Job to Deliver the Latest Hacker News Stories through WhatsApp

.dockerignore
app.py # The main entry point of the Python app
crontab # The interval at which the cron job runs
Dockerfile
hacker_news.py # Interacts with the Hacker News API to pull the latest popular stories
LICENSE
llm.py # Interacts with gpt-oss-20b to create a conversational interaction
README.md 
requirements.txt
script.sh # Triggered by the cron job. Runs the Python app

FROM python:3.11-alpine

# Install Supercronic
RUN apk add --no-cache bash curl \
    && curl -fsSLO "https://github.com/aptible/supercronic/releases/latest/download/supercronic-linux-amd64" \
    && chmod +x supercronic-linux-amd64 \
    && mv supercronic-linux-amd64 /usr/bin/supercronic \
    && /usr/bin/supercronic -version

WORKDIR /app

# Copy crontab to a fixed path
COPY crontab /etc/crontab
RUN chmod 0644 /etc/crontab

# Get the required pip packages
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy python app files
COPY ["*.py", "./"]

# Copy script
COPY script.sh ./script.sh
RUN chmod +x ./script.sh

# Use ENTRYPOINT + CMD
ENTRYPOINT ["/usr/bin/supercronic"]
CMD ["/etc/crontab"]

# Check Hacker News every day at 9:00 am
0 9 * * * /app/script.sh

# Alternatively, enable every 5 minutes for testing
# */5 * * * * /app/script.sh

#!/bin/bash
set -e

echo "[$(date)] Running my function..."
python /app/app.py

# Get an array of IDs of the top stories from Hacker News
def get_top_stories(self):
    order_by_value = quote('$key', safe='$')
    try:
        response = requests.get(f'{self.hn_base_url}topstories{self.endpoint_suffix}/?limitToFirst={10}&orderBy="{order_by_value}"')
        response.raise_for_status()
        data = response.json() 
        return data
    except requests.exceptions.RequestException as e:
        print('Error calling Top Stores API:', e)

# Get story information for a given story ID
def get_story_by_id(self, item_id: str):
    try:
        response = requests.get(f'{self.hn_base_url}item/{item_id}{self.endpoint_suffix}')
        response.raise_for_status()
        data = response.json()
        return data
    except requests.exceptions.RequestException as e:
        print('Error calling Item ID API:', e)

# Build array of top stories
def load_top_stories_concurrent(self):
    top_story_ids = self.get_top_stories()
    results = []

    with ThreadPoolExecutor(max_workers=5) as executor:
        future_to_id = {executor.submit(self.get_story_by_id, str(story_id)): story_id for story_id in top_story_ids}

        for future in as_completed(future_to_id):
            story_id = future_to_id[future]
            try:
                story_data = future.result()
                if story_data:
                    story = {
                        'title': story_data.get('title', None),
                        'text': story_data.get('text', None),
                        'url': story_data.get('url', None)
                    }
                    results.append(story)
            except Exception as e:
                print(f'Exception fetching story {story_id}: {e}')

    return results

# Uses the completions function to create a response
def summarize_hn(self, formatted_input: str, temperature: float = 1.0,  max_tokens: int = 500, role: str = 'user'):
    instructions = (
        'You are ChatGPT. Given a set of stories, create a summary message explaining the stories as though you were speaking to a friend.'
        'Include the links to all the stories if present, or just the info from the text if present.' 
    )
    tokenizer = AutoTokenizer.from_pretrained('openai/gpt-oss-20b', trust_remote_code=True)
    # Send a request to the OpenAI API and return the response.
    if not isinstance(formatted_input, str):
        formatted_input = formatted_input.to_string()

    try:
        messages=[{'role': 'system', 'content': instructions},{'role': role, 'content': formatted_input}]
        prompt = tokenizer.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=True,
        )
        response = self.client.completions.create(
            prompt=prompt,
            model=f'/models/{self.model}',
            max_tokens=max_tokens,
            temperature=temperature
        )

        if response.choices:
            chat_response = response.choices[0].text
            match = re.search(r'final(.*)', chat_response, re.DOTALL)
            # Remove newlines
            final_result = match.group(1).replace(r'\\n', ' ').replace('\\', '').strip()
            # Remove markdown formatting from URLs
            final_result = re.sub(r'\[([^\]]+)\]\((https?://[^\)]+)\)', r'\1: \2', final_result)
            return final_result if final_result else None
    except Exception as e:
        print(f'Request failed: {e}')
        return None

client = Client(account_sid, auth_token)
llm = GptOpenAi()
hacker_news = HackerNews()

# Get the ten top stories
top_stories = hacker_news.load_top_stories_concurrent()
# Have gpt-oss-20b summarize the stories
top_summary = llm.summarize_hn(f'{top_stories}', max_tokens=4000)

# Send messages to your WhatsApp number
def send_whatsapp_message_in_chunks(body, chunk_size=1600):
    # Split gpt-oss-20b response into 1600 char chunks to meet the delivery limits of WhatsApp with Twilio
    part_num = 1

    while body:
        if len(body) > chunk_size:
            split_point = body.rfind(' ', 0, chunk_size)
            if split_point == -1:
                split_point = chunk_size
        else:
            split_point = len(body)  # final chunk size or smaller

        part = body[:split_point]
        body = body[split_point:].lstrip()

        # Send the chunk
        message = client.messages.create(
            to=f'whatsapp:{whatsapp_to}',
            from_=f'whatsapp:{whatsapp_from}',
            body=part
        )
        print(f'Sent part {part_num} SID: {message.sid}')
        part_num += 1

messages = send_whatsapp_message_in_chunks(top_summary)