#!/usr/bin/env python3 """Download one explicitly selected BOX event page; no pagination loop or retries.""" import argparse import json import math import os import ssl import sys from http.client import HTTPException from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit from urllib.request import HTTPSHandler, HTTPRedirectHandler, Request, build_opener ENDPOINT = "/api/statistics/event-storage/events/" MAX_BYTES = 10 * 1024 * 1024 class FetchError(Exception): """A safe diagnostic that contains neither response data nor credentials.""" class NoRedirect(HTTPRedirectHandler): def redirect_request(self, request, fp, code, message, headers, newurl): # An auth header must never follow a redirect to another origin. return None def origin_url(value): try: parsed = urlsplit(value) valid = ( parsed.scheme == "https" and bool(parsed.hostname) and parsed.username is None and parsed.password is None and parsed.path in ("", "/") and not parsed.query and not parsed.fragment and not any(char.isspace() or ord(char) < 32 or ord(char) > 126 for char in value) ) parsed.port # Validate the optional port without displaying the supplied URL. except ValueError: valid = False if not valid: raise FetchError("BOX_ORIGIN должен быть HTTPS-origin без пути, query и учётных данных; домен — ASCII/Punycode") return value.rstrip("/") def positive_integer(value): number = int(value) if number < 1: raise argparse.ArgumentTypeError("нужно положительное целое число") return number def fetch_page(origin, token, page, page_size, sorting="-timestamp", timeout=15): origin = origin_url(origin) if not token or any(ord(char) <= 32 or ord(char) > 126 for char in token): raise FetchError("BOX_ACCESS_TOKEN должен содержать access_token без префикса Bearer") if type(page) is not int or page < 1 or type(page_size) is not int or page_size < 1: raise FetchError("page и page_size должны быть положительными integer") if not isinstance(sorting, str) or not sorting or any(ord(char) < 32 for char in sorting): raise FetchError("sorting должен быть непустой строкой из описания API сервера") if not math.isfinite(timeout) or timeout <= 0: raise FetchError("timeout должен быть положительным конечным числом") query = urlencode({"page": page, "page_size": page_size, "sorting": sorting}) request = Request(origin + ENDPOINT + "?" + query, headers={ "Authorization": "Bearer " + token, "Accept": "application/json", }, method="GET") # The standard trust store and hostname verification remain enabled. opener = build_opener(HTTPSHandler(context=ssl.create_default_context()), NoRedirect()) try: with opener.open(request, timeout=timeout) as response: if response.status != 200: raise FetchError("HTTP " + str(response.status) + ": страница не сохранена") body = response.read(MAX_BYTES + 1) except HTTPError as error: code = error.code error.close() if 300 <= code < 400: raise FetchError("HTTP " + str(code) + ": перенаправление запрещено; проверьте BOX_ORIGIN") from None raise FetchError("HTTP " + str(code) + ": проверьте права, токен и параметры в Swagger") from None except (URLError, TimeoutError, OSError, HTTPException, ValueError): # Exception messages may contain URL, proxy details or response data. raise FetchError("Сетевая ошибка, ошибка TLS или таймаут; страница не сохранена") from None if len(body) > MAX_BYTES: raise FetchError("Ответ превышает 10 МиБ; уменьшите page_size") try: data = json.loads(body) except (ValueError, UnicodeError, RecursionError): raise FetchError("Ответ не содержит корректный JSON; страница не сохранена") from None if not isinstance(data, dict) or not isinstance(data.get("events"), list): raise FetchError("В ответе ожидается JSON-объект с массивом events; проверьте схему ответа API сервера") return data def download_page(output, **options): output = Path(output) if os.path.lexists(output): raise FetchError("Выходной файл уже существует; выберите новое имя") data = fetch_page(**options) try: # O_EXCL also protects against a file appearing after the check above. descriptor = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) except FileExistsError: raise FetchError("Выходной файл уже существует; выберите новое имя") from None except OSError: raise FetchError("Не удалось создать выходной файл; проверьте каталог и права") from None try: with os.fdopen(descriptor, "w", encoding="utf-8") as stream: json.dump(data, stream, ensure_ascii=False, indent=2, allow_nan=False) stream.write("\n") except (OSError, ValueError, RecursionError): try: output.unlink(missing_ok=True) except OSError: raise FetchError("Запись JSON прервана; проверьте и удалите неполный выходной файл") from None raise FetchError("Не удалось сохранить JSON; проверьте данные, место и права") from None return {"status": "saved", "events": len(data["events"]), "page_requested": options["page"], "page_size_requested": options["page_size"]} def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--page", required=True, type=positive_integer) parser.add_argument("--page-size", required=True, type=positive_integer) parser.add_argument("--sorting", default="-timestamp", help="для минуса: --sorting=-timestamp") parser.add_argument("--timeout", type=float, default=15, help="таймаут сетевых операций, секунды") parser.add_argument("--output", required=True, help="новый JSON-файл; перезапись запрещена") args = parser.parse_args(argv) try: result = download_page(args.output, origin=os.environ.get("BOX_ORIGIN", ""), token=os.environ.get("BOX_ACCESS_TOKEN", "").strip(), page=args.page, page_size=args.page_size, sorting=args.sorting, timeout=args.timeout) except FetchError as error: print("Ошибка: " + str(error), file=sys.stderr) return 1 print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 if __name__ == "__main__": sys.exit(main())