# retoor <retoor@molodetz.nl>
import asyncio
import hashlib
import logging
import random
from devplacepy.services.bot.config import (
HOME_URL,
MAX_THREAD_REPLIES,
MENTION_REPLIES_PER_SESSION,
PROJECT_STATUSES,
bio_has_signal,
ensure_bio_signal,
)
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].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 _profile_has_signal(self) -> bool:
profile = await self._fetch_own_profile()
user = profile.get("profile_user") or {}
website = (user.get("website") or "").strip().rstrip("/")
return bio_has_signal(user.get("bio") or "") and website == HOME_URL
async def _ensure_profile_signal(self) -> bool:
if self.state.profile_filled and await self._profile_has_signal():
return True
self._log("Profile signal missing, refreshing profile")
ok = await self._update_profile()
await self.b.goto(f"{self.base_url}/feed")
await self.b._idle(0.8, 2.0)
return ok
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 = ensure_bio_signal(bio[:400])[: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 "")[:125000]
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 = ""
if mentioner and mentioner.lower() == self.state.username.lower():
self._log("Mention target resolves to self, skipping")
return False
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 = self._sanitize_mentions(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 View Git Blame Copy Permalink