A Telegram job alert bot in 60 lines of Python, on live ATS data

작성자

카테고리:

← 피드로
DEV Community · Oleg Starnikov · 2026-09-13 개발(SW)

Oleg Starnikov

Job boards go stale fast. By the time a weekly digest reaches you, the interesting roles have been open
for a month. So let’s build the smallest thing that fixes that: a Telegram bot that wakes up, asks one
API for everything posted in the last seven days, and sends you the new ones.

Sixty lines, no framework, no database.

The data

We’ll read HiringIndex — job postings taken from
the catalogue APIs that applicant tracking systems publish, so every row carries the employer’s own
apply link rather than an aggregator’s redirect.

One call tells you what a slice looks like before you write any code. Two slices, both measured live at
2026-09-13 04:47–04:50 UTC:

All remote, last 7 days Python roles, remote, last 7 days Postings 10,860 9 Companies 2,227 8 Median days open 4 — Posted in the last 7 days 100% 100%

The second column is the filter the bot below actually uses, and it is deliberately narrow — nine
postings is a morning’s worth of reading, not a firehose. Widen it by dropping job_titles and you
are back to the first column.

Salaries do not come from search at all: they come from /jobs/insights, per currency, and each
percentile says how many disclosures it rests on — for the wide slice, 744 USD postings give p25
115,000–150,000, p50 155,650–200,000, p75 199,000–260,000. Percentiles appear only where at least
30 salaries are disclosed, so a number you get is computed from real bands. (insights computes a
slice in the background the first time it is asked — see the snippet at the end for the two-line
202 handling.)

The bot

import os
import time

import requests

API = "https://hiringindex.p.rapidapi.com/jobs/search"
HEADERS = {
    "X-RapidAPI-Key": os.environ["RAPIDAPI_KEY"],
    "X-RapidAPI-Host": "hiringindex.p.rapidapi.com",
    "Content-Type": "application/json",
}
TELEGRAM = f"https://api.telegram.org/bot{os.environ['TG_BOT_TOKEN']}/sendMessage"
CHAT_ID = os.environ["TG_CHAT_ID"]

FILTER = {
    "job_titles": ["Python Developer"],
    "remote_flag": ["true"],
    "days_ago": 7,
    "limit": 20,
}


def fetch(page=1):
    r = requests.post(API, headers=HEADERS, json=FILTER | {"page": page}, timeout=60)
    if r.status_code == 503:                     # capacity, not an error on your side
        time.sleep(float(r.headers.get("Retry-After", 5)))
        return fetch(page)
    r.raise_for_status()
    return r.json()


def line(job):
    bits = [job.get("country_code"), job.get("employment_type"), job.get("seniority")]
    tail = " · ".join(b for b in bits if b)
    return (f"<b>{job['title']}</b>\n{job.get('company_name') or job.get('handle')}"
            f"{' · ' + tail if tail else ''}\nposted {job.get('posted_at')}\n"
            f"{job.get('apply_url') or job.get('posting_url')}")


def send(text):
    requests.post(TELEGRAM, json={"chat_id": CHAT_ID, "text": text,
                                  "parse_mode": "HTML", "disable_web_page_preview": True}, timeout=30)


def main():
    seen = set(open("seen.txt").read().split()) if os.path.exists("seen.txt") else set()
    data = fetch()
    fresh = [j for j in data["jobs"] if j["_id"] not in seen]
    for job in fresh:
        send(line(job))
        seen.add(job["_id"])
    with open("seen.txt", "w") as f:
        f.write("\n".join(seen))
    print(f"{len(fresh)} new of {data['total_count']} matching")


if __name__ == "__main__":
    main()

Enter fullscreen mode Exit fullscreen mode

Run it from cron every morning. seen.txt is the whole state: a posting is sent once and never again.

Three things worth knowing before you build on it

days_ago filters by the employer’s publication date, not by when the index saw the row. That is
why the median above is four days and not four hours — the filter is about the job, not about our
crawler.

Rows are not postings. A role advertised in three cities is three rows in search, one posting in
the aggregate. If you count anything, count with /jobs/insights, which deduplicates; total_count
on search is rows.

Errors arrive as one envelope. Branch on error, never on the message text:

{"error": "invalid_request", "message": "days_ago: expected an integer", "meta": {"request_id": "..."}}

Enter fullscreen mode Exit fullscreen mode

An unknown filter key is rejected with 422 and the message names the valid keys, so a typo never
returns a silently wrong slice. An empty jobs array with total_count: 0 is a normal 200 — no
matches is not an error.

Making it yours

Swap the filter and the bot changes job:

  • {"keywords": ["kubernetes"], "days_ago": 3} — a stack watch
  • {"handles": ["walmart:wd504:WalmartExternal"]} — one employer’s careers page, by ATS board handle
  • {"cities": ["Berlin"], "seniority": ["senior"], "salary": {"min": 80000}} — a narrow local search

Send the same filter to /jobs/insights instead of /jobs/search and you get the market around your
search: how many companies are hiring for it, where they are, and what they pay.

One thing to handle there, because aggregates are computed over the whole matching set rather than the
page you asked for: a filter nobody has requested yet comes back as 202 with Retry-After: 30
while the numbers are computed in the background. Ask again after the pause and you get 200 with a
meta.computed_at stamp. Measured 2026-09-13: a fresh filter answered 202 in 495 ms, and the same
filter returned 200 about a minute later.

INSIGHTS = "https://hiringindex.p.rapidapi.com/jobs/insights"   # same headers as above


def insights(body, attempts=4):
    for _ in range(attempts):
        r = requests.post(INSIGHTS, headers=HEADERS, json=body, timeout=60)
        # 202: the slice is being computed, not an error. A wide slice can answer 202 more than once,
        # so loop rather than retry exactly once. 429 is the plan limit, 503 is capacity.
        if r.status_code in (202, 429, 503):
            time.sleep(float(r.headers.get("Retry-After", 30)))
            continue
        r.raise_for_status()
        return r.json()
    raise RuntimeError("insights did not return data in time")


market = insights(FILTER)                      # same filter, no limit and no page
head, fresh = market["headline"], market.get("freshness") or {}
print(head["job_count"], "postings,", head["company_count"], "employers")
print("median days open:", fresh.get("median_days_live", "—"))   # empty slice: counts are 0, median None

Enter fullscreen mode Exit fullscreen mode

The free plan is 200 postings and 5 insights calls a month, no card:
get a key.

Written with an AI assistant; every number and code snippet was run against the live HiringIndex API by the team before publishing. <!– ai-disclosure –>

원문에서 계속 ↗