如何在 Windows 上运行 FastAPI 验证码识别服务

pip install capskip fastapi uvicorn

# Where CapSkip listens and which key it expects.
$env:CAPSKIP_HOST = "127.0.0.1"
$env:CAPSKIP_PORT = "8080"
$env:CAPSKIP_API_KEY = "YOUR_CAPSKIP_KEY"
# Match Max. Threads in the CapSkip settings.
$env:CAPSKIP_MAX_THREADS = "10"
# The key your callers must send to this service.
$env:SOLVER_API_KEY = "YOUR_API_KEY"
# Seconds a caller may wait, queue time included.
$env:SOLVE_DEADLINE = "120"

import asyncio
import os
from contextlib import asynccontextmanager

from capskip import AsyncCapSkip
from fastapi import FastAPI

# The SDK reads no env vars, so pass each setting in by hand.
HOST = os.getenv("CAPSKIP_HOST", "127.0.0.1")
PORT = int(os.getenv("CAPSKIP_PORT", "8080"))
KEY = os.getenv("CAPSKIP_API_KEY", "capskip")
MAX_THREADS = int(os.getenv("CAPSKIP_MAX_THREADS", "10"))
DEADLINE = float(os.getenv("SOLVE_DEADLINE", "120"))

@asynccontextmanager
async def lifespan(app: FastAPI):
    # One client and one gate, shared by every request.
    app.state.cap = AsyncCapSkip(
        apiKey=KEY, host=HOST, port=PORT,
        recaptchaTimeout=DEADLINE,
    )
    app.state.slots = asyncio.Semaphore(MAX_THREADS)
    yield


app = FastAPI(lifespan=lifespan)

import secrets

from fastapi import Depends, Header, HTTPException

SERVICE_KEY = os.environ["SOLVER_API_KEY"]
if not SERVICE_KEY:
    raise SystemExit("SOLVER_API_KEY must not be empty")


def require_key(x_api_key: str = Header(default="")):
    good = secrets.compare_digest(
        x_api_key.encode(), SERVICE_KEY.encode())
    if not good:
        raise HTTPException(
            401, "missing or wrong X-API-Key",
            headers={"WWW-Authenticate": "APIKey"})

from fastapi import Request
from pydantic import BaseModel, Field


class SolveRequest(BaseModel):
    sitekey: str = Field(min_length=1)
    url: str = Field(pattern=r"^https?://")


async def solve_v2(state, sitekey: str, url: str) -> str:
    # The deadline also covers the wait for a free slot.
    async with asyncio.timeout(DEADLINE):
        async with state.slots:
            result = await state.cap.recaptcha(sitekey, url)
    return result["code"]

@app.post("/solve", dependencies=[Depends(require_key)])
async def solve(body: SolveRequest, request: Request):
    token = await solve_v2(request.app.state, body.sitekey, body.url)
    return {"token": token}

import logging

from capskip import ApiException, CapSkipError, NetworkException
from capskip import TimeoutException, ValidationException
from fastapi.responses import JSONResponse

STATUS = {
    NetworkException: (503, "capskip_unreachable"),
    TimeoutException: (504, "capskip_timeout"),
    ValidationException: (422, "bad_parameters"),
    ApiException: (502, "capskip_error"),
}

@app.exception_handler(CapSkipError)
async def capskip_error(request: Request, exc: CapSkipError):
    # Log the detail here, keep it away from the caller.
    logging.getLogger("uvicorn.error").warning("%r", exc)
    status, code = STATUS.get(type(exc), (502, "capskip_error"))
    if "UNSOLVABLE" in str(exc):
        status, code = 422, "unsolvable"
    return JSONResponse({"error": code}, status)

@app.exception_handler(TimeoutError)
async def deadline_hit(request: Request, exc: TimeoutError):
    # asyncio.timeout raises the built-in TimeoutError. The SDK's
    # TimeoutException is a different class, so it needs its own entry.
    return JSONResponse({"error": "deadline_exceeded"}, 504)

@app.get("/health")
async def health():
    # CapSkip has no ping call, so check that its port accepts a
    # connection.
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(HOST, PORT), timeout=3)
    except (OSError, TimeoutError):
        return JSONResponse({"ok": False}, 503)
    writer.close()
    return {"ok": True}

python -m uvicorn main:app --host 127.0.0.1 --port 8000

Invoke-RestMethod -Method Post -Uri http://192.168.1.50:8000/solve `
  -Headers @{ "X-API-Key" = "YOUR_API_KEY" } `
  -ContentType "application/json" `
  -Body '{"sitekey":"YOUR_SITEKEY","url":"https://example.com/page-with-recaptcha"}'

# curl.exe reads the JSON from stdin, so no quote escaping is needed.
$body = '{"sitekey":"YOUR_SITEKEY","url":"https://example.com/page-with-recaptcha"}'
$body | curl.exe -s http://192.168.1.50:8000/solve `
  -H "X-API-Key: YOUR_API_KEY" `
  -H "Content-Type: application/json" `
  --data-binary "@-"

import asyncio
import logging
import os
import secrets
from contextlib import asynccontextmanager

from capskip import ApiException, AsyncCapSkip, CapSkipError
from capskip import NetworkException, TimeoutException, ValidationException
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

# The SDK reads no env vars, so pass each setting in by hand.
HOST = os.getenv("CAPSKIP_HOST", "127.0.0.1")
PORT = int(os.getenv("CAPSKIP_PORT", "8080"))
KEY = os.getenv("CAPSKIP_API_KEY", "capskip")
MAX_THREADS = int(os.getenv("CAPSKIP_MAX_THREADS", "10"))
DEADLINE = float(os.getenv("SOLVE_DEADLINE", "120"))


@asynccontextmanager
async def lifespan(app: FastAPI):
    # One client and one gate, shared by every request.
    app.state.cap = AsyncCapSkip(
        apiKey=KEY, host=HOST, port=PORT,
        recaptchaTimeout=DEADLINE,
    )
    app.state.slots = asyncio.Semaphore(MAX_THREADS)
    yield


app = FastAPI(lifespan=lifespan)

SERVICE_KEY = os.environ["SOLVER_API_KEY"]
if not SERVICE_KEY:
    raise SystemExit("SOLVER_API_KEY must not be empty")


def require_key(x_api_key: str = Header(default="")):
    good = secrets.compare_digest(
        x_api_key.encode(), SERVICE_KEY.encode())
    if not good:
        raise HTTPException(
            401, "missing or wrong X-API-Key",
            headers={"WWW-Authenticate": "APIKey"})


class SolveRequest(BaseModel):
    sitekey: str = Field(min_length=1)
    url: str = Field(pattern=r"^https?://")


async def solve_v2(state, sitekey: str, url: str) -> str:
    # The deadline also covers the wait for a free slot.
    async with asyncio.timeout(DEADLINE):
        async with state.slots:
            result = await state.cap.recaptcha(sitekey, url)
    return result["code"]


@app.post("/solve", dependencies=[Depends(require_key)])
async def solve(body: SolveRequest, request: Request):
    token = await solve_v2(request.app.state, body.sitekey, body.url)
    return {"token": token}


STATUS = {
    NetworkException: (503, "capskip_unreachable"),
    TimeoutException: (504, "capskip_timeout"),
    ValidationException: (422, "bad_parameters"),
    ApiException: (502, "capskip_error"),
}


@app.exception_handler(CapSkipError)
async def capskip_error(request: Request, exc: CapSkipError):
    # Log the detail here, keep it away from the caller.
    logging.getLogger("uvicorn.error").warning("%r", exc)
    status, code = STATUS.get(type(exc), (502, "capskip_error"))
    if "UNSOLVABLE" in str(exc):
        status, code = 422, "unsolvable"
    return JSONResponse({"error": code}, status)


@app.exception_handler(TimeoutError)
async def deadline_hit(request: Request, exc: TimeoutError):
    # asyncio.timeout raises the built-in TimeoutError. The SDK's
    # TimeoutException is a different class, so it needs its own entry.
    return JSONResponse({"error": "deadline_exceeded"}, 504)


@app.get("/health")
async def health():
    # CapSkip has no ping call, so check that its port accepts a
    # connection.
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(HOST, PORT), timeout=3)
    except (OSError, TimeoutError):
        return JSONResponse({"ok": False}, 503)
    writer.close()
    return {"ok": True}