api.py

import os
from pathlib import Path
import asyncio
from flask import Flask, jsonify, request, logging as flog
from flask_limiter.util import get_remote_address
from playwright.async_api import async_playwright

import hcaptcha_solver


async def getcontext():
    async with async_playwright() as p:
        context0 = p.firefox.launch_persistent_context(
            user_data_dir=context_dir,
            headless=True,
            locale="en-US"
        )
    return context0


user_data_dir = Path(__file__).parent.joinpath("user_data_dir")
context_dir = user_data_dir.joinpath("context")
contexts = getcontext()
app = Flask(__name__)


def get_ipaddr():
    if request.access_route:
        print(request.access_route[0])
        return request.access_route[0]
    else:
        return request.remote_addr or '127.0.0.1'


handler = flog.default_handler


def get_token():
    default_token = "init_token"
    if os.path.exists("token"):
        return open("token", "r").read().strip()
    return default_token


def check_request(required_data, data):
    token = get_token()
    if not data or any(key not in data for key in required_data):
        print("Error:Invalid Request Data\n" + str(data))
        return False
    if data["token"] != token:
        print("Error:Invalid Token\n" + str(data))
        return False
    return True


@app.errorhandler(429)
def rate_limit_exceeded(e):
    print(get_remote_address())
    return jsonify(msg="Too many request"), 429


@app.errorhandler(405)
def method_not_allowed(e):
    print(get_remote_address())
    return jsonify(msg="Unauthorized Request"), 405


@app.route("/", methods=["GET"])
def index():
    return jsonify(status_code=200, ip=get_ipaddr())


@app.route("/update/token", methods=["POST"])
def update_token():
    require_data = ["token", "new_token"]
    data = request.get_json(force=True, silent=True)
    if not check_request(require_data, data):
        return jsonify(msg="Unauthorized Request"), 403
    token = open("token", "w+")
    token.write(data["new_token"])
    token.close()
    return jsonify(msg="Token updated successfully", success=True)


@app.route("/api/solve", methods=["POST"])
def solver_captcha():
    require_data = ["token", "host", "site_key"]
    data = request.get_json(force=True, silent=True)
    if not check_request(require_data, data):
        return jsonify(msg="Unauthorized Request"), 403
    return asyncio.run(hcaptcha_solver.bytedance(contexts, data["host"], data["site_key"]))


app.run(host="0.0.0.0", port=7860)