Hell13Cat
118 lines
4.5 KiB
Python
118 lines
4.5 KiB
Python
from system_module import db, gelbooru, logger
|
|
from flask import Flask, render_template, jsonify, send_from_directory, render_template_string, request
|
|
import json
|
|
import requests
|
|
from pathlib import Path
|
|
import os
|
|
|
|
CONFIG_ROOT = json.load(open("config.json", "r", encoding="utf-8"))
|
|
CONFIG_ROOT_S = json.load(open("config_gb.json", "r", encoding="utf-8"))
|
|
CONFIG_PORT = CONFIG_ROOT["server"]["port"]
|
|
PAGINATION_COUNT = CONFIG_ROOT["server"]["count_page"]
|
|
GB_ID = CONFIG_ROOT_S["gb"]["id"]
|
|
GB_HASH = CONFIG_ROOT_S["gb"]["hash"]
|
|
GB_HEADERS = CONFIG_ROOT["gb"]["headers"]
|
|
FILE_FOLDERS = CONFIG_ROOT["file"]["repos"]
|
|
THUMB_FOLDER = "data_test/thumb"
|
|
|
|
gb = gelbooru.GB(GB_ID, GB_HASH, GB_HEADERS)
|
|
log = logger.Logger()
|
|
database = db.DB(True)
|
|
app = Flask(__name__)
|
|
|
|
def find_file(filename, file_folders):
|
|
for folder in file_folders:
|
|
folder_path = Path(folder)
|
|
if not folder_path.exists():
|
|
continue
|
|
for file_path in folder_path.rglob(filename):
|
|
return folder_path
|
|
return None
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return render_template("index.html")
|
|
|
|
@app.route("/analytics")
|
|
def analytics():
|
|
return render_template("analytics.html")
|
|
|
|
@app.route("/post/<path:post_id>")
|
|
def post(post_id):
|
|
return render_template("post.html")
|
|
|
|
@app.route('/file/<path:filename>')
|
|
def serve_image(filename):
|
|
folder_path = find_file(filename, FILE_FOLDERS)
|
|
return send_from_directory(folder_path, filename)
|
|
|
|
@app.route('/thumb/<path:filename>')
|
|
def serve_thumb(filename):
|
|
return send_from_directory(THUMB_FOLDER, filename)
|
|
|
|
@app.route("/dw_api", methods=['POST'])
|
|
def dw_api():
|
|
data = request.get_json()
|
|
if not data or 'id' not in data:
|
|
return jsonify({'status': 'error', 'message': 'URL not get'}), 400
|
|
image_id = data["id"]
|
|
try:
|
|
data_get, url_file, file_name = gb.get_from_id(image_id)
|
|
if os.path.exists("dw/"+file_name):
|
|
message = f"File '{file_name}' already exists. Skipping."
|
|
log.send("warning", "all", message)
|
|
return jsonify({'status': 'skipped', 'message': message})
|
|
with requests.get(url_file, stream=True, headers=GB_HEADERS, timeout=30) as r:
|
|
r.raise_for_status()
|
|
content_type = r.headers.get('Content-Type', '')
|
|
if 'text/html' in content_type:
|
|
log.send("error", "all", "The server returned HTML instead of an image. Bot protection.")
|
|
return False
|
|
with open("dw/"+file_name, 'wb') as f:
|
|
for chunk in r.iter_content(chunk_size=8192):
|
|
f.write(chunk)
|
|
database.add_raw(data_get)
|
|
return jsonify({'status': 'success'})
|
|
except requests.exceptions.RequestException as e:
|
|
error_message = f"Network error {image_id}: {e}"
|
|
log.send("error", "all", error_message)
|
|
return jsonify({'status': 'error', 'message': error_message}), 500
|
|
except Exception as e:
|
|
error_message = f"Unknown error: {e}"
|
|
log.send("error", "all", error_message)
|
|
return jsonify({'status': 'error', 'message': error_message}), 500
|
|
|
|
@app.route("/api", methods=["GET", "POST"])
|
|
def api():
|
|
data = request.get_json()
|
|
if data["type"] == "search":
|
|
search_query = data["query"]
|
|
search_rating = data["rating"]
|
|
if search_query == "" and search_rating == "e+q+s+g":
|
|
results = database.get_all()
|
|
else:
|
|
search_query = search_query.split(" ")
|
|
search_rating = search_rating.split("+")
|
|
results = database.get_search(search_rating, search_query)
|
|
if len(results) <= PAGINATION_COUNT:
|
|
results_send = {"results_info":{"total":len(results), "page":int(data["page"]), "count_page":PAGINATION_COUNT}, "results":results}
|
|
return jsonify(results_send)
|
|
else:
|
|
results_slile = results[int(data["page"])*PAGINATION_COUNT:(int(data["page"])+1)*PAGINATION_COUNT]
|
|
results_send = {"results_info":{"total":len(results), "page":int(data["page"]), "count_page":PAGINATION_COUNT}, "results":results_slile}
|
|
return jsonify(results_send)
|
|
elif data["type"] == "autocomplete":
|
|
tag_slise = data["q"]
|
|
result_autocomplete = database.get_autocomplete(tag_slise)
|
|
return jsonify(result_autocomplete)
|
|
elif data["type"] == "post":
|
|
post_id = data["post_id"]
|
|
result_post = database.get_from_id(post_id)
|
|
return jsonify(result_post)
|
|
else:
|
|
pass
|
|
return "", 204
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=CONFIG_PORT) |