487 lines
14 KiB
Python
487 lines
14 KiB
Python
#!/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 = """<!doctype html>
|
||
<html lang="ru">
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||
<title>Sber History Export</title>
|
||
<style>
|
||
:root {
|
||
color-scheme: light;
|
||
--bg: #f5f7fa;
|
||
--panel: #ffffff;
|
||
--text: #17202a;
|
||
--muted: #5f6b7a;
|
||
--line: #d9e0e8;
|
||
--accent: #12805c;
|
||
--accent-dark: #0b6248;
|
||
--danger: #a33131;
|
||
--danger-bg: #fff1f1;
|
||
}
|
||
|
||
* {
|
||
box-sizing: border-box;
|
||
}
|
||
|
||
body {
|
||
margin: 0;
|
||
min-height: 100vh;
|
||
background: var(--bg);
|
||
color: var(--text);
|
||
font: 16px/1.5 -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
|
||
}
|
||
|
||
main {
|
||
width: min(920px, calc(100% - 32px));
|
||
margin: 0 auto;
|
||
padding: 40px 0;
|
||
}
|
||
|
||
.shell {
|
||
background: var(--panel);
|
||
border: 1px solid var(--line);
|
||
border-radius: 8px;
|
||
box-shadow: 0 12px 28px rgba(31, 45, 61, 0.08);
|
||
overflow: hidden;
|
||
}
|
||
|
||
header {
|
||
padding: 28px 32px 22px;
|
||
border-bottom: 1px solid var(--line);
|
||
}
|
||
|
||
h1 {
|
||
margin: 0 0 8px;
|
||
font-size: 28px;
|
||
line-height: 1.15;
|
||
letter-spacing: 0;
|
||
}
|
||
|
||
p {
|
||
margin: 0;
|
||
color: var(--muted);
|
||
}
|
||
|
||
form {
|
||
padding: 28px 32px 32px;
|
||
display: grid;
|
||
gap: 22px;
|
||
}
|
||
|
||
label {
|
||
display: grid;
|
||
gap: 8px;
|
||
font-weight: 650;
|
||
}
|
||
|
||
.steps {
|
||
margin: 0;
|
||
padding-left: 22px;
|
||
color: var(--text);
|
||
}
|
||
|
||
.steps li {
|
||
margin: 6px 0;
|
||
}
|
||
|
||
.steps a {
|
||
color: var(--accent-dark);
|
||
font-weight: 650;
|
||
}
|
||
|
||
textarea,
|
||
input,
|
||
select {
|
||
width: 100%;
|
||
border: 1px solid var(--line);
|
||
border-radius: 6px;
|
||
color: var(--text);
|
||
background: #fff;
|
||
font: inherit;
|
||
}
|
||
|
||
textarea {
|
||
min-height: 220px;
|
||
resize: vertical;
|
||
padding: 14px;
|
||
font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace;
|
||
font-size: 14px;
|
||
line-height: 1.45;
|
||
}
|
||
|
||
input,
|
||
select {
|
||
min-height: 44px;
|
||
padding: 0 12px;
|
||
}
|
||
|
||
.grid {
|
||
display: grid;
|
||
grid-template-columns: minmax(0, 1fr) minmax(180px, 240px);
|
||
gap: 18px;
|
||
align-items: end;
|
||
}
|
||
|
||
.grid.single {
|
||
grid-template-columns: minmax(180px, 320px);
|
||
}
|
||
|
||
.hint {
|
||
color: var(--muted);
|
||
font-size: 14px;
|
||
font-weight: 400;
|
||
}
|
||
|
||
.hint a {
|
||
color: var(--accent-dark);
|
||
font-weight: 650;
|
||
}
|
||
|
||
.cookie-status {
|
||
min-height: 22px;
|
||
font-size: 14px;
|
||
font-weight: 500;
|
||
}
|
||
|
||
.cookie-status.ok {
|
||
color: var(--accent-dark);
|
||
}
|
||
|
||
.cookie-status.error {
|
||
color: var(--danger);
|
||
}
|
||
|
||
.actions {
|
||
display: flex;
|
||
gap: 12px;
|
||
align-items: center;
|
||
flex-wrap: wrap;
|
||
}
|
||
|
||
button {
|
||
min-height: 46px;
|
||
border: 0;
|
||
border-radius: 6px;
|
||
padding: 0 18px;
|
||
background: var(--accent);
|
||
color: #fff;
|
||
font: inherit;
|
||
font-weight: 700;
|
||
cursor: pointer;
|
||
}
|
||
|
||
button:hover {
|
||
background: var(--accent-dark);
|
||
}
|
||
|
||
.notice {
|
||
border: 1px solid var(--line);
|
||
border-radius: 6px;
|
||
padding: 12px 14px;
|
||
background: #f9fbfc;
|
||
color: var(--muted);
|
||
font-size: 14px;
|
||
}
|
||
|
||
.error {
|
||
border-color: #e7b5b5;
|
||
background: var(--danger-bg);
|
||
color: var(--danger);
|
||
}
|
||
|
||
@media (max-width: 680px) {
|
||
main {
|
||
width: min(100% - 20px, 920px);
|
||
padding: 18px 0;
|
||
}
|
||
|
||
header,
|
||
form {
|
||
padding-left: 18px;
|
||
padding-right: 18px;
|
||
}
|
||
|
||
.grid {
|
||
grid-template-columns: 1fr;
|
||
}
|
||
|
||
h1 {
|
||
font-size: 24px;
|
||
}
|
||
}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<main>
|
||
<section class="shell">
|
||
<header>
|
||
<h1>Экспорт операций Сбера</h1>
|
||
<p>Вставьте cookies в формате Netscape, JSON-массива или обычного Cookie header.</p>
|
||
</header>
|
||
<form method="post" action="/export">
|
||
__MESSAGE__
|
||
<div>
|
||
<ol class="steps">
|
||
<li>Установите расширение <a href="https://chromewebstore.google.com/detail/get-cookiestxt-locally/cclelndahbckbenkjhflpdbgdldlbecc?pli=1" target="_blank" rel="noopener noreferrer">Get cookies.txt LOCALLY</a>.</li>
|
||
<li>Откройте <a href="https://online.sberbank.ru/CSAFront/index.do" target="_blank" rel="noopener noreferrer">СберБанк Онлайн</a> и авторизуйтесь.</li>
|
||
<li>Скопируйте cookies через расширение и вставьте их в поле ниже.</li>
|
||
</ol>
|
||
</div>
|
||
<label>
|
||
Cookies
|
||
<textarea name="cookies" spellcheck="false" autocomplete="off" required placeholder="session_id=...; other_cookie=..."></textarea>
|
||
<span class="cookie-status" id="cookie-status" role="status" aria-live="polite"></span>
|
||
<span class="hint">
|
||
Cookies используются только для запроса к Сберу и не сохраняются.
|
||
Получить `cookies.txt` можно через
|
||
<a href="https://chromewebstore.google.com/detail/get-cookiestxt-locally/cclelndahbckbenkjhflpdbgdldlbecc?pli=1" target="_blank" rel="noopener noreferrer">Get cookies.txt LOCALLY</a>.
|
||
</span>
|
||
</label>
|
||
<div class="grid single">
|
||
<label>
|
||
Период
|
||
<input type="month" name="period" value="__PERIOD__" required>
|
||
</label>
|
||
</div>
|
||
<div class="actions">
|
||
<button type="submit">Экспортировать ZIP</button>
|
||
<span class="hint">В архиве будут Excel и JSON со списком всех операций.</span>
|
||
</div>
|
||
</form>
|
||
</section>
|
||
</main>
|
||
<script>
|
||
const cookiesInput = document.querySelector('textarea[name="cookies"]');
|
||
const statusNode = document.querySelector('#cookie-status');
|
||
let validateTimer = null;
|
||
|
||
function setCookieStatus(text, mode) {
|
||
statusNode.textContent = text;
|
||
statusNode.className = mode ? `cookie-status ${mode}` : 'cookie-status';
|
||
}
|
||
|
||
async function validateCookies() {
|
||
const cookies = cookiesInput.value.trim();
|
||
if (!cookies) {
|
||
setCookieStatus('', '');
|
||
return;
|
||
}
|
||
|
||
try {
|
||
const response = await fetch('/validate-cookies', {
|
||
method: 'POST',
|
||
headers: { 'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8' },
|
||
body: new URLSearchParams({ cookies }),
|
||
});
|
||
const result = await response.json();
|
||
if (result.ok) {
|
||
setCookieStatus(`Формат распознан, cookies: ${result.count}`, 'ok');
|
||
} else {
|
||
setCookieStatus(result.error || 'Cookies не удалось распарсить.', 'error');
|
||
}
|
||
} catch {
|
||
setCookieStatus('Не удалось проверить cookies.', 'error');
|
||
}
|
||
}
|
||
|
||
cookiesInput.addEventListener('input', () => {
|
||
clearTimeout(validateTimer);
|
||
validateTimer = setTimeout(validateCookies, 250);
|
||
});
|
||
</script>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
|
||
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'<div class="notice error">{html.escape(message)}</div>'
|
||
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())
|