161 lines
4.4 KiB
Python
161 lines
4.4 KiB
Python
import json
|
|
import logging
|
|
import traceback
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
from urllib.parse import urlsplit
|
|
|
|
from camoufox.async_api import AsyncCamoufox
|
|
from camoufox.fingerprints import generate_context_fingerprint
|
|
from camoufox.utils import determine_ua_os
|
|
from fastapi import Cookie
|
|
from fastapi import FastAPI
|
|
from fastapi import Request
|
|
from fastapi import Response
|
|
from playwright.async_api import ProxySettings
|
|
from ua_parser import user_agent_parser
|
|
|
|
from engine import actions
|
|
from engine.schemas import SolveRequest
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger.setLevel(logging.DEBUG)
|
|
logger.addHandler(logging.StreamHandler())
|
|
BROWSER_SLOTS = 10
|
|
|
|
IDENTITY_FILE = Path("identity.json")
|
|
|
|
|
|
def parse_proxy(proxy_url: str) -> ProxySettings:
|
|
parts = urlsplit(proxy_url)
|
|
server = f"{parts.scheme}://{parts.hostname}:{parts.port}"
|
|
settings: ProxySettings = {"server": server}
|
|
if parts.username:
|
|
settings["username"] = parts.username
|
|
if parts.password:
|
|
settings["password"] = parts.password
|
|
return settings
|
|
|
|
|
|
def load_or_create_identity(user_data_dir, os=None, ff_version=None, locale=None, time_zone=None, user_agent=None) -> dict:
|
|
identity_file = user_data_dir / IDENTITY_FILE
|
|
|
|
if identity_file.exists():
|
|
return json.loads(identity_file.read_text())
|
|
|
|
kwargs = {}
|
|
if user_agent:
|
|
os, ff_version = parse_os_and_ff(user_agent)
|
|
kwargs["config_overrides"] = {"navigator.UserAgent": user_agent}
|
|
|
|
fp = generate_context_fingerprint(os=os, ff_version=ff_version, locale=locale, timezone=time_zone)
|
|
identity_file.write_text(json.dumps(fp))
|
|
return fp
|
|
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Hello World"}
|
|
|
|
|
|
_OS_MAP = {'win': 'windows', 'mac': 'macos', 'lin': 'linux'}
|
|
|
|
|
|
def parse_os_and_ff(user_agent) -> tuple[str, str]:
|
|
os_name = _OS_MAP[determine_ua_os(user_agent)]
|
|
parsed = user_agent_parser.ParseUserAgent(user_agent)
|
|
ff_version = int(parsed["major"]) if (
|
|
parsed.get("family") == "Firefox" and parsed.get("major")) else 150
|
|
return os_name, str(ff_version)
|
|
|
|
|
|
@app.post("/v1/solve")
|
|
async def solve(
|
|
request: Request,
|
|
payload: SolveRequest,
|
|
response: Response,
|
|
session_id: Annotated[str | None, Cookie()] = None
|
|
):
|
|
api_key = request.headers.get('X-API-Key')
|
|
|
|
logger.info("Incoming payload: [%s] %s", session_id, payload)
|
|
global BROWSER_SLOTS
|
|
|
|
if BROWSER_SLOTS <= 0:
|
|
return {
|
|
"ok": False,
|
|
"error": "No slots available",
|
|
}
|
|
else:
|
|
BROWSER_SLOTS -= 1
|
|
|
|
if not session_id:
|
|
session_id = str(uuid.uuid4())
|
|
response.set_cookie("session_id", session_id)
|
|
logger.info("Setup new session with session_id: %s", session_id)
|
|
|
|
scr_w, scr_h = payload.screen.split("x")
|
|
scr_size = int(scr_w), int(scr_h)
|
|
proxy = parse_proxy(payload.proxy)
|
|
|
|
user_data_dir = Path(__file__).parent / f"../extra/user_data_dir/{session_id}"
|
|
if not user_data_dir.exists():
|
|
user_data_dir.mkdir(parents=True)
|
|
|
|
config = load_or_create_identity(user_data_dir, user_agent=payload.user_agent)
|
|
ctx_opts = {**config["context_options"], "config": config["config"]}
|
|
|
|
async with AsyncCamoufox(
|
|
persistent_context=True,
|
|
user_data_dir=user_data_dir,
|
|
proxy=proxy,
|
|
window=scr_size,
|
|
i_know_what_im_doing=True,
|
|
geoip=True,
|
|
humanize=True,
|
|
**ctx_opts,
|
|
) as camoufo:
|
|
await camoufo.add_init_script(config["init_script"])
|
|
page = camoufo.pages[0] if camoufo.pages else await camoufo.new_page()
|
|
if payload.cookies:
|
|
await camoufo.add_cookies(payload.cookies)
|
|
page.set_default_timeout(30_000)
|
|
page.set_default_navigation_timeout(60_000)
|
|
|
|
await page.goto(url=payload.url)
|
|
|
|
actions_result = await actions.execute(
|
|
page, payload.actions, payload.captcha,
|
|
antibot_name=payload.captcha_type, success_locator=payload.captcha_success_locator
|
|
)
|
|
|
|
status = not "error" in actions_result[-1]
|
|
error = ["Something error", None][status]
|
|
cookies = await camoufo.cookies()
|
|
html = await page.inner_html('html')
|
|
# TODO: добавить как опцию. передачу скрина в base64.
|
|
# await page.screenshot(
|
|
# path=(
|
|
# f"/home/sokol/PycharmProjects/WebRoboApi/extra/screenshots/"
|
|
# f"{datetime.timestamp(datetime.now())}.png"
|
|
# )
|
|
# )
|
|
|
|
BROWSER_SLOTS += 1
|
|
result = {
|
|
"status": ["err", "ok"][status],
|
|
"user_agent": ctx_opts["user_agent"],
|
|
"cookies": cookies,
|
|
"actions": actions_result,
|
|
}
|
|
if payload.html:
|
|
result["html"] = html
|
|
if error:
|
|
result["error"] = error
|
|
logger.info("Outgoing payload: %s", result)
|
|
return result
|