ربات تلگرام قیمت دلار با پایتون؛ از کلید تا کانال

در یک گروه تلگرامی خانوادگی یا کانال یک فروشگاه کوچک، هر روز یک نفر می‌پرسد «دلار امروز چند است؟» و نفر دیگری عدد را از یک سایت کپی می‌کند. دو روز بعد همان عدد کپی‌شده مال دیروز است، کسی ساعتش را ننوشته و بحث سر این‌که کدام عدد درست است بالا می‌گیرد. رباتی که خودش جواب بدهد و کنار هر عدد ساعتش را بنویسد، این بحث را تمام می‌کند.

در این راهنما چنین رباتی را با پایتون می‌سازیم. ربات نرخ را از وب‌سرویس نرخ ارز نِت اَرز می‌گیرد، پاسخ را چند دقیقه نگه می‌دارد تا سهمیه نسوزد، مبلغ را به تومان تبدیل می‌کند و اگر بخواهید هر ساعت نرخ را در کانالتان می‌گذارد. شکل دقیق پاسخ‌ها در مستندات نرخ‌ها آمده و کد این مقاله فقط همان فیلدهایی را می‌خواند که آن‌جا نوشته شده‌اند.

یک راه کوتاه‌تر هم هست و بهتر است اول آن را بدانید. اگر فقط می‌خواهید نرخ چند ارز اصلی در کانال یا گروهتان بیاید، ربات آمادهٔ @netarzbot را به آن اضافه کنید. رایگان است و هر شش ساعت نرخ دلار، یورو، پوند و درهم را می‌فرستد. ساختن ربات خودتان وقتی می‌ارزد که ارزهای دیگر، دستورهای خودتان یا زمان‌بندی دیگری لازم دارید.

پیش از کد: چهار چیزی که لازم دارید

• توکن ربات. در تلگرام به @BotFather پیام بدهید، دستور /newbot را بفرستید و نام ربات را انتخاب کنید. توکنی که می‌گیرید مثل رمز عبور است؛ آن را داخل فایل کد ننویسید.
• کلید اپ نرخ ارز. در پنل «API نرخ ارز» در نِت اَرز یک اپ بسازید. کلیدی که با fx-ntz-v1- شروع می‌شود فقط یک بار نمایش داده می‌شود، پس همان لحظه کپی‌اش کنید.
• یک دامنه که مالکش هستید. هر اپ به یک دامنهٔ تأییدشده بسته است، حتی اگر فقط از سرور صدایش بزنید. برای تأیید، یک رکورد DNS یا یک فایل کوچک می‌گذارید و این کار یک بار برای همیشه است. اگر هنوز دامنه ندارید، دامنه و هاست خارجی گزینه‌ها را کنار هم گذاشته است.
• سروری که به تلگرام برسد. ربات باید شبانه‌روز روشن بماند و به api.telegram.org وصل شود. از سرورهای داخل ایران این اتصال معمولاً برقرار نیست، پس یک سرور مجازی بیرون از ایران ساده‌ترین انتخاب است. اگر با آن آشنا نیستید، سرور مجازی چیست پایه را توضیح می‌دهد.

قفل IP: چرا ربات اول کار ۴۰۳ می‌گیرد

کلید نرخ ارز به دامنه و IP شما قفل است. درخواستی که از مرورگر می‌آید با دامنهٔ صفحه سنجیده می‌شود و درخواستی که از سرور می‌آید، مثل همین ربات، با IP فرستنده. ربات سربرگ Origin نمی‌فرستد، پس تا IP سرورش را در فهرست «IPهای مجاز» اپ ننویسید، پاسخ 403 origin_required می‌گیرد. IP خروجی سرور را روی خود سرور با این دستور ببینید:

curl -4 https://api.ipify.org
# 203.0.113.25   <- add exactly this address to the app's allowed IPs

اگر سرور هم IPv4 دارد و هم IPv6، ممکن است درخواست از راه IPv6 برود و با وجود ثبت IPv4 خطای ip_not_allowed بگیرید. فیلد ip در همان پاسخ خطا می‌گوید نِت اَرز چه نشانی‌ای دیده؛ همان را اضافه کنید. همهٔ حالت‌های این خطا را در خطای ip_not_allowed در API نرخ ارز باز کرده‌ایم.

کد ربات: نرخ، کش و سه دستور

از کتابخانهٔ python-telegram-bot نسخهٔ ۲۱ و requests استفاده می‌کنیم. روی سرور یک محیط مجازی بسازید و هر دو را نصب کنید. بخش [job-queue] برای ارسال زمان‌بندی‌شده در کانال است که پایین‌تر به آن می‌رسیم:

python3 -m venv .venv && . .venv/bin/activate
pip install "python-telegram-bot[job-queue]==21.*" requests

کلید و توکن را در متغیر محیطی بگذارید، نه در خود کد:

export TELEGRAM_BOT_TOKEN="123456:AA..."
export NETARZ_FX_KEY="fx-ntz-v1-..."

حالا خود ربات در فایل bot.py. قلب کار تابع board() است: حداکثر هر پنج دقیقه یک بار سراغ سرویس می‌رود و بقیهٔ وقت از حافظه جواب می‌دهد. اگر سرویس یک بار جواب ندهد، آخرین نرخ سالم را نگه می‌دارد و یک دقیقه بعد دوباره امتحان می‌کند.

import asyncio, logging, os, threading, time
from datetime import datetime

import requests
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

API = "https://netarz.ir/api/fx/v1"
CODES = "USD,EUR,AED,TRY,GBP,CNY"
CACHE_SECONDS = 300          # the board changes every few minutes; never call per message

log = logging.getLogger("fxbot")
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['NETARZ_FX_KEY']}"
_cache = {"at": 0.0, "body": None}
_lock = threading.Lock()

FA = str.maketrans("0123456789,.", "۰۱۲۳۴۵۶۷۸۹٬٫")
LATIN = str.maketrans("۰۱۲۳۴۵۶۷۸۹٫", "0123456789.")


def board() -> dict:
    """The rate board, with at most one API call per CACHE_SECONDS."""
    with _lock:
        if _cache["body"] and time.time() - _cache["at"] < CACHE_SECONDS:
            return _cache["body"]
        try:
            r = session.get(f"{API}/rates", params={"codes": CODES}, timeout=10)
            body = r.json()
            if r.status_code != 200:
                raise RuntimeError(body.get("error", {}).get("code", r.status_code))
            _cache.update(at=time.time(), body=body)
        except Exception as exc:
            log.warning("FX call failed: %s", exc)
            if _cache["body"] is None:
                raise
            _cache["at"] = time.time() - CACHE_SECONDS + 60   # keep the last good board, retry in a minute
        return _cache["body"]


def fa(n: float) -> str:
    return f"{round(n):,}".translate(FA)


def stamp(meta: dict) -> str:
    t = datetime.fromisoformat(meta["as_of"]).strftime("%H:%M").translate(FA)
    late = f"، با {fa(meta['delayed_minutes'])} دقیقه تأخیر" if meta.get("is_delayed") else ""
    return f"نرخ ساعت {t}{late}"


def line(row: dict) -> str:
    unit = f" (هر {fa(row['unit'])} واحد)" if row["unit"] > 1 else ""
    return f"{row['name']}{unit}: خرید {fa(row['buy'])} / فروش {fa(row['sell'])} تومان"


async def rows():
    body = await asyncio.to_thread(board)          # requests is blocking; keep it off the event loop
    return {r["code"]: r for r in body["data"]}, body["meta"]


async def dollar(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    table, meta = await rows()
    await update.effective_message.reply_text(f"{line(table['USD'])}\n{stamp(meta)}")


async def rate(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    table, meta = await rows()
    code = (context.args[0] if context.args else "USD").upper()
    if code not in table:
        await update.effective_message.reply_text("ارزهای این ربات: " + "، ".join(table))
        return
    await update.effective_message.reply_text(f"{line(table[code])}\n{stamp(meta)}")


async def convert(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    table, meta = await rows()
    try:
        amount = float(context.args[0].translate(LATIN))
        code = (context.args[1] if len(context.args) > 1 else "USD").upper()
        row = table[code]
    except (IndexError, ValueError, KeyError):
        await update.effective_message.reply_text("نمونه: /convert 100 USD")
        return
    toman = amount * row["sell"] / row["unit"]
    shown = f"{amount:g}".translate(FA)
    await update.effective_message.reply_text(
        f"{shown} {row['name']} ≈ {fa(toman)} تومان، با نرخ فروش\n{stamp(meta)}")


async def on_error(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
    log.error("handler failed: %s", context.error)
    if isinstance(update, Update) and update.effective_message:
        await update.effective_message.reply_text("فعلاً نرخ در دسترس نیست؛ چند دقیقهٔ دیگر دوباره بپرسید.")


def main() -> None:
    logging.basicConfig(level=logging.INFO)
    app = Application.builder().token(os.environ["TELEGRAM_BOT_TOKEN"]).build()
    app.add_handler(CommandHandler(["start", "dollar"], dollar))
    app.add_handler(CommandHandler("rate", rate))
    app.add_handler(CommandHandler("convert", convert))
    app.add_error_handler(on_error)
    app.run_polling()


if __name__ == "__main__":
    main()

سه دستوری که ربات می‌فهمد:

چند تصمیم کوچک در این کد هست که بهتر است دلیلشان را بدانید. تبدیل مبلغ را از همان تابلوی کش‌شده حساب می‌کنیم، نه با صدا زدن مسیر /convert برای هر پیام؛ نتیجه یکی است و سهمیه‌ای خرج نمی‌شود. خرید و فروش از دید صرافی نوشته شده: buy یعنی صرافی می‌خرد و sell یعنی صرافی می‌فروشد. کسی که می‌پرسد «صد درهم چند تومان است»، معمولاً می‌خواهد بداند برای خریدنش چقدر لازم دارد، پس تبدیل با نرخ فروش است. و قیمت هر ارز برای unit واحد است؛ برای ارزی مثل دینار عراق که قیمتش برای صد واحد اعلام می‌شود، کد پیش از ضرب بر unit تقسیم می‌کند. ورودی با رقم فارسی هم پذیرفته می‌شود، چون بیشتر کاربرها همان‌طور تایپ می‌کنند.

سهمیه و تأخیر طرح رایگان را درست حساب کنید

طرح رایگان API نرخ ارز برای همین ربات کافی است، به یک شرط: کش. هر درخواست، موفق یا ناموفق، یک واحد از سهمیهٔ روزانهٔ اپ است. اگر ربات برای هر پیام یک بار سرویس را صدا بزند، یک گروه شلوغ می‌تواند سهمیهٔ روز را پیش از ظهر تمام کند و از آن به بعد ربات فقط خطای 429 daily_quota_exceeded می‌بیند. با کش پنج‌دقیقه‌ای، ربات در کل شبانه‌روز حداکثر ۲۸۸ درخواست می‌فرستد، هر چقدر هم کاربر داشته باشد؛ این کسر کوچکی از سهمیهٔ روزانهٔ طرح رایگان است. عدد دقیق سهمیه‌ها و محدودیت هر دقیقه را در صفحهٔ طرح‌ها ببینید.

نکتهٔ دوم تأخیر است. در طرح رایگان نرخ با چند دقیقه تأخیر می‌آید: همان عدد و همان دقت، فقط عقب‌تر. پاسخ سرویس این را پنهان نمی‌کند. فیلد meta.as_of زمان واقعی نرخ را می‌گوید و meta.delayed_minutes مقدار تأخیر را، و تابع stamp() هر دو را زیر هر پاسخ می‌نویسد تا کسی که عدد را می‌بیند بداند مال چه ساعتی است. برای گروه خانوادگی یا کانالی که روزی چند بار نرخ می‌گذارد، این تأخیر مشکلی نیست. اگر ربات شما پایهٔ تصمیم خرید و فروش است و هر دقیقه اهمیت دارد، طرح پرو را بردارید.

ارسال خودکار نرخ در کانال

برای این‌که ربات هر ساعت نرخ را در کانال بگذارد، اول آن را به کانال اضافه و مدیر (ادمین) کنید، با اجازهٔ ارسال پیام. بعد شناسهٔ کانال را در متغیر محیطی FX_CHANNEL بگذارید: برای کانال عمومی نام کاربری با @ کافی است (مثل @my_channel) و برای کانال خصوصی شناسهٔ عددی‌ای که با -100 شروع می‌شود. زمان‌بندی را صف کارهای خود کتابخانه (JobQueue) انجام می‌دهد. این چند خط را به bot.py اضافه کنید:

CHANNEL = os.environ.get("FX_CHANNEL")          # "@my_channel" or "-100..."


async def post_rates(context: ContextTypes.DEFAULT_TYPE) -> None:
    table, meta = await rows()
    text = "\n".join(line(table[c]) for c in ("USD", "EUR", "AED") if c in table)
    await context.bot.send_message(chat_id=CHANNEL, text=f"{text}\n{stamp(meta)}")

# in main(), just before app.run_polling():
#     if CHANNEL:
#         app.job_queue.run_repeating(post_rates, interval=3600, first=30)

پست کانال از همان کش می‌خواند، پس تقریباً درخواست اضافه‌ای نمی‌سازد. فاصله را کمتر از یک ساعت نگذارید؛ کانالی که هر ده دقیقه یک عدد تکراری می‌فرستد، زود عضوهایش را از دست می‌دهد.

اجرا روی سرور، بدون این‌که با بستن ترمینال خاموش شود

run_polling() خودش از تلگرام پیام تازه می‌پرسد، پس به دامنه، گواهی SSL یا وب‌هوک نیاز ندارید. فرق این روش با وب‌هوک را در ربات تلگرام هوش مصنوعی با وب‌سرویس باز کرده‌ایم؛ برای ربات نرخ، همین روش ساده کافی است. فقط باید برنامه بعد از قطع اتصال SSH و بعد از ری‌استارت سرور هم روشن بماند، و systemd همین کار را می‌کند:

# /etc/systemd/system/fxbot.service
[Unit]
Description=Telegram dollar-rate bot
After=network-online.target

[Service]
WorkingDirectory=/opt/fxbot
EnvironmentFile=/opt/fxbot/.env
ExecStart=/opt/fxbot/.venv/bin/python bot.py
Restart=always
User=fxbot

[Install]
WantedBy=multi-user.target

فایل .env همان متغیرها را بدون کلمهٔ export دارد، هر کدام در یک خط (NETARZ_FX_KEY=fx-ntz-v1-...). بعد با systemctl enable --now fxbot ربات را روشن کنید و با journalctl -u fxbot -f گزارشش را ببینید. دسترسی این فایل را فقط به کاربر ربات بدهید، چون هر دو کلید داخلش است.

اگر ربات جواب نداد

• origin_required یا ip_not_allowed در گزارش: IP سرور در فهرست مجاز اپ نیست یا درخواست از نشانی دیگری رفته. همان نشانی‌ای را که در پاسخ خطا آمده اضافه کنید.
• domain_not_verified: مالکیت دامنهٔ اپ هنوز تأیید نشده. رکورد DNS یا فایل تأیید را بگذارید و در پنل «بررسی و تأیید» را بزنید.
• invalid_api_key: کلید ناقص کپی شده یا کلید را در پنل عوض کرده‌اید. کلید تازه بسازید و فایل .env را به‌روز کنید.
• ربات اصلاً پیام‌ها را نمی‌بیند: اگر در گزارش خطای شبکه (NetworkError) هست، سرور به تلگرام نمی‌رسد. این مشکل به سرویس نرخ ارز ربطی ندارد و با جابه‌جایی سرور حل می‌شود.

قدم بعدی

همین کلید در جاهای دیگر هم به کارتان می‌آید. اگر همین نرخ‌ها را در صفحه‌گسترده می‌خواهید، نرخ دلار در گوگل شیت و اکسل فرمول آماده دارد، و برای سایت، نمایش نرخ ارز زنده در سایت سه روش را مقایسه کرده است. نمونه‌های سمت سرور در زبان‌های دیگر هم در API نرخ ارز در PHP، پایتون و Node.js آمده‌اند. اگر بعداً خواستید ربات به پرسش‌های آزاد هم جواب بدهد، ربات تلگرام فارسی با API هوش مصنوعی از همین نقطه ادامه می‌دهد.