popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

flask webui looking good

Browse Source
This commit is contained in:
Dobromir Popov 2024-10-13 01:37:10 +03:00
parent 49384accf6
commit a655c5bd88
8 changed files with 257 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
app_data.db Normal file
View File

Binary file not shown.

2
crypto/sol/app.py
View File

@ -1465,6 +1465,7 @@ async def main():
from modules.webui import init_app
from modules.storage import init_db
async def run_flask():
# loop = asyncio.get_running_loop()
@ -1475,6 +1476,7 @@ async def run_flask():
async def run_all():
await asyncio.gather(
init_db(),
main(),
run_flask()
)

167
crypto/sol/modules/storage.py
View File

@ -1,8 +1,12 @@
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import aiosqlite
import json
from datetime import datetime
DATABASE_FILE = "app_data.db"
DATABASE_FILE = "./app_data.db"
async def init_db():
async with aiosqlite.connect(DATABASE_FILE) as db:
@ -80,59 +84,6 @@ async def store_transaction(wallet_id, transaction_type, sell_currency, sell_amo
""", (wallet_id, datetime.now().isoformat(), transaction_type, sell_currency, sell_amount, sell_value, buy_currency, buy_amount, buy_value, solana_signature, json.dumps(details or {})))
await db.commit()
# async def get_new_transactions(wallet_address, rpc_url):
# async with AsyncClient(rpc_url) as client:
# last_tx = await get_last_stored_transaction(wallet_address)
# if last_tx:
# last_signature, last_timestamp = last_tx
# else:
# # If no transactions are stored, we'll fetch all transactions
# last_signature = None
# last_timestamp = None
# new_transactions = []
# # Get the transaction history for the wallet
# tx_history = await client.get_signatures_for_address(wallet_address, before=last_signature)
# for tx in tx_history.value:
# # Check if the transaction is newer than the last stored one
# if not last_timestamp or tx.block_time > datetime.fromisoformat(last_timestamp).timestamp():
# # Fetch the full transaction details
# tx_details = await client.get_transaction(tx.signature, commitment=Confirmed)
# new_transactions.append(tx_details)
# return new_transactions
# async def process_new_transactions(wallet_id, wallet_address, rpc_url):
# new_transactions = await get_new_transactions(wallet_address, rpc_url)
# for tx in new_transactions:
# # Process the transaction and extract relevant information
# # This is a placeholder - you'll need to implement the actual logic based on your requirements
# transaction_type = "swap" # Determine the type based on the transaction data
# sell_currency = "SOL" # Extract from transaction data
# sell_amount = 1.0 # Extract from transaction data
# sell_value = 100.0 # Extract from transaction data
# buy_currency = "USDC" # Extract from transaction data
# buy_amount = 100.0 # Extract from transaction data
# buy_value = 100.0 # Extract from transaction data
# solana_signature = tx.transaction.signatures[0]
# # Store the transaction in the database
# await store_transaction(
# wallet_id, transaction_type, sell_currency, sell_amount, sell_value,
# buy_currency, buy_amount, buy_value, solana_signature
# )
# # Update holdings
# await update_holdings(wallet_id, sell_currency, -sell_amount)
# await update_holdings(wallet_id, buy_currency, buy_amount)
# # After processing all new transactions, close completed transactions
# await close_completed_transactions(wallet_id)
async def update_holdings(wallet_id, currency, amount_change):
async with aiosqlite.connect(DATABASE_FILE) as db:
@ -258,6 +209,114 @@ async def get_profit_loss(wallet_id, currency, start_date=None, end_date=None):
result = await cursor.fetchone()
return result[0] if result else 0
# # # # # # USERS
# For this example, we'll use a simple dictionary to store users
users = {
"db": {"id": 1, "username": "db", "email": "user1@example.com", "password": "db"},
"popov": {"id": 2, "username": "popov", "email": "user2@example.com", "password": "popov"}
}
def get_or_create_user(email, google_id):
user = next((u for u in users.values() if u['email'] == email), None)
if not user:
user_id = max(u['id'] for u in users.values()) + 1
username = email.split('@')[0] # Use the part before @ as username
user = {
'id': user_id,
'username': username,
'email': email,
'google_id': google_id
}
users[username] = user
return user
def authenticate_user(username, password):
"""
Authenticate a user based on username and password.
Returns user data if authentication is successful, None otherwise.
"""
user = users.get(username)
if user and user['password'] == password:
return {"id": user['id'], "username": user['username'], "email": user['email']}
return None
def get_user_by_id(user_id):
"""
Retrieve a user by their ID.
"""
for user in users.values():
if user['id'] == int(user_id):
return {"id": user['id'], "username": user['username'], "email": user['email']}
return None
def store_api_key(user_id, api_key):
"""
Store the generated API key for a user.
"""
# In a real application, you would store this in a database
# For this example, we'll just print it
print(f"Storing API key {api_key} for user {user_id}")
# async def get_new_transactions(wallet_address, rpc_url):
# async with AsyncClient(rpc_url) as client:
# last_tx = await get_last_stored_transaction(wallet_address)
# if last_tx:
# last_signature, last_timestamp = last_tx
# else:
# # If no transactions are stored, we'll fetch all transactions
# last_signature = None
# last_timestamp = None
# new_transactions = []
# # Get the transaction history for the wallet
# tx_history = await client.get_signatures_for_address(wallet_address, before=last_signature)
# for tx in tx_history.value:
# # Check if the transaction is newer than the last stored one
# if not last_timestamp or tx.block_time > datetime.fromisoformat(last_timestamp).timestamp():
# # Fetch the full transaction details
# tx_details = await client.get_transaction(tx.signature, commitment=Confirmed)
# new_transactions.append(tx_details)
# return new_transactions
# async def process_new_transactions(wallet_id, wallet_address, rpc_url):
# new_transactions = await get_new_transactions(wallet_address, rpc_url)
# for tx in new_transactions:
# # Process the transaction and extract relevant information
# # This is a placeholder - you'll need to implement the actual logic based on your requirements
# transaction_type = "swap" # Determine the type based on the transaction data
# sell_currency = "SOL" # Extract from transaction data
# sell_amount = 1.0 # Extract from transaction data
# sell_value = 100.0 # Extract from transaction data
# buy_currency = "USDC" # Extract from transaction data
# buy_amount = 100.0 # Extract from transaction data
# buy_value = 100.0 # Extract from transaction data
# solana_signature = tx.transaction.signatures[0]
# # Store the transaction in the database
# await store_transaction(
# wallet_id, transaction_type, sell_currency, sell_amount, sell_value,
# buy_currency, buy_amount, buy_value, solana_signature
# )
# # Update holdings
# await update_holdings(wallet_id, sell_currency, -sell_amount)
# await update_holdings(wallet_id, buy_currency, buy_amount)
# # After processing all new transactions, close completed transactions
# await close_completed_transactions(wallet_id)
# Example usage
if __name__ == "__main__":
import asyncio

149
crypto/sol/modules/webui.py
View File

@ -1,75 +1,112 @@
from flask import Flask, jsonify, request, render_template, redirect, url_for
# from flask_oauthlib.client import OAuth
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user
import secrets
from modules import storage
import os
app = Flask(__name__, template_folder='../templates', static_folder='../static')
app.config['SECRET_KEY'] = 'your-secret-key'
login_manager = LoginManager(app)
login_manager.login_view = 'login'
def init_app():
app = Flask(__name__, template_folder='../templates', static_folder='../static')
app.config['SECRET_KEY'] = 'your-secret-key'
login_manager = LoginManager(app)
login_manager.login_view = 'login'
class User(UserMixin):
def __init__(self, id, username, email):
self.id = id
self.username = username
self.email = email
# oauth = OAuth(app)
# google = oauth.remote_app(
# 'google',
# consumer_key='YOUR_GOOGLE_CLIENT_ID',
# consumer_secret='YOUR_GOOGLE_CLIENT_SECRET',
# request_token_params={
# 'scope': 'email'
# },
# base_url='https://www.googleapis.com/oauth2/v1/',
# request_token_url=None,
# access_token_method='POST',
# access_token_url='https://accounts.google.com/o/oauth2/token',
# authorize_url='https://accounts.google.com/o/oauth2/auth',
# )
@login_manager.user_loader
def load_user(user_id):
user_data = storage.get_user_by_id(user_id)
if user_data:
return User(id=user_data['id'], username=user_data['username'], email=user_data['email'])
return None
@app.route('/')
def index():
return render_template('index.html')
login_manager = LoginManager()
login_manager.init_app(app)
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
@app.route('/login/google/authorized')
def authorized():
# resp = google.authorized_response()
# if resp is None or resp.get('access_token') is None:
# return 'Access denied: reason={} error={}'.format(
# request.args['error_reason'],
# request.args['error_description']
# )
# session['google_token'] = (resp['access_token'], '')
# user_info = google.get('userinfo')
# user = storage.get_or_create_user(user_info.data['email'], user_info.data['id'])
# login_user(user)
return redirect(url_for('index'))
user = storage.authenticate_user(username, password)
if user:
login_user(User(id=user['id'], username=user['username'], email=user['email']))
return redirect(url_for('dashboard'))
else:
return render_template('login.html', error='Invalid credentials')
return render_template('login.html')
@app.route('/logout')
@login_required
def logout():
logout_user()
return redirect(url_for('index'))
class User(UserMixin):
def __init__(self, id, username, email):
self.id = id
self.username = username
self.email = email
@app.route('/dashboard')
@login_required
def dashboard():
return render_template('dashboard.html')
@login_manager.user_loader
def load_user(user_id):
user_data = storage.get_user_by_id(user_id)
if user_data:
return User(id=user_data['id'], username=user_data['username'], email=user_data['email'])
return None
@app.route('/generate_api_key', methods=['POST'])
@login_required
def generate_api_key():
api_key = secrets.token_urlsafe(32)
storage.store_api_key(current_user.id, api_key)
return jsonify({'api_key': api_key})
@app.route('/')
def index():
return render_template('index.html')
@app.route('/wallet/<int:wallet_id>/transactions', methods=['GET'])
@login_required
def get_transactions(wallet_id):
transactions = storage.get_transactions(wallet_id)
return jsonify(transactions)
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
user = storage.authenticate_user(username, password)
if user:
login_user(User(id=user['id'], username=user['username'], email=user['email']))
return redirect(url_for('dashboard'))
else:
return render_template('login.html', error='Invalid credentials')
elif request.args.get('google'):
return google.authorize(callback=url_for('authorized', _external=True))
return render_template('login.html')
@app.route('/wallet/<int:wallet_id>/holdings', methods=['GET'])
@login_required
def get_holdings(wallet_id):
holdings = storage.get_holdings(wallet_id)
return jsonify(holdings)
@app.route('/logout')
@login_required
def logout():
logout_user()
return redirect(url_for('index'))
@app.route('/dashboard')
@login_required
def dashboard():
return render_template('dashboard.html')
@app.route('/generate_api_key', methods=['POST'])
@login_required
def generate_api_key():
api_key = secrets.token_urlsafe(32)
storage.store_api_key(current_user.id, api_key)
return jsonify({'api_key': api_key})
@app.route('/wallet/<int:wallet_id>/transactions', methods=['GET'])
@login_required
def get_transactions(wallet_id):
transactions = storage.get_transactions(wallet_id)
return jsonify(transactions)
@app.route('/wallet/<int:wallet_id>/holdings', methods=['GET'])
@login_required
def get_holdings(wallet_id):
holdings = storage.get_holdings(wallet_id)
return jsonify(holdings)
# Implement other routes for reports, price alerts, following accounts, etc.
def init_app():
return app

12
crypto/sol/requirements.in
View File

@ -1,12 +0,0 @@
aiohttp==3.10.9
aiohttp==3.10.5
base58==2.1.1
dexscreener==1.1
Flask==3.0.3
jupiter_python_sdk==0.0.2.0
python-dotenv==1.0.1
python-telegram-bot==21.6
Requests==2.32.3
solana==0.34.3
solders==0.21.0
websockets==10.4

2
crypto/sol/requirements.txt
View File

@ -1,8 +1,10 @@
aiohttp==3.10.9
aiosqlite
base58==2.1.1
dexscreener==1.1
Flask==3.0.3
flask-login
flask-oauthlib
jupiter_python_sdk==0.0.2.0
python-dotenv==1.0.1
python-telegram-bot==21.6

67
crypto/sol/static/js/app.js
View File

@ -1,39 +1,42 @@
document.getElementById('connectWallet').addEventListener('click', async () => {
try {
const { solana } is window;
if (solana && solana.isPhantom) {
const response = await solana.connect({ onlyIfTrusted: true });
console.log('Connected with Public Key:', response.publicKey.toString());
} else {
alert('Phantom wallet not found. Please install it.');
}
} catch (error) {
console.error(error);
alert('Connection to Phantom Wallet failed');
}
});
document.getElementById('swapToken').addEventListener('click', () => {
const tokenName = document.getElementById('tokenName').value;
const amount = document.getElementById('amount').value;
fetch('/swap', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({token_name: tokenName, amount: amount})
})
.then(response => response.json())
.then(data => alert(data.message));
});
// Add your custom JavaScript here
document.addEventListener('DOMContentLoaded', () => {
const connectWalletButton = document.getElementById('connectWallet');
const swapTokenButton = document.getElementById('swapToken');
const generateApiKeyButton = document.getElementById('generate-api-key');
const apiKeyDisplay = document.getElementById('api-key-display');
if (connectWalletButton) {
connectWalletButton.addEventListener('click', async () => {
try {
const { solana } = window;
if (solana && solana.isPhantom) {
const response = await solana.connect({ onlyIfTrusted: true });
console.log('Connected with Public Key:', response.publicKey.toString());
} else {
alert('Phantom wallet not found. Please install it.');
}
} catch (error) {
console.error(error);
alert('Connection to Phantom Wallet failed');
}
});
}
if (swapTokenButton) {
swapTokenButton.addEventListener('click', () => {
const tokenName = document.getElementById('tokenName').value;
const amount = document.getElementById('amount').value;
fetch('/swap', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({token_name: tokenName, amount: amount})
})
.then(response => response.json())
.then(data => alert(data.message));
});
}
if (generateApiKeyButton) {
generateApiKeyButton.addEventListener('click', async () => {
const response = await fetch('/generate_api_key', { method: 'POST' });

21
crypto/sol/templates/login.html
View File

@ -2,16 +2,17 @@
{% block content %}
<h1>Login</h1>
<form method="POST" action="{{ url_for('login') }}">
<label for="username">Username:</label>
<input type="text" id="username" name="username" required>
<label for="password">Password:</label>
<input type="password" id="password" name="password" required>
<button type="submit">Login</button>
</form>
{% if error %}
<p class="error">{{ error }}</p>
<p style="color: red;">{{ error }}</p>
{% endif %}
<form method="POST">
<label for="username">Username:</label>
<input type="text" id="username" name="username" required><br><br>
<label for="password">Password:</label>
<input type="password" id="password" name="password" required><br><br>
<input type="submit" value="Login">
</form>
<br>
<a href="{{ url_for('login', google=1) }}">Login with Google</a>
{% endblock %}