tm216ux/zkt25
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Aktualizovat sk1/backend/app.py

Browse Source
This commit is contained in:
Tetiana Mohorian 2025-04-30 09:54:15 +00:00
parent 71e32912cd
commit 55dc649360
1 changed files with 170 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

342
sk1/backend/app.py
View File

@ -1,172 +1,170 @@
from flask import Flask, request, jsonify, Response, send_from_directory
from flask import Flask, request, jsonify, Response, send_from_directory from flask_cors import CORS
from flask_cors import CORS from flask_caching import Cache
from flask_caching import Cache import json
import json import os
import os import hashlib
import hashlib import re
import re import psycopg2
import psycopg2 from datetime import datetime
from datetime import datetime import pytz
import pytz import torch
import torch from transformers import AutoModelForSequenceClassification, AutoTokenizer
from transformers import AutoModelForSequenceClassification, AutoTokenizer
app = Flask(__name__)
app = Flask(__name__) CORS(app)
CORS(app) app.config['CACHE_TYPE'] = 'SimpleCache'
app.config['CACHE_TYPE'] = 'SimpleCache' cache = Cache(app)
cache = Cache(app)
# Database connection
# Database connection DATABASE_URL = os.environ.get('DATABASE_URL')
DATABASE_URL = os.environ.get('DATABASE_URL') conn = psycopg2.connect(DATABASE_URL)
conn = psycopg2.connect(DATABASE_URL) cursor = conn.cursor()
cursor = conn.cursor()
# Initialize DB
# Initialize DB cursor.execute('''
cursor.execute(''' CREATE TABLE IF NOT EXISTS history (
CREATE TABLE IF NOT EXISTS history ( id SERIAL PRIMARY KEY,
id SERIAL PRIMARY KEY, text TEXT NOT NULL,
text TEXT NOT NULL, prediction TEXT NOT NULL,
prediction TEXT NOT NULL, timestamp TIMESTAMP NOT NULL
timestamp TIMESTAMP NOT NULL )
) ''')
''') conn.commit()
conn.commit()
model_path = "tetianamohorian/hate_speech_model"
model_path = "tetianamohorian/hate_speech_model" tokenizer = AutoTokenizer.from_pretrained(model_path)
tokenizer = AutoTokenizer.from_pretrained(model_path) model = AutoModelForSequenceClassification.from_pretrained(model_path)
model = AutoModelForSequenceClassification.from_pretrained(model_path) model.eval()
model.eval()
HISTORY_FILE = "history.json"
HISTORY_FILE = "history.json"
def generate_text_hash(text):
def generate_text_hash(text): return hashlib.md5(text.encode('utf-8')).hexdigest()
return hashlib.md5(text.encode('utf-8')).hexdigest()
def get_current_time():
def get_current_time(): tz = pytz.timezone('Europe/Bratislava')
tz = pytz.timezone('Europe/Bratislava') return datetime.now(tz)
return datetime.now(tz)
def sync_history_file():
def sync_history_file(): # Synchronizácia history.json so všetkými záznamami z databázy
# Synchronizácia history.json so všetkými záznamami z databázy cursor.execute("SELECT text, prediction, timestamp FROM history ORDER BY timestamp DESC")
cursor.execute("SELECT text, prediction, timestamp FROM history ORDER BY timestamp DESC") rows = cursor.fetchall()
rows = cursor.fetchall() history = [
history = [ {
{ "text": r[0],
"text": r[0], "prediction": r[1],
"prediction": r[1], "timestamp": r[2].strftime("%d.%m.%Y %H:%M:%S")
"timestamp": r[2].strftime("%d.%m.%Y %H:%M:%S") } for r in rows
} for r in rows ]
] with open(HISTORY_FILE, "w", encoding="utf-8") as f:
with open(HISTORY_FILE, "w", encoding="utf-8") as f: json.dump(history, f, ensure_ascii=False, indent=2)
json.dump(history, f, ensure_ascii=False, indent=2)
def save_to_history(text, prediction_label):
def save_to_history(text, prediction_label): timestamp = get_current_time()
timestamp = get_current_time() try:
try: cursor.execute(
# Проверка на дубликат "SELECT 1 FROM history WHERE text = %s AND prediction = %s",
cursor.execute( (text, prediction_label)
"SELECT 1 FROM history WHERE text = %s AND prediction = %s", )
(text, prediction_label) if cursor.fetchone() is None:
) cursor.execute(
if cursor.fetchone() is None: "INSERT INTO history (text, prediction, timestamp) VALUES (%s, %s, %s)",
cursor.execute( (text, prediction_label, timestamp)
"INSERT INTO history (text, prediction, timestamp) VALUES (%s, %s, %s)", )
(text, prediction_label, timestamp) conn.commit()
) sync_history_file()
conn.commit() except Exception as e:
sync_history_file() print("Nepodarilo sa uložiť do databázy alebo prepísať history.json:", e)
except Exception as e:
print("Nepodarilo sa uložiť do databázy alebo prepísať history.json:", e) @app.route("/")
def serve_frontend():
@app.route("/") return send_from_directory("static", "index.html")
def serve_frontend():
return send_from_directory("static", "index.html") @app.route("/<path:path>")
def serve_static_files(path):
@app.route("/<path:path>") return send_from_directory("static", path)
def serve_static_files(path):
return send_from_directory("static", path) @app.route("/api/predict", methods=["POST"])
def predict():
@app.route("/api/predict", methods=["POST"]) try:
def predict(): data = request.json
try: text = data.get("text", "")
data = request.json
text = data.get("text", "") if not text:
return jsonify({"error": "Text nesmie byť prázdny."}), 400
if not text: if len(text) > 512:
return jsonify({"error": "Text nesmie byť prázdny."}), 400 return jsonify({"error": "Text je príliš dlhý. Maximálne 512 znakov."}), 400
if len(text) > 512: if re.search(r"[а-яА-ЯёЁ]", text):
return jsonify({"error": "Text je príliš dlhý. Maximálne 512 znakov."}), 400 return jsonify({"error": "Text nesmie obsahovať azbuku (cyriliku)."}), 400
if re.search(r"[а-яА-ЯёЁ]", text):
return jsonify({"error": "Text nesmie obsahovať azbuku (cyriliku)."}), 400 text_hash = generate_text_hash(text)
cached_result = cache.get(text_hash)
text_hash = generate_text_hash(text) if cached_result:
cached_result = cache.get(text_hash) save_to_history(text, cached_result)
if cached_result: return jsonify({"prediction": cached_result}), 200
save_to_history(text, cached_result)
return jsonify({"prediction": cached_result}), 200 inputs = tokenizer(text, return_tensors="pt", truncation=True, padding=True)
inputs = tokenizer(text, return_tensors="pt", truncation=True, padding=True) with torch.no_grad():
outputs = model(**inputs)
with torch.no_grad(): predictions = torch.argmax(outputs.logits, dim=1).item()
outputs = model(**inputs)
predictions = torch.argmax(outputs.logits, dim=1).item() prediction_label = "Pravdepodobne toxický" if predictions == 1 else "Neutrálny text"
cache.set(text_hash, prediction_label)
prediction_label = "Pravdepodobne toxický" if predictions == 1 else "Neutrálny text"
cache.set(text_hash, prediction_label) save_to_history(text, prediction_label)
save_to_history(text, prediction_label) return jsonify({"prediction": prediction_label}), 200
return jsonify({"prediction": prediction_label}), 200 except Exception as e:
return jsonify({"error": str(e)}), 500
except Exception as e:
return jsonify({"error": str(e)}), 500 @app.route("/api/history", methods=["GET"])
def get_history():
@app.route("/api/history", methods=["GET"]) try:
def get_history(): with open(HISTORY_FILE, "r", encoding="utf-8") as f:
try: content = json.load(f)
with open(HISTORY_FILE, "r", encoding="utf-8") as f: return Response(
content = json.load(f) json.dumps(content, ensure_ascii=False, indent=2),
return Response( mimetype="application/json"
json.dumps(content, ensure_ascii=False, indent=2), )
mimetype="application/json" except Exception as e:
) return jsonify({"error": str(e)}), 500
except Exception as e:
return jsonify({"error": str(e)}), 500 @app.route("/api/history/reset", methods=["POST"])
def reset_history():
@app.route("/api/history/reset", methods=["POST"]) try:
def reset_history(): cursor.execute("DELETE FROM history")
try: conn.commit()
cursor.execute("DELETE FROM history") with open(HISTORY_FILE, "w", encoding="utf-8") as f:
conn.commit() json.dump([], f, ensure_ascii=False, indent=2)
with open(HISTORY_FILE, "w", encoding="utf-8") as f: return jsonify({"message": "História bola úspešne vymazaná."}), 200
json.dump([], f, ensure_ascii=False, indent=2) except Exception as e:
return jsonify({"message": "História bola úspešne vymazaná."}), 200 return jsonify({"error": str(e)}), 500
except Exception as e:
return jsonify({"error": str(e)}), 500 @app.route("/api/history/db", methods=["GET"])
def get_history_from_db():
@app.route("/api/history/db", methods=["GET"]) try:
def get_history_from_db(): cursor.execute("SELECT text, prediction, timestamp FROM history ORDER BY timestamp DESC")
try: rows = cursor.fetchall()
cursor.execute("SELECT text, prediction, timestamp FROM history ORDER BY timestamp DESC") history = [
rows = cursor.fetchall() {
history = [ "text": r[0],
{ "prediction": r[1],
"text": r[0], "timestamp": r[2].strftime("%d.%m.%Y %H:%M:%S")
"prediction": r[1], } for r in rows
"timestamp": r[2].strftime("%d.%m.%Y %H:%M:%S") ]
} for r in rows return Response(
] json.dumps(history, ensure_ascii=False, indent=2),
return Response( mimetype="application/json"
json.dumps(history, ensure_ascii=False, indent=2), )
mimetype="application/json" except Exception as e:
) return jsonify({"error": str(e)}), 500
except Exception as e:
return jsonify({"error": str(e)}), 500 print("✅ Flask is starting...")
sync_history_file()
print("✅ Flask is starting...")
sync_history_file() if __name__ == "__main__":
port = int(os.environ.get("PORT", 8080))
if __name__ == "__main__": app.run(host="0.0.0.0", port=port)
port = int(os.environ.get("PORT", 8080))
app.run(host="0.0.0.0", port=port)