564 lines
19 KiB
Python
564 lines
19 KiB
Python
import os
|
|
import time
|
|
import json
|
|
import uuid
|
|
import math
|
|
import re
|
|
from datetime import datetime
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
from werkzeug.utils import secure_filename
|
|
from flask import Flask, request, jsonify, session, send_from_directory
|
|
from flask_cors import CORS
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = "your-secret-key-here-change-in-production" # needed for sessions
|
|
CORS(app, supports_credentials=True) # allow cookies
|
|
|
|
# Directories
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DATA_DIR = os.path.join(BASE_DIR, "data")
|
|
USERS_FILE = os.path.join(DATA_DIR, "users.json")
|
|
JOURNEYS_FILE = os.path.join(DATA_DIR, 'journeys.json')
|
|
UPLOAD_DIR = os.path.join(BASE_DIR, "uploads")
|
|
ALLOWED_IMAGE_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}
|
|
VALID_VISIBILITIES = {"private", "public", "shared"}
|
|
MAX_MARKERS_PER_JOURNEY = 500
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
|
|
|
|
|
class ValidationError(ValueError):
|
|
pass
|
|
|
|
|
|
@app.errorhandler(ValidationError)
|
|
def handle_validation_error(error):
|
|
return jsonify({"error": str(error)}), 400
|
|
|
|
|
|
def get_json_object():
|
|
data = request.get_json(silent=True)
|
|
if not isinstance(data, dict):
|
|
raise ValidationError("Request body must be a JSON object")
|
|
return data
|
|
|
|
|
|
def clean_text(value, field, max_length, required=False, strip=True):
|
|
if value is None:
|
|
value = ""
|
|
if not isinstance(value, str):
|
|
raise ValidationError(f"{field} must be text")
|
|
|
|
# Keep newlines/tabs for Markdown, but remove non-printing control characters.
|
|
value = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", value)
|
|
if strip:
|
|
value = value.strip()
|
|
if required and not value:
|
|
raise ValidationError(f"{field} is required")
|
|
if len(value) > max_length:
|
|
raise ValidationError(f"{field} must be at most {max_length} characters")
|
|
return value
|
|
|
|
|
|
def clean_number(value, field, minimum, maximum):
|
|
if isinstance(value, bool):
|
|
raise ValidationError(f"{field} must be a number")
|
|
try:
|
|
value = float(value)
|
|
except (TypeError, ValueError):
|
|
raise ValidationError(f"{field} must be a number")
|
|
if not math.isfinite(value) or not minimum <= value <= maximum:
|
|
raise ValidationError(f"{field} must be between {minimum} and {maximum}")
|
|
return value
|
|
|
|
|
|
def clean_visibility(value):
|
|
if value not in VALID_VISIBILITIES:
|
|
raise ValidationError("Invalid journey visibility")
|
|
return value
|
|
|
|
|
|
def clean_images(images):
|
|
if images is None:
|
|
return []
|
|
if not isinstance(images, list):
|
|
raise ValidationError("Marker images must be a list")
|
|
if len(images) > 20:
|
|
raise ValidationError("A marker can contain at most 20 images")
|
|
|
|
cleaned = []
|
|
for image in images:
|
|
if isinstance(image, str):
|
|
cleaned.append(clean_text(image, "Image URL", 2048, required=True))
|
|
continue
|
|
if not isinstance(image, dict):
|
|
raise ValidationError("Invalid marker image")
|
|
url = clean_text(image.get("url"), "Image URL", 2048, required=True)
|
|
cleaned.append({
|
|
"filename": clean_text(image.get("filename"), "Image filename", 255),
|
|
"originalName": clean_text(image.get("originalName"), "Original image name", 255),
|
|
"url": url,
|
|
})
|
|
return cleaned
|
|
|
|
|
|
def clean_markers(markers):
|
|
if markers is None:
|
|
return []
|
|
if not isinstance(markers, list):
|
|
raise ValidationError("Markers must be a list")
|
|
if len(markers) > MAX_MARKERS_PER_JOURNEY:
|
|
raise ValidationError(f"A journey can contain at most {MAX_MARKERS_PER_JOURNEY} markers")
|
|
|
|
cleaned = []
|
|
for marker in markers:
|
|
if not isinstance(marker, dict):
|
|
raise ValidationError("Invalid marker")
|
|
cleaned.append({
|
|
"lat": clean_number(marker.get("lat"), "Marker latitude", -90, 90),
|
|
"lng": clean_number(marker.get("lng"), "Marker longitude", -180, 180),
|
|
"title": clean_text(marker.get("title"), "Marker title", 200),
|
|
"date": clean_text(marker.get("date"), "Marker date", 20),
|
|
"description": clean_text(marker.get("description"), "Marker description", 10000),
|
|
"images": clean_images(marker.get("images", [])),
|
|
})
|
|
return cleaned
|
|
|
|
|
|
|
|
# ==================== User helpers ====================
|
|
def require_login():
|
|
if "user_id" not in session:
|
|
return False
|
|
return True
|
|
|
|
def get_current_user_id():
|
|
return session.get("user_id")
|
|
|
|
def load_users():
|
|
try:
|
|
if os.path.exists(USERS_FILE):
|
|
with open(USERS_FILE, "r") as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
print(f"Error loading users: {e}")
|
|
return []
|
|
|
|
|
|
def save_users(users):
|
|
try:
|
|
with open(USERS_FILE, "w") as f:
|
|
json.dump(users, f, indent=2)
|
|
except Exception as e:
|
|
print(f"Error saving users: {e}")
|
|
|
|
|
|
def get_next_user_id(users):
|
|
if not users:
|
|
return 1
|
|
return max(u["id"] for u in users) + 1
|
|
|
|
|
|
def get_user_by_username(username):
|
|
users = load_users()
|
|
return next((u for u in users if u["username"] == username), None)
|
|
|
|
|
|
def get_user_by_id(user_id):
|
|
users = load_users()
|
|
return next((u for u in users if u["id"] == user_id), None)
|
|
|
|
|
|
def public_user(user):
|
|
return {
|
|
"id": user["id"],
|
|
"username": user["username"],
|
|
}
|
|
|
|
|
|
def normalize_user_ids(user_ids):
|
|
users = load_users()
|
|
valid_ids = {user["id"] for user in users}
|
|
normalized = []
|
|
for user_id in user_ids or []:
|
|
try:
|
|
user_id = int(user_id)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if user_id in valid_ids and user_id not in normalized:
|
|
normalized.append(user_id)
|
|
return normalized
|
|
|
|
|
|
def allowed_image_file(filename):
|
|
if "." not in filename:
|
|
return False
|
|
extension = filename.rsplit(".", 1)[1].lower()
|
|
return extension in ALLOWED_IMAGE_EXTENSIONS
|
|
|
|
# ==================== Authentication endpoints ====================
|
|
@app.route("/api/register", methods=["POST"])
|
|
def register():
|
|
data = get_json_object()
|
|
username = clean_text(data.get("username"), "Username", 50, required=True)
|
|
password = clean_text(data.get("password"), "Password", 200, required=True, strip=False)
|
|
if len(password) < 4:
|
|
raise ValidationError("Password must be at least 4 characters")
|
|
|
|
# Check if username already exists
|
|
if get_user_by_username(username):
|
|
return jsonify({"error": "Username already taken"}), 409
|
|
|
|
users = load_users()
|
|
new_id = get_next_user_id(users)
|
|
|
|
hashed = generate_password_hash(password)
|
|
new_user = {
|
|
"id": new_id,
|
|
"username": username,
|
|
"password_hash": hashed,
|
|
"created_at": datetime.now().isoformat(),
|
|
}
|
|
users.append(new_user)
|
|
save_users(users)
|
|
|
|
# Log the user in automatically
|
|
session["user_id"] = new_id
|
|
|
|
return jsonify(
|
|
{"id": new_id, "username": username, "message": "Registration successful"}
|
|
), 201
|
|
|
|
|
|
@app.route("/api/login", methods=["POST"])
|
|
def login():
|
|
data = get_json_object()
|
|
username = clean_text(data.get("username"), "Username", 50, required=True)
|
|
password = clean_text(data.get("password"), "Password", 200, required=True, strip=False)
|
|
|
|
user = get_user_by_username(username)
|
|
if not user or not check_password_hash(user["password_hash"], password):
|
|
return jsonify({"error": "Invalid username or password"}), 401
|
|
|
|
session["user_id"] = user["id"]
|
|
return jsonify(
|
|
{"id": user["id"], "username": user["username"], "message": "Login successful"}
|
|
)
|
|
|
|
|
|
@app.route("/api/logout", methods=["POST"])
|
|
def logout():
|
|
session.pop("user_id", None)
|
|
return jsonify({"message": "Logged out"})
|
|
|
|
|
|
@app.route("/api/me", methods=["GET"])
|
|
def me():
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"error": "Not logged in"}), 401
|
|
user = get_user_by_id(user_id)
|
|
if not user:
|
|
# Should not happen, but clean session
|
|
session.pop("user_id", None)
|
|
return jsonify({"error": "User not found"}), 401
|
|
return jsonify({"id": user["id"], "username": user["username"]})
|
|
|
|
|
|
@app.route("/api/users", methods=["GET"])
|
|
def get_users():
|
|
if not require_login():
|
|
return jsonify({"error": "Authentication required"}), 401
|
|
current_user_id = get_current_user_id()
|
|
users = [
|
|
public_user(user)
|
|
for user in load_users()
|
|
if user["id"] != current_user_id
|
|
]
|
|
return jsonify(users)
|
|
|
|
# ==================== Journey helper functions ====================
|
|
|
|
def load_all_journeys():
|
|
try:
|
|
if os.path.exists(JOURNEYS_FILE):
|
|
with open(JOURNEYS_FILE, 'r') as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
print(f"Error loading journeys: {e}")
|
|
return []
|
|
|
|
def save_all_journeys(journeys):
|
|
try:
|
|
with open(JOURNEYS_FILE, 'w') as f:
|
|
json.dump(journeys, f, indent=2)
|
|
except Exception as e:
|
|
print(f"Error saving journeys: {e}")
|
|
|
|
def get_next_journey_id(journeys):
|
|
if not journeys:
|
|
return 1
|
|
return max(j['id'] for j in journeys) + 1
|
|
|
|
def get_journey_by_id(journey_id):
|
|
journeys = load_all_journeys()
|
|
return next((j for j in journeys if j['id'] == journey_id), None)
|
|
|
|
def user_can_view_journey(journey, user_id):
|
|
if journey['owner_id'] == user_id:
|
|
return True
|
|
if journey.get('visibility') == 'public':
|
|
return True
|
|
if journey.get('visibility') == 'shared' and user_id in journey.get('shared_read', []):
|
|
return True
|
|
if journey.get('visibility') == 'shared' and user_id in journey.get('shared_edit', []):
|
|
return True
|
|
return False
|
|
|
|
def user_can_edit_journey(journey, user_id):
|
|
if journey['owner_id'] == user_id:
|
|
return True
|
|
if journey.get('visibility') == 'shared' and user_id in journey.get('shared_edit', []):
|
|
return True
|
|
return False
|
|
|
|
|
|
# ==================== Journey endpoints ====================
|
|
@app.route('/api/journeys', methods=['GET'])
|
|
def get_journeys():
|
|
if not require_login():
|
|
return jsonify({'error': 'Authentication required'}), 401
|
|
user_id = get_current_user_id()
|
|
all_journeys = load_all_journeys()
|
|
result = []
|
|
for j in all_journeys:
|
|
if user_can_view_journey(j, user_id):
|
|
# Add extra flags for the frontend
|
|
j_copy = j.copy()
|
|
j_copy['can_edit'] = user_can_edit_journey(j, user_id)
|
|
result.append(j_copy)
|
|
return jsonify(result)
|
|
|
|
@app.route('/api/journeys', methods=['POST'])
|
|
def create_journey():
|
|
if not require_login():
|
|
return jsonify({'error': 'Authentication required'}), 401
|
|
data = get_json_object()
|
|
|
|
title = clean_text(data.get('title'), "Journey title", 200, required=True)
|
|
|
|
user_id = get_current_user_id()
|
|
journeys = load_all_journeys()
|
|
new_id = get_next_journey_id(journeys)
|
|
|
|
new_journey = {
|
|
'id': new_id,
|
|
'owner_id': user_id,
|
|
'title': title,
|
|
'description': clean_text(data.get('description'), "Journey description", 20000),
|
|
'markers': clean_markers(data.get('markers', [])),
|
|
'created_at': datetime.now().isoformat(),
|
|
'visibility': clean_visibility(data.get('visibility', 'private')),
|
|
'shared_read': normalize_user_ids(data.get('shared_read', [])),
|
|
'shared_edit': normalize_user_ids(data.get('shared_edit', [])),
|
|
'comments': []
|
|
}
|
|
|
|
journeys.append(new_journey)
|
|
save_all_journeys(journeys)
|
|
return jsonify(new_journey), 201
|
|
|
|
@app.route('/api/journeys/<int:journey_id>', methods=['GET'])
|
|
def get_journey(journey_id):
|
|
if not require_login():
|
|
return jsonify({'error': 'Authentication required'}), 401
|
|
user_id = get_current_user_id()
|
|
journey = get_journey_by_id(journey_id)
|
|
if journey is None:
|
|
return jsonify({'error': 'Journey not found'}), 404
|
|
if not user_can_view_journey(journey, user_id):
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
return jsonify(journey)
|
|
|
|
@app.route('/api/journeys/<int:journey_id>', methods=['PUT'])
|
|
def update_journey(journey_id):
|
|
if not require_login():
|
|
return jsonify({'error': 'Authentication required'}), 401
|
|
user_id = get_current_user_id()
|
|
journeys = load_all_journeys()
|
|
journey = next((j for j in journeys if j['id'] == journey_id), None)
|
|
if journey is None:
|
|
return jsonify({'error': 'Journey not found'}), 404
|
|
if not user_can_edit_journey(journey, user_id):
|
|
return jsonify({'error': 'Not authorized to edit this journey'}), 403
|
|
|
|
data = get_json_object()
|
|
if 'title' in data:
|
|
journey['title'] = clean_text(data['title'], "Journey title", 200, required=True)
|
|
if 'description' in data:
|
|
journey['description'] = clean_text(data['description'], "Journey description", 20000)
|
|
if 'markers' in data:
|
|
journey['markers'] = clean_markers(data['markers'])
|
|
sharing_fields = {'visibility', 'shared_read', 'shared_edit'}
|
|
if sharing_fields.intersection(data.keys()):
|
|
if journey['owner_id'] != user_id:
|
|
return jsonify({'error': 'Only the owner can update sharing settings'}), 403
|
|
if 'visibility' in data:
|
|
journey['visibility'] = clean_visibility(data['visibility'])
|
|
if 'shared_read' in data:
|
|
journey['shared_read'] = normalize_user_ids(data['shared_read'])
|
|
if 'shared_edit' in data:
|
|
journey['shared_edit'] = normalize_user_ids(data['shared_edit'])
|
|
save_all_journeys(journeys)
|
|
return jsonify(journey)
|
|
|
|
@app.route('/api/journeys/<int:journey_id>', methods=['DELETE'])
|
|
def delete_journey(journey_id):
|
|
if not require_login():
|
|
return jsonify({'error': 'Authentication required'}), 401
|
|
user_id = get_current_user_id()
|
|
journeys = load_all_journeys()
|
|
journey = next((j for j in journeys if j['id'] == journey_id), None)
|
|
if journey is None:
|
|
return jsonify({'error': 'Journey not found'}), 404
|
|
if journey['owner_id'] != user_id:
|
|
return jsonify({'error': 'Only the owner can delete this journey'}), 403
|
|
|
|
journeys = [j for j in journeys if j['id'] != journey_id]
|
|
save_all_journeys(journeys)
|
|
return jsonify({'message': 'Journey deleted successfully', 'journey': journey})
|
|
|
|
# ==================== Comments (stored inside journeys) ====================
|
|
def save_journey(journey):
|
|
journeys = load_all_journeys()
|
|
for i, j in enumerate(journeys):
|
|
if j['id'] == journey['id']:
|
|
journeys[i] = journey
|
|
break
|
|
save_all_journeys(journeys)
|
|
|
|
@app.route('/api/journeys/<int:journey_id>/comments', methods=['GET'])
|
|
def get_journey_comments(journey_id):
|
|
user_id = session.get('user_id')
|
|
if not user_id:
|
|
return jsonify({'error': 'Authentication required'}), 401
|
|
|
|
journey = get_journey_by_id(journey_id)
|
|
if not journey:
|
|
return jsonify({'error': 'Journey not found'}), 404
|
|
if not user_can_view_journey(journey, user_id):
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
return jsonify(journey.get('comments', []))
|
|
|
|
@app.route('/api/journeys/<int:journey_id>/comments', methods=['POST'])
|
|
def add_journey_comment(journey_id):
|
|
user_id = session.get('user_id')
|
|
if not user_id:
|
|
return jsonify({'error': 'Authentication required'}), 401
|
|
data = get_json_object()
|
|
text = clean_text(data.get('text'), "Comment", 2000, required=True)
|
|
|
|
journey = get_journey_by_id(journey_id)
|
|
if not journey:
|
|
return jsonify({'error': 'Journey not found'}), 404
|
|
if not user_can_view_journey(journey, user_id):
|
|
return jsonify({'error': 'Access denied'}), 403
|
|
|
|
comment = {
|
|
'id': int(time.time() * 1000),
|
|
'author_id': user_id,
|
|
'author_name': get_user_by_id(user_id)['username'],
|
|
'text': text,
|
|
'created_at': datetime.now().isoformat(),
|
|
}
|
|
if 'comments' not in journey:
|
|
journey['comments'] = []
|
|
journey['comments'].append(comment)
|
|
save_journey(journey)
|
|
|
|
return jsonify(comment), 201
|
|
|
|
@app.route('/api/comments/<int:comment_id>', methods=['DELETE'])
|
|
def delete_comment(comment_id):
|
|
user_id = session.get('user_id')
|
|
if not user_id:
|
|
return jsonify({'error': 'Authentication required'}), 401
|
|
|
|
journeys = load_all_journeys()
|
|
for journey in journeys:
|
|
if 'comments' in journey:
|
|
for i, c in enumerate(journey['comments']):
|
|
if c['id'] == comment_id:
|
|
# Check permissions: comment author or journey owner can delete
|
|
if c['author_id'] == user_id or journey['owner_id'] == user_id:
|
|
del journey['comments'][i]
|
|
save_journey(journey)
|
|
return jsonify({'message': 'Comment deleted'})
|
|
else:
|
|
return jsonify({'error': 'Not authorized'}), 403
|
|
return jsonify({'error': 'Comment not found'}), 404
|
|
|
|
|
|
# ==================== Uploads ====================
|
|
@app.route("/api/uploads/images", methods=["POST"])
|
|
def upload_images():
|
|
if not require_login():
|
|
return jsonify({"error": "Authentication required"}), 401
|
|
|
|
files = request.files.getlist("images")
|
|
if not files:
|
|
return jsonify({"error": "No images provided"}), 400
|
|
|
|
uploaded = []
|
|
for file in files:
|
|
if not file or file.filename == "":
|
|
continue
|
|
if not allowed_image_file(file.filename):
|
|
return jsonify({"error": f"Unsupported image type: {file.filename}"}), 400
|
|
|
|
original_name = secure_filename(file.filename)
|
|
extension = original_name.rsplit(".", 1)[1].lower()
|
|
filename = f"{uuid.uuid4().hex}.{extension}"
|
|
file.save(os.path.join(UPLOAD_DIR, filename))
|
|
uploaded.append(
|
|
{
|
|
"filename": filename,
|
|
"originalName": original_name,
|
|
"url": f"/uploads/{filename}",
|
|
}
|
|
)
|
|
|
|
if not uploaded:
|
|
return jsonify({"error": "No valid images provided"}), 400
|
|
return jsonify({"images": uploaded}), 201
|
|
|
|
|
|
@app.route("/uploads/<path:filename>")
|
|
def uploaded_file(filename):
|
|
return send_from_directory(UPLOAD_DIR, filename)
|
|
|
|
# ==================== Health and root ====================
|
|
@app.route("/api/journeys/health", methods=["GET"])
|
|
def health_check():
|
|
return jsonify({"status": "healthy", "timestamp": datetime.now().isoformat()})
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return """
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head><title>Journey Mapper API</title></head>
|
|
<body>
|
|
<h1>Journey Mapper API</h1>
|
|
<p>Backend is running. Use <code>/api/journeys</code> endpoints.</p>
|
|
<p>Authentication required: register, login, then use session cookies.</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(debug=True, port=5000)
|