DemonGatanjieuandClaude Opus 5.5
103ee6a25d
New users did not know to scan first. Until anything is scanned, Home shows a card with four steps and "Set up for me": GET /anomalous/civitai_check (api/network_check.py, local requests only) says whether Civitai answers and whether a set key is refused; when it does not, the card offers Retry (with a proxy) or an offline scan. The scan never renames files or refetches, whatever the scan page was set to; the card follows its progress. Never a popup; "Later" hides it per browser and Home's "New here?" line brings it back. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
44 lines
1.9 KiB
Python
44 lines
1.9 KiB
Python
"""GET /anomalous/civitai_check: whether this computer reaches Civitai now, for the first-run setup
|
|
before an online scan (web/modules/ui_first_run.js). 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
|
|
|
|
PROBE_URL = "https://civitai.com/api/v1/models?limit=1"
|
|
TIMEOUT = 8
|
|
|
|
|
|
def check_civitai():
|
|
"""{reachable, key ('set' | 'none' | 'refused'), ms, error}."""
|
|
headers = civitai_headers()
|
|
key = "set" if "Authorization" in headers else "none"
|
|
started = time.monotonic()
|
|
try:
|
|
with urllib.request.urlopen(urllib.request.Request(PROBE_URL, headers=headers), timeout=TIMEOUT) as response:
|
|
response.read(1024)
|
|
return {"reachable": True, "key": key, "ms": round((time.monotonic() - started) * 1000), "error": ""}
|
|
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 {"reachable": True, "key": "refused", "ms": round((time.monotonic() - started) * 1000), "error": f"HTTP {error.code}"}
|
|
return {"reachable": error.code < 500, "key": key, "ms": round((time.monotonic() - started) * 1000), "error": f"HTTP {error.code}"}
|
|
except (urllib.error.URLError, OSError, ValueError) as error:
|
|
reason = getattr(error, "reason", None) or error
|
|
return {"reachable": False, "key": key, "ms": round((time.monotonic() - started) * 1000), "error": str(reason)[:200]}
|
|
|
|
|
|
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)
|