forked from ITStudio_OUC/OucPosts
backend implemented
This commit is contained in:
nyaco3
+711
@@ -0,0 +1,711 @@
|
||||
from datetime import datetime, timedelta
|
||||
from flask import Blueprint, request, jsonify
|
||||
from flask_jwt_extended import create_access_token, jwt_required, get_jwt_identity, get_jwt
|
||||
from Crypto.PublicKey import RSA
|
||||
from werkzeug.security import generate_password_hash, check_password_hash
|
||||
import uuid
|
||||
from models import db, User, UserRole, PendingPassage, Passage, Paper, PaperPassagesRelation
|
||||
|
||||
bp_api_v1 = Blueprint('api/v1', __name__)
|
||||
SESSION_EXPIRATION = timedelta(minutes=5)
|
||||
sessions = {}
|
||||
|
||||
|
||||
def cleanup_sessions():
|
||||
now = datetime.now()
|
||||
expired_keys = [sid for sid, info in sessions.items() if now - info['created_at'] > SESSION_EXPIRATION]
|
||||
for sid in expired_keys:
|
||||
sessions.pop(sid, None)
|
||||
|
||||
|
||||
@bp_api_v1.route('/ping')
|
||||
def ping():
|
||||
return 'pong'
|
||||
|
||||
|
||||
@bp_api_v1.route('/get_key')
|
||||
def get_key():
|
||||
cleanup_sessions()
|
||||
|
||||
session_id = str(uuid.uuid4())
|
||||
private_key = RSA.generate(2048)
|
||||
sessions[session_id] = {
|
||||
'private_key': private_key,
|
||||
'created_at': datetime.now()
|
||||
}
|
||||
|
||||
public_key_pem = private_key.publickey().export_key(format='PEM').decode('utf-8')
|
||||
return jsonify({'pkey': public_key_pem, 'session_id': session_id}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/register', methods=['POST'])
|
||||
def register():
|
||||
cleanup_sessions()
|
||||
data = request.get_json()
|
||||
|
||||
session_id = data.get('session_id')
|
||||
if not session_id or session_id not in sessions:
|
||||
return jsonify({'error': 'Invalid session_id'}), 400
|
||||
|
||||
username = data.get('username')
|
||||
if not username or len(username) == 0 or len(username) > 80:
|
||||
return jsonify({'error': 'Username is required'}), 400
|
||||
if User.query.filter_by(username=username).first():
|
||||
return jsonify({'error': 'Username already exists'}), 400
|
||||
|
||||
nickname = data.get('nickname')
|
||||
if nickname and len(nickname) > 80:
|
||||
return jsonify({'error': 'Nickname is too long'}), 400
|
||||
|
||||
phone = data.get('phone')
|
||||
email = data.get('email')
|
||||
qq = data.get('qq')
|
||||
if phone and (len(phone) != 11 or not phone.isdigit()):
|
||||
return jsonify({'error': 'Invalid phone number'}), 400
|
||||
if email and (not '@' in email or len(email) > 120):
|
||||
return jsonify({'error': 'Invalid email address'}), 400
|
||||
if qq and (len(qq) < 5 or not qq.isdigit() or len(qq) > 10):
|
||||
return jsonify({'error': 'Invalid QQ number'}), 400
|
||||
if not phone and not email and not qq:
|
||||
return jsonify({'error': 'At least one contact method is required'}), 400
|
||||
|
||||
password_encrypted = data.get('password')
|
||||
if not password_encrypted:
|
||||
return jsonify({'error': 'Password is required'}), 400
|
||||
|
||||
private_key = sessions[session_id]['private_key']
|
||||
try:
|
||||
password_decrypted = private_key.decrypt(bytes.fromhex(password_encrypted)).decode('utf-8')
|
||||
password_hash = generate_password_hash(password_decrypted)
|
||||
except Exception:
|
||||
return jsonify({'error': 'Failed to decrypt password'}), 400
|
||||
|
||||
sessions.pop(session_id, None)
|
||||
new_user = User(username=username, nickname=nickname, password=password_hash, role=UserRole.USER, phone=phone, email=email, qq=qq)
|
||||
db.session.add(new_user)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'User registered successfully'}), 201
|
||||
|
||||
|
||||
@bp_api_v1.route('/login', methods=['POST'])
|
||||
def login():
|
||||
cleanup_sessions()
|
||||
data = request.get_json()
|
||||
|
||||
session_id = data.get('session_id')
|
||||
if not session_id or session_id not in sessions:
|
||||
return jsonify({'error': 'Invalid session_id'}), 400
|
||||
|
||||
username = data.get('username')
|
||||
password_encrypted = data.get('password')
|
||||
if not username or not password_encrypted:
|
||||
return jsonify({'error': 'Username and password are required'}), 400
|
||||
|
||||
user = User.query.filter_by(username=username).first()
|
||||
if not user:
|
||||
return jsonify({'error': 'User has not been registered'}), 401
|
||||
|
||||
private_key = sessions[session_id]['private_key']
|
||||
try:
|
||||
password_decrypted = private_key.decrypt(bytes.fromhex(password_encrypted)).decode('utf-8')
|
||||
except Exception:
|
||||
return jsonify({'error': 'Failed to decrypt password'}), 400
|
||||
|
||||
if not check_password_hash(user.password, password_decrypted):
|
||||
return jsonify({'error': 'Incorrect username or password'}), 401
|
||||
|
||||
sessions.pop(session_id, None)
|
||||
access_token = create_access_token(identity=user.id, expires_delta=timedelta(days=30 * 6), additional_claims={
|
||||
'role': user.role.value,
|
||||
'username': user.username,
|
||||
'phone': user.phone,
|
||||
'email': user.email,
|
||||
'qq': user.qq
|
||||
})
|
||||
return jsonify({'access_token': access_token}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/profile')
|
||||
@jwt_required()
|
||||
def profile():
|
||||
user_id = get_jwt_identity()
|
||||
user = User.query.get(user_id)
|
||||
if not user:
|
||||
return jsonify({'error': 'User not found'}), 404
|
||||
|
||||
return jsonify({
|
||||
'username': user.username,
|
||||
'role': user.role.value,
|
||||
'created_at': user.created_at.isoformat(),
|
||||
'phone': user.phone,
|
||||
'email': user.email,
|
||||
'qq': user.qq
|
||||
}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/passages/<int:passage_id>', methods=['GET'])
|
||||
@jwt_required(optional=True)
|
||||
def get_passage(passage_id):
|
||||
passage = Passage.query.get(passage_id)
|
||||
if not passage:
|
||||
return jsonify({'error': 'Passage not found'}), 404
|
||||
|
||||
if not passage.is_posted:
|
||||
jwt_data = get_jwt()
|
||||
if not jwt_data or jwt_data.get('role') != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Passage is not posted yet'}), 403
|
||||
|
||||
return jsonify({
|
||||
'id': passage.id,
|
||||
'title': passage.title,
|
||||
'content': passage.content,
|
||||
'image_url': passage.image_url,
|
||||
'authors': passage.authors,
|
||||
'created_at': passage.created_at.isoformat(),
|
||||
'is_outlink': passage.is_outlink,
|
||||
'sender': {
|
||||
'id': passage.sender.id,
|
||||
'username': passage.sender.username
|
||||
}
|
||||
}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/passages', methods=['GET'])
|
||||
@jwt_required()
|
||||
def list_passages():
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
try:
|
||||
page = int(request.args.get('page', 1))
|
||||
per_page = int(request.args.get('per_page', 10))
|
||||
except ValueError:
|
||||
return jsonify({'error': 'Invalid pagination parameters'}), 400
|
||||
|
||||
query = Passage.query.paginate(page=page, per_page=per_page, error_out=False)
|
||||
passages = query.items
|
||||
return jsonify([{
|
||||
'id': p.id,
|
||||
'title': p.title,
|
||||
'image_url': p.image_url,
|
||||
'authors': p.authors,
|
||||
'created_at': p.created_at.isoformat(),
|
||||
'is_outlink': p.is_outlink,
|
||||
'sender': {
|
||||
'id': p.sender.id,
|
||||
'username': p.sender.username
|
||||
},
|
||||
'has_next': query.has_next
|
||||
} for p in passages]), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/papers/<int:paper_id>', methods=['GET'])
|
||||
def get_paper(paper_id):
|
||||
paper = Paper.query.get(paper_id)
|
||||
if not paper:
|
||||
return jsonify({'error': 'Paper not found'}), 404
|
||||
|
||||
return jsonify({
|
||||
'id': paper.id,
|
||||
'title': paper.title,
|
||||
'description': paper.description,
|
||||
'image_url': paper.image_url,
|
||||
'created_at': paper.created_at.isoformat(),
|
||||
'passages': [p.id for p in paper.passages]
|
||||
}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/papers', methods=['GET'])
|
||||
def list_papers():
|
||||
try:
|
||||
page = int(request.args.get('page', 1))
|
||||
per_page = int(request.args.get('per_page', 10))
|
||||
except ValueError:
|
||||
return jsonify({'error': 'Invalid pagination parameters'}), 400
|
||||
|
||||
query = Paper.query.paginate(page=page, per_page=per_page, error_out=False)
|
||||
papers = query.items
|
||||
return jsonify([{
|
||||
'id': p.id,
|
||||
'title': p.title,
|
||||
'description': p.description,
|
||||
'image_url': p.image_url,
|
||||
'created_at': p.created_at.isoformat(),
|
||||
'has_next': query.has_next
|
||||
} for p in papers]), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/create_passage', methods=['POST'])
|
||||
@jwt_required()
|
||||
def create_passage():
|
||||
user_id = get_jwt_identity()
|
||||
user = User.query.get(user_id)
|
||||
if not user:
|
||||
return jsonify({'error': 'User not found'}), 404
|
||||
|
||||
data = request.get_json()
|
||||
title = data.get('title')
|
||||
content = data.get('content')
|
||||
image_url = data.get('image_url')
|
||||
authors = data.get('authors')
|
||||
is_outlink = data.get('is_outlink', False)
|
||||
|
||||
if not title or len(title) == 0 or len(title) > 200:
|
||||
return jsonify({'error': 'Invalid title'}), 400
|
||||
if not content or len(content) == 0:
|
||||
return jsonify({'error': 'Invalid content'}), 400
|
||||
if image_url and len(image_url) > 500:
|
||||
return jsonify({'error': 'Image URL is too long'}), 400
|
||||
if not authors or len(authors) == 0 or len(authors) > 60:
|
||||
return jsonify({'error': 'Invalid authors'}), 400
|
||||
|
||||
new_passage = PendingPassage(
|
||||
sender_id=user.id,
|
||||
title=title,
|
||||
content=content,
|
||||
image_url=image_url,
|
||||
authors=authors,
|
||||
is_outlink=is_outlink
|
||||
)
|
||||
db.session.add(new_passage)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Passage submitted successfully'}), 201
|
||||
|
||||
|
||||
@bp_api_v1.route('/pending_passages', methods=['GET'])
|
||||
@jwt_required()
|
||||
def list_pending_passages():
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
try:
|
||||
page = int(request.args.get('page', 1))
|
||||
per_page = int(request.args.get('per_page', 10))
|
||||
except ValueError:
|
||||
return jsonify({'error': 'Invalid pagination parameters'}), 400
|
||||
|
||||
query = PendingPassage.query.paginate(page=page, per_page=per_page, error_out=False)
|
||||
passages = query.items
|
||||
return jsonify([{
|
||||
'id': p.id,
|
||||
'title': p.title,
|
||||
'content': p.content,
|
||||
'image_url': p.image_url,
|
||||
'authors': p.authors,
|
||||
'created_at': p.created_at.isoformat(),
|
||||
'sender': {
|
||||
'id': p.sender.id,
|
||||
'username': p.sender.username
|
||||
},
|
||||
'has_next': query.has_next
|
||||
} for p in passages]), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/pending_passages/approve/<int:passage_id>', methods=['POST'])
|
||||
@jwt_required()
|
||||
def approve_passage(passage_id):
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
passage = PendingPassage.query.get(passage_id)
|
||||
if not passage:
|
||||
return jsonify({'error': 'Pending passage not found'}), 404
|
||||
|
||||
new_passage = Passage(
|
||||
sender_id=passage.sender_id,
|
||||
processor_id=get_jwt_identity(),
|
||||
title=passage.title,
|
||||
content=passage.content,
|
||||
image_url=passage.image_url,
|
||||
authors=passage.authors,
|
||||
is_outlink=passage.is_outlink
|
||||
)
|
||||
db.session.add(new_passage)
|
||||
db.session.delete(passage)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Passage approved successfully'}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/pending_passages/reject/<int:passage_id>', methods=['POST'])
|
||||
@jwt_required()
|
||||
def reject_passage(passage_id):
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
passage = PendingPassage.query.get(passage_id)
|
||||
if not passage:
|
||||
return jsonify({'error': 'Pending passage not found'}), 404
|
||||
|
||||
db.session.delete(passage)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Passage rejected successfully'}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/update_passage/<int:passage_id>', methods=['POST'])
|
||||
@jwt_required()
|
||||
def update_passage(passage_id):
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
passage = Passage.query.get(passage_id)
|
||||
if not passage:
|
||||
return jsonify({'error': 'Passage not found'}), 404
|
||||
|
||||
data = request.get_json()
|
||||
title = data.get('title')
|
||||
content = data.get('content')
|
||||
image_url = data.get('image_url')
|
||||
authors = data.get('authors')
|
||||
is_outlink = data.get('is_outlink')
|
||||
|
||||
if title and (len(title) == 0 or len(title) > 200):
|
||||
return jsonify({'error': 'Invalid title'}), 400
|
||||
if content and len(content) == 0:
|
||||
return jsonify({'error': 'Invalid content'}), 400
|
||||
if image_url and len(image_url) > 500:
|
||||
return jsonify({'error': 'Image URL is too long'}), 400
|
||||
if authors and (len(authors) == 0 or len(authors) > 60):
|
||||
return jsonify({'error': 'Invalid authors'}), 400
|
||||
|
||||
if title:
|
||||
passage.title = title
|
||||
if content:
|
||||
passage.content = content
|
||||
if image_url is not None:
|
||||
passage.image_url = image_url
|
||||
if authors:
|
||||
passage.authors = authors
|
||||
if is_outlink is not None:
|
||||
passage.is_outlink = is_outlink
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Passage updated successfully'}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/delete_passage/<int:passage_id>', methods=['POST'])
|
||||
@jwt_required()
|
||||
def delete_passage(passage_id):
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
passage = Passage.query.get(passage_id)
|
||||
if not passage:
|
||||
return jsonify({'error': 'Passage not found'}), 404
|
||||
|
||||
db.session.delete(passage)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Passage deleted successfully'}), 200
|
||||
|
||||
@bp_api_v1.route('/update_pending_passage/<int:passage_id>', methods=['POST'])
|
||||
@jwt_required()
|
||||
def update_pending_passage(passage_id):
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value and (get_jwt_identity() != PendingPassage.query.get(passage_id).sender_id):
|
||||
return jsonify({'error': 'Permission denied'}), 403
|
||||
|
||||
passage = PendingPassage.query.get(passage_id)
|
||||
if not passage:
|
||||
return jsonify({'error': 'Pending passage not found'}), 404
|
||||
|
||||
data = request.get_json()
|
||||
title = data.get('title')
|
||||
content = data.get('content')
|
||||
image_url = data.get('image_url')
|
||||
authors = data.get('authors')
|
||||
is_outlink = data.get('is_outlink')
|
||||
|
||||
if title and (len(title) == 0 or len(title) > 200):
|
||||
return jsonify({'error': 'Invalid title'}), 400
|
||||
if content and len(content) == 0:
|
||||
return jsonify({'error': 'Invalid content'}), 400
|
||||
if image_url and len(image_url) > 500:
|
||||
return jsonify({'error': 'Image URL is too long'}), 400
|
||||
if authors and (len(authors) == 0 or len(authors) > 60):
|
||||
return jsonify({'error': 'Invalid authors'}), 400
|
||||
|
||||
if title:
|
||||
passage.title = title
|
||||
if content:
|
||||
passage.content = content
|
||||
if image_url is not None:
|
||||
passage.image_url = image_url
|
||||
if authors:
|
||||
passage.authors = authors
|
||||
if is_outlink is not None:
|
||||
passage.is_outlink = is_outlink
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Pending passage updated successfully'}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/delete_pending_passage/<int:passage_id>', methods=['POST'])
|
||||
@jwt_required()
|
||||
def delete_pending_passage(passage_id):
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value and (get_jwt_identity() != PendingPassage.query.get(passage_id).sender_id):
|
||||
return jsonify({'error': 'Permission denied'}), 403
|
||||
|
||||
passage = PendingPassage.query.get(passage_id)
|
||||
if not passage:
|
||||
return jsonify({'error': 'Pending passage not found'}), 404
|
||||
|
||||
db.session.delete(passage)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Pending passage deleted successfully'}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/create_paper', methods=['POST'])
|
||||
@jwt_required()
|
||||
def create_paper():
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
data = request.get_json()
|
||||
title = data.get('title')
|
||||
description = data.get('description')
|
||||
image_url = data.get('image_url')
|
||||
passage_ids = data.get('passage_ids', [])
|
||||
|
||||
if not title or len(title) == 0 or len(title) > 200:
|
||||
return jsonify({'error': 'Invalid title'}), 400
|
||||
if description and len(description) > 500:
|
||||
return jsonify({'error': 'Description is too long'}), 400
|
||||
if image_url and len(image_url) > 500:
|
||||
return jsonify({'error': 'Image URL is too long'}), 400
|
||||
|
||||
relations = []
|
||||
for pid in passage_ids:
|
||||
passage = Passage.query.get(pid)
|
||||
if passage:
|
||||
relation = PaperPassagesRelation(paper_id=new_paper.id, passage_id=pid)
|
||||
relations.append(relation)
|
||||
passage.is_posted = True
|
||||
else:
|
||||
db.session.rollback()
|
||||
return jsonify({'error': f'Passage with id {pid} not found'}), 400
|
||||
|
||||
|
||||
new_paper = Paper(
|
||||
title=title,
|
||||
description=description,
|
||||
image_url=image_url
|
||||
)
|
||||
db.session.add(new_paper)
|
||||
db.session.commit()
|
||||
|
||||
for relation in relations:
|
||||
db.session.add(relation)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Paper created successfully'}), 201
|
||||
|
||||
|
||||
@bp_api_v1.route('/delete_paper/<int:paper_id>', methods=['POST'])
|
||||
@jwt_required()
|
||||
def delete_paper(paper_id):
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
paper = Paper.query.get(paper_id)
|
||||
if not paper:
|
||||
return jsonify({'error': 'Paper not found'}), 404
|
||||
|
||||
relations = PaperPassagesRelation.query.filter_by(paper_id=paper_id).all()
|
||||
for relation in relations:
|
||||
passage = Passage.query.get(relation.passage_id)
|
||||
if passage:
|
||||
if PaperPassagesRelation.query.filter_by(passage_id=passage.id).count() == 1:
|
||||
passage.is_posted = False
|
||||
db.session.delete(relation)
|
||||
|
||||
db.session.delete(paper)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Paper deleted successfully'}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/update_paper/<int:paper_id>', methods=['POST'])
|
||||
@jwt_required()
|
||||
def update_paper(paper_id):
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
paper = Paper.query.get(paper_id)
|
||||
if not paper:
|
||||
return jsonify({'error': 'Paper not found'}), 404
|
||||
|
||||
data = request.get_json()
|
||||
title = data.get('title')
|
||||
description = data.get('description')
|
||||
image_url = data.get('image_url')
|
||||
passage_ids = data.get('passage_ids')
|
||||
|
||||
if title and (len(title) == 0 or len(title) > 200):
|
||||
return jsonify({'error': 'Invalid title'}), 400
|
||||
if description and len(description) > 500:
|
||||
return jsonify({'error': 'Description is too long'}), 400
|
||||
if image_url and len(image_url) > 500:
|
||||
return jsonify({'error': 'Image URL is too long'}), 400
|
||||
|
||||
add_relations = []
|
||||
remove_relations = []
|
||||
if passage_ids is not None:
|
||||
existing_relations = PaperPassagesRelation.query.filter_by(paper_id=paper_id).all()
|
||||
existing_passage_ids = {r.passage_id for r in existing_relations}
|
||||
new_passage_ids = set(passage_ids)
|
||||
|
||||
for pid in new_passage_ids:
|
||||
if pid not in existing_passage_ids:
|
||||
passage = Passage.query.get(pid)
|
||||
if passage:
|
||||
add_relations.append(PaperPassagesRelation(paper_id=paper_id, passage_id=pid))
|
||||
passage.is_posted = True
|
||||
else:
|
||||
db.session.rollback()
|
||||
return jsonify({'error': f'Passage with id {pid} not found'}), 400
|
||||
|
||||
for relation in existing_relations:
|
||||
if relation.passage_id not in new_passage_ids:
|
||||
remove_relations.append(relation)
|
||||
passage = Passage.query.get(relation.passage_id)
|
||||
if passage and PaperPassagesRelation.query.filter_by(passage_id=passage.id).count() == 1:
|
||||
passage.is_posted = False
|
||||
|
||||
if title:
|
||||
paper.title = title
|
||||
if description is not None:
|
||||
paper.description = description
|
||||
if image_url is not None:
|
||||
paper.image_url = image_url
|
||||
|
||||
for relation in add_relations:
|
||||
db.session.add(relation)
|
||||
for relation in remove_relations:
|
||||
db.session.delete(relation)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Paper updated successfully'}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/update_profile', methods=['POST'])
|
||||
@jwt_required()
|
||||
def update_profile():
|
||||
current_user_id = get_jwt_identity()
|
||||
user = User.query.get(current_user_id)
|
||||
if not user:
|
||||
return jsonify({'error': 'User not found'}), 404
|
||||
|
||||
if user.role != UserRole.ADMIN.value and user.id != current_user_id:
|
||||
return jsonify({'error': 'Permission denied'}), 403
|
||||
|
||||
data = request.get_json()
|
||||
nickname = data.get('nickname')
|
||||
phone = data.get('phone')
|
||||
email = data.get('email')
|
||||
qq = data.get('qq')
|
||||
|
||||
if nickname is not None:
|
||||
if len(nickname) > 80:
|
||||
return jsonify({'error': 'Nickname is too long'}), 400
|
||||
user.nickname = nickname
|
||||
|
||||
if phone is not None:
|
||||
user.phone = phone
|
||||
if email is not None:
|
||||
user.email = email
|
||||
if qq is not None:
|
||||
user.qq = qq
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Profile updated successfully'}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/change_password', methods=['POST'])
|
||||
@jwt_required()
|
||||
def change_password():
|
||||
cleanup_sessions()
|
||||
current_user_id = get_jwt_identity()
|
||||
user = User.query.get(current_user_id)
|
||||
if not user:
|
||||
return jsonify({'error': 'User not found'}), 404
|
||||
|
||||
data = request.get_json()
|
||||
|
||||
session_id = data.get('session_id')
|
||||
if not session_id or session_id not in sessions:
|
||||
return jsonify({'error': 'Invalid session_id'}), 400
|
||||
|
||||
old_password_encrypted = data.get('old_password')
|
||||
new_password_encrypted = data.get('new_password')
|
||||
if not old_password_encrypted or not new_password_encrypted:
|
||||
return jsonify({'error': 'Old and new passwords are required'}), 400
|
||||
|
||||
private_key = sessions[session_id]['private_key']
|
||||
try:
|
||||
old_password_decrypted = private_key.decrypt(bytes.fromhex(old_password_encrypted)).decode('utf-8')
|
||||
new_password_decrypted = private_key.decrypt(bytes.fromhex(new_password_encrypted)).decode('utf-8')
|
||||
except Exception:
|
||||
return jsonify({'error': 'Failed to decrypt passwords'}), 400
|
||||
|
||||
if not check_password_hash(user.password, old_password_decrypted):
|
||||
return jsonify({'error': 'Incorrect old password'}), 401
|
||||
|
||||
sessions.pop(session_id, None)
|
||||
user.password = generate_password_hash(new_password_decrypted)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Password changed successfully'}), 200
|
||||
|
||||
|
||||
@bp_api_v1.route('/admin/reset_password/<int:user_id>', methods=['POST'])
|
||||
@jwt_required()
|
||||
def admin_reset_password(user_id):
|
||||
cleanup_sessions()
|
||||
role = get_jwt()['role']
|
||||
if role != UserRole.ADMIN.value:
|
||||
return jsonify({'error': 'Admin access required'}), 403
|
||||
|
||||
user = User.query.get(user_id)
|
||||
if not user:
|
||||
return jsonify({'error': 'User not found'}), 404
|
||||
|
||||
data = request.get_json()
|
||||
|
||||
session_id = data.get('session_id')
|
||||
if not session_id or session_id not in sessions:
|
||||
return jsonify({'error': 'Invalid session_id'}), 400
|
||||
|
||||
new_password_encrypted = data.get('new_password')
|
||||
if not new_password_encrypted:
|
||||
return jsonify({'error': 'New password is required'}), 400
|
||||
|
||||
private_key = sessions[session_id]['private_key']
|
||||
try:
|
||||
new_password_decrypted = private_key.decrypt(bytes.fromhex(new_password_encrypted)).decode('utf-8')
|
||||
except Exception:
|
||||
return jsonify({'error': 'Failed to decrypt password'}), 400
|
||||
|
||||
sessions.pop(session_id, None)
|
||||
user.password = generate_password_hash(new_password_decrypted)
|
||||
db.session.commit()
|
||||
|
||||
return jsonify({'message': 'Password reset successfully'}), 200
|
||||
Reference in New Issue
Block a user