#!/usr/bin/env python3
"""A local, standard-library Responses API configuration checker."""

import argparse
import datetime
import getpass
import http.client
import ipaddress
import json
import os
import re
import socket
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import warnings

VERSION = "0.1.1"
KEY_ENV = "ELYSIA_REQUEST_DOCTOR_API_KEY"
DEFAULT_BASE = "https://api.elysia.earth/v1"
DEFAULT_MODEL = "gpt-6.1-terra"
MAX_BODY = 1_048_576

MESSAGES = {
    "url_format": "接続先を完全な URL で指定してください。",
    "url_credentials": "ユーザー名・パスワードを含む URL は使えません。",
    "url_query": "URL のクエリーとフラグメントを取り除いてください。",
    "url_https": "HTTPS が必要です。HTTP は localhost のみ使えます。",
    "url_duplicate_v1": "/v1 が重複しています。base URL を見直してください。",
    "url_path": "base URL はルート、/v1、または /v1/responses にしてください。",
    "url_root": "ルート URL に /v1/responses を補います。独自パスは明示してください。",
    "model_format": "モデル ID の文字と長さを確認してください。",
    "credential_required": "端末からキーを入力するか、専用環境変数を設定してください。",
    "credential_format": "キーに空白や改行を含めないでください。",
}


class ConfigError(ValueError):
    def __init__(self, code):
        self.code = code
        super().__init__(code)


class SafeParser(argparse.ArgumentParser):
    def error(self, message):
        # argparse's ordinary message can echo an accidentally supplied key.
        self.print_usage(sys.stderr)
        self.exit(2, "引数を確認してください。キーはコマンドラインで指定できません。\n")


def endpoint_for(base_url):
    """Validate without resolving DNS or making a request."""
    if not base_url or base_url != base_url.strip() or re.search(r"[\s\\\x00-\x1f\x7f]", base_url):
        raise ConfigError("url_format")
    try:
        parsed = urllib.parse.urlsplit(base_url)
        host, port = parsed.hostname, parsed.port
    except ValueError:
        raise ConfigError("url_format") from None
    if parsed.username is not None or parsed.password is not None:
        raise ConfigError("url_credentials")
    if parsed.query or parsed.fragment or "?" in base_url or "#" in base_url:
        raise ConfigError("url_query")
    if not host or parsed.scheme not in ("https", "http") or port == 0:
        raise ConfigError("url_format")
    try:
        localhost = host.lower() == "localhost" or ipaddress.ip_address(host).is_loopback
    except ValueError:
        localhost = host.lower() == "localhost"
    if parsed.scheme != "https" and not localhost:
        raise ConfigError("url_https")
    # Encoded delimiters, dot segments and repeated slashes hide mistakes.
    path = parsed.path.rstrip("/")
    if "%" in path or "//" in path or any(p in (".", "..") for p in path.split("/")):
        raise ConfigError("url_path")
    if sum(segment.lower() == "v1" for segment in path.split("/")) > 1:
        raise ConfigError("url_duplicate_v1")
    warning = None
    if not path:
        path, warning = "/v1/responses", "url_root"
    elif path.endswith("/v1"):
        path += "/responses"
    elif not path.endswith("/v1/responses"):
        raise ConfigError("url_path")
    return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, path, "", "")), warning


def model_valid(model):
    return bool(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:/-]{0,127}", model)) and not model.lower().startswith(("sk-", "bearer"))


def clean_usage(raw):
    """Only explicitly known token counters are shareable."""
    if not isinstance(raw, dict):
        return {}
    result = {}
    for name in ("input_tokens", "output_tokens", "total_tokens"):
        value = raw.get(name)
        if type(value) is int and 0 <= value <= 10**12:
            result[name] = value
    for group, field in (("input_tokens_details", "cached_tokens"), ("output_tokens_details", "reasoning_tokens")):
        detail = raw.get(group)
        value = detail.get(field) if isinstance(detail, dict) else None
        if type(value) is int and 0 <= value <= 10**12:
            result[group] = {field: value}
    return result


def new_report(mode):
    return {
        "tool": "responses-request-doctor",
        "version": VERSION,
        "generated_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),
        "mode": mode,
        "checks": [],
    }


def check_config(base_url, model):
    report = new_report("offline")
    endpoint = None
    try:
        endpoint, warning = endpoint_for(base_url)
        report["checks"].append({"id": "endpoint", "status": "pass"})
        if warning:
            report["checks"].append({"id": warning, "status": "warning"})
    except ConfigError as error:
        report["checks"].append({"id": error.code, "status": "fail"})
    report["checks"].append({"id": "model_format" if not model_valid(model) else "model", "status": "pass" if model_valid(model) else "fail"})
    return endpoint, report


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, request, fp, code, msg, headers, newurl):
        return None  # Never forward the Authorization header to a redirect.


def http_category(code):
    if code in (401, 403):
        return "authentication_or_access"
    if code == 404:
        return "endpoint_or_model_not_found"
    if code == 429:
        return "rate_limit_quota_or_capacity"
    if 300 <= code < 400:
        return "redirect_blocked"
    if code >= 500:
        return "upstream_or_server_error"
    return "http_error"


def send_once(endpoint, model, key, timeout=30, opener=None):
    # Fixed public test phrase. Do not accept customer prompts or echo bodies.
    payload = {"model": model, "input": "Reply with OK only.", "max_output_tokens": 64, "stream": False}
    request = urllib.request.Request(endpoint, data=json.dumps(payload).encode("utf-8"), headers={"Authorization": "Bearer " + key, "Content-Type": "application/json", "User-Agent": "Responses-Request-Doctor/" + VERSION}, method="POST")
    start = time.perf_counter()
    result = {}
    try:
        client = opener if opener is not None else urllib.request.build_opener(NoRedirect())
        with client.open(request, timeout=timeout) as response:
            code = response.status
            result["http_status"] = code
            if not 200 <= code < 300:
                result["category"] = http_category(code)
            else:
                body = response.read(MAX_BODY + 1)
                if len(body) > MAX_BODY:
                    result["category"] = "response_too_large"
                else:
                    try:
                        parsed = json.loads(body)
                    except (ValueError, UnicodeError):
                        parsed = None
                    if not isinstance(parsed, dict):
                        result["category"] = "unexpected_response"
                    else:
                        state = parsed.get("status")
                        result["category"] = "completed" if state == "completed" else "incomplete" if state == "incomplete" else "failed" if state == "failed" else "unexpected_response"
                        result["usage"] = clean_usage(parsed.get("usage"))
    except urllib.error.HTTPError as error:
        result.update(http_status=error.code, category=http_category(error.code))
        error.close()
    except (TimeoutError, socket.timeout):
        result["category"] = "timeout"
    except urllib.error.URLError as error:
        result["category"] = "timeout" if isinstance(error.reason, (TimeoutError, socket.timeout)) else "connection_failed"
    except (OSError, ValueError, http.client.HTTPException):
        result["category"] = "connection_failed"
    result["elapsed_seconds"] = round(time.perf_counter() - start, 3)
    return result


def self_test():
    import io
    import unittest

    class Tests(unittest.TestCase):
        def test_paths(self):
            self.assertEqual(endpoint_for("https://example.com/v1/")[0], "https://example.com/v1/responses")
            self.assertEqual(endpoint_for("https://example.com/openai/v1/responses")[0], "https://example.com/openai/v1/responses")
            self.assertIsNotNone(endpoint_for("https://example.com")[1])

        def test_duplicate(self):
            with self.assertRaisesRegex(ConfigError, "url_duplicate_v1"):
                endpoint_for("https://example.com/v1/v1")

        def test_credentials_and_queries(self):
            for url in ("https://user:secret@example.com/v1", "https://example.com/v1?api_key=secret", "https://example.com/v1#secret"):
                with self.assertRaises(ConfigError):
                    endpoint_for(url)

        def test_http_only_loopback(self):
            for url in ("http://localhost:8000/v1", "http://127.0.0.1:8000/v1", "http://[::1]:8000/v1"):
                endpoint_for(url)
            for url in ("http://example.com/v1", "http://localhost.evil.test/v1", "https://example.com/a/../v1"):
                with self.assertRaises(ConfigError):
                    endpoint_for(url)

        def test_model(self):
            self.assertTrue(model_valid("gpt-6.1-terra"))
            for value in ("sk-secret", "", "a\nb", "x" * 129):
                self.assertFalse(model_valid(value))

        def test_usage_whitelist(self):
            self.assertEqual(clean_usage({"input_tokens": 3, "output_tokens": True, "total_tokens": -1, "secret": "sk-SECRET", "input_tokens_details": {"cached_tokens": 2, "prompt": "SECRET"}}), {"input_tokens": 3, "input_tokens_details": {"cached_tokens": 2}})

        def test_offline_summary_redaction(self):
            _, report = check_config("https://user:SECRET@example.com/v1", "sk-SECRET")
            text = json.dumps(report)
            for secret in ("SECRET", "example.com", "user:"):
                self.assertNotIn(secret, text)

        def test_live_summary_and_single_request(self):
            class FakeResponse(io.BytesIO):
                status = 200
            class FakeOpener:
                calls = 0
                def open(self, req, timeout):
                    self.calls += 1
                    return FakeResponse(json.dumps({"status": "completed", "output": [{"text": "SECRET"}], "usage": {"input_tokens": 4, "output_tokens": 2, "private": "SECRET"}, "api_key": "SECRET"}).encode())
            fake = FakeOpener()
            result = send_once("https://example.com/v1/responses", "model", "SECRET", opener=fake)
            self.assertEqual(fake.calls, 1)
            self.assertEqual(result["category"], "completed")
            self.assertNotIn("SECRET", json.dumps(result))

        def test_errors_and_no_retry(self):
            for code in (401, 404, 429, 503, 302):
                class FakeOpener:
                    calls = 0
                    def open(self, req, timeout):
                        self.calls += 1
                        raise urllib.error.HTTPError(req.full_url, code, "SECRET", {}, io.BytesIO(b"SECRET"))
                fake = FakeOpener()
                result = send_once("https://example.com/v1/responses", "model", "SECRET", opener=fake)
                self.assertEqual(fake.calls, 1)
                self.assertEqual(result["http_status"], code)
                self.assertNotIn("SECRET", json.dumps(result))
            self.assertIsNone(NoRedirect().redirect_request(None, None, 302, "", {}, "https://evil.test"))

        def test_timeout_and_malformed_response(self):
            class TimeoutOpener:
                def open(self, req, timeout):
                    raise urllib.error.URLError(socket.timeout("SECRET"))
            self.assertEqual(send_once("https://example.com/v1/responses", "model", "SECRET", opener=TimeoutOpener())["category"], "timeout")
            class FakeResponse(io.BytesIO):
                status = 200
            class BadBodyOpener:
                def open(self, req, timeout):
                    return FakeResponse(b"SECRET not json")
            result = send_once("https://example.com/v1/responses", "model", "SECRET", opener=BadBodyOpener())
            self.assertEqual(result["category"], "unexpected_response")
            self.assertNotIn("SECRET", json.dumps(result))

    return 0 if unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(Tests)).wasSuccessful() else 1


def main():
    parser = SafeParser(description="Responses API の接続設定をローカルで確認します。既定では通信しません。")
    parser.add_argument("--base-url", default=DEFAULT_BASE)
    parser.add_argument("--model", default=DEFAULT_MODEL)
    parser.add_argument("--live", action="store_true", help="有料となり得る固定テストを一回送信します")
    parser.add_argument("--report", metavar="FILE", help="入力や本文を含まない共有用 JSON を保存")
    parser.add_argument("--self-test", action="store_true", help="通信もキー読取もしない自己テスト")
    args = parser.parse_args()
    if args.self_test:
        return self_test()
    endpoint, report = check_config(args.base_url, args.model)
    valid = not any(item["status"] == "fail" for item in report["checks"])
    if args.live and valid:
        report["mode"] = "live"
        print("固定の短文を一回送信します。利用残高を消費する場合があります。自動リトライはしません。")
        key = os.environ.get(KEY_ENV, "")
        if not key and sys.stdin.isatty():
            try:
                with warnings.catch_warnings():
                    warnings.simplefilter("error", getpass.GetPassWarning)
                    key = getpass.getpass("API key (非表示): ")
            except (EOFError, KeyboardInterrupt, getpass.GetPassWarning):
                key = ""
        if not key:
            report["result"] = {"category": "credential_required"}
        elif re.search(r"[\s\x00-\x1f\x7f]", key):
            report["result"] = {"category": "credential_format"}
        else:
            report["result"] = send_once(endpoint, args.model, key)
        key = ""
    print("モード:", report["mode"])
    for check in report["checks"]:
        print(check["status"].upper() + ": " + MESSAGES.get(check["id"], check["id"]))
    if "result" in report:
        print(json.dumps(report["result"], ensure_ascii=False, indent=2))
        if report["result"]["category"] in ("timeout", "connection_failed"):
            print("再試行の前に利用履歴を確認してください。応答がなくても処理済みの場合があります。")
    else:
        print("通信・キーの読み取りは行っていません。これは接続成功の証明ではありません。")
    if args.report:
        try:
            with open(args.report, "w", encoding="utf-8") as output:
                json.dump(report, output, ensure_ascii=False, indent=2)
                output.write("\n")
            print("共有用 JSON を保存しました。")
        except OSError:
            print("共有用 JSON を保存できませんでした。", file=sys.stderr)
            return 2
    return 0 if valid and (not args.live or report.get("result", {}).get("category") == "completed") else 1


if __name__ == "__main__":
    raise SystemExit(main())