#!/usr/bin/env python3 """Small web UI for exporting Sber operations to XLSX and JSON.""" from __future__ import annotations import argparse import html import json import re import zipfile from datetime import datetime from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from tempfile import TemporaryDirectory from urllib.parse import parse_qs from uuid import uuid4 from load_data import SberMonthHistoryExporter from netscape_cookies.parse_netscape_cookies import build_cookie_header, parse_cookies MAX_FORM_BYTES = 2 * 1024 * 1024 EXPORT_DIR = Path("data") / "exports" PERIOD_RE = re.compile(r"^(\d{4})-(\d{2})(?:-\d{2})?$") PAGE_HTML = """ Sber History Export

Экспорт операций Сбера

Вставьте cookies в формате Netscape, JSON-массива или обычного Cookie header.

__MESSAGE__
  1. Установите расширение Get cookies.txt LOCALLY.
  2. Откройте СберБанк Онлайн и авторизуйтесь.
  3. Скопируйте cookies через расширение и вставьте их в поле ниже.
В архиве будут Excel и JSON со списком всех операций.
""" class ExportHandler(BaseHTTPRequestHandler): server_version = "SberExportHTTP/1.0" def do_GET(self) -> None: if self.path not in ("/", "/index.html"): self.send_error(HTTPStatus.NOT_FOUND, "Not found") return self._send_page() def do_POST(self) -> None: if self.path == "/validate-cookies": self._handle_validate_cookies() return if self.path != "/export": self.send_error(HTTPStatus.NOT_FOUND, "Not found") return try: fields = self._read_form() cookies_text = self._field(fields, "cookies").strip() year, month = parse_period(self._field(fields, "period")) cookie_header, _ = parse_cookie_header_from_text(cookies_text) archive_name, archive_bytes, count = export_archive(cookie_header, year, month) except Exception as exc: self._send_page(str(exc), status=HTTPStatus.BAD_REQUEST) return self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "application/zip") self.send_header("Content-Disposition", f'attachment; filename="{archive_name}"') self.send_header("Content-Length", str(len(archive_bytes))) self.send_header("X-Operations-Count", str(count)) self.end_headers() self.wfile.write(archive_bytes) def _handle_validate_cookies(self) -> None: try: fields = self._read_form() _, cookie_count = parse_cookie_header_from_text(self._field(fields, "cookies").strip()) payload = {"ok": True, "count": cookie_count} status = HTTPStatus.OK except Exception as exc: payload = {"ok": False, "error": str(exc)} status = HTTPStatus.BAD_REQUEST self._send_json(payload, status=status) def log_message(self, format: str, *args: object) -> None: print(f"{self.address_string()} - {format % args}") def _read_form(self) -> dict[str, list[str]]: content_length = int(self.headers.get("Content-Length", "0")) if content_length > MAX_FORM_BYTES: raise ValueError("Слишком большой ввод cookies.") raw_body = self.rfile.read(content_length).decode("utf-8") return parse_qs(raw_body, keep_blank_values=True) def _field(self, fields: dict[str, list[str]], name: str, default: str = "") -> str: values = fields.get(name) return values[0] if values else default def _send_page(self, message: str = "", status: HTTPStatus = HTTPStatus.OK) -> None: now = datetime.now() notice = "" if message: notice = f'
{html.escape(message)}
' body = ( PAGE_HTML .replace("__MESSAGE__", notice) .replace("__PERIOD__", f"{now.year}-{now.month:02d}") ) encoded = body.encode("utf-8") self.send_response(status) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) def _send_json(self, payload: dict[str, object], status: HTTPStatus = HTTPStatus.OK) -> None: encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) def parse_cookie_header_from_text(cookies_text: str) -> tuple[str, int]: if not cookies_text: raise ValueError("Вставьте cookies.") cookies = parse_cookies(cookies_text, input_format="auto") cookie_header = build_cookie_header(cookies) if not cookie_header: raise ValueError("Не удалось получить ни одной cookie из введенного текста.") return cookie_header, len(cookies) def parse_period(value: str) -> tuple[int, int]: match = PERIOD_RE.match(value.strip()) if not match: raise ValueError("Выберите период в формате ГГГГ-ММ.") year = int(match.group(1)) month = int(match.group(2)) if month < 1 or month > 12: raise ValueError("Месяц должен быть от 1 до 12.") return year, month def export_archive(cookie_header: str, year: int, month: int) -> tuple[str, bytes, int]: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") base_name = f"sber_operations_{year}_{month:02d}_{timestamp}_{uuid4().hex[:8]}" EXPORT_DIR.mkdir(parents=True, exist_ok=True) archive_path = EXPORT_DIR / f"{base_name}.zip" with TemporaryDirectory() as tmp_dir: tmp_path = Path(tmp_dir) xlsx_path = tmp_path / f"{base_name}.xlsx" json_path = tmp_path / f"{base_name}.json" exporter = SberMonthHistoryExporter(cookie=cookie_header) _, _, count = exporter.save_month_bundle(year, month, xlsx_path, json_path) with zipfile.ZipFile(archive_path, "w", compression=zipfile.ZIP_DEFLATED) as archive: archive.write(xlsx_path, arcname=xlsx_path.name) archive.write(json_path, arcname=json_path.name) return archive_path.name, archive_path.read_bytes(), count def create_server(host: str, port: int) -> ThreadingHTTPServer: return ThreadingHTTPServer((host, port), ExportHandler) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run web UI for Sber operations export.") parser.add_argument("--host", default="127.0.0.1", help="Host to bind. Default: 127.0.0.1") parser.add_argument("--port", type=int, default=8000, help="Port to bind. Default: 8000") return parser.parse_args() def main() -> int: args = parse_args() server = create_server(args.host, args.port) print(f"Open http://{args.host}:{args.port}") try: server.serve_forever() except KeyboardInterrupt: print("\nStopping server") finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())