feat(agent): detect and scrape explicitly requested URLs

Add URL visit detection that triggers a direct page fetch when the
user asks to enter, open, analyze, or read a site along with a URL.
Extend URL extraction to match bare domains (e.g. example.com.br),
deduplicate results, and normalize missing protocol prefixes. Update
the system prompt to instruct the model to treat visited webpage
content as the primary source when [WEBSITE CONTENT FROM ...] tags
are present.

Also rotate conversation context logs for 2026-10-27.
This commit is contained in:
2026-10-27 17:29:08 -03:00
parent ff10b642a7
commit 752bb94d48
5 changed files with 129 additions and 505 deletions
+97 -1
View File
@@ -2961,6 +2961,7 @@ def build_system_prompt(profile: Dict[str, Any]) -> str:
"\n- If you don't have access to search, say you can't check right now."
"\n- NEVER invent URLs, links, or specific references. Only cite sources that are explicitly provided in the web search results."
"\n- If web search results are provided, base your answer ONLY on those results. Do not mix in unverified information."
"\n- When webpage content from a visited URL is included below (marked with [WEBSITE CONTENT FROM ...] tags), base your answer specifically on that content. Use it as primary source for answering questions about that website."
)
except Exception:
pass
@@ -3058,6 +3059,71 @@ def ensure_min_lines(response: str, user_message: str, profile: Dict[str, Any])
return response
# ===================== URL VISIT DETECTION =====================
_URL_VISIT_TRIGGERS = [
"entra",
"entre",
"acesse",
"acessar",
"visite",
"visitar",
"abre",
"abrir",
"analisa",
"analisar",
"leia",
"leitura",
"le",
"busca",
"buscar",
"pesquise",
"pesquisar",
"olha",
"olhar",
"ver",
"veja",
"mostra",
"mostrar",
"check",
"visit",
"browse",
"scrape",
"site",
"pagina",
"página",
"url",
"page",
]
def _looks_like_url_visit_request(user_message: str) -> bool:
"""Detect if message contains a URL + intent to visit/scrape/analyze it.
Args:
user_message: Raw user message.
Returns:
True when a direct URL fetch should be attempted.
"""
q = user_message.lower().strip()
# Must contain at least one URL pattern
urls = _extract_urls(q)
if not urls:
return False
# Must contain at least one action trigger (allow short messages with URLs since the URL itself implies intent)
if len(q.split()) >= 2:
return any(t in q for t in _URL_VISIT_TRIGGERS)
return False
def _normalize_url(raw_url: str) -> str:
"""Ensure URL has a protocol prefix so requests can handle it."""
stripped = raw_url.strip().rstrip(".,;:")
if not stripped.startswith(("http://", "https://")):
stripped = "https://" + stripped
return stripped
# ===================== VALIDATION =====================
def _extract_urls(text: str) -> List[str]:
@@ -3073,11 +3139,20 @@ def _extract_urls(text: str) -> List[str]:
patterns = [
r'https?://[^\s)>\]"\']+',
r'www\.[^\s)>\]"\']+',
r'[a-zA-Z0-9](?:[a-zA-Z0-9-]*[a-zA-Z0-9])?\.(?:com|br|net|org|io|dev|co|info|biz|gov|edu|cloud|ai)[^\s)>\]"\']+',
]
urls = []
for pattern in patterns:
urls.extend(re.findall(pattern, text))
return urls
# Deduplicate while preserving order (normalize for comparison)
seen = set()
unique = []
for u in urls:
normalized = _normalize_url(u)
if normalized not in seen:
seen.add(normalized)
unique.append(u)
return unique
def _validate_search_response(
@@ -6887,6 +6962,27 @@ async def process_text_message(update: Update, context: ContextTypes.DEFAULT_TYP
]
else:
web_results = []
# Direct URL visit/scrape when user explicitly asks to enter or analyze a site
direct_url_context = ""
if _looks_like_url_visit_request(user_message):
raw_urls = _extract_urls(user_message)
for raw in raw_urls:
normalized = _normalize_url(raw)
page_text = _fetch_page_content(normalized, max_chars=4000)
if page_text:
domain = ""
try:
from urllib.parse import urlparse
domain = urlparse(normalized).netloc
except Exception:
domain = normalized
direct_url_context += f"\n\n[WEBSITE CONTENT FROM {domain}]\n{page_text}\n[END OF WEBSITE CONTENT]"
print(f"[URL_VISIT] scraped url={normalized} chars={len(page_text)}")
else:
print(f"[URL_VISIT] failed to fetch url={normalized}")
if direct_url_context:
modified = modified + direct_url_context
response = await get_psagentai_response(modified)
if response: