Сбор данных из социальной сети Тинькофф Пульс с помощью python
Вступление
В этой статье я хочу поделиться с вами опытом о том, как просто можно забрать (спарсить) данные из социальной сети Тинькофф “Пульс” c помощью python и sqlite3
Начнем:
Для начала следует понимать, что данные (т.к. они идентичны для всех устройств и хранятся в одном месте), как на сайт, так и в приложение попадают через единую точку REST API - от нас требуется только найти ее через инструменты разработчка в браузере.
instrument и видим несколько запросов, с характерой частью api в строке запроса. 
Давайте скопируем адрес одной из ссылок:https://www.tinkoff.ru/api/invest-gw/social/v1/post/instrume..., как мы видим - строка запроса состоит из нескольких частей:
- Основная (доменное имя)
https://www.tinkoff.ru - Машрутизация
/api/invest-gw/social/v1/post/instrument/AAPL?cursor=3899034&limit=30
В ней нас интересуют части AAPL - название тикера, ?cursor=3899034 - курсор для “обновления” новых постов в момент, когда мы прокручиваем ленту вниз и &limit=30 - кол-во постов, которые будут получены в рамках 1 запроса.
Можно попробовать сделать запрос к API прямо из браузера: Запрос! - работает, двигаемся дальше.
Переходим к коду
Для комфортной работы нам потребуется 2 допольнительных библиотеки: pytz и requests.
Предварительно создадим отдельное виртуальное окружение env.
python -m venv env ; \ source env/bin/activate Установка:
pip install requests && pip install pytz Базовые импорты:
import requests import sqlite3 from time import sleep from datetime import datetime from pytz import timezone Для того, чтобы сервер не заметил того, что мы не являемся человеком и запросы отправляются скриптом - добавим несколько HTTP заголовков в запрос.
headers = { 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.
36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36', 'accept': '*/*', 'content-type': 'application/x-www-form-urlencoded', 'accept-encoding': 'gzip, deflate, br', 'accept-language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7' } Первая функция на запрос и обработку ответа сервера
Заранее подготовим шаблон строки запроса, в который будут подставляться значения
(на месте {})'https://www.tinkoff.ru/api/invest-gw/social/v1/post/instrument/{}?limit=50&appName=invest&platform=web&cursor={}'
def get_data_from_api(url: str, ticker: str, cursor_number: str) -> None: link = url.format(ticker, cursor_number) session = requests.Session() data = session.get(link, headers=headers, stream=True) raw_data = data.json().get('payload').get('items') for i in raw_data: my_data = { 'ticker': ticker, 'id': i.get('id')[:7], 'inserted': convert_inserted(i.get('inserted')), 'instruments': ', '.join(get_instruments(i.get('instruments'))), 'likesCount': i.get('likesCount'), 'nickname': i.get('nickname'), 'commentsCount': i.get('commentsCount'), 'parse_date': datetime.today().strftime('%Y-%m-%d %H:%M:%S') } print(my_data) insert('dt_pulse', my_data) Как видно из кода фукнции - 3м аргументом принимается некий cursor_number, это как раз и есть номер для фиксации нового среза данных. Для его корректного определения напишем отдельную функцию.
def get_cursor(url: str, ticker: str, cursor_number: str) -> str: link = url.format(ticker, cursor_number) session = requests.Session() data = session.get(link, headers=headers, stream=True) cursor_number = data.json().get('payload').get('nextCursor') return cursor_number Определение курсора - есть, функция для получения данных с одного запроса - тоже есть, осталось их объединить в одно целое. Как пример - будем забирать 100 постов с тикера AAPL.
def get_data_from_ticker(ticker: str) -> None: max_cursor = '99999999999' for _ in range(1, 3): get_data_from_api(link, ticker, max_cursor) sleep(0.5) max_cursor = get_cursor(link, ticker, max_cursor) sleep(0.8) # RUN get_data_from_ticker('AAPL') Отлично! Данные по отдельным постам формируются в отдельный dict, давайте перенесем их в базу данных. Как пример - будем использовать sqlite3, которая есть “из коробки” на любой unix машине.
Добавим sql-файл create_table.sql с кодом на создание таблицы:
create table dt_pulse( src_id int primary_key, ticker varchar(10) not null, id char(7) not null, inserted timestamp not null, instruments varchar(1000) not null, likesCount int not null, nickname varchar(55) not null, commentsCount int not null, parse_date timestamp not null ); Инициализация БД и запуск нашего файла.
conn = sqlite3.connect('pulse-parser.db') cursor = conn.cursor() def _init_database(): with open('create_table.sql', 'r') as f: sql = f.read() cursor.execute(sql) conn.commit() Универсалья функция для вставки insert данных в нашу таблицу, которая принимает 2 аргумента:
- Название таблицы
table_name -
Dictс даннымиcolumn_data, из которого сформируется набор столбцов
def insert(table_name: str, column_data: dict) -> None: columns = ', '.join(column_data.keys()) values = [tuple(column_data.values())] placeholders = ', '.join('?' * len(column_data.keys())) cursor.executemany( f'insert into {table_name}({columns}) values({placeholders})', values ) conn.commit() Просмотр полученных данных
Откроем наш файл с бд pulse_lesson.db с помощью команды
sqlite3 pulse_lesson.db Внутри консоли выполним:
-
.mode box- для вывода данных в табличном виде -
.tables- для просмотра списка таблиц
select id , ticker , inserted , likesCount , nickname , commentsCount from dt_pulse limit 5; Результат:
Добавим немного группировки
select ticker , max(likesCount) as max_likes , max(commentsCount) as max_cnts from dt_pulse group by ticker; 
Распределение по кол-ву лайков AAPL
with t1 as ( select likesCount , count(*) as cnt from dt_pulse where ticker = 'AAPL' group by likesCount ) select 'AAPL' as ticker , likesCount , cnt , printf('%.' || (cnt*30 / (select max(cnt) from t1)) || 'c', '*') as bar from t1 order by likesCount asc limit 15; Результат:
Распределение по кол-ву лайков SBER
with t1 as ( select likesCount , count(*) as cnt from dt_pulse where ticker = 'SBER' group by likesCount ) select 'SBER' as ticker , likesCount , cnt , printf('%.' || (cnt*30 / (select max(cnt) from t1)) || 'c', '*') as bar from t1 order by likesCount asc limit 15 Результат:
Полный код (ровно 100 строк)
import requests import sqlite3 from time import sleep from pytz import timezone from datetime import datetime headers = { 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36', 'accept': '*/*', 'content-type': 'application/x-www-form-urlencoded', 'accept-encoding': 'gzip, deflate, br', 'accept-language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7' } link = 'https://www.tinkoff.ru/api/invest-gw/social/v1/post/instrument/{}?limit=50&appName=invest&platform=web&cursor={}' conn = sqlite3.connect('pulse_lesson.db') cursor = conn.cursor() def _init_database(): with open('create_table.sql', 'r') as f: sql = f.read() cursor.execute(sql) conn.commit() def insert(table_name: str, column_data: dict) -> None: columns = ', '.join(column_data.keys()) values = [tuple(column_data.values())] placeholders = ', '.join('?' * len(column_data.keys())) cursor.executemany( f'insert into {table_name}({columns}) values({placeholders})', values ) conn.commit() def get_instruments(lst: list): data = [] for item in lst: line = f"{item.get('briefName')} !! {item.get('ticker')} !! {item.get('price')} !! {item.get('lastPrice')}" data.append(line) return data def convert_inserted(line): utc_time = datetime.strptime(line, '%Y-%m-%dT%H:%M:%S.%f%z')\ .replace(tzinfo=timezone('utc')).strftime('%Y-%m-%d %H:%M:%S') return utc_time def get_cursor(url: str, ticker: str, cursor_number: str) -> str: link = url.format(ticker, cursor_number) session = requests.Session() data = session.get(link, headers=headers, stream=True) cursor_number = data.json().get('payload').get('nextCursor') return cursor_number def get_data_from_api(url: str, ticker: str, cursor_number: str) -> None: link = url.format(ticker, cursor_number) session = requests.Session() data = session.get(link, headers=headers, stream=True) raw_data = data.json().get('payload').get('items') for i in raw_data: my_data = { 'ticker': ticker, 'id': i.get('id')[:7], 'inserted': convert_inserted(i.get('inserted')), 'instruments': ', '.join(get_instruments(i.get('instruments'))), 'likesCount': i.get('likesCount'), 'nickname': i.get('nickname'), 'commentsCount': i.get('commentsCount'), 'parse_date': datetime.today().strftime('%Y-%m-%d %H:%M:%S') } print(my_data) insert('dt_pulse', my_data) def get_data_from_ticker(ticker: str): max_cursor = '99999999999' for _ in range(1, 3): get_data_from_api(link, ticker, max_cursor) sleep(0.5) max_cursor = get_cursor(link, ticker, max_cursor) sleep(0.8) def main(): _init_database() get_data_from_ticker('AAPL') get_data_from_ticker('AMZN') get_data_from_ticker('TSLA') get_data_from_ticker('SBER') if __name__ == '__main__': main() Если вам интересны подобные рассуждения, подписывайтесь на мой канал artydev & Co.