Spaces:
Sleeping
Sleeping
File size: 2,533 Bytes
d0e5bf9 d172b0d d0e5bf9 d172b0d d0e5bf9 fce9511 1ec2ad3 d172b0d 94cc940 d172b0d 94cc940 d0e5bf9 94cc940 d0e5bf9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 |
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)
|