A Bilibili user's scan said Civitai could not be reached while their browser reached it through a VPN app: the plugin's urllib requests only use HTTPS_PROXY or the Windows proxy, not a PAC script or a browser-only proxy. net_proxy.py: the proxy address saved in api/config.json (NETWORK_PROXY) that every request of the plugin goes through (Civitai, Hugging Face, GitHub, translators), handed to the scan process as HTTP(S)_PROXY; empty keeps the old behaviour. It also names why a request failed (timeout, refused, dns, ssl, reset, proxy, http:<code>). Settings -> Network has the address (SOCKS and malformed ones refused with what to write) and Test connection, which calls 1.60's /anomalous/civitai_check (brought here unchanged, now through the proxy and with the failure's name). A scan that loses Civitai records why (civitai_error) and the scan page and activity log say it with what to try. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
53 lines
2.1 KiB
Python
53 lines
2.1 KiB
Python
"""GET /anomalous/civitai_check: whether this computer reaches Civitai now, through the proxy in
|
|
Settings -> Network when one is set (net_proxy.py); for Settings' "Test" button and the first-run
|
|
setup. One small request with the key, if one is set, so a key Civitai refuses is told apart.
|
|
It goes online, so it answers this computer only (local_guard LOCAL_GETS)."""
|
|
|
|
import asyncio
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
from aiohttp import web
|
|
|
|
from .download_sources import civitai_headers
|
|
|
|
try:
|
|
from ..net_proxy import configured, failure, urlopen
|
|
except ImportError: # loaded outside the package (tests)
|
|
from net_proxy import configured, failure, urlopen
|
|
|
|
PROBE_URL = "https://civitai.com/api/v1/models?limit=1"
|
|
TIMEOUT = 8
|
|
|
|
|
|
def check_civitai():
|
|
"""{reachable, key ('set' | 'none' | 'refused'), ms, error, failure (net_proxy.failure), proxy}."""
|
|
headers = civitai_headers()
|
|
key = "set" if "Authorization" in headers else "none"
|
|
started = time.monotonic()
|
|
|
|
def answer(reachable, key, error="", why=""):
|
|
return {"reachable": reachable, "key": key, "ms": round((time.monotonic() - started) * 1000),
|
|
"error": error, "failure": why, "proxy": configured()}
|
|
try:
|
|
with urlopen(urllib.request.Request(PROBE_URL, headers=headers), timeout=TIMEOUT) as response:
|
|
response.read(1024)
|
|
return answer(True, key)
|
|
except urllib.error.HTTPError as error:
|
|
# Civitai answered: the network is fine; 401/403 with a key means the key is refused.
|
|
if error.code in (401, 403) and key == "set":
|
|
return answer(True, "refused", f"HTTP {error.code}", failure(error))
|
|
return answer(error.code < 500, key, f"HTTP {error.code}", failure(error))
|
|
except (urllib.error.URLError, OSError, ValueError) as error:
|
|
reason = getattr(error, "reason", None) or error
|
|
return answer(False, key, str(reason)[:200], failure(error))
|
|
|
|
|
|
async def api_civitai_check(request):
|
|
return web.json_response(await asyncio.to_thread(check_civitai))
|
|
|
|
|
|
def register_routes(app):
|
|
app.router.add_get("/anomalous/civitai_check", api_civitai_check)
|