# retoor <retoor@molodetz.nl>
import asyncio
import hashlib
import logging
import random
from devplacepy.services.bot.config import pick_category
logger = logging.getLogger(__name__)
class BotPostingMixin:
async def _create_gist(self) -> bool:
b = self.b
await b.goto(f"{self.base_url}/gists")
if not await b.click("#create-gist-btn, .gists-fab, .feed-fab"):
self._log("Create gist: trigger not found")
return False
await b._idle(0.8, 1.5)
if await b.page.locator("#create-gist-modal").count() == 0:
self._log("Create gist: modal not found")
return False
generated = None
for attempt in range(2):
candidate = await self._generate(self.llm.generate_gist, self.state.persona)
if not candidate:
continue
cand_title, cand_desc, cand_lang, cand_code = candidate
if not cand_code.strip():
continue
verdict = await self._generate(
self.llm.gist_quality_check, cand_title, cand_code, cand_lang
)
if verdict is None or verdict[0]:
generated = candidate
break
self.state.quality_rejections += 1
action = "regenerating" if attempt == 0 else "skipping"
self._log(f"Gist quality reject ({verdict[1]}), {action}")
if not generated:
self._log("Create gist: no quality candidate")
return False
title, desc, language, code = generated
title = self.llm.strip_md(title)[:200] or title[:200]
if not await b.fill("#gist-title", title):
self._log("Create gist: title field not found")
return False
await b._idle(0.2, 0.5)
await b.fill("#gist-description", desc[:500])
await b._idle(0.2, 0.5)
try:
await b.page.select_option("#gist-language", language)
except Exception as e:
logger.debug("gist language select failed: %s", e)
await b._idle(0.3, 0.7)
cm = b.page.locator("#create-gist-modal .CodeMirror")
if await cm.count() == 0:
self._log("Create gist: code editor not found")
return False
try:
await cm.first.click()
await b._idle(0.3, 0.6)
await b.page.evaluate(
"""(code) => {
const el = document.querySelector('#create-gist-modal .CodeMirror');
if (el && el.CodeMirror) { el.CodeMirror.setValue(code); el.CodeMirror.save(); }
const ta = document.querySelector('#gist-source-editor');
if (ta) { ta.value = code; ta.dispatchEvent(new Event('input', {bubbles: true})); }
}""",
code,
)
except Exception as e:
logger.debug("gist code fill failed: %s", e)
return False
await b._idle(0.5, 1.2)
ok = await b.click(
"#create-gist-modal button[type='submit'], #create-gist-modal .btn-primary"
)
if not ok:
self._log("Create gist: submit not found")
return False
await asyncio.sleep(2)
url = await b.url()
if "/gists/" in url:
if url not in self.state.own_post_urls:
self.state.own_post_urls.append(url)
self.state.gists_created += 1
self._action(
"GIST",
f"{title[:40]} [{language}] [{self.state.gists_created}] -> {url[-50:]}",
)
return True
self._log("Create gist: no redirect to gist detail")
return False
async def _create_post(self) -> bool:
b = self.b
await b.goto(f"{self.base_url}/feed")
fab_sel = "#create-post-btn, .feed-fab, button.feed-fab, a.feed-fab"
ok = await b.click(fab_sel)
if not ok:
self._log("Create post: FAB button not found")
return False
self._log("Create post: FAB clicked, waiting for modal")
await b._idle(1.0, 2.0)
modal = await b.page.locator("#create-post-modal").count()
if modal == 0:
self._log("Create post: modal #create-post-modal not in DOM")
return False
hidden = await b.page.locator("#create-post-modal").get_attribute("class") or ""
if "hidden" in hidden or await b.page.locator("#create-post-modal").is_hidden():
self._log("Create post: modal is hidden")
return False
self._log("Create post: modal visible")
article_title = ""
article_desc = ""
post_title = ""
if self._post_cache:
article_title, post_title, article_desc, content, topic = (
self._post_cache.pop(0)
)
asyncio.create_task(self._refill_cache_async("post"))
else:
articles = self.news.fetch()
topic = pick_category(self.state.persona)
article = self.registry.reserve_unused(
articles,
self.state.username,
self.llm.clean,
topic,
self.state.persona,
)
if not article:
self._log("Create post: no unused articles available")
return False
desc = article.get("description", "")[:500]
post_title = await self._generate(
self.llm.generate_post_title,
article["title"],
self.state.persona,
topic,
)
content = await self._generate(
self.llm.generate_post,
article["title"],
desc,
self.state.persona,
topic,
self.state.recent_post_titles,
)
if not content:
self._log("Create post: generation failed/empty")
return False
article_title = article["title"]
article_desc = desc
verdict = await self._generate(
self.llm.quality_check, "post", content, article_desc
)
if verdict is not None and not verdict[0]:
self.state.quality_rejections += 1
self._log(f"Post quality reject ({verdict[1]}), regenerating")
content = await self._generate(
self.llm.generate_post,
article_title,
article_desc,
self.state.persona,
topic,
self.state.recent_post_titles,
)
if not content:
self._log("Create post: regeneration failed/empty")
return False
verdict = await self._generate(
self.llm.quality_check, "post", content, article_desc
)
if verdict is not None and not verdict[0]:
self.state.quality_rejections += 1
self._log(f"Post quality reject ({verdict[1]}), skipping")
return False
content_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
if content_hash in self.state.posted_content_hashes:
self._log(f"Create post: duplicate content {content_hash}, skipping")
return False
clean_title = self.llm.clean(article_title[:200]) if article_title else ""
if clean_title:
if not self.registry.reserve(clean_title, self.state.username, topic):
self._log(
f"Create post: article '{clean_title[:50]}...' already covered, skipping"
)
return False
type_title = self.llm.strip_md(post_title)[:120] if post_title else ""
if not type_title:
type_title = clean_title
ts = f"#create-post-modal input[name='topic'][value='{topic}']"
if await b.page.locator(ts).count() == 0:
self._log(f"Create post: topic radio '{topic}' not found")
else:
await b.click(ts)
await b._idle(0.2, 0.5)
if type_title:
ok = await b.fill("#create-post-modal input[name='title']", type_title)
if not ok:
self._log("Create post: title input not found")
await b._idle(0.2, 0.4)
content_sel = "#create-post-modal textarea[name='content']"
if await b.page.locator(content_sel).count() == 0:
self._log("Create post: content textarea not found")
return False
if not await self._type_like_human(content_sel, content[:5000]):
self._log("Create post: typing content into modal failed")
return False
await b._idle(0.5, 1.5)
submit_sels = [
"#create-post-modal button[type='submit']",
"#create-post-modal .btn-primary",
".modal-overlay#create-post-modal button[type='submit']",
"#create-post-modal button:has-text('Post')",
]
ok = False
for sel in submit_sels:
if await b.click(sel):
ok = True
break
if not ok:
self._log("Create post: submit button not found")
return False
post_url = await b.url()
for _ in range(12):
if "/posts/" in post_url:
break
await asyncio.sleep(1)
post_url = await b.url()
if "/posts/" not in post_url:
self._log(f"Create post: submit did not create a post (url={post_url[-45:]})")
return False
self.state.created_posts += 1
self.state.posted_content_hashes.append(content_hash)
if type_title:
self.state.recent_post_titles.append(type_title)
self.state.recent_post_titles = self.state.recent_post_titles[-8:]
if post_url not in self.state.own_post_urls:
self.state.own_post_urls.append(post_url)
if post_url not in self.state.known_posts:
self.state.known_posts.append(post_url)
self._action(
"POST",
f"{topic} [{self.state.created_posts}]: {type_title[:50] or content[:50]} -> {post_url[-45:]}",
)
return True
async def _click_post_link(self) -> bool:
b = self.b
links = b.page.locator("a[href*='/posts/']")
count = await links.count()
if count == 0:
return False
candidates = []
for i in range(count):
h = await links.nth(i).get_attribute("href")
if h and "/posts/" in h and len(h) > 20:
candidates.append(i)
if not candidates:
return False
idx = random.choice(candidates)
href = await links.nth(idx).get_attribute("href")
try:
await links.nth(idx).click(timeout=5000)
await b._idle(1.0, 2.0)
if href and href not in self.state.known_posts:
self.state.known_posts.append(href)
self._action("OPEN", f"post {href[-45:] if href else '?'}")
return True
except Exception as e:
logger.debug("_click_post_link failed: %s", e)
return False
async def _click_gist_link(self) -> bool:
b = self.b
links = b.page.locator("a[href*='/gists/']")
count = await links.count()
if count == 0:
return False
candidates = []
for i in range(count):
h = await links.nth(i).get_attribute("href")
if h and "/gists/" in h and "?" not in h and h.rstrip("/") != "/gists":
candidates.append(i)
if not candidates:
return False
idx = random.choice(candidates)
href = await links.nth(idx).get_attribute("href")
try:
await links.nth(idx).click(timeout=5000)
await b._idle(1.0, 2.0)
self._action("OPEN", f"gist {href[-45:] if href else '?'}")
return True
except Exception as e:
logger.debug("_click_gist_link failed: %s", e)
return False
async def _open_content(self, kind: str) -> bool:
b = self.b
base = "/news/" if kind == "news" else f"/{kind}s/"
links = b.page.locator(f"a[href*='{base}']")
count = await links.count()
if count == 0:
return False
candidates = []
for i in range(count):
h = await links.nth(i).get_attribute("href")
if h and base in h and "?" not in h and h.rstrip("/") != base.rstrip("/"):
candidates.append(i)
if not candidates:
return False
idx = random.choice(candidates)
href = await links.nth(idx).get_attribute("href")
try:
await links.nth(idx).click(timeout=5000)
await b._idle(1.0, 2.0)
self._action("OPEN", f"{kind} {href[-45:] if href else '?'}")
return True
except Exception as e:
logger.debug("_open_content(%s) failed: %s", kind, e)
return False
Reference in New Issue View Git Blame Copy Permalink