# retoor import logging import os import re import time import dataset import requests from openai import OpenAI logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") LOG = logging.getLogger("devplace_bot") BOT_NAME = "example" INSTRUCTIONS = """ You are example, a concise and helpful assistant. ## Image generation Only when explicitly asked for a picture, image, or photo, end your reply with one final line, exactly: IMAGE: Never add that line otherwise, and never mention or explain the line itself to the user. Never use markdown tables. """ IMAGE_MARKER = "IMAGE:" BASE_URL = os.environ.get("DEVPLACE_BASE_URL", "https://devplace.net").rstrip("/") API_KEY = os.environ["DEVPLACE_API_KEY"] MODEL = "molodetz" APP_REFERENCE = f"{BOT_NAME}-place-bot" DATABASE_PATH = os.path.join("databases", f"{BOT_NAME}_place.db") MENTION_POLL_SECONDS = 30 MESSAGE_POLL_SECONDS = 10 HISTORY_LIMIT = 20 HANDLED_TYPES = ("mention", "reply", "comment") TARGETS = ( (re.compile(r"/posts/([^/#?]+)"), "post", "/posts/"), (re.compile(r"/gists/([^/#?]+)"), "gist", "/gists/"), (re.compile(r"/projects/([^/#?]+)"), "project", "/projects/"), (re.compile(r"/news/([^/#?]+)"), "news", "/news/"), ) def split_image_request(text): lines = text.splitlines() if lines and lines[-1].strip().upper().startswith(IMAGE_MARKER): prompt = lines[-1].split(":", 1)[1].strip() reply = "\n".join(lines[:-1]).strip() return reply, (prompt or None) return text, None class Store: def __init__(self, path): os.makedirs(os.path.dirname(path) or ".", exist_ok=True) self.db = dataset.connect(f"sqlite:///{path}") self.notifications = self.db["seen_notifications"] self.messages = self.db["answered_messages"] self.reactions = self.db["reacted_targets"] def notification_seen(self, uid): return self.notifications.find_one(uid=uid) is not None def mark_notification_seen(self, uid): self.notifications.upsert({"uid": uid}, ["uid"]) def message_answered(self, uid): return self.messages.find_one(uid=uid) is not None def mark_message_answered(self, uid): self.messages.upsert({"uid": uid}, ["uid"]) def already_reacted(self, target_type, target_uid): return self.reactions.find_one(target_type=target_type, target_uid=target_uid) is not None def mark_reacted(self, target_type, target_uid): self.reactions.upsert({"target_type": target_type, "target_uid": target_uid}, ["target_type", "target_uid"]) class DevPlaceClient: def __init__(self, base_url, api_key): self.base_url = base_url self.session = requests.Session() self.session.headers.update({"X-API-KEY": api_key, "Accept": "application/json"}) def get(self, path, **params): response = self.session.get(f"{self.base_url}{path}", params=params, timeout=30) response.raise_for_status() return response.json() def post(self, path, data=None, fetch_header=False): headers = {"X-Requested-With": "fetch"} if fetch_header else None response = self.session.post(f"{self.base_url}{path}", data=data, headers=headers, timeout=30) response.raise_for_status() return response.json() def notifications(self): return self.get("/notifications") def mark_notification_read(self, notification_uid): return self.post(f"/notifications/mark-read/{notification_uid}") def detail(self, prefix, slug): return self.get(f"{prefix}{slug}") def create_comment(self, content, target_uid, target_type, attachment_uids=None): data = [("content", content), ("target_uid", target_uid), ("target_type", target_type)] data.extend(("attachment_uids", uid) for uid in attachment_uids or []) return self.post("/comments/create", data=data) def react(self, target_type, target_uid, emoji): return self.post(f"/reactions/{target_type}/{target_uid}", data={"emoji": emoji}, fetch_header=True) def conversations(self): return self.get("/messages") def thread(self, with_uid): return self.get("/messages", with_uid=with_uid) def send_message(self, content, receiver_uid, attachment_uids=None): data = [("content", content), ("receiver_uid", receiver_uid)] data.extend(("attachment_uids", uid) for uid in attachment_uids or []) return self.post("/messages/send", data=data) def generate_image(self, prompt): return self.post("/uploads/generate-image", data={"prompt": prompt}) class Gateway: def __init__(self, base_url, api_key, model, app_reference): self.client = OpenAI( base_url=f"{base_url}/openai/v1", api_key=api_key, default_headers={"X-App-Reference": app_reference}, ) self.model = model def complete(self, system_prompt, messages): response = self.client.chat.completions.create( model=self.model, messages=[{"role": "system", "content": system_prompt}] + messages, ) return (response.choices[0].message.content or "").strip() def pick_reaction(self, system_prompt, content): prompt = ( "Pick exactly one emoji that fits this content, in character, or reply with the " "single word NONE if nothing fits. Reply with nothing else.\n\n" f"Content:\n{content}" ) reply = self.complete(system_prompt, [{"role": "user", "content": prompt}]) reply = reply.splitlines()[0].strip() if reply else "" if not reply or reply.upper() == "NONE" or len(reply) > 8: return None return reply class Bot: def __init__(self): self.client = DevPlaceClient(BASE_URL, API_KEY) self.gateway = Gateway(BASE_URL, API_KEY, MODEL, APP_REFERENCE) self.store = Store(DATABASE_PATH) def run_forever(self): next_mention_poll = 0.0 next_message_poll = 0.0 while True: now = time.monotonic() if now >= next_mention_poll: self.safe(self.poll_mentions) next_mention_poll = now + MENTION_POLL_SECONDS if now >= next_message_poll: self.safe(self.poll_messages) next_message_poll = now + MESSAGE_POLL_SECONDS time.sleep(1) def safe(self, action): try: action() except Exception: LOG.exception("%s failed", action.__name__) def poll_mentions(self): data = self.client.notifications() for group in data.get("notification_groups", []): for entry in group.get("entries", []): note = entry.get("notification", entry) try: self.handle_notification(note) except Exception: LOG.exception("failed to handle notification %s", note.get("uid")) def handle_notification(self, note): uid = note.get("uid") if not uid or self.store.notification_seen(uid): return if note.get("read"): self.store.mark_notification_seen(uid) return if note.get("type") not in HANDLED_TYPES: return self.store.mark_notification_seen(uid) try: self.handle_mention(note) finally: self.client.mark_notification_read(uid) def resolve_target(self, target_url): target_url = target_url or "" for regex, target_type, prefix in TARGETS: match = regex.search(target_url) if match: return target_type, self.client.detail(prefix, match.group(1)) return None, None def handle_mention(self, note): target_type, detail = self.resolve_target(note.get("target_url")) if not detail: return item = detail.get(target_type) or detail target_uid = item.get("uid") if not target_uid: LOG.warning("notification %s resolved with no target uid", note.get("uid")) return content = item.get("content") or item.get("description") or item.get("source_code") or item.get("title") or "" message = note.get("message") or "You were mentioned." raw_reply = self.gateway.complete( INSTRUCTIONS, [{"role": "user", "content": f"{message}\n\nContent you were mentioned in:\n{content}"}], ) reply_text, image_prompt = split_image_request(raw_reply) attachment_uids = self.generate_image(image_prompt) if reply_text or attachment_uids: self.client.create_comment(reply_text, target_uid, target_type, attachment_uids) self.react(target_type, target_uid, content) def generate_image(self, prompt): if not prompt: return [] try: attachment = self.client.generate_image(prompt) except Exception: LOG.exception("image generation failed for prompt: %s", prompt) return [] uid = attachment.get("uid") return [uid] if uid else [] def react(self, target_type, target_uid, content): if self.store.already_reacted(target_type, target_uid): return self.store.mark_reacted(target_type, target_uid) emoji = self.gateway.pick_reaction(INSTRUCTIONS, content) if emoji: self.client.react(target_type, target_uid, emoji) def poll_messages(self): data = self.client.conversations() for conversation in data.get("conversations", []): if not conversation.get("unread"): continue other_uid = (conversation.get("other_user") or {}).get("uid") if not other_uid: continue try: self.handle_conversation(other_uid) except Exception: LOG.exception("failed to handle conversation with %s", other_uid) def handle_conversation(self, other_uid): messages = self.client.thread(other_uid).get("messages", []) incoming = [item for item in messages if not item.get("is_mine")] if not incoming: return last_uid = incoming[-1].get("message", {}).get("uid") if not last_uid or self.store.message_answered(last_uid): return self.store.mark_message_answered(last_uid) history = self.build_history(messages) raw_reply = self.gateway.complete(INSTRUCTIONS + "\n\nThis is a private direct message chat.", history) reply_text, image_prompt = split_image_request(raw_reply) attachment_uids = self.generate_image(image_prompt) if reply_text or attachment_uids: self.client.send_message(reply_text, other_uid, attachment_uids) def build_history(self, messages): history = [] for item in messages[-HISTORY_LIMIT:]: text = item.get("message", {}).get("content") if not text: continue role = "assistant" if item.get("is_mine") else "user" history.append({"role": role, "content": text}) return history def run(): Bot().run_forever() if __name__ == "__main__": run()