869 lines
37 KiB
Python
869 lines
37 KiB
Python
from app.utils.browser_utils import safe_browser_operation
|
|
import asyncio
|
|
from urllib.parse import unquote
|
|
import time
|
|
from playwright.async_api import async_playwright
|
|
from app.config import CUSTOM_USER_AGENT
|
|
|
|
# Global tracking for active pages and their creation times
|
|
active_pages = set()
|
|
page_creation_times = {}
|
|
|
|
async def visit_url_service(decoded_url):
|
|
"""Service function to visit a URL and get its content"""
|
|
print(f"Visiting URL: {decoded_url}")
|
|
|
|
# Define the operation to perform with the browser
|
|
async def visit_operation(page):
|
|
try:
|
|
response = await page.goto(decoded_url, wait_until='networkidle', timeout=30000)
|
|
if not response:
|
|
print(f"Warning: No response object returned for {decoded_url}")
|
|
|
|
cookie_buttons = [
|
|
'Accept all',
|
|
'Accepteer',
|
|
'Accepteren',
|
|
'Accept'
|
|
]
|
|
|
|
try:
|
|
for button in cookie_buttons:
|
|
element = await page.query_selector(f'text="{button}"')
|
|
if element:
|
|
await element.click()
|
|
print(f"{button} button found and clicked")
|
|
await asyncio.sleep(1)
|
|
break
|
|
|
|
except Exception as e:
|
|
print(f"Error during cookie acceptance: {e}")
|
|
print("No cookies to accept")
|
|
|
|
# Get page content
|
|
content = await page.content()
|
|
return {"status": "success", "content": content}
|
|
except Exception as e:
|
|
print(f"Error during page navigation: {e}")
|
|
# Try to get content anyway
|
|
try:
|
|
content = await page.content()
|
|
return {"status": "partial", "content": content, "error": str(e)}
|
|
except:
|
|
raise Exception(f"Failed to get page content: {str(e)}")
|
|
|
|
# Perform the operation
|
|
return await safe_browser_operation(decoded_url, visit_operation)
|
|
|
|
async def screenshot_url_service(decoded_url, full_page=True):
|
|
"""Service function to visit a URL and return a screenshot as PNG bytes"""
|
|
print(f"Taking screenshot of: {decoded_url}")
|
|
|
|
async def screenshot_operation(page):
|
|
try:
|
|
# Start navigation early, then wait for progressively stronger load signals.
|
|
await page.goto(decoded_url, wait_until='domcontentloaded', timeout=30000)
|
|
await page.wait_for_load_state('load', timeout=30000)
|
|
|
|
# Wait until the browser reports full document readiness.
|
|
await page.wait_for_function("document.readyState === 'complete'", timeout=30000)
|
|
|
|
# Ensure network has gone quiet so dynamic requests can finish.
|
|
await page.wait_for_load_state('networkidle', timeout=30000)
|
|
|
|
# Scroll to trigger lazy-loaded/infinite content until height stabilizes.
|
|
await page.evaluate("""async () => {
|
|
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
|
let previousHeight = 0;
|
|
let stableRounds = 0;
|
|
const maxRounds = 30;
|
|
|
|
for (let i = 0; i < maxRounds; i++) {
|
|
const currentHeight = Math.max(
|
|
document.body.scrollHeight,
|
|
document.documentElement.scrollHeight
|
|
);
|
|
|
|
window.scrollTo(0, currentHeight);
|
|
await sleep(700);
|
|
|
|
const newHeight = Math.max(
|
|
document.body.scrollHeight,
|
|
document.documentElement.scrollHeight
|
|
);
|
|
|
|
if (newHeight <= previousHeight && newHeight <= currentHeight) {
|
|
stableRounds += 1;
|
|
if (stableRounds >= 2) {
|
|
break;
|
|
}
|
|
} else {
|
|
stableRounds = 0;
|
|
}
|
|
|
|
previousHeight = newHeight;
|
|
}
|
|
|
|
// Return to top for a predictable final render before capture.
|
|
window.scrollTo(0, 0);
|
|
await sleep(700);
|
|
}""")
|
|
|
|
# Extra settle time so animations/transitions can complete.
|
|
await page.wait_for_timeout(5000)
|
|
screenshot_bytes = await page.screenshot(full_page=full_page, type='png')
|
|
return screenshot_bytes
|
|
except Exception as e:
|
|
print(f"Error during screenshot capture: {e}")
|
|
return {"status": "error", "url": decoded_url, "error": str(e)}
|
|
|
|
return await safe_browser_operation(decoded_url, screenshot_operation)
|
|
|
|
async def click_selector_service(decoded_url: str, selector: str, return_type: str):
|
|
"""Navigate to a URL, click a CSS selector, and return PNG bytes or resulting HTML."""
|
|
print(f"Clicking selector on {decoded_url}: {selector}")
|
|
|
|
async def click_operation(page):
|
|
try:
|
|
# Navigate and wait for initial DOM.
|
|
await page.goto(decoded_url, wait_until='domcontentloaded', timeout=30000)
|
|
await page.wait_for_load_state('load', timeout=30000)
|
|
|
|
# Ensure the page is fully loaded (helps with JS-driven UIs).
|
|
await page.wait_for_function("document.readyState === 'complete'", timeout=30000)
|
|
|
|
# Best-effort cookie banner handling (helps avoid click interception).
|
|
cookie_buttons = [
|
|
'Accept all',
|
|
'Accepteer',
|
|
'Accepteren',
|
|
'Accept'
|
|
]
|
|
for button in cookie_buttons:
|
|
try:
|
|
element = await page.query_selector(f'text="{button}"')
|
|
if element:
|
|
await element.click()
|
|
await page.wait_for_timeout(1000)
|
|
break
|
|
except Exception:
|
|
# Ignore cookie/banner interaction errors; click may still work.
|
|
pass
|
|
|
|
# Wait for the target selector, scroll it into view, then click.
|
|
await page.wait_for_selector(selector, timeout=30000)
|
|
locator = page.locator(selector)
|
|
await locator.scroll_into_view_if_needed(timeout=30000)
|
|
await locator.click(force=True, timeout=30000)
|
|
|
|
# Give the page a moment to react to the click.
|
|
await page.wait_for_timeout(1000)
|
|
try:
|
|
await page.wait_for_load_state('networkidle', timeout=10000)
|
|
except Exception:
|
|
pass
|
|
|
|
if return_type == "html":
|
|
return await page.content()
|
|
|
|
# Return a full-page screenshot of the resulting view.
|
|
return await page.screenshot(full_page=True, type='png')
|
|
|
|
except Exception as e:
|
|
print(f"Error during click capture: {e}")
|
|
return {"status": "error", "url": decoded_url, "error": str(e)}
|
|
|
|
return await safe_browser_operation(decoded_url, click_operation)
|
|
|
|
async def interactions_service(
|
|
decoded_url: str,
|
|
interactions,
|
|
return_type: str,
|
|
viewport=None,
|
|
scroll_x: int | None = None,
|
|
scroll_y: int | None = None,
|
|
):
|
|
"""Open `decoded_url`, execute click/type interactions in order, then return screenshot, HTML, or resulting_url."""
|
|
print(f"Running interactions on {decoded_url} (return_type={return_type})")
|
|
|
|
async def interactions_operation(page):
|
|
try:
|
|
# Interactions endpoint policy:
|
|
# - Disable all scrolling features (ignore scrollX/scrollY and do not auto-scroll).
|
|
# - Force viewport width to 1280.
|
|
# - Return a full-height screenshot (full_page=True) when return_type != "html".
|
|
forced_viewport_width = 1280
|
|
|
|
async def resolve_viewport_point(px: int, py: int) -> tuple[float, float]:
|
|
"""
|
|
`document.elementFromPoint(x, y)` uses *viewport* (CSS pixel) coordinates.
|
|
|
|
With scrolling disabled, we keep the page scrolled to the top and expand the
|
|
viewport height to the document height (best-effort). That makes document and
|
|
viewport coordinates equivalent for typical pages.
|
|
"""
|
|
return float(px), float(py)
|
|
|
|
async def element_from_point_retry(vx: float, vy: float, attempts: int = 10):
|
|
"""Retry `elementFromPoint` to avoid timing issues around scroll/render."""
|
|
last_error = None
|
|
for _ in range(attempts):
|
|
handle = None
|
|
try:
|
|
handle = await page.evaluate_handle(
|
|
"(p) => document.elementFromPoint(p.x, p.y)",
|
|
{"x": vx, "y": vy},
|
|
)
|
|
element = handle.as_element()
|
|
if element is not None:
|
|
saved_handle = handle
|
|
# Prevent `finally` from disposing the backing JSHandle;
|
|
# the caller will dispose after clicking/focusing.
|
|
handle = None
|
|
return element, saved_handle
|
|
last_error = f"No element found at viewport point ({vx}, {vy})"
|
|
except Exception as e:
|
|
last_error = str(e)
|
|
finally:
|
|
try:
|
|
if handle is not None:
|
|
await handle.dispose()
|
|
except Exception:
|
|
pass
|
|
await page.wait_for_timeout(100)
|
|
raise ValueError(last_error or f"No element found at viewport point ({vx}, {vy})")
|
|
|
|
# Force viewport width to 1280 (ignore payload viewport for interactions endpoint).
|
|
try:
|
|
vp_size = page.viewport_size or {"width": forced_viewport_width, "height": 720}
|
|
await page.set_viewport_size({"width": forced_viewport_width, "height": int(vp_size.get("height", 720))})
|
|
except Exception as e:
|
|
print(f"Error setting forced viewport width: {e}")
|
|
|
|
# Navigate and wait for initial DOM.
|
|
await page.goto(decoded_url, wait_until="domcontentloaded", timeout=30000)
|
|
await page.wait_for_load_state("load", timeout=30000)
|
|
|
|
# Ensure the page is fully loaded (helps with JS-driven UIs).
|
|
await page.wait_for_function("document.readyState === 'complete'", timeout=30000)
|
|
|
|
# Best-effort cookie/banner handling (helps avoid click interception).
|
|
cookie_buttons = [
|
|
"Accept all",
|
|
"Accepteer",
|
|
"Accepteren",
|
|
"Accept",
|
|
]
|
|
for button in cookie_buttons:
|
|
try:
|
|
element = await page.query_selector(f'text="{button}"')
|
|
if element:
|
|
await element.click()
|
|
await page.wait_for_timeout(1000)
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
# Disable scrolling: always keep page at the top.
|
|
try:
|
|
await page.evaluate("() => window.scrollTo({ left: 0, top: 0, behavior: 'auto' })")
|
|
await page.wait_for_timeout(250)
|
|
except Exception:
|
|
pass
|
|
|
|
# Best-effort: expand viewport height to the document height (capped by Chromium).
|
|
# This keeps coordinate interactions working without scrolling.
|
|
try:
|
|
max_vp_h = 16384
|
|
full_h = await page.evaluate(
|
|
"() => Math.max(document.body?.scrollHeight || 0, document.documentElement?.scrollHeight || 0)"
|
|
)
|
|
target_h = int(min(max(int(full_h or 0), 1), max_vp_h))
|
|
await page.set_viewport_size({"width": forced_viewport_width, "height": target_h})
|
|
await page.wait_for_timeout(250)
|
|
except Exception as e:
|
|
print(f"Error resizing viewport for full-height interactions: {e}")
|
|
|
|
for interaction in interactions:
|
|
# Support both Pydantic model instances and raw dicts.
|
|
if isinstance(interaction, dict):
|
|
action = interaction.get("action")
|
|
selector = interaction.get("selector")
|
|
xpath_selector = interaction.get("xpath_selector")
|
|
text = interaction.get("text")
|
|
else:
|
|
action = getattr(interaction, "action", None)
|
|
selector = getattr(interaction, "selector", None)
|
|
xpath_selector = getattr(interaction, "xpath_selector", None)
|
|
text = getattr(interaction, "text", None)
|
|
# Coordinates may be present on interaction models.
|
|
x = getattr(interaction, "x", None)
|
|
y = getattr(interaction, "y", None)
|
|
|
|
if isinstance(interaction, dict):
|
|
x = interaction.get("x")
|
|
y = interaction.get("y")
|
|
|
|
if not action:
|
|
raise ValueError("Interaction must contain an action")
|
|
|
|
async def wait_for_xpath(xp: str, timeout_ms: int = 30000) -> None:
|
|
# Accept both raw XPath ("//div") and Playwright-style ("xpath=//div").
|
|
normalized = xp[len("xpath=") :] if xp.startswith("xpath=") else xp
|
|
await page.wait_for_function(
|
|
"""(xpath) => {
|
|
try {
|
|
return !!document.evaluate(
|
|
xpath,
|
|
document,
|
|
null,
|
|
XPathResult.FIRST_ORDERED_NODE_TYPE,
|
|
null
|
|
).singleNodeValue;
|
|
} catch (e) {
|
|
return false;
|
|
}
|
|
}""",
|
|
arg=normalized,
|
|
timeout=timeout_ms,
|
|
)
|
|
|
|
# Prefer selector-based interactions when selectors are provided (CSS first, then XPath).
|
|
if selector or xpath_selector:
|
|
if action == "click":
|
|
if selector:
|
|
await page.wait_for_selector(selector, timeout=30000)
|
|
# Avoid Playwright auto-scrolling: click via DOM.
|
|
await page.evaluate(
|
|
"""(sel) => {
|
|
const el = document.querySelector(sel);
|
|
if (!el) throw new Error(`No element for selector: ${sel}`);
|
|
el.click();
|
|
}""",
|
|
selector,
|
|
)
|
|
else:
|
|
if not xpath_selector:
|
|
raise ValueError("xpath_selector must be a non-empty string")
|
|
await wait_for_xpath(xpath_selector, timeout_ms=30000)
|
|
normalized_xpath = xpath_selector[len("xpath=") :] if xpath_selector.startswith("xpath=") else xpath_selector
|
|
# Use Playwright's click to generate real pointer/mouse events.
|
|
locator = page.locator(f"xpath={normalized_xpath}").first
|
|
await locator.wait_for(timeout=30000)
|
|
await locator.click(force=True, timeout=30000)
|
|
|
|
elif action == "type":
|
|
if text is None:
|
|
raise ValueError("Type interaction must contain text")
|
|
if selector:
|
|
await page.wait_for_selector(selector, timeout=30000)
|
|
# Avoid Playwright auto-scrolling: focus/fill via DOM.
|
|
await page.evaluate(
|
|
"""({ sel, value }) => {
|
|
const el = document.querySelector(sel);
|
|
if (!el) throw new Error(`No element for selector: ${sel}`);
|
|
el.focus?.();
|
|
if ('value' in el) {
|
|
el.value = value;
|
|
el.dispatchEvent(new Event('input', { bubbles: true }));
|
|
el.dispatchEvent(new Event('change', { bubbles: true }));
|
|
}
|
|
}""",
|
|
{"sel": selector, "value": text},
|
|
)
|
|
else:
|
|
if not xpath_selector:
|
|
raise ValueError("xpath_selector must be a non-empty string")
|
|
await wait_for_xpath(xpath_selector, timeout_ms=30000)
|
|
normalized_xpath = xpath_selector[len("xpath=") :] if xpath_selector.startswith("xpath=") else xpath_selector
|
|
locator = page.locator(f"xpath={normalized_xpath}").first
|
|
await locator.wait_for(timeout=30000)
|
|
# Prefer fill/type so frameworks see real input events.
|
|
await locator.click(force=True, timeout=30000)
|
|
await locator.fill(text, timeout=30000)
|
|
|
|
else:
|
|
raise ValueError(f"Unknown interaction action: {action}")
|
|
|
|
# Fallback to coordinate-based interactions when no selector is given.
|
|
elif x is not None and y is not None:
|
|
if action == "click":
|
|
# Use DOM elementFromPoint so React/SPA components receive real click events.
|
|
try:
|
|
vx, vy = await resolve_viewport_point(x, y)
|
|
element, handle = await element_from_point_retry(vx, vy)
|
|
try:
|
|
await element.click(force=True, timeout=30000)
|
|
finally:
|
|
try:
|
|
await handle.dispose()
|
|
except Exception:
|
|
pass
|
|
except Exception as e:
|
|
raise ValueError(f"Click via elementFromPoint failed at ({x}, {y}): {e}")
|
|
|
|
elif action == "type":
|
|
if text is None:
|
|
raise ValueError("Type interaction must contain text")
|
|
# Focus the element at the given coordinates, then type.
|
|
vx, vy = await resolve_viewport_point(x, y)
|
|
element, handle = await element_from_point_retry(vx, vy)
|
|
try:
|
|
await element.click(force=True, timeout=30000)
|
|
# Ensure the click actually put focus on the intended control.
|
|
try:
|
|
await element.focus(timeout=30000)
|
|
except Exception:
|
|
pass
|
|
await page.wait_for_timeout(100)
|
|
await page.keyboard.type(text)
|
|
finally:
|
|
try:
|
|
await handle.dispose()
|
|
except Exception:
|
|
pass
|
|
|
|
else:
|
|
raise ValueError(f"Unknown interaction action: {action}")
|
|
|
|
else:
|
|
raise ValueError("Interaction must contain selector, xpath_selector, or x/y coordinates")
|
|
|
|
# Let the UI settle after each action.
|
|
await page.wait_for_timeout(1000)
|
|
try:
|
|
await page.wait_for_load_state("networkidle", timeout=10000)
|
|
except Exception:
|
|
pass
|
|
|
|
if return_type == "resulting_url":
|
|
final_url = unquote(page.url or "")
|
|
return {
|
|
"status": "success",
|
|
"original_url": decoded_url,
|
|
"resulting_url": final_url,
|
|
"redirected": final_url != decoded_url,
|
|
}
|
|
|
|
if return_type == "html":
|
|
return await page.content()
|
|
|
|
# Always return full-page screenshot (full height).
|
|
return await page.screenshot(full_page=True, type="png")
|
|
|
|
except Exception as e:
|
|
print(f"Error during interactions capture: {e}")
|
|
return {"status": "error", "url": decoded_url, "error": str(e)}
|
|
|
|
return await safe_browser_operation(decoded_url, interactions_operation)
|
|
|
|
async def extract_seo_service(decoded_url):
|
|
"""Service function to extract SEO information from a website"""
|
|
print(f"Extracting SEO from: {decoded_url}")
|
|
|
|
# Define the operation to perform with the browser
|
|
async def seo_operation(page):
|
|
try:
|
|
response = await page.goto(decoded_url, wait_until='networkidle', timeout=30000)
|
|
|
|
# Extract SEO information
|
|
seo_data = await page.evaluate('''() => {
|
|
const data = {
|
|
title: document.title || '',
|
|
description: '',
|
|
canonical: '',
|
|
h1: [],
|
|
h2: [],
|
|
images: 0,
|
|
links: 0
|
|
};
|
|
|
|
// Get meta description
|
|
const metaDescription = document.querySelector('meta[name="description"]');
|
|
if (metaDescription) {
|
|
data.description = metaDescription.getAttribute('content') || '';
|
|
}
|
|
|
|
// Get canonical link
|
|
const canonicalLink = document.querySelector('link[rel="canonical"]');
|
|
if (canonicalLink) {
|
|
data.canonical = canonicalLink.getAttribute('href') || '';
|
|
}
|
|
|
|
// Get h1 tags
|
|
document.querySelectorAll('h1').forEach(h1 => {
|
|
const text = h1.innerText.trim();
|
|
if (text) data.h1.push(text);
|
|
});
|
|
|
|
// Get h2 tags
|
|
document.querySelectorAll('h2').forEach(h2 => {
|
|
const text = h2.innerText.trim();
|
|
if (text) data.h2.push(text);
|
|
});
|
|
|
|
// Count images
|
|
data.images = document.querySelectorAll('img').length;
|
|
|
|
// Count links
|
|
data.links = document.querySelectorAll('a').length;
|
|
|
|
return data;
|
|
}''')
|
|
|
|
result = {
|
|
"status": "success",
|
|
"url": decoded_url,
|
|
"seo": seo_data
|
|
}
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"Error during SEO extraction: {e}")
|
|
return {"status": "error", "url": decoded_url, "error": str(e)}
|
|
|
|
# Perform the operation
|
|
return await safe_browser_operation(decoded_url, seo_operation)
|
|
|
|
async def extract_meta_tags_service(decoded_url):
|
|
"""Service function to extract meta tags from a website"""
|
|
print(f"Extracting meta tags from: {decoded_url}")
|
|
|
|
# Define the operation to perform with the browser
|
|
async def meta_tags_operation(page):
|
|
try:
|
|
response = await page.goto(decoded_url, wait_until='networkidle', timeout=30000)
|
|
|
|
# Extract meta tags using JavaScript
|
|
meta_data = await page.evaluate('''() => {
|
|
const data = {
|
|
meta_tags: [],
|
|
open_graph: {},
|
|
twitter_card: {},
|
|
title: document.title || ''
|
|
};
|
|
|
|
// Extract all meta tags
|
|
document.querySelectorAll('meta').forEach(meta => {
|
|
const attributes = {};
|
|
for (let attr of meta.attributes) {
|
|
attributes[attr.name] = attr.value;
|
|
}
|
|
data.meta_tags.push(attributes);
|
|
});
|
|
|
|
// Extract Open Graph tags
|
|
document.querySelectorAll('meta[property^="og:"]').forEach(meta => {
|
|
data.open_graph[meta.getAttribute('property')] = meta.getAttribute('content') || '';
|
|
});
|
|
|
|
// Extract Twitter card tags
|
|
document.querySelectorAll('meta[name^="twitter:"]').forEach(meta => {
|
|
data.twitter_card[meta.getAttribute('name')] = meta.getAttribute('content') || '';
|
|
});
|
|
|
|
return data;
|
|
}''')
|
|
|
|
result = {
|
|
"status": "success",
|
|
"url": decoded_url,
|
|
"meta_tags": meta_data["meta_tags"],
|
|
"open_graph": meta_data["open_graph"],
|
|
"twitter_card": meta_data["twitter_card"],
|
|
"title": meta_data["title"]
|
|
}
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"Error during meta tag extraction: {e}")
|
|
return {"status": "error", "url": decoded_url, "error": str(e)}
|
|
|
|
# Perform the operation
|
|
return await safe_browser_operation(decoded_url, meta_tags_operation)
|
|
|
|
async def capture_outgoing_calls_service(decoded_url):
|
|
"""Service function to capture outgoing API calls from a website"""
|
|
print(f"Capturing outgoing calls from: {decoded_url}")
|
|
|
|
# Define the operation to perform with the browser
|
|
async def outgoing_calls_operation(page):
|
|
try:
|
|
# List to store all network requests
|
|
network_requests = []
|
|
|
|
# Set up network request listener
|
|
async def handle_request(request):
|
|
# Only capture API-like requests (not static assets)
|
|
url = request.url
|
|
method = request.method
|
|
headers = request.headers
|
|
|
|
# Skip static assets and common non-API requests
|
|
static_extensions = ['.css', '.js', '.png', '.jpg', '.jpeg', '.gif', '.svg', '.ico', '.woff', '.woff2', '.ttf', '.eot']
|
|
if any(url.endswith(ext) for ext in static_extensions):
|
|
return
|
|
|
|
# Skip data URLs and blob URLs
|
|
if url.startswith(('data:', 'blob:')):
|
|
return
|
|
|
|
# Skip same-origin requests that are likely static assets
|
|
if url.startswith(decoded_url) and any(static_ext in url.lower() for static_ext in static_extensions):
|
|
return
|
|
|
|
# Capture the request details
|
|
request_data = {
|
|
"url": url,
|
|
"method": method,
|
|
"headers": dict(headers),
|
|
"timestamp": None # Will be set when request is finished
|
|
}
|
|
|
|
# Store request for later processing
|
|
network_requests.append(request_data)
|
|
|
|
# Listen to all requests
|
|
page.on("request", handle_request)
|
|
|
|
# Navigate to the URL and wait for network to be idle
|
|
try:
|
|
await page.goto(decoded_url, wait_until='networkidle', timeout=30000)
|
|
except Exception as e:
|
|
print(f"Error during page navigation: {e}")
|
|
# Continue anyway to capture any requests that were made
|
|
|
|
# Wait a bit more to catch any delayed requests
|
|
await page.wait_for_timeout(2000)
|
|
|
|
# Process and categorize the requests
|
|
api_calls = []
|
|
for req in network_requests:
|
|
# Determine if this looks like an API call
|
|
is_api_call = False
|
|
api_type = "unknown"
|
|
|
|
# Check for common API patterns
|
|
if any(pattern in req["url"].lower() for pattern in ['/api/', '/rest/', '/graphql', '/json', '/xml']):
|
|
is_api_call = True
|
|
api_type = "rest"
|
|
elif req["url"].endswith('.json'):
|
|
is_api_call = True
|
|
api_type = "json"
|
|
elif 'application/json' in req["headers"].get('content-type', '').lower():
|
|
is_api_call = True
|
|
api_type = "json"
|
|
elif 'application/xml' in req["headers"].get('content-type', '').lower():
|
|
is_api_call = True
|
|
api_type = "xml"
|
|
elif req["method"] in ['POST', 'PUT', 'PATCH', 'DELETE']:
|
|
# These methods are typically API calls
|
|
is_api_call = True
|
|
api_type = "rest"
|
|
elif any(domain in req["url"] for domain in ['api.', 'rest.', 'graphql.']):
|
|
is_api_call = True
|
|
api_type = "rest"
|
|
|
|
# Include all requests but mark API calls specifically
|
|
call_info = {
|
|
"url": req["url"],
|
|
"method": req["method"],
|
|
"is_api_call": is_api_call,
|
|
"api_type": api_type if is_api_call else None,
|
|
"headers": {k: v for k, v in req["headers"].items() if k.lower() not in ['user-agent', 'accept-encoding', 'accept-language', 'cache-control']}
|
|
}
|
|
|
|
api_calls.append(call_info)
|
|
|
|
# Sort by whether it's an API call (API calls first), then by URL
|
|
api_calls.sort(key=lambda x: (not x["is_api_call"], x["url"]))
|
|
|
|
result = {
|
|
"status": "success",
|
|
"url": decoded_url,
|
|
"total_requests": len(api_calls),
|
|
"api_calls": [call for call in api_calls if call["is_api_call"]],
|
|
"other_requests": [call for call in api_calls if not call["is_api_call"]],
|
|
"summary": {
|
|
"api_calls_count": len([call for call in api_calls if call["is_api_call"]]),
|
|
"other_requests_count": len([call for call in api_calls if not call["is_api_call"]]),
|
|
"unique_domains": len(set([call["url"].split('/')[2] for call in api_calls if len(call["url"].split('/')) > 2]))
|
|
}
|
|
}
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"Error during outgoing calls capture: {e}")
|
|
return {"status": "error", "url": decoded_url, "error": str(e)}
|
|
|
|
# Perform the operation
|
|
return await safe_browser_operation(decoded_url, outgoing_calls_operation)
|
|
async def get_resulting_url_service(decoded_url):
|
|
"""Service function to get the resulting URL after navigation (handles redirects)"""
|
|
print(f"Getting resulting URL for: {decoded_url}")
|
|
|
|
# Define the operation to perform with the browser
|
|
async def resulting_url_operation(page):
|
|
try:
|
|
# Navigate to the URL and wait for network to be idle
|
|
response = await page.goto(decoded_url, wait_until='networkidle', timeout=30000)
|
|
|
|
# Get the final URL after any redirects and decode it
|
|
final_url = page.url
|
|
decoded_final_url = unquote(final_url)
|
|
|
|
# Get response status if available
|
|
status_code = response.status if response else None
|
|
|
|
result = {
|
|
"status": "success",
|
|
"original_url": decoded_url,
|
|
"resulting_url": decoded_final_url,
|
|
"status_code": status_code
|
|
}
|
|
|
|
# Check if there was a redirect
|
|
if decoded_final_url != decoded_url:
|
|
result["redirected"] = True
|
|
else:
|
|
result["redirected"] = False
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"Error during URL navigation: {e}")
|
|
return {"status": "error", "original_url": decoded_url, "error": str(e)}
|
|
|
|
# Perform the operation
|
|
return await safe_browser_operation(decoded_url, resulting_url_operation)
|
|
|
|
async def safe_browser_operation(url, operation_func):
|
|
"""Safely perform browser operations with proper cleanup and timeouts"""
|
|
browser = None
|
|
context = None
|
|
page = None
|
|
playwright = None
|
|
|
|
try:
|
|
# Get or create playwright instance with timeout
|
|
playwright = await asyncio.wait_for(async_playwright().start(), timeout=30.0)
|
|
|
|
browser = await asyncio.wait_for(playwright.chromium.launch(
|
|
headless=True,
|
|
args=['--no-sandbox', '--disable-setuid-sandbox', '--max_old_space_size=256'],
|
|
), timeout=30.0)
|
|
|
|
# Create context and page with timeout
|
|
context = await asyncio.wait_for(
|
|
browser.new_context(
|
|
user_agent=CUSTOM_USER_AGENT,
|
|
viewport={"width": 1280, "height": 720},
|
|
ignore_https_errors=True,
|
|
),
|
|
timeout=10.0,
|
|
)
|
|
|
|
page = await asyncio.wait_for(context.new_page(), timeout=10.0)
|
|
page.set_default_timeout(30000) # 30 second timeout
|
|
|
|
# Track page creation time for force cleanup
|
|
page_creation_times[page] = time.time()
|
|
active_pages.add(page)
|
|
|
|
# Call the operation function that uses the page with timeout
|
|
result = await asyncio.wait_for(operation_func(page), timeout=60.0)
|
|
|
|
return result
|
|
|
|
except asyncio.TimeoutError as e:
|
|
print(f"Browser operation timeout for {url}: {e}")
|
|
return {
|
|
"status": "error",
|
|
"error": f"Operation timeout: {str(e)}",
|
|
"url": url
|
|
}
|
|
except Exception as e:
|
|
print(f"Error during browser operation for {url}: {e}")
|
|
# Return error result instead of re-raising to allow graceful handling
|
|
return {
|
|
"status": "error",
|
|
"error": str(e),
|
|
"url": url
|
|
}
|
|
finally:
|
|
# Cleanup page tracking with timeout
|
|
if page:
|
|
try:
|
|
# Remove from tracking
|
|
if page in active_pages:
|
|
active_pages.remove(page)
|
|
if page in page_creation_times:
|
|
del page_creation_times[page]
|
|
await asyncio.wait_for(page.close(), timeout=5.0)
|
|
except (asyncio.TimeoutError, Exception) as e:
|
|
print(f"Error closing page: {e}")
|
|
try:
|
|
await page.close(force=True)
|
|
except:
|
|
pass
|
|
|
|
# Ensure context is closed properly with timeout
|
|
if context:
|
|
try:
|
|
await asyncio.wait_for(context.close(), timeout=5.0)
|
|
except (asyncio.TimeoutError, Exception) as e:
|
|
print(f"Error closing context: {e}")
|
|
|
|
# Ensure browser is closed properly with timeout
|
|
if browser:
|
|
try:
|
|
await asyncio.wait_for(browser.close(), timeout=10.0)
|
|
except asyncio.TimeoutError:
|
|
try:
|
|
await browser.close(force=True)
|
|
except:
|
|
pass
|
|
except Exception as e:
|
|
print(f"Error closing browser: {e}")
|
|
|
|
# Ensure playwright is closed properly with timeout - THIS IS THE KEY FIX
|
|
if playwright:
|
|
try:
|
|
await asyncio.wait_for(playwright.stop(), timeout=10.0)
|
|
except (asyncio.TimeoutError, Exception) as e:
|
|
print(f"Error closing playwright: {e}")
|
|
|
|
async def force_cleanup_old_pages():
|
|
"""Force cleanup old page instances created by this module"""
|
|
from app.config import BROWSER_INSTANCE_TIMEOUT_MINUTES
|
|
|
|
print(f"Checking for old page instances in browser service (timeout: {BROWSER_INSTANCE_TIMEOUT_MINUTES} minutes)...")
|
|
|
|
current_time = time.time()
|
|
timeout_seconds = BROWSER_INSTANCE_TIMEOUT_MINUTES * 60
|
|
cleaned_pages = 0
|
|
|
|
# Clean up old pages
|
|
pages_to_cleanup = []
|
|
for page in list(active_pages):
|
|
creation_time = page_creation_times.get(page, 0)
|
|
if current_time - creation_time > timeout_seconds:
|
|
pages_to_cleanup.append(page)
|
|
print(f"Marking page for cleanup (age: {(current_time - creation_time)/60:.1f} minutes)")
|
|
|
|
for page in pages_to_cleanup:
|
|
try:
|
|
if page in active_pages:
|
|
active_pages.remove(page)
|
|
if page in page_creation_times:
|
|
del page_creation_times[page]
|
|
await page.close()
|
|
cleaned_pages += 1
|
|
except Exception as e:
|
|
print(f"Error cleaning up old page: {e}")
|
|
|
|
print(f"Force cleanup completed: {cleaned_pages} pages cleaned")
|