post_x.py

import sys, os, time, uuid, hmac, hashlib, base64, urllib.parse, urllib.request, json

def percent_encode(s):
    return urllib.parse.quote(str(s), safe='')

def oauth_header(method, url, extra_params, key, secret, token, token_secret):
    oauth_params = {
        'oauth_consumer_key': key,
        'oauth_nonce': uuid.uuid4().hex,
        'oauth_signature_method': 'HMAC-SHA1',
        'oauth_timestamp': str(int(time.time())),
        'oauth_token': token,
        'oauth_version': '1.0',
    }
    params = {**oauth_params, **extra_params}
    param_str = '&'.join(f"{percent_encode(k)}={percent_encode(params[k])}" for k in sorted(params))
    base_str = '&'.join([method.upper(), percent_encode(url), percent_encode(param_str)])
    signing_key = f"{percent_encode(secret)}&{percent_encode(token_secret)}"
    signature = base64.b64encode(hmac.new(signing_key.encode(), base_str.encode(), hashlib.sha1).digest()).decode()
    oauth_params['oauth_signature'] = signature
    return 'OAuth ' + ', '.join(f'{percent_encode(k)}="{percent_encode(v)}"' for k, v in sorted(oauth_params.items()))

def upload_media(image_path, keys):
    url = 'https://upload.twitter.com/1.1/media/upload.json'
    with open(image_path, 'rb') as f:
        data = f.read()
    b64data = base64.b64encode(data).decode()
    header = oauth_header('POST', url, {}, keys['key'], keys['secret'], keys['token'], keys['token_secret'])
    boundary = uuid.uuid4().hex
    body = (f'--{boundary}\r\nContent-Disposition: form-data; name="media_data"\r\n\r\n{b64data}\r\n'
            f'--{boundary}--\r\n').encode()
    req = urllib.request.Request(url, data=body, method='POST')
    req.add_header('Authorization', header)
    req.add_header('Content-Type', f'multipart/form-data; boundary={boundary}')
    try:
        with urllib.request.urlopen(req) as resp:
            result = json.loads(resp.read())
    except urllib.error.HTTPError as e:
        print('MEDIA_UPLOAD_ERROR', e.code, e.read().decode())
        raise
    return result['media_id_string']

def post_tweet(text, media_id, keys):
    url = 'https://api.twitter.com/2/tweets'
    header = oauth_header('POST', url, {}, keys['key'], keys['secret'], keys['token'], keys['token_secret'])
    payload = {'text': text}
    if media_id:
        payload['media'] = {'media_ids': [media_id]}
    body = json.dumps(payload).encode()
    req = urllib.request.Request(url, data=body, method='POST')
    req.add_header('Authorization', header)
    req.add_header('Content-Type', 'application/json')
    try:
        with urllib.request.urlopen(req) as resp:
            result = json.loads(resp.read())
    except urllib.error.HTTPError as e:
        print('TWEET_POST_ERROR', e.code, e.read().decode())
        raise
    return result

if __name__ == '__main__':
    image_path = sys.argv[1]
    text_path = sys.argv[2]
    with open(text_path, encoding='utf-8') as f:
        text = f.read()
    keys = {
        'key': os.environ['X_API_KEY'],
        'secret': os.environ['X_API_SECRET'],
        'token': os.environ['X_ACCESS_TOKEN'],
        'token_secret': os.environ['X_ACCESS_SECRET'],
    }
    media_id = upload_media(image_path, keys) if image_path != '-' else None
    result = post_tweet(text, media_id, keys)
    print(json.dumps(result))