Python ile SMS Kodu Alma

python3 -m venv .venv && source .venv/bin/activate
pip install requests          # HTTP client used for every API call
pip install flask             # only needed for the webhook receiver below
export TURONAY_API_KEY="paste-your-api-key-here"

import os
import requests

API = "https://turonay.com/stubs/handler_api.php"

resp = requests.get(
    API,
    params={"api_key": os.environ["TURONAY_API_KEY"], "action": "getBalance"},
    timeout=20,
)
text = resp.text.strip()            # plain text, not JSON: "ACCESS_BALANCE:125.4000"
if text.startswith("ACCESS_BALANCE:"):
    print("Balance (TRY):", text.split(":", 1)[1])
else:
    print("Error:", text)            # e.g. BAD_KEY

#!/usr/bin/env python3
"""sms_flow.py - buy a number, wait for the SMS code, finish or cancel.  pip install requests"""
import os
import sys
import time
from decimal import Decimal

import requests

API = "https://turonay.com/stubs/handler_api.php"
API_KEY = os.environ["TURONAY_API_KEY"]


class ApiError(Exception):
    """The API answered with a plain-text error code (NO_BALANCE, NO_NUMBERS, ...)."""


class Transient(ApiError):
    """Network trouble, HTTP 429 or 5xx, NO_CONNECTION: safe to retry later."""


def call(action, **params):
    query = {k: v for k, v in params.items() if v is not None}
    query.update(api_key=API_KEY, action=action)
    try:
        resp = requests.get(API, params=query, timeout=20)
    except requests.RequestException as exc:
        # Do not chain the original error: its message contains your api_key.
        raise Transient("network error: " + type(exc).__name__) from None
    if resp.status_code != 200:
        raise Transient("HTTP %d" % resp.status_code)
    text = resp.text.strip()            # always strip: bodies are plain text
    if text == "NO_CONNECTION":
        raise Transient(text)
    return text


def get_balance():
    text = call("getBalance")
    if not text.startswith("ACCESS_BALANCE:"):
        raise ApiError(text)
    return Decimal(text.split(":", 1)[1])


def get_number(service, country, max_price=None):
    """Returns (activation_id, phone). The price is only RESERVED here, not charged."""
    text = call("getNumber", service=service, country=country,
                maxPrice=None if max_price is None else "%.2f" % max_price)
    parts = text.split(":")
    if parts[0] == "ACCESS_NUMBER" and len(parts) >= 3 and parts[1] and parts[2]:
        return parts[1], parts[2]
    raise ApiError(text)    # NO_BALANCE, NO_NUMBERS, BAD_SERVICE, WRONG_MAX_PRICE:<price> ...


def get_status(activation_id):
    """Returns ("wait" | "ok" | "cancel", code_or_None)."""
    text = call("getStatus", id=activation_id)
    if text.startswith("STATUS_OK:"):
        return "ok", text.split(":", 1)[1]
    if text == "STATUS_CANCEL":
        return "cancel", None
    if text in ("STATUS_WAIT_CODE", "STATUS_WAIT_RESEND") or text.startswith("STATUS_WAIT_RETRY:"):
        return "wait", None    # WAIT_RETRY carries the OLD code: still waiting for the new one
    raise ApiError(text)       # NO_ACTIVATION, BAD_KEY ...


def set_status(activation_id, status):
    return call("setStatus", id=activation_id, status=status)


def request_another_sms(activation_id):
    return set_status(activation_id, 3)    # free, ACCESS_RETRY_GET


def finish(activation_id):
    return set_status(activation_id, 6)    # ACCESS_ACTIVATION, only valid after a code


def cancel(activation_id):
    return set_status(activation_id, 8)    # ACCESS_CANCEL, or EARLY_CANCEL_DENIED after a code


def wait_for_code(activation_id, timeout=1200, interval=5):
    """Poll every `interval` seconds for at most `timeout` seconds (20 minutes)."""
    deadline = time.monotonic() + timeout
    delay = interval
    while time.monotonic() < deadline:
        try:
            state, code = get_status(activation_id)
            delay = interval
            if state == "ok":
                return code
            if state == "cancel":
                raise ApiError("STATUS_CANCEL")
        except Transient:
            delay = min(delay * 2, 60)       # back off on HTTP 429 / network trouble
        time.sleep(max(0, min(delay, deadline - time.monotonic())))
    raise TimeoutError("no SMS within %d seconds" % timeout)


def main():
    service, country = "wa", "62"            # WhatsApp, Turkey (see getPrices / getCountries)
    try:
        print("Balance (TRY):", get_balance())
        activation_id, phone = get_number(service, country, max_price=15.00)
    except Transient as exc:
        sys.exit("Temporary problem, try again later: %s" % exc)
    except ApiError as exc:
        reason = str(exc)
        if reason.startswith("WRONG_MAX_PRICE"):
            print("Current price is above your limit:", reason.partition(":")[2] or "?")
        elif reason == "NO_NUMBERS":
            print("No numbers in stock right now, try again in a minute")
        elif reason == "NO_BALANCE":
            print("Not enough balance, top up first")
        else:
            print("API error:", reason)
        sys.exit(1)

    print("Use this number: %s (activation %s)" % (phone, activation_id))
    try:
        code = wait_for_code(activation_id)
    except TimeoutError:
        reply = cancel(activation_id)        # nothing is charged when no code arrived
        if reply != "EARLY_CANCEL_DENIED":
            sys.exit("Timed out, activation cancelled: %s" % reply)
        code = get_status(activation_id)[1]  # the code arrived in the last second
    except ApiError as exc:                  # STATUS_CANCEL: the supplier cancelled, nothing charged
        sys.exit("Activation cancelled: %s" % exc)

    print("SMS code:", code)
    print(finish(activation_id))             # ACCESS_ACTIVATION


if __name__ == "__main__":
    main()

import json


def get_number_v2(service, country, max_price=None):
    text = call("getNumberV2", service=service, country=country,
                maxPrice=None if max_price is None else "%.2f" % max_price)
    try:
        data = json.loads(text)
    except ValueError:
        raise ApiError(text)    # errors stay plain text, even in V2
    return data                 # activationId, phoneNumber, activationCost, activationEndTime ...


def get_status_v2(activation_id):
    """dict with the SMS once it arrived, otherwise the plain-text status (a str)."""
    text = call("getStatusV2", id=activation_id)
    try:
        data = json.loads(text)
    except ValueError:
        return text             # STATUS_WAIT_CODE, STATUS_WAIT_RETRY:<old>, STATUS_CANCEL ...
    sms = (data or {}).get("sms") or {}
    return {"code": sms.get("code"), "text": sms.get("text")}


order = get_number_v2("wa", "62", max_price=15.00)
print(order["activationId"], order["phoneNumber"], order["activationCost"], "until", order["activationEndTime"])

# The first code has arrived and you need a second SMS on the same number (free).
first_code = wait_for_code(activation_id)
reply = request_another_sms(activation_id)        # setStatus status=3 -> ACCESS_RETRY_GET
if reply == "ACCESS_RETRY_GET":
    # While the new SMS is on its way getStatus answers STATUS_WAIT_RETRY:<first code>,
    # which wait_for_code() treats as "still waiting".
    second_code = wait_for_code(activation_id)
    print("First:", first_code, "Second:", second_code)
finish(activation_id)                             # close only after the LAST code

# webhook_app.py   pip install flask   |   flask --app webhook_app run --port 8080
import hashlib
import hmac
import os
import time

from flask import Flask, request

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 64 * 1024            # events are tiny
SECRET = os.environ["TURONAY_WEBHOOK_SECRET"]           # whsec_... from your profile page
SEEN = set()                                            # demo only: use Redis/DB with a TTL


def verify_webhook_signature(secret, timestamp, raw_body, signature, tolerance=300):
    """sha256=HMAC_SHA256(secret, timestamp + "." + raw body), 5 minute replay window."""
    try:
        age = abs(time.time() - int(timestamp))
    except (TypeError, ValueError):
        return False
    if age > tolerance:
        return False
    mac = hmac.new(secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256)
    expected = ("sha256=" + mac.hexdigest()).encode()
    return hmac.compare_digest(expected, (signature or "").strip().encode())   # constant time


@app.post("/turonay/webhook")
def turonay_webhook():
    raw = request.get_data()            # the exact bytes, before any JSON parsing
    if not verify_webhook_signature(
        SECRET,
        request.headers.get("X-Turonay-Timestamp", ""),
        raw,
        request.headers.get("X-Turonay-Signature", ""),
    ):
        return "", 401
    event = request.get_json(force=True, silent=True)
    if not isinstance(event, dict):
        return "", 400
    if event.get("id") in SEEN:         # retries repeat the same delivery id
        return "", 200
    SEEN.add(event.get("id"))
    if event.get("event") == "code_received":
        print("code", event["code"], "for activation", event["activationId"], event["phone"])
    elif event.get("event") == "cancelled":
        print("activation", event["activationId"], "cancelled, nothing charged")
    return "", 200                      # answer 2xx within 5 seconds; do slow work elsewhere

import os

# Before: BASE_URL = "https://provider.example.com/stubs/handler_api.php"
BASE_URL = os.environ.get("SMS_API_URL", "https://turonay.com/stubs/handler_api.php")
API_KEY = os.environ["SMS_API_KEY"]        # your Turonay API key

# Libraries and bots with a base-URL option need only these two values, e.g.:
# client = SomeHandlerApiClient(api_key=API_KEY, base_url=BASE_URL)