153 lines
4.4 KiB
Python
153 lines
4.4 KiB
Python
"""Proje yönetimi: yükleme, listeleme, indirme, silme (yalnızca admin)."""
|
||
import mimetypes
|
||
import os
|
||
import re
|
||
import uuid
|
||
|
||
from flask import (
|
||
Blueprint,
|
||
abort,
|
||
current_app,
|
||
flash,
|
||
redirect,
|
||
render_template,
|
||
request,
|
||
session,
|
||
send_from_directory,
|
||
url_for,
|
||
)
|
||
from werkzeug.utils import secure_filename
|
||
|
||
import security
|
||
from db import UPLOAD_DIR, get_db
|
||
|
||
project_bp = Blueprint("projects", __name__)
|
||
|
||
ALLOWED_EXT = {
|
||
"png", "jpg", "jpeg", "gif", "webp",
|
||
"pdf", "zip", "rar", "7z", "tar", "gz",
|
||
"txt", "md", "html", "htm",
|
||
"doc", "docx", "ppt", "pptx", "xls", "xlsx",
|
||
}
|
||
MAX_UPLOAD = 25 * 1024 * 1024
|
||
|
||
|
||
def _is_admin():
|
||
return session.get("role") == "admin"
|
||
|
||
|
||
@project_bp.before_request
|
||
def _guard():
|
||
if request.path.startswith("/admin") and not _is_admin():
|
||
abort(403)
|
||
|
||
|
||
@project_bp.get("/admin/projeler")
|
||
def admin_panel():
|
||
db = get_db()
|
||
projects = db.execute(
|
||
"SELECT * FROM projects ORDER BY created_at DESC"
|
||
).fetchall()
|
||
return render_template("admin/projects.html", projects=projects)
|
||
|
||
|
||
@project_bp.post("/admin/projeler")
|
||
@security.rate_limited("upload", 12, 900)
|
||
def do_upload():
|
||
title = security.clean_text(request.form.get("title"), 150)
|
||
description = security.clean_multiline(request.form.get("description"), 5000)
|
||
tags = security.clean_text(request.form.get("tags"), 300)
|
||
link = security.validate_link(request.form.get("link"))
|
||
|
||
if not title:
|
||
flash("Proje başlığı zorunlu.", "error")
|
||
return redirect(url_for("projects.admin_panel"))
|
||
if link is None:
|
||
flash("Bağlantı geçersiz (http/https olmalı).", "error")
|
||
return redirect(url_for("projects.admin_panel"))
|
||
|
||
stored_name = None
|
||
orig_name = ""
|
||
mime = None
|
||
size = None
|
||
file = request.files.get("file")
|
||
if file and file.filename:
|
||
orig = secure_filename(file.filename) or "dosya"
|
||
ext = orig.rsplit(".", 1)[-1].lower() if "." in orig else ""
|
||
if ext not in ALLOWED_EXT:
|
||
flash("Bu dosya türüne izin verilmiyor.", "error")
|
||
return redirect(url_for("projects.admin_panel"))
|
||
payload = file.read(MAX_UPLOAD + 1)
|
||
if len(payload) > MAX_UPLOAD:
|
||
flash("Dosya 25 MB sınırını aşıyor.", "error")
|
||
return redirect(url_for("projects.admin_panel"))
|
||
stored_name = uuid.uuid4().hex + "." + ext
|
||
with open(os.path.join(UPLOAD_DIR, stored_name), "wb") as fh:
|
||
fh.write(payload)
|
||
orig_name = orig
|
||
mime = mimetypes.guess_type(orig)[0] or "application/octet-stream"
|
||
size = len(payload)
|
||
|
||
db = get_db()
|
||
db.execute(
|
||
"""INSERT INTO projects (title, description, tags, link, stored_name,
|
||
orig_name, mime, size)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
(title, description, tags, link, stored_name, orig_name, mime, size),
|
||
)
|
||
db.commit()
|
||
flash("Proje yüklendi.", "success")
|
||
return redirect(url_for("projects.admin_panel"))
|
||
|
||
|
||
@project_bp.post("/admin/projeler/<int:pid>/sil")
|
||
def do_delete(pid):
|
||
db = get_db()
|
||
row = db.execute("SELECT * FROM projects WHERE id = ?", (pid,)).fetchone()
|
||
if row is None:
|
||
abort(404)
|
||
if row["stored_name"]:
|
||
path = os.path.join(UPLOAD_DIR, row["stored_name"])
|
||
if os.path.isfile(path):
|
||
os.remove(path)
|
||
db.execute("DELETE FROM projects WHERE id = ?", (pid,))
|
||
db.commit()
|
||
flash("Proje kaldırıldı.", "info")
|
||
return redirect(url_for("projects.admin_panel"))
|
||
|
||
|
||
# ---------------- herkese açık uçlar ----------------
|
||
|
||
def _get_public_project(pid):
|
||
db = get_db()
|
||
return db.execute("SELECT * FROM projects WHERE id = ?", (pid,)).fetchone()
|
||
|
||
|
||
@project_bp.get("/projeler/<int:pid>/indir")
|
||
def download(pid):
|
||
row = _get_public_project(pid)
|
||
if row is None or not row["stored_name"]:
|
||
abort(404)
|
||
return send_from_directory(
|
||
UPLOAD_DIR,
|
||
row["stored_name"],
|
||
as_attachment=True,
|
||
download_name=row["orig_name"] or "proje",
|
||
)
|
||
|
||
|
||
@project_bp.get("/projeler/<int:pid>/goster")
|
||
def show_file(pid):
|
||
row = _get_public_project(pid)
|
||
if row is None or not row["stored_name"]:
|
||
abort(404)
|
||
if not (row["mime"] or "").startswith("image/"):
|
||
abort(404)
|
||
return send_from_directory(UPLOAD_DIR, row["stored_name"])
|
||
|
||
|
||
def public_projects():
|
||
db = get_db()
|
||
return db.execute(
|
||
"SELECT * FROM projects ORDER BY created_at DESC"
|
||
).fetchall() |