bluebubbles-bot/bin/bluebubbles_bot

#!python3

import os
import sys
import yaml
import uuid
import logging
import argparse
import persona
from typing import List
from dataclasses import dataclass
import datetime
import requests
from fastapi import FastAPI
import uvicorn

class LoggingFormatter(logging.Formatter):
	def format(self, record):
		module_max_width = 30
		datefmt='%Y/%m/%d/ %H:%M:%S'
		level = f'[{record.levelname}]'.ljust(9)
		if 'log_module' not in dir(record) :
			modname = str(record.module)+'.'+str(record.name)
		else :
			modname = record.log_module
		modname = (f'{modname}'[:module_max_width-1] + ']').ljust(module_max_width)
		final = "%-7s %s [%s %s" % (self.formatTime(record, self.datefmt), level, modname, record.getMessage())
		return final

bot = FastAPI()

if __name__ == '__main__':

	# Command-line client
	# Define constants
	config_template = {'bluebubbles_bot': {}}

	# Gather Argument options
	EXAMPLE_TEXT='Example:\n\tbluebubbles_bot -h'
	parser = argparse.ArgumentParser(epilog=EXAMPLE_TEXT,formatter_class=argparse.RawDescriptionHelpFormatter)
	parser.add_argument('-l', '--log', action='store', help='Specify a file to log to.')
	parser.add_argument('-v', '--verbose', action='count', help='Include verbose information in the output. Add \'v\'s for more output.',default=0)
	args = parser.parse_args()

	log = logging.getLogger(__name__)
	log = logging.LoggerAdapter(log,{'log_module':'bluebubbles_bot'})

	# Configure logging
	log_options = [logging.ERROR, logging.WARNING, logging.INFO, logging.DEBUG]
	if not args.verbose :
		args.verbose = 0
	if args.verbose > 3 :
		args.verbose = 3
	if args.log :
		logging.basicConfig(level=log_options[args.verbose],filename=args.log)
		logging.getLogger().addHandler(logging.StreamHandler(sys.stderr))
	else :
		logging.basicConfig(level=log_options[args.verbose])
	logging.getLogger().handlers[0].setFormatter(LoggingFormatter())
	logging.propagate=True

	# Check for missing environment variables
	for required_var in ['BB_SERVER_URL','BB_SERVER_PASSWORD','BIND_PORT'] :
		if required_var not in os.environ.keys() :
			log.error(f'Missing required ENV variable {required_var}')
			exit(1)

	def send_message(message) :
		"""Send a message to the server, optionally with attachments."""
		# Use the private API instead of applescript
		if 'USE_PRIVATE_API' in os.environ.keys() and os.environ['USE_PRIVATE_API'].lower() in ['true', 'yes'] :
			method = 'private-api'
		else :
			method = 'apple-script'
		uid = str(uuid.uuid4()).upper()
		text = message.text
		chat_guid = message.chat_identifier
		url = os.environ['BB_SERVER_URL'].rstrip('/') + '/api/v1/message/text'
		params = {'password': os.environ['BB_SERVER_PASSWORD']}
		effect_id = ''
		subject = ''
		if 'effectId' in message.meta.keys() :
			effect_id = message.meta['effectId']
		if 'subject' in message.meta.keys() :
			subject = message.meta['subject']
		payload = {
			'chatGuid': chat_guid,
			'message': text,
			'tempGuid': uid,
			'method': method,
			'effectId': effect_id,
			'subject': subject,
			'selectedMessageGuid': ''
		}
		if len(message.attachments) > 0 :
			payload.update({'name': uid})
			payload.pop('text', None)
			attachments = []
			for attachment in message.attachments :
				file = {'file': (uid, message.attachments[attachment].data, message.attachments[attachment].mime_type)}
				attachments.append(file)
			payload.update({'attachments':attachments})
			url = os.environ['BB_SERVER_URL'].rstrip('/') + '/api/v1/message/attachment'
		requests.post(url,params=params,json=payload)

	def get_full_attachments(message: persona.Message) -> persona.Message :
		"""Given a message with basic attachment descriptions, fetch the full attachment payloads from BlueBubbles"""
		for attachment in message.attachments :
			params = {'password': os.environ['BB_SERVER_PASSWORD']}
			url = os.environ['BB_SERVER_URL'].rstrip('/') + '/api/v1/attachment/' + attachment.data
			# log.debug(requests.get(url,params=params).text)
			url = os.environ['BB_SERVER_URL'].rstrip('/') + '/api/v1/attachment/' + attachment.data + '/download'
			content = requests.get(url,params=params).content
			attachment.data = content
		return message
	# Create persona instance
	current_persona = persona.Persona()

	# Create a fastAPI instance
	@bot.post('/message')
	async def message(content: dict):
		log.debug(content)
		if content['type'] == 'new-message' :
			message = content['data']
			# Determine sender and receiver
			if message['isFromMe'] :
				sender = message['handle']['address']
				recipients = []
			else :
				sender = None
				recipients = [message['handle']['address']]
			# Resolve attachments
			attachments = []
			for attachment in message['attachments'] :
				attachments.append(persona.Attachment(mime_type=attachment['mimeType'],data=attachment['guid']))
			# Get the date sent
			date_sent = datetime.datetime.fromtimestamp(message['dateCreated']/1000)
			# Get any effects or subjects
			subject = ''
			effect_id = ''
			if 'subject' in message.keys() :
				subject = message['subject']
			if 'expressiveSendStyleId' in message.keys() :
				effect_id = message['expressiveSendStyleId']
			# Craft the message
			persona_message = persona.Message(
				text=message['text'],
				sender_identifier=sender,
				chat_identifier=message['chats'][-1]['guid'],
				identifier=message['guid'],
				timestamp=date_sent,
				recipients=recipients,
				attachments=attachments,
				meta={'subject': subject, 'effectId': effect_id}
				)
			persona_message = get_full_attachments(persona_message)
			responses = current_persona.receive_message(persona_message)
			for response in responses :
				prompt = persona_message
				send_message(response)
		return content

	bind_port = int(os.environ['BIND_PORT'])
	uvicorn.run(bot,host='0.0.0.0', port=bind_port)