feat(proxy): integrate Scrapling for enhanced web scraping capabilities

- Replaced Playwright with Scrapling's fetchers in the web crawling and YouTube processing modules for improved performance and flexibility.
- Updated proxy configuration to support dynamic proxy selection via environment variables.
- Enhanced logging to track performance metrics during web scraping operations.
- Refactored related modules to utilize the new proxy utilities and streamline the scraping process.
This commit is contained in:
DESKTOP-RTLN3BA\$punk 2026-06-09 00:15:10 -07:00
parent 41a93ca8fb
commit 640ef5f15d
16 changed files with 5770 additions and 4886 deletions

View file

@ -1,86 +1,20 @@
"""
Residential proxy configuration utility.
"""Backward-compatible shim for the proxy helpers.
Reads proxy credentials from the application Config and provides helper
functions that return proxy configs in the format expected by different
HTTP libraries (requests, httpx, aiohttp, Playwright).
The implementation now lives in the modular :mod:`app.utils.proxy` package.
Existing imports of ``app.utils.proxy_config`` keep working via these re-exports.
Prefer importing from ``app.utils.proxy`` (and ``get_proxy_url``) in new code.
"""
import base64
import json
import logging
from app.utils.proxy import (
get_playwright_proxy,
get_proxy_url,
get_requests_proxies,
get_residential_proxy_url,
)
from app.config import Config
logger = logging.getLogger(__name__)
def _build_password_b64() -> str | None:
"""
Build the base64-encoded password dict required by anonymous-proxies.net.
Returns ``None`` when the required config values are not set.
"""
password = Config.RESIDENTIAL_PROXY_PASSWORD
if not password:
return None
password_dict = {
"p": password,
"l": Config.RESIDENTIAL_PROXY_LOCATION,
"t": Config.RESIDENTIAL_PROXY_TYPE,
}
return base64.b64encode(json.dumps(password_dict).encode("utf-8")).decode("utf-8")
def get_residential_proxy_url() -> str | None:
"""
Return the fully-formed residential proxy URL, or ``None`` when not
configured.
The URL format is::
http://<username>:<base64_password>@<hostname>/
"""
username = Config.RESIDENTIAL_PROXY_USERNAME
hostname = Config.RESIDENTIAL_PROXY_HOSTNAME
password_b64 = _build_password_b64()
if not all([username, hostname, password_b64]):
return None
return f"http://{username}:{password_b64}@{hostname}/"
def get_requests_proxies() -> dict[str, str] | None:
"""
Return a ``{"http": , "https": }`` dict suitable for
``requests.Session.proxies`` and ``aiohttp`` ``proxy=`` kwarg,
or ``None`` when not configured.
"""
proxy_url = get_residential_proxy_url()
if proxy_url is None:
return None
return {"http": proxy_url, "https": proxy_url}
def get_playwright_proxy() -> dict[str, str] | None:
"""
Return a Playwright-compatible proxy dict::
{"server": "http://host:port", "username": "", "password": ""}
or ``None`` when not configured.
"""
username = Config.RESIDENTIAL_PROXY_USERNAME
hostname = Config.RESIDENTIAL_PROXY_HOSTNAME
password_b64 = _build_password_b64()
if not all([username, hostname, password_b64]):
return None
return {
"server": f"http://{hostname}",
"username": username,
"password": password_b64,
}
__all__ = [
"get_playwright_proxy",
"get_proxy_url",
"get_requests_proxies",
"get_residential_proxy_url",
]