Spaces:
Running
Running
File size: 3,572 Bytes
3546e19 5eb5225 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 |
import os, jwt, datetime, json
from flask import Flask, request, jsonify
from werkzeug.security import generate_password_hash, check_password_hash
from models import db, User, SearchHistory
from flask_cors import CORS
from functools import wraps
from datetime import datetime as dt
SECRET_KEY = os.getenv("SECRET_KEY", "ultra_secret_dev_key")
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///users.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db.init_app(app)
CORS(app)
with app.app_context():
db.create_all()
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = None
if "Authorization" in request.headers:
token = request.headers["Authorization"].split(" ")[-1]
if not token:
return jsonify({"message": "Token is missing!"}), 401
try:
data = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
current_user = User.query.get(data["user_id"])
except Exception:
return jsonify({"message": "Token is invalid!"}), 401
return f(current_user, *args, **kwargs)
return decorated
@app.route("/signup", methods=["POST"])
def signup():
data = request.json
if User.query.filter_by(username=data["username"]).first():
return jsonify({"message": "User already exists"}), 400
hashed_pw = generate_password_hash(data["password"])
new_user = User(username=data["username"], password_hash=hashed_pw)
db.session.add(new_user)
db.session.commit()
return jsonify({"message": "User created"}), 201
@app.route("/login", methods=["POST"])
def login():
data = request.json
user = User.query.filter_by(username=data["username"]).first()
if not user or not check_password_hash(user.password_hash, data["password"]):
return jsonify({"message": "Invalid credentials"}), 401
token = jwt.encode({
"user_id": user.id,
"exp": datetime.datetime.utcnow() + datetime.timedelta(hours=2)
}, SECRET_KEY, algorithm="HS256")
return jsonify({"token": token})
@app.route("/search", methods=["POST"])
@token_required
def secure_search(current_user):
data = request.get_json()
query = data.get("query", "")
filters = data.get("filters", {})
# Save search to history
entry = SearchHistory(
query=query,
filters=json.dumps(filters),
user_id=current_user.id,
timestamp=dt.utcnow()
)
db.session.add(entry)
db.session.commit()
# Simulate filtered results
results = [
{"title": f"{query} ({', '.join(k for k, v in filters.items() if v)})", "url": "#"}
]
return jsonify({
"user": current_user.username,
"results": results,
"ai_preview": "https://fake-preview.com/image.png"
})
@app.route("/history", methods=["GET"])
@token_required
def get_history(current_user):
history = SearchHistory.query.filter_by(user_id=current_user.id)\
.order_by(SearchHistory.timestamp.desc()).limit(20).all()
return jsonify([
{
"query": h.query,
"filters": json.loads(h.filters or '{}'),
"timestamp": h.timestamp.isoformat()
}
for h in history
])
@app.route("/history", methods=["DELETE"])
@token_required
def clear_history(current_user):
deleted = SearchHistory.query.filter_by(user_id=current_user.id).delete()
db.session.commit()
return jsonify({"message": f"{deleted} entries deleted."}), 200
if __name__ == "__main__":
app.run(host="0.0.0.0", port=7860))
|