vtbreg_parser.py - master - Ged52/vtbreg_parser

vtbreg_parser.py
227 строк8 KB
Ged52
+ date_report
05 янв 2026, 17:06
05 янв 2026, 17:066529a58
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Почему-то новая версия кабинета не запрашивает код подтверждения.

import os
import json
import time
import logging
import configparser
import requests
from getpass import getpass
from datetime import datetime

# === НАСТРОЙКИ ===
CONFIG_FILE = "config.ini"
SESSION_FILE = "vtbreg_session.json"
OUTPUT_FILE = "vtbreg_portfolio_dividends.json"
BASE_URL = "https://pos.vtbreg.ru/marketplace"

# === ЛОГГИРОВАНИЕ ===
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%H:%M:%S"
)
logger = logging.getLogger(__name__)

# === ЗАГРУЗКА КОНФИГУРАЦИИ ===
def load_config(path=CONFIG_FILE):
    email = None
    password = None

    if os.path.exists(path):
        config = configparser.ConfigParser()
        config.read(path, encoding="utf-8")
        if "vtbreg" in config:
            email = config["vtbreg"].get("email", "").strip()
            password = config["vtbreg"].get("password", "").strip()
            # Если пароль пустой — сбросим его
            if not password:
                password = None

    # Запрашиваем недостающие данные
    if not email:
        email = input("📧 Email (логин): ").strip()
    if not password:
        password = getpass("🔑 Пароль (секретный ввод): ")

    return email, password

# === ОСТАЛЬНОЙ КОД БЕЗ ИЗМЕНЕНИЙ ===

def fix_encoding(text):
    if isinstance(text, str):
        try:
            return text.encode('latin1').decode('utf-8')
        except (UnicodeEncodeError, UnicodeDecodeError):
            return text
    return text

def deep_fix_encoding(obj):
    if isinstance(obj, dict):
        return {k: deep_fix_encoding(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [deep_fix_encoding(item) for item in obj]
    elif isinstance(obj, str):
        return fix_encoding(obj)
    else:
        return obj

def save_session(session_id, path=SESSION_FILE):
    with open(path, "w", encoding="utf-8") as f:
        json.dump({"session_id": session_id}, f, ensure_ascii=False, indent=2)

def load_session(path=SESSION_FILE):
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f).get("session_id")
    return None

def api_request(session, url, params=None, method='GET', session_id=None):
    headers = {
        "accept": "application/json, text/plain, */*",
        "accept-language": "ru,en;q=0.9",
        "content-type": "application/json;charset=UTF-8",
        "origin": "https://pos.vtbreg.ru",
        "referer": "https://pos.vtbreg.ru/marketplace/v2/auth/login",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 YaBrowser/25.10.0.0 Safari/537.36",
        "sec-fetch-site": "same-origin",
        "sec-fetch-mode": "cors",
        "sec-fetch-dest": "empty"
    }
    if session_id:
        headers["X-Session-Id"] = session_id

    try:
        logger.debug(f"➡️ ЗАПРОС: {method} {url}")
        if session_id:
            logger.debug(f"   X-Session-Id: {session_id}")
        if method == 'POST' and params:
            logger.debug(f"   Тело: {json.dumps(params, ensure_ascii=False)}")

        if method == 'GET':
            resp = session.get(url, params=params, headers=headers, timeout=15)
        else:
            resp = session.post(url, json=params, headers=headers, timeout=15)

        logger.debug(f"⬅️ ОТВЕТ: {resp.status_code} {resp.reason}")
        if resp.content:
            try:
                data = resp.json()
                fixed_data = deep_fix_encoding(data)
                logger.debug(f"   Тело: {json.dumps(fixed_data, ensure_ascii=False, indent=2)}")
                return fixed_data
            except ValueError:
                text = resp.text
                logger.debug(f"   Тело (не JSON): {text[:500]}{'...' if len(text) > 500 else ''}")
                return text
        else:
            logger.debug("   Тело: <пусто>")
            return None

    except Exception as e:
        logger.error(f"❌ Ошибка запроса {url}: {e}")
        return None

def authenticate_with_credentials(email, password):
    logger.info("🔐 Авторизация по логину и паролю...")
    session = requests.Session()
    data = api_request(session, f"{BASE_URL}/api/mp/v2/auth-web/login",
                       params={"email": email, "password": password},
                       method='POST')
    if not data or "session" not in data:
        raise RuntimeError("Не удалось авторизоваться. Проверьте логин и пароль.")
    session_id = data["session"]
    logger.info("✅ Авторизация успешна")
    return session, session_id

def check_auth(session, session_id):
    logger.debug("🔍 Проверка активности сессии...")
    data = api_request(session, f"{BASE_URL}/api/mp/v2/auth-web/check", session_id=session_id)
    ok = data and data.get("result") == "OK"
    logger.debug(f"   Сессия {'активна' if ok else 'просрочена'}")
    return ok

def fetch_user_info(session, session_id):
    logger.info("📥 Получение информации о пользователе...")
    return api_request(session, f"{BASE_URL}/api/mp/v2/user", session_id=session_id)

def fetch_portfolio(session, session_id):
    logger.info("💼 Загрузка портфеля...")
    count_data = api_request(session, f"{BASE_URL}/api/mp/v2/portfolio/securities/count", session_id=session_id)
    if not count_data or not count_data.get("allCount", 0):
        return []

    data = api_request(session, f"{BASE_URL}/api/mp/v2/portfolio/securities", {
        "pageNum": 1,
        "pageSize": 100,
        "sort": "emitent",
        "descend": "true"
    }, session_id=session_id)
    return data.get("list", []) if data else []

def fetch_dividends(session, session_id):
    logger.info("🧾 Получение истории дивидендов...")
    data = api_request(session, f"{BASE_URL}/api/mp/v2/portfolio/dividends", {
        "sort": "emitent",
        "descend": "true"
    }, session_id=session_id)
    return data or []

def main():
    email, password = load_config()
    session_id = load_session()
    session = requests.Session()

    if session_id:
        if check_auth(session, session_id):
            logger.info("✅ Сессия восстановлена из файла")
        else:
            logger.warning("⚠️ Сессия устарела. Требуется повторная авторизация.")
            session_id = None

    if not session_id:
        session, session_id = authenticate_with_credentials(email, password)
        save_session(session_id)

    user_info = fetch_user_info(session, session_id)
    portfolio = fetch_portfolio(session, session_id)
    dividends = fetch_dividends(session, session_id)

    now = datetime.now()
    result = {
        'date_report': now.strftime("%Y-%m-%d %H:%M:%S"),
        "user": user_info,
        "portfolio": portfolio,
        "dividends": dividends,
        "exported_at": time.strftime("%Y-%m-%d %H:%M:%S")
    }

    with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2)

    logger.info(f"✅ Данные сохранены в '{OUTPUT_FILE}'")

    print("\n" + "="*60)
    print("💼 ПОРТФЕЛЬ")
    print("="*60)
    for item in portfolio:
        name = item["name"]
        qty = item["quantity"]
        total = item["totalPrice"]
        print(f"{name:<50} {qty:>8} шт.  на сумму {total}")

    print("\n" + "="*60)
    print("🧾 ДИВИДЕНДЫ")
    print("="*60)
    for d in dividends:
        emitent = d["emitent"]["name"]
        period = d["period"]
        amount = d["amount"]
        date = d["paymentDate"]
        print(f"{date} | {emitent:<30} | {period:<15} | {amount}")

if __name__ == "__main__":
    main()