devplacepy/devplacepy/services/bot/agent.py

# retoor <retoor@molodetz.nl>

import logging
import random

logger = logging.getLogger(__name__)


class BotAgentMixin:
    def _build_menu(self, page_state: dict) -> list[dict]:
        page = page_state["page"]
        menu: list[dict] = []

        def add(action: str, desc: str) -> None:
            menu.append({"action": action, "desc": desc})

        if page == "post":
            if not page_state["own_post"] or page_state["mentioned"]:
                add("comment", "write a comment on this post")
                add("reply", "reply to an existing comment in the thread")
                add("vote", "upvote this post")
                add("vote_comment", "upvote a good comment in the thread")
                add("vote_poll", "answer the attached poll if there is one")
                add("react", "add an emoji reaction to this post")
        elif page == "news_detail":
            add("comment", "write a comment reacting to this news article")
        elif page == "project_detail":
            add("vote", "star this project")
            add("react", "add an emoji reaction")
            add("comment", "write a comment on this project")
        elif page == "gist_detail":
            add("comment", "write a comment on this code snippet")
            add("vote", "star this gist")
            add("react", "add an emoji reaction")
        elif page == "profile":
            if page_state.get("own_profile"):
                if not self.state.profile_filled:
                    add("update_profile", "fill in your own profile bio and links")
                add("open_post", "open one of your posts to read it")
            else:
                if not page_state["follows"]:
                    add("follow", "follow this developer")
                if page_state["can_message"]:
                    add("message", "send this developer a direct message")
                add("open_post", "open one of their posts to read it")
        elif page == "projects_list":
            add("vote", "star a project in the list")
            if page_state["can_project"]:
                add("create_project", "create a new project of your own")
            add("open_project", "open a project to read it")
        elif page == "gists_list":
            add("vote", "star a gist in the list")
            if not page_state["gists"]:
                add("create_gist", "publish a new code snippet of your own")
            add("open_gist", "open a gist to read it")
        elif page == "news_list":
            add("open_news", "open a news article to read it")
        elif page == "notifications":
            add("check_notifications", "read and act on your notifications")
        else:
            add("open_post", "open a post from the feed to read it")
            add("vote", "upvote a post in the feed")
            add("react", "react to a post in the feed")
            add("vote_poll", "answer a poll in the feed")
            if page_state["can_post"] and not page_state["did_post"]:
                add("create_post", "write a new post reacting to a tech news article")
            if not page_state["gists"]:
                add("create_gist", "publish a new code snippet of your own")
            add("open_profile", "open another developer's profile")
            add("search", "search the platform for a topic you care about")
            add("browse", "browse a category or section")
        if page != "notifications" and page_state.get("unread_notifications"):
            add(
                "check_notifications",
                "open your notifications and reply to anyone who mentioned or replied to you",
            )
        add("navigate", "move to another section: feed, projects, gists, or news")
        return menu

    def _record_history(self, action: str, rationale: str) -> None:
        note = action.replace("_", " ")
        if rationale:
            note = f"{note} ({rationale[:60]})"
        self._session_history.append(note)
        if len(self._session_history) > 20:
            self._session_history = self._session_history[-20:]

    async def _vote_for_page(self, page: str) -> bool:
        if page in ("project_detail", "projects_list"):
            return await self._vote_on_project()
        if page in ("gist_detail", "gists_list"):
            return await self._vote_on_gist()
        return await self._vote_on_feed()

    async def _navigate_to(self, target: str) -> bool:
        routes = {
            "feed": "/feed",
            "projects": "/projects",
            "gists": "/gists",
            "news": "/news",
            "notifications": "/notifications",
        }
        path = routes.get((target or "").lower())
        if path:
            await self.b.goto(f"{self.base_url}{path}")
            return True
        return await self._random_nav_click()

    async def _dispatch_action(self, action: str, target: str, page_state: dict) -> bool:
        if action == "navigate":
            return await self._navigate_to(target)
        if action == "vote":
            return await self._vote_for_page(page_state["page"])
        handlers = {
            "comment": self._comment_on_post,
            "reply": self._reply_to_random_comment,
            "vote_comment": self._vote_on_comments,
            "vote_poll": self._vote_on_poll,
            "react": self._react_to_object,
            "create_post": self._create_post,
            "create_gist": self._create_gist,
            "create_project": self._click_create_project,
            "update_profile": self._update_profile,
            "follow": self._follow_user,
            "message": self._send_message,
            "open_post": self._click_post_link,
            "open_gist": self._click_gist_link,
            "open_profile": self._click_profile_link,
            "open_project": lambda: self._open_content("project"),
            "open_news": lambda: self._open_content("news"),
            "check_notifications": self._check_notifications,
            "search": lambda: self._search("feed"),
            "browse": self._browse_section,
        }
        handler = handlers.get(action)
        if handler is None:
            return False
        try:
            result = await handler()
        except Exception as e:
            logger.debug("dispatch %s failed: %s", action, e)
            return False
        return True if result is None else bool(result)

    async def _run_plan(
        self, plan: list[dict], page_state: dict
    ) -> tuple[int, bool, int, int, int]:
        b = self.b
        actions = 0
        did_post = False
        projs = 0
        follows = 0
        gists = 0
        if not plan:
            await self._navigate_to("feed")
            return 0, did_post, projs, follows, gists
        for entry in plan:
            action = entry["action"]
            if action == "create_post" and (did_post or page_state["did_post"]):
                continue
            if action == "create_gist" and (gists or page_state["gists"]):
                continue
            if action == "create_project" and projs:
                continue
            if await self._dispatch_action(action, entry.get("target", ""), page_state):
                actions += 1
                self._record_history(action, entry.get("rationale", ""))
                if action == "create_post":
                    did_post = True
                elif action == "create_project":
                    projs += 1
                elif action == "follow":
                    follows += 1
                elif action == "create_gist":
                    gists += 1
            await b._idle(0.5, 1.5)
        return actions, did_post, projs, follows, gists

    async def _cycle_ai(
        self,
        mood: str = "normal",
        *,
        session_posted: bool = False,
        session_projects: int = 0,
        session_follows: int = 0,
        session_gists: int = 0,
        can_project: bool = False,
        can_post: bool = False,
        can_message: bool = False,
    ) -> tuple[int, bool, int, int, int]:
        b = self.b
        curl = await b.url()
        page_text = (await b.html()).lower()
        page_state = await self._page_state(
            curl,
            page_text,
            did_post=session_posted,
            follows=session_follows,
            gists=session_gists,
            can_project=can_project,
            can_post=can_post,
            can_message=can_message,
        )
        if page_state["page"] in (
            "post",
            "news_detail",
            "project_detail",
            "gist_detail",
        ):
            await self._read_like_human(4, 90)
        menu = self._build_menu(page_state)
        decision = await self._generate(
            self.llm.decide,
            self.state.identity,
            page_state,
            menu,
            self._session_history,
            self._decision_temperature,
        )
        if not decision:
            decision = {"plan": [], "energy": self._session_energy, "stop_after": 0}
        self._session_energy = decision.get("energy", self._session_energy)
        self._session_stop_after = decision.get("stop_after", 0)
        actions, did_post, projs, follows, gists = await self._run_plan(
            decision["plan"], page_state
        )
        return (
            actions,
            did_post or session_posted,
            session_projects + projs,
            session_follows + follows,
            session_gists + gists,
        )

    def _scaled_pause(self) -> int:
        base = random.randint(self._pause_min, self._pause_max)
        factor = 1.4 - 0.8 * self._session_energy
        return max(1, int(base * factor))