Поверхнево досліджуючи тему отримання настроїв щодо очікувань від криптовалюти
із Steemit, монети Steem. Способом автоматизованим новими технологіями. Виникла
ще одна хитра думка, щоб таким чином аналізувати дописи. Тобто застосовуючи
якусь там нейронну мережу чи ще щось подібне, тобто якусь штуковину розумну.
Діючу по вказаному для аналізу тексту за заданими параметрами, наприклад
позитивність тексту по ключовим тегам. З цього вийшло не те що хотілось, бо
вихоплювало по API й показувало лише декілька дописів із багатьох опублікованих
у спільноті. Причина проста, треба розум мати як це все налаштувати й навчити, а
там дрімучий ліс технічних знань. Жаль, бо був хитрий задум, що це може
дозволити курувати дописи не читаючи їх, орієнтуючись на підходящі для мене
патерни. Та й все підряд теж не годиться апвотить.
Згодом намір думки трішки змінив вектор, де було б добре отримувати всі свіжі
дописи за вказаним тегом спільноти та мати змогу проголосувати.
Тепер про плазунів. Представниць цього року. Символічне комбо. У всесвіті Гаррі Поттера є така мова - парселтанг (чи якось так), що дозволяє розмовляти зі зміями. Рідкість не бачена. Проте для простих маглів у нашій реальності є інша мова - пітон (Python), яка відкриває безліч можливостей у цифровому світі. Раніше була доступна лише обраним чарівникам, носіям таємних знань і заклиначів програмного коду. Тепер нею може послуговуватись більше бажаючих, однак із допомогою ШІ. І цей во, не такі масштабні проєкти будуть, бо там кумекать треба. Однак щось простіше дійсно можна брати на озброєння навіть із ніяким рівнем знань.
Парсинг Steem
Парсинг - якщо коротко, то це автоматизований збір даних в
інтернеті. Переважно текст для аналізу, але графіку та дещо інше теж можна
витягнути.
Отже, сумісними зусиллями із ШІ. З нього код і обчислення, а з мене питання (не
завжди гострі), запити та перевірки результату. Вийшло щось таке.
Мінімальний набір інформації про автора, дату (дописи не старше 7 днів),
посилання (трохи не дороблено), кнопка голосування, слайдер сили голосу, текст
у спойлері та помітка, що за допис вже було проголосовано. Компактно та швидко
переглядати й залишати голос.
Відкривання вмісту в один клік. Маркдаун теж підтримує й відоражає, проте html не хоче (або потрібно підналаштувати щось).
Голосування. Не так швидко, як на зображенні, що залежить від різних факторів, в
тому числі швидкість інтернету. Не вийшло реалізувати цю функцію steem
бібліотекою для python. Тож для цього задіяно javascript Node.js. На сайті
для розробки https://developers.steem.io/tutorials/ є необхідні дані, щодо
виконання команд.
Код
Всі залучені бібліотеки показано в самому кодові, проте може ще що потрібно встановлювати для його роботи, або навпаки не потрібно. Бо це результат багатьох
спроб та змін, які проводив ШІ на мої запити й свої міркування, як має бути
правильно, щоб працювало. Тож дещо може бути зайвим. Для візуалізації та
взаємодії у веббраузер використовується Streamlit
import requests
import streamlit as st
from textblob import TextBlob
import nltk
from nltk.corpus import stopwords
from nltk.tokenize import RegexpTokenizer
from nltk.sentiment import SentimentIntensityAnalyzer
import steem
import steembase
import subprocess
import json
from datetime import datetime, timedelta
import os
import time
import random
import re
from bs4 import BeautifulSoup
steembase.chains.known_chains['STEEM'] = {
'chain_id': '79276aea5d4877d9a25892eaa01b0adf019d3e5cb12a97478df3298ccdd01673',
'prefix': 'STX', 'steem_symbol': 'STEEM', 'sbd_symbol': 'SBD', 'vests_symbol': 'VESTS'
}
try:
nltk.data.find('corpora/stopwords')
nltk.data.find('sentiment/vader_lexicon')
except LookupError:
st.warning("Зараз завантажуються NLTK ресурси, це може зайняти певний час...")
nltk.download('stopwords')
nltk.download('vader_lexicon')
stop_words = set(stopwords.words('english'))
tokenizer = RegexpTokenizer(r'\b\w[\w-]+\b')
analyzer = SentimentIntensityAnalyzer()
def get_steemit_posts(tags, limit=100, last_permlink=None, address=None):
all_posts = []
now = datetime.utcnow()
if not isinstance(tags, list):
tags = [tags]
if not tags:
return all_posts
while True:
for tag in tags:
params = {"tag": tag, "limit": limit}
if last_permlink:
params["start_permlink"] = last_permlink
try:
response = requests.post('https://api.steemit.com', json={
"jsonrpc": "2.0",
"method": "condenser_api.get_discussions_by_created",
"params": [params],
"id": random.randint(1, 100000)
})
response.raise_for_status()
data = response.json().get('result', [])
if not data:
continue
for post in data:
created_time = datetime.strptime(post['created'], '%Y-%m-%dT%H:%M:%S')
age = now - created_time
if age <= timedelta(days=7):
if address:
if address in f"https://steemit.com/{post['category']}/@{post['author']}/{post['permlink']}":
all_posts.append(post)
else:
all_posts.append(post)
if len(data) < limit:
continue
last_permlink = data[-1]['permlink']
except requests.exceptions.RequestException as e:
st.error(f"Помилка запиту до API: {e}")
print(f"Помилка запиту до API: {e}")
return []
except json.JSONDecodeError as e:
st.error(f"Помилка обробки JSON: {e}")
print(f"Помилка обробки JSON: {e}")
return []
except Exception as e:
st.error(f"Невідома помилка: {e}")
print(f"Невідома помилка: {e}")
return []
if not data or len(data) < limit:
break
return all_posts
def get_steemit_posts_by_link(url, use_api=False):
all_posts = []
try:
if use_api:
match = re.search(r'@(?P<author>[\w-]+)/(?P<permlink>[\w-]+)$', url)
if match:
author = match.group('author')
permlink = match.group('permlink')
response = requests.post('https://api.steemit.com', json={
"jsonrpc": "2.0",
"method": "condenser_api.get_content",
"params": [author, permlink],
"id": random.randint(1, 100000)
})
response.raise_for_status()
data = response.json().get('result', {})
if data:
all_posts.append(data)
else:
st.error(f"Некоректне посилання: {url}")
print(f"Некоректне посилання: {url}")
else:
response = requests.get(url)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
posts = soup.find_all('article', class_='Post')
if posts:
for post in posts:
title_element = post.find('h1', class_='Post__title')
author_element = post.find('a', class_='Post__author')
date_element = post.find('span', class_='Post__date')
body_element = post.find('div', class_='Post__content')
if title_element and author_element and date_element and body_element:
permlink_element = post.find('a', class_='Post__link')
permlink = permlink_element.get('href').split('/')[-1] if permlink_element else ""
category = permlink_element.get('href').split('/')[-2] if permlink_element else ""
all_posts.append({
'title': title_element.get_text(strip=True),
'url': f"{url}",
'author': author_element.get_text(strip=True),
'created': date_element.get_text(strip=True),
'body': body_element.get_text(strip=True),
'permlink': permlink,
'category': category
})
return all_posts
except requests.exceptions.RequestException as e:
st.error(f"Помилка запиту до URL: {e}")
print(f"Помилка запиту до URL: {e}")
return []
except Exception as e:
st.error(f"Невідома помилка: {e}")
print(f"Невідома помилка: {e}")
return []
def analyze_text(text):
try:
sentiment_tb = TextBlob(text).sentiment.polarity
tokens = tokenizer.tokenize(text.lower())
filtered_keywords = [w for w in tokens if w not in stop_words]
sentiment_nltk = analyzer.polarity_scores(text)
return sentiment_tb, filtered_keywords, sentiment_nltk
except Exception as e:
st.error(f"Помилка в analyze_text: {e}")
print(f"Помилка в analyze_text: {e}")
return 0, [], {'compound': 0}
def vote_js(username, posting_key, author, permlink, weight):
js_code = f"""
const steem = require("steem");
steem.api.setOptions({{ url: 'https://api.steemit.com' }});
steem.broadcast.vote(
'{posting_key}',
'{username}',
'{author}',
'{permlink}',
{weight} * 100,
function(err, result) {{
if (err) {{
console.error(err);
}} else {{
console.log(result);
}}
}}
);
"""
try:
node_executable = '/usr/bin/node'
process = subprocess.run([node_executable, '-e', js_code], capture_output=True, text=True, check=True)
return process.stdout
except subprocess.CalledProcessError as e:
return f"Помилка NodeJS: {e.stderr}"
except FileNotFoundError:
return "Помилка: Не знайдено Node.js. Будь ласка, перевірте встановлення Node.js і його шлях."
def has_voted(username, author, permlink, steem_instance):
if not steem_instance:
return False
try:
result = steem_instance.get_active_votes(author, permlink)
return any(vote['voter'] == username for vote in result)
except Exception as e:
print(f"Error checking vote status {e}")
return False
def main():
st.title("Аналіз дописів Steemit")
default_tags = ['ukraine', 'technology', 'health', 'travel']
if 'tag' not in st.session_state:
st.session_state.tag = ""
if 'link' not in st.session_state:
st.session_state.link = ""
if 'address' not in st.session_state:
st.session_state.address = ""
if 'username' not in st.session_state:
st.session_state.username = ""
if 'posting_key' not in st.session_state:
st.session_state.posting_key = ""
if 'use_white_filter' not in st.session_state:
st.session_state.use_white_filter = False
if 'white_list' not in st.session_state:
st.session_state.white_list = ""
if 'use_black_filter' not in st.session_state:
st.session_state.use_black_filter = False
if 'black_list' not in st.session_state:
st.session_state.black_list = ""
if 'use_enhanced_sentiment' not in st.session_state:
st.session_state.use_enhanced_sentiment = False
if 'limit' not in st.session_state:
st.session_state.limit = 100
if 'results' not in st.session_state:
st.session_state.results = []
if 'last_permlink' not in st.session_state:
st.session_state.last_permlink = None
if 'show_all' not in st.session_state:
st.session_state.show_all = False
tag_options = st.selectbox("Виберіть тег:", default_tags + ["Ввести вручну"], key="tag_select")
if tag_options == "Ввести вручну":
tag = st.text_input("Введіть теги через кому:", key="tag_input", value=st.session_state.tag)
else:
tag = tag_options
st.session_state.tag = tag
st.session_state.link = st.text_input("Введіть посилання на допис або на сторінку каталогу:", key="link_input",
value=st.session_state.link)
st.session_state.address = st.text_input("Фільтрувати за адресою (наприклад, @user):",
value=st.session_state.address)
st.session_state.username = st.text_input("Ваш нікнейм:", value=st.session_state.username)
st.session_state.posting_key = st.text_input("Ваш приватний ключ для постингів:", type="password",
value=st.session_state.posting_key)
st.session_state.use_white_filter = st.checkbox("Увімкнути білий список авторів",
value=st.session_state.use_white_filter)
st.session_state.white_list = st.text_input(
"Введіть нікнейми білого списку через кому (наприклад: user1, user2):",
value=st.session_state.white_list).lower() if st.session_state.use_white_filter else ""
st.session_state.use_black_filter = st.checkbox("Увімкнути чорний список авторів",
value=st.session_state.use_black_filter)
st.session_state.black_list = st.text_input(
"Введіть нікнейми чорного списку через кому (наприклад: bad_user1, bad_user2):",
value=st.session_state.black_list).lower() if st.session_state.use_black_filter else ""
st.session_state.use_enhanced_sentiment = st.checkbox("Увімкнути розширений аналіз тональності",
value=st.session_state.use_enhanced_sentiment)
st.session_state.show_all = st.checkbox("Показати усі дописи (без аналізу тональності)", value = st.session_state.show_all)
st.session_state.limit = st.number_input("Кількість дописів для отримання (не впливає на відображення):", min_value=1,
max_value=200, value=st.session_state.limit)
if st.button("Аналізувати"):
with st.spinner("Завантаження дописів..."):
tags = [tag.strip() for tag in st.session_state.tag.split(',') if tag.strip()]
all_posts = []
if st.session_state.link:
if 'steemit.com' in st.session_state.link:
all_posts = get_steemit_posts_by_link(st.session_state.link)
elif 'hive.blog' in st.session_state.link:
all_posts = get_steemit_posts_by_link(st.session_state.link, use_api = False)
else:
all_posts = get_steemit_posts_by_link(st.session_state.link, use_api = False)
else:
all_posts = get_steemit_posts(tags,
limit=st.session_state.limit,
last_permlink=st.session_state.last_permlink,
address=st.session_state.address)
if not all_posts:
st.warning("Не знайдено дописів за вказаними параметрами.")
else:
filtered_results = []
s = None
if st.session_state.username and st.session_state.posting_key:
try:
s = steem.Steem(keys=[st.session_state.posting_key])
except Exception as e:
st.error(f"Помилка з'єднання до API Steem: {e}")
s = None
for post in all_posts:
try:
author = post['author'].lower()
white_list_stripped = [user.strip() for user in st.session_state.white_list.split(",")] if st.session_state.use_white_filter else []
black_list_stripped = [user.strip() for user in st.session_state.black_list.split(",")] if st.session_state.use_black_filter else []
if st.session_state.use_white_filter and author not in white_list_stripped:
continue
if st.session_state.use_black_filter and author in black_list_stripped:
continue
voted = False
if s:
voted = has_voted(st.session_state.username, post['author'], post['permlink'], s)
if not st.session_state.show_all:
sentiment_tb, keywords, sentiment_nltk = analyze_text(post['body'])
if not st.session_state.use_enhanced_sentiment:
if sentiment_tb <= 0:
continue
else:
compound_score = sentiment_nltk['compound']
if compound_score <= 0.1:
continue
filtered_results.append({
'title': post['title'],
'url': post.get('url', f"https://steemit.com/{post['category']}/@{post['author']}/{post['permlink']}") if post.get('url') else f"https://steemit.com/{post['category']}/@{post['author']}/{post['permlink']}",
'author': post['author'],
'created': post['created'],
'body': post['body'],
'permlink': post['permlink'],
'voted': voted,
'keywords': keywords,
'sentiment_tb': sentiment_tb,
'sentiment_nltk': sentiment_nltk
})
else:
filtered_results.append({
'title': post['title'],
'url': post.get('url', f"https://steemit.com/{post['category']}/@{post['author']}/{post['permlink']}") if post.get('url') else f"https://steemit.com/{post['category']}/@{post['author']}/{post['permlink']}",
'author': post['author'],
'created': post['created'],
'body': post['body'],
'permlink': post['permlink'],
'voted': voted,
})
except Exception as e:
st.error(f"Помилка обробки поста: {e}")
st.session_state.results = filtered_results
if 'results' in st.session_state:
for idx, result in enumerate(st.session_state.results):
st.write("---")
vote_status = "✅" if result.get('voted', False) else ""
created_time = result.get('created', "")
if created_time:
created_time = datetime.strptime(created_time, '%Y-%m-%dT%H:%M:%S') if isinstance(created_time, str) and 'T' in created_time else datetime.strptime(created_time, '%Y-%m-%d %H:%M:%S') if isinstance(created_time, str) else created_time
if created_time:
st.write(f"**{result['title']}** by @{result['author']} - [Перейти]({result['url']}) {vote_status} - {created_time.strftime('%Y-%m-%d %H:%M:%S')}")
else:
st.write(f"**{result['title']}** by @{result['author']} - [Перейти]({result['url']}) {vote_status}")
if not st.session_state.show_all:
st.write(f"Тональність (TextBlob): {result['sentiment_tb']:.2f}, Тональність (NLTK): {result['sentiment_nltk']['compound']:.2f}")
with st.expander("Показати текст допису"):
st.write(result['body'])
if not st.session_state.show_all:
st.write(f"Ключові слова: {', '.join(result['keywords'])}")
if st.session_state.username and st.session_state.posting_key:
weight = st.slider(f"Сила голосу для '{result['title']}' (%):",
min_value=1, max_value=100,
value=100,
key=f"weight_{result['url']}")
if st.button(f"Голосувати за '{result['title']}'", key=f"vote_{result['url']}"):
if st.session_state.username and st.session_state.posting_key:
with st.spinner("Обробка голосу..."):
result_message = vote_js(st.session_state.username, st.session_state.posting_key,
result['author'], result['permlink'], weight)
if "Error" not in result_message and "Помилка" not in result_message:
st.success("Голос успішно подано")
st.session_state.results[idx]['voted'] = True
st.session_state.results = list(st.session_state.results)
else:
st.error(f"Помилка голосування: {result_message}")
else:
st.warning("Будь ласка, введіть нікнейм та приватний ключ!")
if st.button("Оновити список дописів"):
st.session_state.results = []
st.session_state.last_permlink = None
st.rerun()
if __name__ == "__main__":
main()
Перспективи
Якщо вірити поясненням ШІ, то подібні механізми можна впровадити для зручнішої
взаємодії із блокчейном та отримання інформації та її структуруванню.
Наприклад, курування чи просто зруне отримання даних, щоб читати. Тільки треба
доробити багатенько деталей, щоб функціонувало як треба. Бо є питання щодо
оновлення матеріалів при деяких змінах.
- Є різні можливості по збору аналітичних даних.
- Автоматизовний аналіз дописів та автоголосування, що, мабуть, найцікавіше, але
це все треба Доре налаштувати. Навчити модель розпізнавання, щоб орієнтувалась
на певний матеріал і робилась перевірка унікальності та наявність ШІ вмісту.
Основні зручності
- Зручний перегляд дописів;
- Можна фільтрувати за обраними авторами, щоб відразу й зручно поглянути що вони
написали, а не блудити десь сторінками чи купою завалених матеріалів у
підписці, де твориться безлад; - Голосування.
Та багато іншого, якщо доробити та підправити код як треба.
Безпека
Точно не знаю наскільки безпечно використовувати подібний код та ресурси, проте
нікого до цього не закликаю.
Моделі ШІ
Переважну допомогу надали "GPT-4o mini" та "Gemini 2.0 Flash Experimental" із
базою даних до серпня 2024 року та швидкою відповіддю й великим вікном на
відповідь як величиною чату. Доволі метикувата модель і можна безоплатно
користуватись, доки на стадії експериментування. Може видавати велику
кількість коду на одну відповідь, бо інші безоплатні рішення бува, перериваються
й доводиться писати "продовжуй" і складати із частин.
Може це відкриття лише для мене, проте було цікаво дізнатись щось нове.
Технології зараз відіграють все більше ролі багатьох сферах.
#ukraine #python #java #club5050 #steemexclusive #developing #votint #streamlit