builds DALL-E art from Craigslists posts & tweets 'em (https://twitter.com/mistconnectnbot)

craigsdalle.py

# craigsdalle - builds DALL-E art from Missed Connections posts & tweets
# jeffehobbs@gmail.com // November 2022

import asyncio
from pyppeteer import launch
import openai, tweepy, requests, configparser, os, shutil, hashlib
from mastodon import Mastodon

# set up API keys from external config apikeys.txt file
SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__))
config = configparser.ConfigParser()
config.read(SCRIPT_PATH +'/apikeys.txt')
OPENAI_APIKEY = config.get('apikeys', 'openai_apikey')
TWITTER_CONSUMER_KEY = config.get('twitter', 'consumer_key')
TWITTER_CONSUMER_SECRET = config.get('twitter', 'consumer_secret')
TWITTER_ACCESS_TOKEN = config.get('twitter', 'access_token')
TWITTER_ACCESS_TOKEN_SECRET = config.get('twitter', 'access_token_secret')
MASTODON_ACCESS_TOKEN = config.get('mastodon','access_token')

URL = 'https://westernmass.craigslist.org/search/mis#search=1~list~0~0' # base index URL to scrape
LIMIT = 9 # number of index posts to collect
INDEX_NUM = 0 # 0 is newest
PRE_PROMPT = 'A photo of ' # prefix to image generation text prompt
BLOCKLIST = ['DL']

# get content of index
async def get_CL_index(url):
    print("getting posts...")
    data = []
    browser = await launch({
       'executablePath':'/usr/bin/chromium'
    })
    page = await browser.newPage()
    await page.goto(url,{
        'waitUntil': 'networkidle0'}
    )
    posts = await page.querySelectorAll('.cl-search-result')
    for index, element in enumerate(posts):
        meta_el = await element.querySelector('.meta')
        href_el = await element.querySelector('.titlestring')
        title = await page.evaluate('(element) => element.title', element)
        meta = await page.evaluate('(meta_el) => meta_el.textContent', meta_el)
        href = await page.evaluate('(href_el) => href_el.href', href_el)
        post_id = hashlib.md5(str(href).encode('utf-8')).hexdigest()
        location = meta[1:].split('·')[0].lower().replace('(','').replace(')','').replace(', ma','')
        data.append({'title': title.strip(), 'url': href, 'id': post_id, 'location': location})
    return(data)

# get content of post
async def get_CL_article(url, title):
    browser = await launch({
       'executablePath':'/usr/bin/chromium'
    })
    page = await browser.newPage()
    await page.goto(url,{
        'waitUntil': 'networkidle0'}
    )
    content = await page.querySelectorAll('[id*="postingbody"]')
    for index, chunk in enumerate(content):
        text = await chunk.getProperty("textContent")
        #print(await text.jsonValue())
        post_text = await text.jsonValue()
        text_chunks = post_text.splitlines()
        full_content = title + ". "
        for text_chunk in text_chunks:
            if (not text_chunk.isspace()) and ("QR Code" not in text_chunk):
                full_content = full_content + str(text_chunk)
        return(full_content)

# generate image from post text
def get_openai_image(text, num_images):
    openai.api_key = OPENAI_APIKEY
    text = PRE_PROMPT + text.replace(".",",")
    response = openai.Image.create(prompt=text, n=num_images, size="1024x1024")
    image_url = response['data'][0]['url']
    return(image_url)

# tweet that stuff
def send_tweet(status, image_file_path, url):
    media_ids = []
    tweet = status + ' ' + url
    client = tweepy.Client(consumer_key=TWITTER_CONSUMER_KEY,
                           consumer_secret=TWITTER_CONSUMER_SECRET,
                           access_token=TWITTER_ACCESS_TOKEN,
                           access_token_secret=TWITTER_ACCESS_TOKEN_SECRET)
    auth = tweepy.OAuth1UserHandler(
        TWITTER_CONSUMER_KEY,
        TWITTER_CONSUMER_SECRET,
        TWITTER_ACCESS_TOKEN,
        TWITTER_ACCESS_TOKEN_SECRET,
    )
    api = tweepy.API(auth)
    media_upload_response = api.media_upload(image_file_path)
    media_ids.append(media_upload_response.media_id)

    if (len(status) > 256):
        status = status[:253] + "..."
    tweet_text = status + " " + url

    response = client.create_tweet(text=tweet, user_auth=True, media_ids=media_ids)
    return

def send_mastodon(status, image_file_path, url):
    mastodon = Mastodon(
        access_token = MASTODON_ACCESS_TOKEN,
        api_base_url = 'https://botsin.space/'
    )
    media = mastodon.media_post(image_file_path, description="Weather summary")
    mastodon.status_post(status, media_ids=media)
    return

# the plan? 
# get posts, get post content, check if post has been made before; if not, generate art & tweet it
def main():
    print("---")
    data = asyncio.get_event_loop().run_until_complete(get_CL_index(URL))
    content = asyncio.get_event_loop().run_until_complete(get_CL_article(data[0]['url'], data[0]['title']))
    for term in BLOCKLIST:
        if term in content:
            print('blocklisted')
            exit()
    print(f'content    : {content}')
    print("---")
    file_hash = hashlib.md5(str(data[0]['url']).encode('utf-8')).hexdigest()
    file_path = SCRIPT_PATH + '/output/' + file_hash + '.png'
    print(f"file path  : {file_path}")
    file_exists = os.path.isfile(file_path)
    if not file_exists:
        image_url = get_openai_image(str(content), 1)
        response = requests.get(image_url, stream=True)
        with open(file_path, 'wb') as out_file:
            shutil.copyfileobj(response.raw, out_file)
            del response
        send_tweet(content, file_path, data[0]['url'])
        send_mastodon(content, file_path, data[0]['url'])
    else:
        print('file exists!')
        exit()

if __name__ == '__main__':
    main()

#fin