#!/usr/bin/env python3 """Troubleshoot Python JSONDecodeError caused by invalid API responses. This script helps operators inspect an API response before parsing it as JSON. It prints a concise diagnostic report with: - HTTP status code - Content-Type and related headers - Raw body preview - JSON parse outcome - Suggested next steps based on common failure patterns Usage examples: python troubleshoot_jsondecodeerror.py --url https://api.example.com/resource python troubleshoot_jsondecodeerror.py --file response.txt python troubleshoot_jsondecodeerror.py --method POST --url https://api.example.com/resource --header 'Accept: application/json' Notes: - No secrets are stored in the script. - Do not use this as a silent fallback parser; it is intentionally strict. - For production use, validate the response contract before parsing. """ from __future__ import annotations import argparse import json import sys from dataclasses import dataclass from typing import Dict, Iterable, Optional, Tuple from urllib.error import HTTPError, URLError from urllib.parse import urlparse from urllib.request import Request, urlopen DEFAULT_PREVIEW_BYTES = 500 DEFAULT_TIMEOUT = 10 @dataclass class ResponseSnapshot: status: Optional[int] headers: Dict[str, str] body: bytes source: str def parse_headers(header_items: Iterable[str]) -> Dict[str, str]: headers: Dict[str, str] = {} for item in header_items: if ":" not in item: raise ValueError(f"Invalid header format: {item!r}. Expected 'Name: value'.") name, value = item.split(":", 1) name = name.strip() value = value.strip() if not name or not value: raise ValueError(f"Invalid header format: {item!r}. Header name and value are required.") headers[name] = value return headers def load_from_file(path: str) -> ResponseSnapshot: with open(path, "rb") as f: raw = f.read() # Optional simple format: headers and body separated by a blank line. # If the file is raw body only, this still works. header_block, sep, body = raw.partition(b"\n\n") headers: Dict[str, str] = {} status: Optional[int] = None if sep: header_lines = header_block.decode("utf-8", errors="replace").splitlines() for line in header_lines: if line.lower().startswith("status:"): try: status = int(line.split(":", 1)[1].strip()) except ValueError: status = None elif ":" in line: k, v = line.split(":", 1) headers[k.strip()] = v.strip() body = body.lstrip(b"\r\n") else: body = raw return ResponseSnapshot(status=status, headers=headers, body=body, source=f"file:{path}") def fetch_url(url: str, method: str, headers: Dict[str, str], timeout: int) -> ResponseSnapshot: parsed = urlparse(url) if parsed.scheme not in {"http", "https"}: raise ValueError("URL must start with http:// or https://") req = Request(url=url, method=method.upper(), headers=headers) try: with urlopen(req, timeout=timeout) as resp: body = resp.read() status = getattr(resp, "status", None) resp_headers = {k: v for k, v in resp.headers.items()} return ResponseSnapshot(status=status, headers=resp_headers, body=body, source=f"url:{url}") except HTTPError as e: body = e.read() if hasattr(e, "read") else b"" resp_headers = {k: v for k, v in (e.headers.items() if e.headers else [])} return ResponseSnapshot(status=e.code, headers=resp_headers, body=body, source=f"url:{url}") except URLError as e: raise ConnectionError(f"Failed to reach {url}: {e.reason}") from e def preview_text(body: bytes, limit: int) -> str: return body[:limit].decode("utf-8", errors="replace") def content_type_is_json(content_type: str) -> bool: ct = content_type.lower() return "application/json" in ct or ct.endswith("+json") def diagnose(snapshot: ResponseSnapshot, preview_limit: int) -> Tuple[str, int]: body = snapshot.body headers = snapshot.headers status = snapshot.status content_type = headers.get("Content-Type", headers.get("content-type", "")) preview = preview_text(body, preview_limit).strip() score = 0 lines = [] lines.append(f"Source: {snapshot.source}") lines.append(f"Status: {status if status is not None else 'unknown'}") lines.append(f"Content-Type: {content_type or 'missing'}") lines.append(f"Body length: {len(body)} bytes") lines.append(f"Body preview: {preview[:preview_limit]!r}") if status in {204}: lines.append("Signal: 204 No Content; parsing as JSON will fail because the body is empty.") score += 3 if status in {301, 302, 303, 307, 308}: lines.append("Signal: redirect response; verify the final destination returns JSON and the client follows redirects appropriately.") score += 2 if status in {401, 403}: lines.append("Signal: auth/permission failure; the body may be an HTML or plain-text error page instead of JSON.") score += 3 if status == 404: lines.append("Signal: not found; confirm the endpoint path and base URL.") score += 2 if status == 429: lines.append("Signal: rate limiting; the API may return a non-JSON message or truncated payload.") score += 2 if status is not None and status >= 500: lines.append("Signal: server-side error; inspect upstream logs and reverse proxy behavior.") score += 2 if not body: lines.append("Signal: empty response body.") score += 4 else: stripped = body.lstrip() if stripped.startswith(b"<"): lines.append("Signal: body looks like HTML, not JSON.") score += 4 elif stripped[:1] in {b"{", b"["}: lines.append("Signal: body starts like JSON; corruption, truncation, or encoding may still be present.") score += 1 else: lines.append("Signal: body does not look like JSON at first glance.") score += 2 if not content_type: lines.append("Signal: missing Content-Type header.") score += 2 elif not content_type_is_json(content_type): lines.append("Signal: Content-Type is not JSON-compatible.") score += 3 try: decoded = body.decode("utf-8") json.loads(decoded) lines.append("JSON parse check: success; the body is valid JSON.") except UnicodeDecodeError: lines.append("JSON parse check: failed UTF-8 decoding; verify charset and transport integrity.") score += 2 except json.JSONDecodeError as e: lines.append(f"JSON parse check: failed at line {e.lineno}, column {e.colno}: {e.msg}") score += 2 lines.append("") lines.append("Recommended next steps:") if not content_type_is_json(content_type): lines.append("- Gate parsing on a JSON-compatible Content-Type and log unexpected responses.") if status in {401, 403}: lines.append("- Verify authentication, token freshness, and required scopes/headers.") if status in {301, 302, 303, 307, 308}: lines.append("- Inspect redirect targets and ensure the final response contract is JSON.") if status == 429: lines.append("- Check rate limiting headers and use bounded backoff instead of blind retries.") if not body: lines.append("- Check upstream logs, proxy behavior, and whether the endpoint legitimately returns no content.") if body and preview.startswith("<"): lines.append("- Investigate HTML error pages from proxies, WAFs, or application servers.") lines.append("- Compare this response with a known-good baseline request.") lines.append("- If the API contract permits non-JSON error bodies, handle that branch explicitly.") return "\n".join(lines), score def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Troubleshoot JSONDecodeError by inspecting API responses before parsing.", ) source = parser.add_mutually_exclusive_group(required=True) source.add_argument("--url", help="HTTP or HTTPS URL to inspect") source.add_argument("--file", help="Path to a saved response snapshot or raw body file") parser.add_argument("--method", default="GET", help="HTTP method to use when fetching a URL (default: GET)") parser.add_argument("--header", action="append", default=[], help="Extra request header in 'Name: value' format") parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help="Request timeout in seconds") parser.add_argument("--preview-bytes", type=int, default=DEFAULT_PREVIEW_BYTES, help="Number of bytes to preview") return parser def main() -> int: parser = build_arg_parser() args = parser.parse_args() if args.timeout <= 0: parser.error("--timeout must be greater than 0") if args.preview_bytes <= 0: parser.error("--preview-bytes must be greater than 0") try: extra_headers = parse_headers(args.header) if args.url: snapshot = fetch_url(args.url, args.method, extra_headers, args.timeout) else: snapshot = load_from_file(args.file) report, score = diagnose(snapshot, args.preview_bytes) print(report) print("") print(f"Triage confidence: {score}/15 (higher means more likely the issue is response-shape related)") return 0 except Exception as exc: print(f"Error: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())