#!/usr/bin/env python3
"""Python 3.10+, POSIX, standard library. Default: isolated OFFLINE simulation.

Operator-only --execute-once can incur charges and send ONE private reply. Use a
bot you own, dedicated to this example, with NO other poller anywhere. The local
lock cannot exclude remote pollers. Supply separate TELEGRAM_BOT_TOKEN and
API429_API_KEY secrets, plus exact API429_MODEL (initial example:
gemini-3.8-flash). Never put secrets in command arguments or source control.

This is at-most-once ATTEMPTS, not exactly-once delivery. Reserved/send_pending
state after a crash, timeout, bad response or uncertain write stops later runs.
Do not delete/reset that state to retry: an operator must reconcile the outcome
with both services before deciding how to proceed; a reply can be lost. No
automatic generation/send retries, splitting, webhook changes or pending-update
reset. max_tokens=256 is NOT a total spending cap or a billing guarantee.

Run offline: python3 api429-telegram-private-bot.py
Help: python3 api429-telegram-private-bot.py --help
Explicit operator command (secrets already exported securely):
  python3 api429-telegram-private-bot.py --execute-once --expected-bot-id 123 \
    --allowed-user-id 456 --allowed-chat-id 456 --state-dir /private/bot-state \
    --dedicated-bot-no-other-poller

One pass polls limit=1, timeout=0. Only terminal outcomes permit a future offset.
Telegram retains updates <=24h; after that age we omit the old terminal offset,
so a random lower update ID after a week can be processed. Empty polls do not
refresh handled_at. Uncertain states NEVER expire. No negative offsets.

Primary references checked 2026-10-24 16:20 UTC, Bot API 10.3:
https://core.telegram.org/bots/api#making-requests
https://core.telegram.org/bots/api#getme
https://core.telegram.org/bots/api#getwebhookinfo
https://core.telegram.org/bots/api#getupdates
https://core.telegram.org/bots/api#message
https://core.telegram.org/bots/api#sendmessage
API429 public OpenAPI/catalog checked 16:20:41/43 UTC:
https://api429.com/api/public/openapi.json
https://api429.com/models
Public publication is not proof of access under your key or successful inference.
GET /v1/models checks exact model visibility under your key, not future execution.
"""
import argparse
from dataclasses import dataclass
import fcntl
import json
import os
from pathlib import Path
import re
import stat
import sys
import tempfile
import time
from urllib.error import HTTPError
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener

MAX_RESPONSE_BYTES = 262144
MAX_TEXT_UNITS = 4096
MAX_AGE_SECONDS = 600
RETENTION_SECONDS = 86400


class SafeError(Exception):
    """Only static, non-sensitive messages may cross this boundary."""


def positive_int(value):
    return type(value) is int and 0 < value <= 2**53 - 1


def text_units(value):
    try:
        return len(value.encode('utf-16-le')) // 2
    except (UnicodeError, AttributeError):
        return MAX_TEXT_UNITS + 1


@dataclass(frozen=True)
class Config:
    bot_id: int
    user_id: int
    chat_id: int
    model: str

    def identity(self):
        if not all(positive_int(i) for i in (self.bot_id, self.user_id, self.chat_id)):
            raise SafeError('Expected positive private identity IDs.')
        if not isinstance(self.model, str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._:/-]{0,199}', self.model):
            raise SafeError('An exact model ID is required.')
        return {'bot_id': self.bot_id, 'user_id': self.user_id, 'chat_id': self.chat_id, 'model': self.model}


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise SafeError('Redirect refused.') from None


class HttpTransport:
    """Construct only in explicit real mode. No ambient proxies or redirects."""
    def __init__(self, telegram_token, api_key):
        if not re.fullmatch(r'[0-9]+:[A-Za-z0-9_-]+', telegram_token or ''):
            raise SafeError('Missing or invalid Telegram credential.')
        if not api_key or any(ord(c) < 33 or ord(c) > 126 for c in api_key):
            raise SafeError('Missing or invalid API429 credential.')
        self.telegram_token = telegram_token
        self.api_key = api_key
        self.opener = build_opener(ProxyHandler({}), NoRedirect())

    @staticmethod
    def decode(raw, service):
        try:
            if len(raw) > MAX_RESPONSE_BYTES:
                raise ValueError()
            value = json.loads(raw.decode('utf-8'))
            if not isinstance(value, dict):
                raise ValueError()
            if service == 'telegram':
                if value.get('ok') is not True or 'result' not in value:
                    raise ValueError()
                return value['result']
            return value
        except (ValueError, UnicodeError, TypeError, RecursionError):
            raise SafeError('Invalid or oversized service response.') from None

    def __call__(self, service, method, payload):
        telegram_methods = {'getMe', 'getWebhookInfo', 'getUpdates', 'sendMessage'}
        if service == 'telegram' and method in telegram_methods:
            url = 'https://api.telegram.org/bot' + self.telegram_token + '/' + method
            headers = {'Content-Type': 'application/json'}
            verb = 'POST'
        elif service == 'api429' and method in {'models', 'completions'}:
            url = 'https://gateway.api429.com/v1/' + ('models' if method == 'models' else 'chat/completions')
            headers = {'Authorization': 'Bearer ' + self.api_key, 'Content-Type': 'application/json'}
            verb = 'GET' if method == 'models' else 'POST'
        else:
            raise SafeError('Request method refused.')
        try:
            body = None if verb == 'GET' else json.dumps(payload, ensure_ascii=True).encode('utf-8')
            request = Request(url, data=body, headers=headers, method=verb)
            with self.opener.open(request, timeout=30) as response:
                if response.status != 200:
                    raise SafeError('Service request failed; outcome may be uncertain.')
                raw = response.read(MAX_RESPONSE_BYTES + 1)
            return self.decode(raw, service)
        except HTTPError as error:
            error.close()
            raise SafeError('Service request failed; outcome may be uncertain.') from None
        except SafeError:
            raise
        except Exception:
            # Never expose raw URLs (Telegram token), headers, bodies or causes.
            raise SafeError('Transport failed; outcome may be uncertain.') from None


class StateStore:
    """Directory-relative, no-follow private state, held under a local POSIX lock."""
    def __init__(self, path, identity):
        self.path = Path(path)
        self.identity = identity
        self.dir_fd = None
        self.lock_fd = None
        self.state = None

    @staticmethod
    def private_file(fd):
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) != 0o600:
            raise SafeError('State file ownership or permissions are unsafe.')

    def __enter__(self):
        try:
            # Walk each absolute path component without following even parent symlinks.
            absolute = os.path.abspath(self.path)
            parts = Path(absolute).parts[1:]
            if not parts:
                raise SafeError('A private state directory is required.')
            current = os.open('/', os.O_RDONLY | os.O_DIRECTORY)
            try:
                for index, part in enumerate(parts):
                    if index == len(parts)-1:
                        try: os.mkdir(part, 0o700, dir_fd=current)
                        except FileExistsError: pass
                        # Persist the directory entry before any attempt. Also
                        # sync an existing entry: an earlier failed sync may
                        # have left a directory visible only in memory.
                        os.fsync(current)
                    next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=current)
                    os.close(current)
                    current = next_fd
                self.dir_fd = current
                current = None
            finally:
                if current is not None: os.close(current)
            info = os.fstat(self.dir_fd)
            if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
                raise SafeError('State directory must be owned by you with mode 700.')
            self.lock_fd = os.open('lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600, dir_fd=self.dir_fd)
            self.private_file(self.lock_fd)
            fcntl.flock(self.lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            self.state = self.read()
            return self
        except SafeError:
            self.__exit__(None, None, None)
            raise
        except Exception:
            self.__exit__(None, None, None)
            raise SafeError('Cannot open or lock private state safely.') from None

    def read(self):
        try:
            fd = os.open('state.json', os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=self.dir_fd)
        except FileNotFoundError:
            return None
        try:
            self.private_file(fd)
            raw = os.read(fd, 8193)
            if len(raw)>8192: raise ValueError()
            value = json.loads(raw)
            required = {'format', 'version', 'identity', 'phase', 'update_id', 'handled_at'}
            if not isinstance(value, dict) or set(value) != required or value['format'] != 'api429-private-bot' or type(value['version']) is not int or value['version'] != 1:
                raise ValueError()
            stored_identity = value['identity']
            if not isinstance(stored_identity, dict) or set(stored_identity) != set(self.identity): raise ValueError()
            if any(type(stored_identity[key]) is not type(expected) or stored_identity[key] != expected for key, expected in self.identity.items()): raise ValueError()
            if not positive_int(value['update_id']): raise ValueError()
            if value['phase'] not in ('reserved', 'send_pending', 'terminal'): raise ValueError()
            handled = value['handled_at']
            if (value['phase']=='terminal' and not positive_int(handled)) or (value['phase']!='terminal' and handled is not None): raise ValueError()
            return value
        except SafeError:
            raise
        except Exception:
            raise SafeError('State is corrupt, incomplete or identity mismatched.') from None
        finally:
            os.close(fd)

    def write(self, phase, update_id, handled_at=None):
        value = {'format':'api429-private-bot', 'version':1, 'identity':self.identity, 'phase':phase, 'update_id':update_id, 'handled_at':handled_at}
        name = '.state-' + os.urandom(16).hex()
        fd = None
        try:
            # A stale temp file is harmless: replacement is allowed only under lock.
            fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=self.dir_fd)
            self.private_file(fd)
            raw = json.dumps(value, sort_keys=True).encode('utf-8')
            while raw:
                written = os.write(fd, raw)
                if written <= 0: raise OSError()
                raw = raw[written:]
            os.fsync(fd)
            os.close(fd)
            fd = None
            os.replace(name, 'state.json', src_dir_fd=self.dir_fd, dst_dir_fd=self.dir_fd)
            os.fsync(self.dir_fd)
            self.state = value
        except Exception:
            raise SafeError('Durable state write failed; stop and reconcile manually.') from None
        finally:
            if fd is not None: os.close(fd)
            # Do not clear an uncertain durable record or log its content.

    def __exit__(self, *args):
        if self.lock_fd is not None:
            os.close(self.lock_fd)
            self.lock_fd = None
        if self.dir_fd is not None:
            os.close(self.dir_fd)
            self.dir_fd = None


def eligible(update, config, now):
    # A positive update_id is checked separately before any terminal skip.
    if set(update) != {'update_id', 'message'}: return False
    message = update.get('message')
    if not isinstance(message, dict): return False
    # Fail closed for every unfamiliar Message context. Telegram may add fields;
    # review this allowlist rather than automatically widening accepted contexts.
    ordinary_fields = {'message_id', 'date', 'chat', 'from', 'text', 'entities'}
    if set(message) - ordinary_fields: return False
    sender, chat = message.get('from'), message.get('chat')
    if not isinstance(sender, dict) or not isinstance(chat, dict): return False
    if set(sender) - {'id', 'is_bot', 'first_name', 'last_name', 'username', 'language_code', 'is_premium', 'added_to_attachment_menu'}: return False
    if set(chat) - {'id', 'type', 'first_name', 'last_name', 'username'}: return False
    if not positive_int(sender.get('id')) or sender['id'] != config.user_id or sender.get('is_bot') is not False: return False
    if not positive_int(chat.get('id')) or chat['id'] != config.chat_id or chat.get('type') != 'private': return False
    if not positive_int(message.get('message_id')): return False
    date = message.get('date')
    if not positive_int(date) or not 0 <= now-date <= MAX_AGE_SECONDS: return False
    text = message.get('text')
    return isinstance(text, str) and bool(text.strip()) and text_units(text) <= MAX_TEXT_UNITS


def completion_text(value):
    if not isinstance(value, dict): raise SafeError('Completion response is invalid; no retry.')
    choices = value.get('choices')
    if not isinstance(choices, list) or len(choices)!=1 or not isinstance(choices[0], dict): raise SafeError('Completion response is invalid; no retry.')
    choice = choices[0]
    message = choice.get('message')
    if choice.get('finish_reason') != 'stop' or not isinstance(message, dict) or message.get('tool_calls') or message.get('function_call') or message.get('refusal'):
        raise SafeError('Completion is not a finished plain text reply; no retry.')
    text = message.get('content')
    if not isinstance(text,str) or not text.strip() or text_units(text)>MAX_TEXT_UNITS:
        raise SafeError('Reply is empty or exceeds the plain text limit; no retry.')
    return text


def run_once(config, state_dir, transport, clock=time.time):
    identity = config.identity()
    now = int(clock())
    if not positive_int(now): raise SafeError('Invalid clock.')
    with StateStore(state_dir, identity) as store:
        previous = store.state
        if previous is not None and previous['phase'] != 'terminal':
            raise SafeError('Uncertain pending state: manual reconciliation required; no retry.')
        if previous is not None and previous['handled_at']>now:
            raise SafeError('State clock is in the future; stop.')
        bot = transport('telegram','getMe',{})
        if not isinstance(bot,dict) or not positive_int(bot.get('id')) or bot['id']!=config.bot_id or bot.get('is_bot') is not True:
            raise SafeError('Bot identity does not match.')
        webhook = transport('telegram','getWebhookInfo',{})
        if not isinstance(webhook,dict) or webhook.get('url') != '':
            raise SafeError('Webhook active or unknown; no polling.')
        models = transport('api429','models',{})
        data = models.get('data') if isinstance(models,dict) else None
        if not isinstance(data,list) or not any(isinstance(model,dict) and model.get('id')==config.model for model in data):
            raise SafeError('Exact model is absent from the credential-scoped catalog.')
        recent = previous is not None and now-previous['handled_at'] < RETENTION_SECONDS
        payload = {'limit':1,'timeout':0,'allowed_updates':['message']}
        if recent: payload['offset'] = previous['update_id']+1
        updates = transport('telegram','getUpdates',payload)
        if not isinstance(updates,list) or len(updates)>1: raise SafeError('Invalid update batch; no acknowledgement.')
        if not updates: return 'empty'
        update = updates[0]
        if not isinstance(update,dict) or not positive_int(update.get('update_id')): raise SafeError('Unknown update ID; no acknowledgement.')
        update_id = update['update_id']
        if recent and update_id<=previous['update_id']: return 'duplicate'
        if not eligible(update,config,now):
            store.write('terminal',update_id,now)
            return 'skipped'
        store.write('reserved', update_id)
        reply = completion_text(transport('api429','completions',{'model':config.model,'messages':[{'role':'user','content':update['message']['text']}],'max_tokens':256,'stream':False}))
        store.write('send_pending',update_id)
        sent = transport('telegram','sendMessage',{'chat_id':config.chat_id,'text':reply})
        chat = sent.get('chat') if isinstance(sent,dict) else None
        if not isinstance(sent,dict) or not positive_int(sent.get('message_id')) or not isinstance(chat,dict) or not positive_int(chat.get('id')) or chat['id']!=config.chat_id or chat.get('type')!='private':
            raise SafeError('Send outcome is uncertain; manual reconciliation required.')
        store.write('terminal',update_id,int(clock()))
        return 'sent'


def offline_demo():
    # Read no environment here. No real HTTP object is constructed.
    fixture = Path(__file__).with_name('api429-telegram-update-synthetic.json')
    try: update = json.loads(fixture.read_text(encoding='utf-8'))
    except Exception: raise SafeError('Synthetic fixture is missing or invalid.') from None
    now = 1800000000
    config = Config(123,456,456,'gemini-3.8-flash')
    responses = {'getMe':{'id':123,'is_bot':True},'getWebhookInfo':{'url':''},'models':{'data':[{'id':config.model}]},'getUpdates':[update],'completions':{'choices':[{'finish_reason':'stop','message':{'content':'Synthetic offline reply.'}}]},'sendMessage':{'message_id':11,'chat':{'id':456,'type':'private'}}}
    def fake(service, method, payload): return responses[method]
    # macOS /var is a symlink: resolve the trusted freshly-created temp parent.
    with tempfile.TemporaryDirectory(prefix='api429-offline-') as directory:
        result = run_once(config, Path(directory).resolve()/'state', fake, lambda:now)
    print('OFFLINE synthetic cycle: ' + result + '; network requests: 0.')
    return 0


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--execute-once',action='store_true',help='Operator only: one real polling pass, possible charges and one reply.')
    parser.add_argument('--expected-bot-id',type=int)
    parser.add_argument('--allowed-user-id',type=int)
    parser.add_argument('--allowed-chat-id',type=int)
    parser.add_argument('--state-dir',type=Path)
    parser.add_argument('--dedicated-bot-no-other-poller',action='store_true')
    args = parser.parse_args(argv)
    try:
        if not args.execute_once: return offline_demo()
        if not args.dedicated_bot_no_other_poller or args.state_dir is None:
            raise SafeError('Explicit dedicated-bot acknowledgement and private state directory required.')
        config = Config(args.expected_bot_id,args.allowed_user_id,args.allowed_chat_id,os.environ.get('API429_MODEL',''))
        config.identity()
        transport = HttpTransport(os.environ.get('TELEGRAM_BOT_TOKEN',''),os.environ.get('API429_API_KEY',''))
        result = run_once(config,args.state_dir,transport)
        print('Operator pass: '+result+'.')
        return 0
    except SafeError as error:
        print('Stopped: '+str(error),file=sys.stderr)
        return 1
    except Exception:
        print('Stopped: unexpected local failure; reconcile state before proceeding.',file=sys.stderr)
        return 1


if __name__=='__main__':
    sys.exit(main())