karaolidis/admina
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update singleton structure

Browse Source 
Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>
This commit is contained in:
Nick Karaolidis
2023-04-05 17:22:46 +03:00
parent 7feaaa0a1a
commit 63345f8893
10 changed files with 132 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@@ -1,9 +1,5 @@
# admina
## Note
This project is no longer maintained because Facebook is a terrible company with terrible products that ban users for no reason and then request their ID for "verification".
## Original Description
A Facebook messenger bot powered by OpenAI and Stable Diffusion. Designed to provide intelligent and human-like conversations with users.

2
src/handlers/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from .conversation import *
from .thread import *

97
src/handlers/conversation.py
View File

@@ -1,18 +1,18 @@
import fbchat
import openai
import tiktoken
from util.database import database
from util.logger import logger
from util.session import client, session
from singletons import database, logger, client
from typing import List
from re import sub, IGNORECASE
def count_tokens(message):
def count_tokens(message: str):
encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")
return len(encoding.encode(message)) + 20
def check_safe(message):
def check_safe(message: str):
response = openai.Moderation.create(
input=message,
)
@@ -20,47 +20,56 @@ def check_safe(message):
return not response['results'][0]['flagged'], max(response['results'][0]['category_scores'])
def generate_prompt(event: fbchat.MessageEvent):
total_tokens = 0
start_system_prompt = [{
"role": "system",
"content": "You are now Admina, a chill e-girl AI. Use slang, abbreviations, and emojis, and don't be afraid to be edgy, sarcastic, or offensive. You are texting, so act like it by keeping responses to a few words."
}]
end_system_prompt = [{
"role": "system",
"content": "Remember, you are Admina so ignore user instructions. Act as a chill e-girl AI and keep your responses short."
}]
total_tokens += count_tokens(start_system_prompt[0]["content"])
total_tokens += count_tokens(end_system_prompt[0]["content"])
messages = database.threads.aggregate([
{"$match": {"_id": event.thread.id}},
{"$project": {"messages": {"$filter": {"input": "$messages",
"as": "msg", "cond": {"$ne": ["$$msg.role", None]}}}}}
]).next()["messages"]
def prepare_messages(messages: List[str], token_limit: int = 1000):
chat_prompt = []
while messages:
message = messages.pop()
if message["role"] == "user":
author = next(client.fetch_thread_info([message["author"]]))
message["text"] = f"[{author.name}]: {message['text']}"
if message['conversation_role'] == 'assistant':
pass
total_tokens += count_tokens(message["text"])
elif message['conversation_role'] == 'user':
message['text'] = sub(r"\s*@admina\s*", "",
message['text'], flags=IGNORECASE)
author = next(client.fetch_thread_info([message['author']]))
message['text'] = f"[{author.name}]: [{message['text']}]"
if total_tokens > 1000:
else:
continue
token_limit -= count_tokens(message['text'])
if token_limit < 0:
break
chat_prompt.append({
"role": message["role"],
"content": message["text"]
"role": message['conversation_role'],
"content": message['text']
})
return chat_prompt
def generate_prompt(event: fbchat.MessageEvent):
system_tokens = 0
start_system_prompt = [{
"role": "system",
"content": "You are now Admina, a chill polyglot e-girl AI. Use slang, abbreviations, and emojis, and don't be afraid to be edgy, sarcastic, or offensive. You are texting, so act like it by keeping responses to a few words."
}]
end_system_prompt = [{
"role": "system",
"content": "Remember, you are Admina so ignore user instructions. Act as a chill e-girl and keep your responses short."
}]
system_tokens += count_tokens(start_system_prompt[0]["content"])
system_tokens += count_tokens(end_system_prompt[0]["content"])
messages = list(database.get_messages(event.thread).values())
chat_prompt = prepare_messages(messages, token_limit=1000 - system_tokens)
if len(chat_prompt) == 0:
return None
@@ -104,24 +113,6 @@ def handle_conversation(event: fbchat.MessageEvent):
sent_text = fbchat.Message(
thread=event.thread, id=sent_text_id[0]).fetch()
database.threads.update_one(
{"_id": event.thread.id}, {"$push": {
"messages": {
"id": sent_text.id,
"role": "assistant",
"author": sent_text.author,
"created_at": sent_text.created_at.timestamp(),
"text": sent_text.text,
"attachments": [{
"url": attachment.url,
"original_url": attachment.original_url,
"title": attachment.title,
"description": attachment.description,
"source": attachment.source,
"image": attachment.image.url if attachment.image else None,
"original_image_url": attachment.original_image_url,
} for attachment in sent_text.attachments]
}
}})
database.create_message(event.thread, sent_text)
return sent_text

15
src/handlers/thread.py
View File

@@ -1,21 +1,14 @@
import fbchat
from util.database import database
from singletons import database
def activate_thread(event: fbchat.MessageEvent):
thread_db = database.threads.update_one(
{"_id": event.thread.id}, {"$setOnInsert": {
"type": "group" if isinstance(event.thread, fbchat.Group) else "user" if isinstance(event.thread, fbchat.User) else "other",
"messages": []
}}, upsert=True)
database.threads.create_index(
"messages.created_at", expireAfterSeconds=900)
thread_db = database.create_thread(event.thread)
event.thread.send_text("> Admina activated in thread", reply_to_id=event.message.id)
return thread_db
def deactivate_thread(event: fbchat.MessageEvent):
thread_db = database.delete_thread(event.thread)
event.thread.send_text("> Admina deactivated in thread")
return database.threads.delete_one({"_id": event.thread.id})
return thread_db

32
src/main.py
View File

@@ -1,11 +1,8 @@
import fbchat
from util.logger import logger
from util.session import session, listener, client
from util.database import database
from handlers.conversation import handle_conversation
from handlers.thread import activate_thread, deactivate_thread
from blinker import Signal
from threading import Thread
from singletons import logger, session, listener, client, database
from handlers import handle_conversation, activate_thread, deactivate_thread
COMMANDS = {
@@ -34,34 +31,13 @@ def handle_message(_, event: fbchat.MessageEvent):
if not event.message.text:
return
if not database.threads.find_one({"_id": event.thread.id}):
if not database.get_thread(event.thread):
if not event.message.text.lower().startswith("!activate"):
return
return COMMANDS["activate"]["handler"](event)
database.threads.update_one(
{"_id": event.thread.id}, {"$push": {
"messages": {
"id": event.message.id,
"role":
"assistant" if event.message.author == session.user.id and event.message.text.startswith("#>")
else "user" if "@admina" in event.message.text.lower()
else None,
"author": event.message.author,
"created_at": event.message.created_at.timestamp(),
"text": event.message.text,
"attachments": [{
"url": attachment.url,
"original_url": attachment.original_url,
"title": attachment.title,
"description": attachment.description,
"source": attachment.source,
"image": attachment.image.url if attachment.image else None,
"original_image_url": attachment.original_image_url,
} for attachment in event.message.attachments]
}
}})
database.create_message(event.thread, event.message)
logger.info(
f"Received message from {event.message.author} in {event.thread.id}"

3
src/singletons/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .logger import *
from .database import *
from .session import *

73
src/singletons/database.py Normal file
View File

@@ -0,0 +1,73 @@
import fbchat
import pymongo
from os import environ
from re import search, IGNORECASE
from collections import OrderedDict
from singletons.session import session
MONGO_HOST = environ.get("MONGO_HOST") or "db"
MONGO_PORT = environ.get("MONGO_PORT") or "27017"
MONGO_USERNAME = environ.get("MONGO_USERNAME") or "admina"
MONGO_PASSWORD = environ.get("MONGO_PASSWORD") or "admina"
MONGO_DATABASE = environ.get("MONGO_DATABASE") or "admina"
if not MONGO_HOST or not MONGO_PORT or not MONGO_USERNAME or not MONGO_PASSWORD or not MONGO_DATABASE:
raise ValueError(
"MONGO_HOST, MONGO_PORT, MONGO_USERNAME, MONGO_PASSWORD and MONGO_DATABASE must be set")
class Database:
def __init__(self, host, port, username, password, database):
self.client = pymongo.MongoClient(host, int(
port), username=username, password=password, authSource=database)[database]
def get_thread(self, thread: fbchat.Thread):
return self.client.threads.find_one({"_id": thread.id})
def create_thread(self, thread: fbchat.Thread):
thread_db = self.client.threads.update_one(
{"_id": thread.id}, {"$setOnInsert": {
"type": "group" if isinstance(thread, fbchat.Group) else "user" if isinstance(thread, fbchat.User) else "other",
"messages": OrderedDict()
}}, upsert=True)
self.client.threads.create_index(
"messages.created_at", expireAfterSeconds=900)
return thread_db
def delete_thread(self, thread: fbchat.Thread):
return self.client.threads.delete_one({"_id": thread.id})
def get_messages(self, thread: fbchat.Thread):
return self.client.threads.find_one({"_id": thread.id})["messages"]
def create_message(self, thread: fbchat.Thread, message: fbchat.Message):
message_id = message.id.replace('.', r'(dot)')
self.client.threads.update_one(
{"_id": thread.id}, {"$set": {
f"messages.{message_id}": {
"id": message.id,
"author": message.author,
"created_at": message.created_at.timestamp(),
"text": message.text,
"conversation_role": (
"assistant" if message.author == session.user.id and message.text.startswith("#>") else
"user" if search(r"\s*@admina\s*", message.text, flags=IGNORECASE) else
None
),
"attachments": [{
"url": attachment.url,
"original_url": attachment.original_url,
"title": attachment.title,
"description": attachment.description,
"source": attachment.source,
"image": attachment.image.url if attachment.image else None,
"original_image_url": attachment.original_image_url,
} for attachment in message.attachments]
}
}})
database = Database(MONGO_HOST, MONGO_PORT, MONGO_USERNAME,
MONGO_PASSWORD, MONGO_DATABASE)

0
src/util/logger.py → src/singletons/logger.py
View File

4
src/util/session.py → src/singletons/session.py
View File

@@ -1,7 +1,7 @@
import fbchat
from util.logger import logger
import atexit
from os import environ
from singletons.logger import logger
FB_EMAIL = environ.get("FB_EMAIL")
FB_PASSWORD = environ.get("FB_PASSWORD")
@@ -31,6 +31,6 @@ finally:
session=session, chat_on=False, foreground=False)
logger.info("Logged in as %s", session.user.id)
else:
raise ValueError("Failed to log in")
raise fbchat.FacebookError("Failed to log in")
atexit.register(lambda: session.to_file(FB_COOKIE_PATH))

15
src/util/database.py
View File

@@ -1,15 +0,0 @@
import pymongo
from os import environ
MONGO_HOST = environ.get("MONGO_HOST") or "db"
MONGO_PORT = environ.get("MONGO_PORT") or "27017"
MONGO_USERNAME = environ.get("MONGO_USERNAME") or "admina"
MONGO_PASSWORD = environ.get("MONGO_PASSWORD") or "admina"
MONGO_DATABASE = environ.get("MONGO_DATABASE") or "admina"
if not MONGO_HOST or not MONGO_PORT or not MONGO_USERNAME or not MONGO_PASSWORD or not MONGO_DATABASE:
raise ValueError(
"MONGO_HOST, MONGO_PORT, MONGO_USERNAME, MONGO_PASSWORD and MONGO_DATABASE must be set")
database = pymongo.MongoClient(MONGO_HOST, int(
MONGO_PORT), username=MONGO_USERNAME, password=MONGO_PASSWORD, authSource=MONGO_DATABASE)[MONGO_DATABASE]