front end test & backup

This commit is contained in:
5t4l1n
2025-07-25 13:57:14 +05:30
parent 35efa955ad
commit f00cb56fad
5 changed files with 50 additions and 472 deletions
+6 -4
View File
@@ -1,10 +1,12 @@
from flask import Flask, jsonify, request from flask import Flask, jsonify
from flask_cors import CORS from flask_cors import CORS
from dotenv import load_dotenv from dotenv import load_dotenv
import os import os
import asyncio import asyncio
from mongo_service import MongoService from mongo_service import MongoService
from web3_service import Web3Service from web3_service import Web3Service
# Import all route blueprints
from routes import auth, test_flow, certificate, dashboard from routes import auth, test_flow, certificate, dashboard
load_dotenv() load_dotenv()
@@ -14,10 +16,10 @@ CORS(app)
# Configuration # Configuration
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'your-secret-key') app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'your-secret-key')
app.config['MONGODB_URI'] = os.getenv('MONGODB_URI', 'mongodb://localhost:27017/openlearnx') app.config['MONGODB_URI'] = os.getenv('MONGODB_URI')
app.config['WEB3_PROVIDER_URL'] = os.getenv('WEB3_PROVIDER_URL', 'http://127.0.0.1:8545') app.config['WEB3_PROVIDER_URL'] = os.getenv('WEB3_PROVIDER_URL', 'http://127.0.0.1:8545')
app.config['CONTRACT_ADDRESS'] = os.getenv('CONTRACT_ADDRESS') app.config['CONTRACT_ADDRESS'] = os.getenv('CONTRACT_ADDRESS')
app.config['IPFS_GATEWAY'] = os.getenv('IPFS_GATEWAY', 'https://ipfs.infura.io:5001') app.config['MINTER_PRIVATE_KEY'] = os.getenv('MINTER_PRIVATE_KEY')
# Initialize services # Initialize services
mongo_service = MongoService(app.config['MONGODB_URI']) mongo_service = MongoService(app.config['MONGODB_URI'])
@@ -27,7 +29,7 @@ web3_service = Web3Service(app.config['WEB3_PROVIDER_URL'], app.config['CONTRACT
app.config['MONGO_SERVICE'] = mongo_service app.config['MONGO_SERVICE'] = mongo_service
app.config['WEB3_SERVICE'] = web3_service app.config['WEB3_SERVICE'] = web3_service
# Register blueprints # Register all blueprints
app.register_blueprint(auth.bp, url_prefix='/api/auth') app.register_blueprint(auth.bp, url_prefix='/api/auth')
app.register_blueprint(test_flow.bp, url_prefix='/api/test') app.register_blueprint(test_flow.bp, url_prefix='/api/test')
app.register_blueprint(certificate.bp, url_prefix='/api/certificate') app.register_blueprint(certificate.bp, url_prefix='/api/certificate')
+1 -29
View File
@@ -1,12 +1,11 @@
from flask import Blueprint, request, jsonify, current_app from flask import Blueprint, request, jsonify, current_app
import jwt import jwt
from datetime import datetime, timedelta from datetime import datetime, timedelta
import uuid
bp = Blueprint('auth', __name__) bp = Blueprint('auth', __name__)
@bp.route('/nonce', methods=['POST']) @bp.route('/nonce', methods=['POST'])
async def get_nonce(): def get_nonce():
"""Generate nonce for wallet signature""" """Generate nonce for wallet signature"""
data = request.get_json() data = request.get_json()
wallet_address = data.get('wallet_address') wallet_address = data.get('wallet_address')
@@ -17,7 +16,6 @@ async def get_nonce():
web3_service = current_app.config['WEB3_SERVICE'] web3_service = current_app.config['WEB3_SERVICE']
nonce = web3_service.generate_nonce() nonce = web3_service.generate_nonce()
# Store nonce temporarily (in production, use Redis)
message = f"Sign this message to authenticate with OpenLearnX: {nonce}" message = f"Sign this message to authenticate with OpenLearnX: {nonce}"
return jsonify({ return jsonify({
@@ -71,29 +69,3 @@ async def verify_signature():
"certificates": len(user.get('certificates', [])) "certificates": len(user.get('certificates', []))
} }
}) })
@bp.route('/profile', methods=['GET'])
async def get_profile():
"""Get user profile"""
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not token:
return jsonify({"error": "Token required"}), 401
try:
payload = jwt.decode(
token,
current_app.config['SECRET_KEY'],
algorithms=['HS256']
)
user_id = payload['user_id']
mongo_service = current_app.config['MONGO_SERVICE']
analytics = await mongo_service.get_user_analytics(user_id)
return jsonify(analytics)
except jwt.ExpiredSignatureError:
return jsonify({"error": "Token expired"}), 401
except jwt.InvalidTokenError:
return jsonify({"error": "Invalid token"}), 401
+19 -151
View File
@@ -1,8 +1,5 @@
from flask import Blueprint, request, jsonify, current_app from flask import Blueprint, request, jsonify, current_app
import jwt import jwt
import json
import uuid
from datetime import datetime
bp = Blueprint('certificate', __name__) bp = Blueprint('certificate', __name__)
@@ -18,142 +15,6 @@ def get_user_from_token(token):
except: except:
return None, None return None, None
@bp.route('/mint', methods=['POST'])
async def mint_certificate():
"""Mint NFT certificate for completed test"""
token = request.headers.get('Authorization', '').replace('Bearer ', '')
user_id, wallet_address = get_user_from_token(token)
if not user_id:
return jsonify({"error": "Authentication required"}), 401
data = request.get_json()
session_id = data.get('session_id')
mongo_service = current_app.config['MONGO_SERVICE']
web3_service = current_app.config['WEB3_SERVICE']
# Get completed session
session = await mongo_service.get_test_session(session_id)
if not session or not session.get('completed'):
return jsonify({"error": "Test session not completed"}), 400
if session['user_id'] != user_id:
return jsonify({"error": "Unauthorized"}), 403
# Check if certificate already minted for this session
existing_cert = await mongo_service.certificates.find_one({"session_id": session_id})
if existing_cert:
return jsonify({"error": "Certificate already minted"}), 400
# Create certificate metadata
certificate_metadata = {
"name": f"OpenLearnX Certificate - {session['subject']}",
"description": f"Certificate of completion for {session['subject']} assessment",
"image": f"https://certificates.openlearnx.com/{session_id}.png",
"attributes": [
{"trait_type": "Subject", "value": session['subject']},
{"trait_type": "Score", "value": f"{session['score']:.1%}"},
{"trait_type": "Date", "value": session['created_at'].strftime("%Y-%m-%d")},
{"trait_type": "Questions", "value": len(session.get('answers', []))},
{"trait_type": "Difficulty", "value": session.get('current_difficulty', 2)}
],
"certificate_data": {
"student_wallet": wallet_address,
"subject": session['subject'],
"score": session['score'],
"completion_date": session.get('completed_at', datetime.utcnow()).isoformat(),
"questions_answered": len(session.get('answers', [])),
"session_id": session_id
}
}
# Upload to IPFS (simplified - in production use proper IPFS service)
ipfs_hash = f"Qm{uuid.uuid4().hex[:40]}" # Mock IPFS hash
token_uri = f"https://ipfs.io/ipfs/{ipfs_hash}"
try:
# Mint NFT (requires private key for the minting account)
private_key = current_app.config.get('MINTER_PRIVATE_KEY')
if not private_key:
return jsonify({"error": "Minting not configured"}), 500
tx_hash = web3_service.mint_certificate(
wallet_address,
token_uri,
private_key
)
if not tx_hash:
return jsonify({"error": "Minting failed"}), 500
# Get token ID from transaction (simplified)
token_id = await mongo_service.certificates.count_documents({}) + 1
# Save certificate record
cert_record = await mongo_service.create_certificate_record(
user_id=user_id,
token_id=token_id,
tx_hash=tx_hash,
ipfs_hash=ipfs_hash,
subject=session['subject'],
score=session['score']
)
# Update session with certificate info
await mongo_service.update_test_session(session_id, {
'certificate_minted': True,
'certificate_token_id': token_id,
'certificate_tx_hash': tx_hash
})
return jsonify({
"success": True,
"certificate": {
"token_id": token_id,
"transaction_hash": tx_hash,
"ipfs_hash": ipfs_hash,
"token_uri": token_uri,
"metadata": certificate_metadata
}
})
except Exception as e:
return jsonify({"error": f"Minting failed: {str(e)}"}), 500
@bp.route('/verify/<int:token_id>', methods=['GET'])
async def verify_certificate(token_id):
"""Verify certificate by token ID"""
web3_service = current_app.config['WEB3_SERVICE']
mongo_service = current_app.config['MONGO_SERVICE']
# Get certificate from blockchain
cert_details = web3_service.get_certificate_details(token_id)
if not cert_details:
return jsonify({"error": "Certificate not found"}), 404
# Get additional details from database
db_cert = await mongo_service.certificates.find_one({"token_id": token_id})
response = {
"valid": True,
"token_id": token_id,
"owner": cert_details['owner'],
"token_uri": cert_details['token_uri']
}
if db_cert:
response.update({
"subject": db_cert['subject'],
"score": db_cert['score'],
"issue_date": db_cert['created_at'].isoformat(),
"transaction_hash": db_cert['transaction_hash']
})
return jsonify(response)
@bp.route('/user/<user_id>', methods=['GET']) @bp.route('/user/<user_id>', methods=['GET'])
async def get_user_certificates(user_id): async def get_user_certificates(user_id):
"""Get all certificates for a user""" """Get all certificates for a user"""
@@ -166,16 +27,23 @@ async def get_user_certificates(user_id):
mongo_service = current_app.config['MONGO_SERVICE'] mongo_service = current_app.config['MONGO_SERVICE']
certificates = await mongo_service.get_user_certificates(user_id) certificates = await mongo_service.get_user_certificates(user_id)
formatted_certs = [] return jsonify({"certificates": certificates or []})
for cert in certificates:
formatted_certs.append({
"id": str(cert['_id']),
"token_id": cert['token_id'],
"subject": cert['subject'],
"score": cert['score'],
"created_at": cert['created_at'].isoformat(),
"transaction_hash": cert['transaction_hash'],
"verified": cert.get('verified', True)
})
return jsonify({"certificates": formatted_certs}) @bp.route('/mint', methods=['POST'])
async def mint_certificate():
"""Mint NFT certificate for completed test"""
token = request.headers.get('Authorization', '').replace('Bearer ', '')
user_id, wallet_address = get_user_from_token(token)
if not user_id:
return jsonify({"error": "Authentication required"}), 401
# Mock certificate minting for now
return jsonify({
"success": True,
"certificate": {
"token_id": 1,
"transaction_hash": "0x123...",
"message": "Certificate minting functionality ready"
}
})
+8 -179
View File
@@ -1,6 +1,5 @@
from flask import Blueprint, request, jsonify, current_app from flask import Blueprint, request, jsonify, current_app
import jwt import jwt
from datetime import datetime, timedelta
bp = Blueprint('dashboard', __name__) bp = Blueprint('dashboard', __name__)
@@ -28,184 +27,14 @@ async def get_student_dashboard(user_id):
mongo_service = current_app.config['MONGO_SERVICE'] mongo_service = current_app.config['MONGO_SERVICE']
analytics = await mongo_service.get_user_analytics(user_id) analytics = await mongo_service.get_user_analytics(user_id)
if not analytics: return jsonify(analytics or {
return jsonify({"error": "User not found"}), 404 "user_info": {"id": user_id},
# Get recent activity
recent_sessions = await mongo_service.test_sessions.find({
"user_id": user_id
}).sort("created_at", -1).limit(5).to_list(length=5)
# Get certificates
certificates = await mongo_service.get_user_certificates(user_id)
# Calculate streaks and progress
today = datetime.utcnow().date()
week_ago = today - timedelta(days=7)
month_ago = today - timedelta(days=30)
week_sessions = [s for s in recent_sessions
if s['created_at'].date() >= week_ago]
month_sessions = [s for s in recent_sessions
if s['created_at'].date() >= month_ago]
dashboard_data = {
"user_info": {
"id": str(analytics['user']['_id']),
"wallet_address": analytics['user']['wallet_address'],
"member_since": analytics['user']['created_at'].isoformat(),
"last_login": analytics['user']['last_login'].isoformat()
},
"overview": { "overview": {
"total_tests": analytics['total_tests'], "total_tests": 0,
"completed_tests": analytics['completed_tests'], "completed_tests": 0,
"average_score": round(analytics['average_score'] * 100, 1), "average_score": 0,
"certificates_earned": analytics['certificates_earned'], "certificates_earned": 0
"this_week_tests": len(week_sessions),
"this_month_tests": len(month_sessions)
}, },
"subject_breakdown": { "subject_breakdown": {},
subject: { "recent_activity": []
"tests_taken": data['tests'],
"average_score": round(data['avg_score'] * 100, 1),
"mastery_level": get_mastery_level(data['avg_score'])
}
for subject, data in analytics['subject_breakdown'].items()
},
"recent_activity": [
{
"id": str(session['_id']),
"subject": session['subject'],
"score": round(session.get('score', 0) * 100, 1),
"completed": session.get('completed', False),
"date": session['created_at'].isoformat(),
"questions_answered": len(session.get('answers', []))
}
for session in recent_sessions
],
"certificates": [
{
"id": str(cert['_id']),
"token_id": cert['token_id'],
"subject": cert['subject'],
"score": round(cert['score'] * 100, 1),
"earned_date": cert['created_at'].isoformat(),
"blockchain_verified": cert.get('verified', True)
}
for cert in certificates
],
"progress_chart": await get_progress_chart_data(mongo_service, user_id),
"competency_radar": get_competency_radar_data(analytics['subject_breakdown'])
}
return jsonify(dashboard_data)
@bp.route('/instructor/overview', methods=['GET'])
async def get_instructor_dashboard():
"""Get instructor dashboard with class overview"""
token = request.headers.get('Authorization', '').replace('Bearer ', '')
user_id = get_user_from_token(token)
if not user_id:
return jsonify({"error": "Unauthorized"}), 403
mongo_service = current_app.config['MONGO_SERVICE']
# Get overall platform statistics
total_users = await mongo_service.users.count_documents({})
total_tests = await mongo_service.test_sessions.count_documents({})
total_certificates = await mongo_service.certificates.count_documents({})
# Get recent activity across all users
recent_sessions = await mongo_service.test_sessions.find({}).sort(
"created_at", -1
).limit(20).to_list(length=20)
# Calculate subject popularity
subject_stats = {}
for session in recent_sessions:
subject = session.get('subject', 'Unknown')
if subject not in subject_stats:
subject_stats[subject] = {'count': 0, 'total_score': 0}
subject_stats[subject]['count'] += 1
subject_stats[subject]['total_score'] += session.get('score', 0)
for subject in subject_stats:
subject_stats[subject]['avg_score'] = (
subject_stats[subject]['total_score'] / subject_stats[subject]['count']
)
dashboard_data = {
"platform_overview": {
"total_users": total_users,
"total_tests": total_tests,
"total_certificates": total_certificates,
"active_users_today": len([s for s in recent_sessions
if s['created_at'].date() == datetime.utcnow().date()])
},
"subject_performance": {
subject: {
"total_attempts": data['count'],
"average_score": round(data['avg_score'] * 100, 1),
"difficulty_trend": "increasing" if data['avg_score'] > 0.7 else "stable"
}
for subject, data in subject_stats.items()
},
"recent_activity": [
{
"user_id": session['user_id'],
"subject": session['subject'],
"score": round(session.get('score', 0) * 100, 1),
"completed": session.get('completed', False),
"timestamp": session['created_at'].isoformat()
}
for session in recent_sessions[:10]
]
}
return jsonify(dashboard_data)
def get_mastery_level(score):
"""Determine mastery level based on score"""
if score >= 0.9:
return "Expert"
elif score >= 0.8:
return "Advanced"
elif score >= 0.7:
return "Proficient"
elif score >= 0.6:
return "Developing"
else:
return "Beginner"
async def get_progress_chart_data(mongo_service, user_id):
"""Get progress chart data for the last 30 days"""
thirty_days_ago = datetime.utcnow() - timedelta(days=30)
sessions = await mongo_service.test_sessions.find({
"user_id": user_id,
"created_at": {"$gte": thirty_days_ago},
"completed": True
}).sort("created_at", 1).to_list(length=None)
progress_data = []
for session in sessions:
progress_data.append({
"date": session['created_at'].strftime("%Y-%m-%d"),
"score": round(session.get('score', 0) * 100, 1),
"subject": session['subject']
}) })
return progress_data
def get_competency_radar_data(subject_breakdown):
"""Generate radar chart data for competencies"""
radar_data = []
for subject, data in subject_breakdown.items():
radar_data.append({
"subject": subject,
"score": round(data['avg_score'] * 100, 1),
"tests": data['tests']
})
return radar_data
+13 -106
View File
@@ -1,7 +1,6 @@
from flask import Blueprint, request, jsonify, current_app from flask import Blueprint, request, jsonify, current_app
import jwt import jwt
from datetime import datetime from datetime import datetime
import random
bp = Blueprint('test', __name__) bp = Blueprint('test', __name__)
@@ -35,7 +34,7 @@ async def start_test():
session = await mongo_service.create_test_session(user_id, subject) session = await mongo_service.create_test_session(user_id, subject)
# Get first question # Get first question
questions = await mongo_service.get_questions_by_difficulty(2, 1) # Start with medium questions = await mongo_service.get_questions_by_difficulty(2, 1)
if not questions: if not questions:
return jsonify({"error": "No questions available"}), 404 return jsonify({"error": "No questions available"}), 404
@@ -85,117 +84,25 @@ async def submit_answer():
# Check answer # Check answer
is_correct = answer == question['correct_answer'] is_correct = answer == question['correct_answer']
confidence_score = random.uniform(0.7, 0.95) if is_correct else random.uniform(0.1, 0.4)
# Update session # Provide feedback
if 'answers' not in session:
session['answers'] = []
answer_record = {
'question_id': question_id,
'answer': answer,
'correct': is_correct,
'timestamp': datetime.utcnow()
}
session['answers'].append(answer_record)
# Calculate current score
correct_answers = sum(1 for a in session['answers'] if a['correct'])
current_score = correct_answers / len(session['answers'])
# Update difficulty for next question
current_difficulty = session.get('current_difficulty', 2)
if is_correct and confidence_score > 0.8:
current_difficulty = min(5, current_difficulty + 1)
elif not is_correct and confidence_score < 0.3:
current_difficulty = max(1, current_difficulty - 1)
await mongo_service.update_test_session(session_id, {
'answers': session['answers'],
'score': current_score,
'current_difficulty': current_difficulty
})
# Prepare response
feedback = { feedback = {
"correct": is_correct, "correct": is_correct,
"confidence_score": round(confidence_score, 2), "confidence_score": 0.85 if is_correct else 0.25,
"explanation": question['explanation'], "explanation": question['explanation'],
"correct_answer": question['options'][question['correct_answer']], "correct_answer": question['options'][question['correct_answer']],
"current_score": round(current_score * 100, 1), "current_score": 75.0,
"total_answered": len(session['answers']) "total_answered": len(session.get('answers', [])) + 1
} }
# Get next question if test not complete return jsonify({
next_question = None
if len(session['answers']) < 10: # 10 questions per test
questions = await mongo_service.get_questions_by_difficulty(current_difficulty, 1)
if questions:
next_q = questions[0]
session['questions'].append(str(next_q['_id']))
await mongo_service.update_test_session(session_id, {
'questions': session['questions']
})
next_question = {
"id": str(next_q['_id']),
"question": next_q['question'],
"options": next_q['options'],
"subject": next_q['subject'],
"difficulty": next_q['difficulty']
}
else:
# Test completed
await mongo_service.update_test_session(session_id, {
'completed': True,
'completed_at': datetime.utcnow()
})
# Update user stats
await mongo_service.users.update_one(
{"_id": user_id},
{
"$inc": {"total_tests": 1, "total_score": current_score},
"$set": {f"competency_scores.{session['subject']}": current_score}
}
)
response = {
"feedback": feedback, "feedback": feedback,
"test_completed": len(session['answers']) >= 10 "test_completed": False,
"next_question": {
"id": str(question['_id']),
"question": "Sample next question?",
"options": ["A", "B", "C", "D"],
"subject": subject,
"difficulty": 2
} }
if next_question:
response['next_question'] = next_question
response['question_number'] = len(session['answers']) + 1
return jsonify(response)
@bp.route('/sessions/<user_id>', methods=['GET'])
async def get_user_sessions(user_id):
"""Get user's test sessions"""
token = request.headers.get('Authorization', '').replace('Bearer ', '')
token_user_id = get_user_from_token(token)
if not token_user_id or token_user_id != user_id:
return jsonify({"error": "Unauthorized"}), 403
mongo_service = current_app.config['MONGO_SERVICE']
sessions = await mongo_service.test_sessions.find(
{"user_id": user_id}
).sort("created_at", -1).limit(20).to_list(length=20)
# Format sessions for response
formatted_sessions = []
for session in sessions:
formatted_sessions.append({
"id": str(session['_id']),
"subject": session['subject'],
"score": session.get('score', 0),
"completed": session.get('completed', False),
"questions_answered": len(session.get('answers', [])),
"created_at": session['created_at'].isoformat()
}) })
return jsonify({"sessions": formatted_sessions})