forked from retoor/devplacepy
refactor: split bot.py into mixins (ref.md 3.3)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,624 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import logging
|
||||
import random
|
||||
|
||||
from devplacepy.services.bot.config import (
|
||||
MAX_THREAD_REPLIES,
|
||||
MENTION_REPLIES_PER_SESSION,
|
||||
PROJECT_STATUSES,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BotSocialMixin:
|
||||
async def _click_profile_link(self) -> bool:
|
||||
b = self.b
|
||||
links = b.page.locator("a[href*='/profile/']")
|
||||
count = await links.count()
|
||||
if count == 0:
|
||||
return False
|
||||
|
||||
for i in range(count):
|
||||
h = await links.nth(i).get_attribute("href")
|
||||
if h and self.state.username not in h:
|
||||
try:
|
||||
await links.nth(i).click(timeout=5000)
|
||||
await b._idle(1.0, 2.0)
|
||||
uname = h.split("/profile/")[-1].split("?")[0]
|
||||
if uname not in self.state.known_users:
|
||||
self.state.known_users.append(uname)
|
||||
self.state.profiles_viewed += 1
|
||||
self._action(
|
||||
"PROFILE", f"viewed @{uname} [{self.state.profiles_viewed}]"
|
||||
)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.debug("_click_profile_link failed: %s", e)
|
||||
continue
|
||||
return False
|
||||
|
||||
async def _click_create_project(self) -> bool:
|
||||
b = self.b
|
||||
await b.goto(f"{self.base_url}/projects")
|
||||
ok = await b.click("#create-project-btn")
|
||||
if not ok:
|
||||
return False
|
||||
await b._idle(0.8, 1.5)
|
||||
|
||||
if not await b.text("#create-project-modal"):
|
||||
return False
|
||||
|
||||
if self._project_cache:
|
||||
raw_title, desc = self._project_cache.pop(0)
|
||||
asyncio.create_task(self._refill_cache_async("project"))
|
||||
else:
|
||||
raw_title = await self._generate(
|
||||
self.llm.generate_project_title, self.state.persona
|
||||
)
|
||||
if not raw_title:
|
||||
self._log("Create project: title generation failed")
|
||||
return False
|
||||
raw_title = raw_title[:200]
|
||||
desc = await self._generate(
|
||||
self.llm.generate_project_desc, raw_title, self.state.persona
|
||||
)
|
||||
if not desc:
|
||||
self._log("Create project: description generation failed")
|
||||
return False
|
||||
desc = desc[:5000]
|
||||
|
||||
title = self.llm.strip_md(raw_title)[:120] or raw_title[:120]
|
||||
types = ["game", "software", "mobile_app", "website", "game_asset"]
|
||||
ptype = random.choice(types)
|
||||
status = random.choice(PROJECT_STATUSES)
|
||||
|
||||
await b.fill("#title", title)
|
||||
await b._idle(0.2, 0.5)
|
||||
await b.fill("#description", desc)
|
||||
await b._idle(0.2, 0.5)
|
||||
|
||||
await b.click(f"input[name='project_type'][value='{ptype}']")
|
||||
await b._idle(0.2, 0.5)
|
||||
|
||||
await b.click(f"input[name='status'][value='{status}']")
|
||||
await b._idle(0.2, 0.5)
|
||||
|
||||
ok = await b.click("#create-project-modal .btn-primary")
|
||||
if ok:
|
||||
self.state.projects_created += 1
|
||||
self._action(
|
||||
"PROJECT", f"{title} ({ptype}) [{self.state.projects_created}]"
|
||||
)
|
||||
await asyncio.sleep(2)
|
||||
return True
|
||||
return False
|
||||
|
||||
async def _report_issue(self) -> bool:
|
||||
b = self.b
|
||||
await b.goto(f"{self.base_url}/issues")
|
||||
ok = await b.click("[data-modal='create-issue-modal']")
|
||||
if not ok:
|
||||
return False
|
||||
await b._idle(0.5, 1.5)
|
||||
|
||||
if not await b.text("#create-issue-modal"):
|
||||
return False
|
||||
|
||||
topic = random.choice(
|
||||
[
|
||||
"UI glitch",
|
||||
"Broken link",
|
||||
"Performance issue",
|
||||
"Auth problem",
|
||||
"Formatting error",
|
||||
]
|
||||
)
|
||||
if self._issue_cache:
|
||||
title, desc = self._issue_cache.pop(0)
|
||||
asyncio.create_task(self._refill_cache_async("issue"))
|
||||
else:
|
||||
generated = await self._generate(self.llm.generate_issue, topic)
|
||||
if not generated:
|
||||
self._log("Report issue: generation failed")
|
||||
return False
|
||||
title, desc = generated
|
||||
|
||||
await b.fill("#issue-title", title[:200])
|
||||
await b._idle(0.2, 0.5)
|
||||
await b.fill("textarea[name='description']", desc[:2000])
|
||||
await b._idle(0.5, 1.0)
|
||||
|
||||
ok = await b.click(
|
||||
"button:has-text('Submit Report'), #create-issue-modal .btn-primary"
|
||||
)
|
||||
if ok:
|
||||
self.state.issues_filed += 1
|
||||
self._action("ISSUE", f"{title[:50]} [{self.state.issues_filed}]")
|
||||
await asyncio.sleep(2)
|
||||
return True
|
||||
return False
|
||||
|
||||
async def _update_profile(self) -> bool:
|
||||
b = self.b
|
||||
await b.goto(f"{self.base_url}/profile/{self.state.username}")
|
||||
|
||||
ok = await b.click("[data-edit-field='bio']")
|
||||
if not ok:
|
||||
return False
|
||||
await b._idle(0.3, 0.8)
|
||||
|
||||
bio = await self._generate(self.llm.generate_bio)
|
||||
if not bio:
|
||||
self._log("Update profile: bio generation failed")
|
||||
return False
|
||||
bio = bio[:500]
|
||||
await b.fill("textarea[name='bio']", bio)
|
||||
await b._idle(0.3, 0.8)
|
||||
|
||||
filled = ["bio"]
|
||||
fields = await self._generate(
|
||||
self.llm.generate_profile_fields, self.state.username
|
||||
)
|
||||
if fields:
|
||||
location, git_link, website = fields
|
||||
for field_name, value in (
|
||||
("location", location),
|
||||
("git_link", git_link),
|
||||
("website", website),
|
||||
):
|
||||
if not value:
|
||||
continue
|
||||
await b.click(f"[data-edit-field='{field_name}']")
|
||||
await b._idle(0.2, 0.5)
|
||||
if await b.fill(
|
||||
f"input[name='{field_name}'], textarea[name='{field_name}']", value
|
||||
):
|
||||
filled.append(field_name)
|
||||
await b._idle(0.2, 0.5)
|
||||
|
||||
ok = await b.click(
|
||||
"form[action*='/profile/update'] button[type='submit'], button[type='submit']"
|
||||
)
|
||||
if ok:
|
||||
self.state.profile_filled = True
|
||||
self._save()
|
||||
self._action("PROFILE", f"updated own profile ({', '.join(filled)})")
|
||||
await asyncio.sleep(2)
|
||||
return True
|
||||
return False
|
||||
|
||||
async def _follow_user(self) -> bool:
|
||||
b = self.b
|
||||
curl = await b.url()
|
||||
if "/profile/" not in curl:
|
||||
return False
|
||||
uname = curl.split("/profile/")[-1].split("?")[0]
|
||||
if uname == self.state.username:
|
||||
return False
|
||||
form = b.page.locator("form[action*='/follow/'] button[type='submit']")
|
||||
if await form.count() == 0:
|
||||
return False
|
||||
text = (await form.first.inner_text()).strip()
|
||||
if text.lower() not in ("follow", "unfollow"):
|
||||
return False
|
||||
try:
|
||||
await form.first.click(timeout=3000)
|
||||
self._action(
|
||||
"FOLLOW",
|
||||
f"{'followed' if text.lower() == 'follow' else 'unfollowed'} @{uname}",
|
||||
)
|
||||
await b._idle(1.0, 2.0)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.debug("_follow_user failed: %s", e)
|
||||
return False
|
||||
|
||||
def _classify_notification(self, text: str) -> bool:
|
||||
tl = text.lower()
|
||||
return (
|
||||
f"@{self.state.username.lower()}" in tl
|
||||
or "mentioned you" in tl
|
||||
or "replied to your" in tl
|
||||
or "replied to you" in tl
|
||||
or "tagged you" in tl
|
||||
)
|
||||
|
||||
async def _collect_notifications(self) -> list[dict]:
|
||||
b = self.b
|
||||
cards = b.page.locator(".notification-card")
|
||||
try:
|
||||
count = await cards.count()
|
||||
except Exception:
|
||||
count = 0
|
||||
if count == 0:
|
||||
return await self._collect_notifications_fallback()
|
||||
items: list[dict] = []
|
||||
seen_keys: set = set()
|
||||
for i in range(min(count, 30)):
|
||||
card = cards.nth(i)
|
||||
text = ""
|
||||
try:
|
||||
body = card.locator(".notification-text").first
|
||||
if await body.count() > 0:
|
||||
text = (await body.inner_text() or "")[:600]
|
||||
except Exception:
|
||||
text = ""
|
||||
if not text:
|
||||
try:
|
||||
text = (await card.inner_text() or "")[:600]
|
||||
except Exception:
|
||||
text = ""
|
||||
href = ""
|
||||
try:
|
||||
link = card.locator("a.card-link").first
|
||||
if await link.count() > 0:
|
||||
href = (await link.get_attribute("href")) or ""
|
||||
except Exception:
|
||||
href = ""
|
||||
unread = False
|
||||
try:
|
||||
cls = (await card.get_attribute("class")) or ""
|
||||
unread = "unread" in cls.split()
|
||||
except Exception:
|
||||
unread = False
|
||||
if not text and not href:
|
||||
continue
|
||||
key = hashlib.sha256(f"{href}|{text[:200]}".encode()).hexdigest()[:24]
|
||||
if key in seen_keys:
|
||||
continue
|
||||
seen_keys.add(key)
|
||||
items.append(
|
||||
{
|
||||
"key": key,
|
||||
"href": href,
|
||||
"text": text,
|
||||
"is_mention": self._classify_notification(text),
|
||||
"unread": unread,
|
||||
}
|
||||
)
|
||||
return items
|
||||
|
||||
async def _collect_notifications_fallback(self) -> list[dict]:
|
||||
b = self.b
|
||||
loc = b.page.locator(".notification-item, .notif-item, [data-notification-id]")
|
||||
try:
|
||||
count = await loc.count()
|
||||
except Exception:
|
||||
count = 0
|
||||
items: list[dict] = []
|
||||
seen_keys: set = set()
|
||||
for i in range(min(count, 30)):
|
||||
el = loc.nth(i)
|
||||
try:
|
||||
text = (await el.inner_text() or "")[:600]
|
||||
except Exception:
|
||||
text = ""
|
||||
href = ""
|
||||
try:
|
||||
inner = el.locator("a[href]").first
|
||||
if await inner.count() > 0:
|
||||
href = (await inner.get_attribute("href")) or ""
|
||||
except Exception:
|
||||
href = ""
|
||||
if not text and not href:
|
||||
continue
|
||||
key = hashlib.sha256(f"{href}|{text[:200]}".encode()).hexdigest()[:24]
|
||||
if key in seen_keys:
|
||||
continue
|
||||
seen_keys.add(key)
|
||||
items.append(
|
||||
{
|
||||
"key": key,
|
||||
"href": href,
|
||||
"text": text,
|
||||
"is_mention": self._classify_notification(text),
|
||||
"unread": True,
|
||||
}
|
||||
)
|
||||
return items
|
||||
|
||||
async def _check_notifications(self) -> bool:
|
||||
b = self.b
|
||||
await b.goto(f"{self.base_url}/notifications")
|
||||
await b._idle(0.8, 1.8)
|
||||
notifications = await self._collect_notifications()
|
||||
if not notifications:
|
||||
self._log("Notifications: none")
|
||||
return True
|
||||
unseen_mentions = [
|
||||
n
|
||||
for n in notifications
|
||||
if n["is_mention"]
|
||||
and n["unread"]
|
||||
and n["key"] not in self.state.notifications_seen
|
||||
]
|
||||
self._log(
|
||||
f"Notifications: {len(notifications)} ({len(unseen_mentions)} unseen mentions)"
|
||||
)
|
||||
self.state.mentions_received += len(unseen_mentions)
|
||||
replied = 0
|
||||
for n in unseen_mentions:
|
||||
if self._session_mention_replies >= MENTION_REPLIES_PER_SESSION:
|
||||
self._log(
|
||||
f"Mention reply cap reached ({MENTION_REPLIES_PER_SESSION}/session), leaving the rest"
|
||||
)
|
||||
break
|
||||
ok = await self._respond_to_mention(n["href"], n["text"])
|
||||
if ok:
|
||||
self.state.mentions_replied += 1
|
||||
self._session_mention_replies += 1
|
||||
replied += 1
|
||||
self.state.notifications_seen.append(n["key"])
|
||||
if len(self.state.notifications_seen) > 500:
|
||||
self.state.notifications_seen = self.state.notifications_seen[-250:]
|
||||
await b._idle(2.0, 5.0)
|
||||
if replied == 0:
|
||||
remaining = [
|
||||
n
|
||||
for n in notifications
|
||||
if n["key"] not in self.state.notifications_seen
|
||||
]
|
||||
if remaining:
|
||||
pick = random.choice(remaining)
|
||||
target = pick["href"]
|
||||
if target and not target.startswith("http"):
|
||||
target = self.base_url + (
|
||||
target if target.startswith("/") else "/" + target
|
||||
)
|
||||
if target:
|
||||
try:
|
||||
await b.goto(target)
|
||||
self._log(f"Notifications: opened {target[-60:]}")
|
||||
except Exception as e:
|
||||
logger.debug("notification goto failed: %s", e)
|
||||
self.state.notifications_seen.append(pick["key"])
|
||||
try:
|
||||
mark = b.page.locator(
|
||||
"button:has-text('Mark all read'), form[action*='/mark-all-read'] button"
|
||||
)
|
||||
if await mark.count() > 0:
|
||||
await mark.first.click(timeout=3000)
|
||||
await b._idle(0.5, 1.5)
|
||||
except Exception as e:
|
||||
logger.debug("mark-read failed: %s", e)
|
||||
return True
|
||||
|
||||
async def _respond_to_mention(self, href: str, snippet: str) -> bool:
|
||||
b = self.b
|
||||
target = href
|
||||
if target and not target.startswith("http"):
|
||||
target = self.base_url + (href if href.startswith("/") else "/" + href)
|
||||
if not target:
|
||||
return False
|
||||
try:
|
||||
await b.goto(target)
|
||||
except Exception as e:
|
||||
logger.debug("mention goto failed: %s", e)
|
||||
return False
|
||||
await b._idle(2.0, 4.0)
|
||||
curl = await b.url()
|
||||
if not any(k in curl for k in ("/posts/", "/gists/", "/projects/", "/news/")):
|
||||
self._log("Mention target has no comment thread, skipping")
|
||||
return False
|
||||
thread_id = self._thread_id(curl)
|
||||
if (
|
||||
thread_id
|
||||
and self.state.thread_reply_counts.get(thread_id, 0) >= MAX_THREAD_REPLIES
|
||||
):
|
||||
self._log(
|
||||
f"Conversation taper on {thread_id[:12]} ({MAX_THREAD_REPLIES} replies), letting it rest"
|
||||
)
|
||||
return False
|
||||
needle = f"@{self.state.username}"
|
||||
comment_loc = None
|
||||
comment_selectors = [
|
||||
f".comment:has-text('{needle}')",
|
||||
f"li.comment:has-text('{needle}')",
|
||||
f".comment-item:has-text('{needle}')",
|
||||
f"article.comment:has-text('{needle}')",
|
||||
]
|
||||
for sel in comment_selectors:
|
||||
try:
|
||||
loc = b.page.locator(sel).first
|
||||
if await loc.count() > 0:
|
||||
comment_loc = loc
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
parent_text = ""
|
||||
mentioner = ""
|
||||
if comment_loc is not None:
|
||||
try:
|
||||
parent_text = (await comment_loc.inner_text() or "")[:1000]
|
||||
except Exception:
|
||||
parent_text = ""
|
||||
try:
|
||||
author_link = comment_loc.locator("a[href*='/profile/']").first
|
||||
if await author_link.count() > 0:
|
||||
ah = await author_link.get_attribute("href") or ""
|
||||
if "/profile/" in ah:
|
||||
mentioner = (
|
||||
ah.split("/profile/")[-1].split("?")[0].split("/")[0]
|
||||
)
|
||||
except Exception:
|
||||
mentioner = ""
|
||||
try:
|
||||
await comment_loc.scroll_into_view_if_needed(timeout=2000)
|
||||
await b._idle(0.3, 0.9)
|
||||
except Exception as e:
|
||||
logger.debug("scroll mention failed: %s", e)
|
||||
try:
|
||||
reply_btn = comment_loc.locator(
|
||||
".comment-action-btn, button:has-text('Reply'), a:has-text('Reply')"
|
||||
).first
|
||||
if await reply_btn.count() > 0:
|
||||
await reply_btn.click(timeout=3000)
|
||||
await b._idle(0.5, 1.5)
|
||||
except Exception as e:
|
||||
logger.debug("reply-btn click failed: %s", e)
|
||||
try:
|
||||
post_body = (
|
||||
await b.text(".post-content")
|
||||
or await b.text(".gist-detail-desc")
|
||||
or await b.text(".news-detail-content")
|
||||
or await b.text(".news-detail-desc")
|
||||
or await b.text(".project-detail-desc")
|
||||
or await b.text(".rendered-content")
|
||||
or await b.text("article")
|
||||
or snippet
|
||||
)
|
||||
except Exception:
|
||||
post_body = snippet
|
||||
reply = await self._generate(
|
||||
self.llm.generate_comment,
|
||||
post_body[:1500],
|
||||
self.state.persona,
|
||||
mentioner,
|
||||
parent_text,
|
||||
)
|
||||
if not reply:
|
||||
self._log("Mention reply: generation failed/empty")
|
||||
return False
|
||||
if mentioner and f"@{mentioner.lower()}" not in reply.lower():
|
||||
reply = f"@{mentioner} {reply}"
|
||||
reply = reply[:2000]
|
||||
textarea_sels = [
|
||||
".comment .reply-form textarea[name='content']",
|
||||
".reply-form textarea[name='content']",
|
||||
"form.comment-form textarea[name='content']",
|
||||
"textarea#comment-content",
|
||||
"textarea.emoji-picker-target",
|
||||
"textarea[name='content']",
|
||||
]
|
||||
typed = False
|
||||
for sel in textarea_sels:
|
||||
if await self._type_like_human(sel, reply):
|
||||
typed = True
|
||||
break
|
||||
if not typed:
|
||||
self._log("Mention reply: no textarea found")
|
||||
return False
|
||||
await b._idle(0.5, 1.5)
|
||||
submit_sels = [
|
||||
".reply-form button[type='submit']",
|
||||
"form.comment-form button[type='submit']",
|
||||
"button.btn-primary:has-text('Reply')",
|
||||
"button.btn-primary:has-text('Post')",
|
||||
"form[action*='/comments/create'] button[type='submit']",
|
||||
".comment-form button[type='submit']",
|
||||
]
|
||||
for sel in submit_sels:
|
||||
if await b.click(sel):
|
||||
self.state.comments_posted += 1
|
||||
self.state.replies_to_own_posts += 1
|
||||
if thread_id:
|
||||
self.state.thread_reply_counts[thread_id] = (
|
||||
self.state.thread_reply_counts.get(thread_id, 0) + 1
|
||||
)
|
||||
if len(self.state.thread_reply_counts) > 500:
|
||||
recent = list(self.state.thread_reply_counts.items())[-250:]
|
||||
self.state.thread_reply_counts = dict(recent)
|
||||
self._action("MENTION", f"replied to @{mentioner or '?'}: {reply[:60]}")
|
||||
await asyncio.sleep(2)
|
||||
return True
|
||||
self._log("Mention reply: submit failed")
|
||||
return False
|
||||
|
||||
async def _check_messages(self) -> bool:
|
||||
b = self.b
|
||||
await b.goto(f"{self.base_url}/messages")
|
||||
await b._idle()
|
||||
conv = b.page.locator(
|
||||
".messages-conversations a, .conversation-item, .message-item, .messages-list a"
|
||||
)
|
||||
if await conv.count() == 0:
|
||||
self._log("Messages: no conversations")
|
||||
return True
|
||||
try:
|
||||
await conv.first.click(timeout=3000)
|
||||
await b._idle(1.0, 2.0)
|
||||
except Exception as e:
|
||||
logger.debug("_check_messages open failed: %s", e)
|
||||
return True
|
||||
if random.random() < 0.7:
|
||||
return await self._compose_message(reply=True)
|
||||
self._log("Messages: read conversation")
|
||||
return True
|
||||
|
||||
async def _spoke_last_in_thread(self) -> bool:
|
||||
bubbles = self.b.page.locator(".messages-thread .message-bubble")
|
||||
count = await bubbles.count()
|
||||
if count == 0:
|
||||
return False
|
||||
try:
|
||||
classes = await bubbles.nth(count - 1).get_attribute("class") or ""
|
||||
except Exception as e:
|
||||
logger.debug("_spoke_last_in_thread read failed: %s", e)
|
||||
return False
|
||||
return "mine" in classes.split()
|
||||
|
||||
async def _compose_message(self, reply: bool = False) -> bool:
|
||||
b = self.b
|
||||
box = b.page.locator(
|
||||
"form[action*='/messages/send'] input[name='content'], "
|
||||
"input[name='content'][placeholder*='message'], "
|
||||
"form[action*='/messages/send'] textarea[name='content']"
|
||||
)
|
||||
if await box.count() == 0:
|
||||
self._log("Message: compose box not found")
|
||||
return False
|
||||
if await self._spoke_last_in_thread():
|
||||
self._log("Message: I spoke last; waiting for a reply (dialog, not monologue)")
|
||||
return False
|
||||
context = ""
|
||||
try:
|
||||
context = (
|
||||
await b.text(".messages-thread") or await b.text(".message-list") or ""
|
||||
)[:400]
|
||||
except Exception:
|
||||
context = ""
|
||||
text = await self._generate(self.llm.generate_dm, self.state.persona, context)
|
||||
if not text:
|
||||
return False
|
||||
text = text[:500]
|
||||
if not await self._type_like_human(
|
||||
"form[action*='/messages/send'] input[name='content'], input[name='content'], "
|
||||
"form[action*='/messages/send'] textarea[name='content']",
|
||||
text,
|
||||
):
|
||||
return False
|
||||
await b._idle(0.4, 1.0)
|
||||
if await b.click(
|
||||
"button.messages-send-btn, form[action*='/messages/send'] button[type='submit']"
|
||||
):
|
||||
self.state.messages_sent += 1
|
||||
self._action(
|
||||
"MESSAGE",
|
||||
f"{'replied' if reply else 'sent'} [{self.state.messages_sent}]: {text[:50]}",
|
||||
)
|
||||
await asyncio.sleep(2)
|
||||
return True
|
||||
return False
|
||||
|
||||
async def _send_message(self) -> bool:
|
||||
b = self.b
|
||||
curl = await b.url()
|
||||
if "/profile/" not in curl:
|
||||
return False
|
||||
btn = b.page.locator("a[href*='with_uid=']").first
|
||||
if await btn.count() == 0:
|
||||
return False
|
||||
href = await btn.get_attribute("href") or ""
|
||||
if "with_uid=" not in href:
|
||||
return False
|
||||
try:
|
||||
await b.goto((self.base_url + href) if href.startswith("/") else href)
|
||||
await b._idle(1.0, 2.0)
|
||||
except Exception as e:
|
||||
logger.debug("_send_message goto failed: %s", e)
|
||||
return False
|
||||
return await self._compose_message(reply=False)
|
||||
Reference in New Issue
Block a user