MrAkells/mario_scraper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: auto update

Browse Source
This commit is contained in:
MrAkells
2025-04-18 10:41:32 +03:00
parent e7e6382a10
commit bb12fe5ca7
9 changed files with 1478 additions and 1045 deletions
Download Patch File Download Diff File Expand all files Collapse all files

406
web_interface.py
View File

@@ -1,5 +1,20 @@
from flask import Flask, render_template, request, jsonify, send_from_directory, redirect, url_for
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user
from flask import (
Flask,
render_template,
request,
jsonify,
send_from_directory,
redirect,
url_for,
)
from flask_login import (
LoginManager,
UserMixin,
login_user,
login_required,
logout_user,
current_user,
)
from werkzeug.security import generate_password_hash, check_password_hash
import re
from euro_scraper import create_retry_session, fetch_products, setup_selenium
@@ -12,12 +27,14 @@ from feed_generator import RobotVacuumYMLGenerator
from pathlib import Path
from werkzeug.exceptions import NotFound
from urllib.parse import urljoin
BASE_URL = "https://mario.mrakells.pp.ua"
from apscheduler.schedulers.background import BackgroundScheduler
import time
from config import BASE_URL
# Добавляем в начало файла
login_manager = LoginManager()
login_manager.login_view = 'login'
login_manager.login_view = "login"
class User(UserMixin):
def __init__(self, id, username, password_hash):
@@ -25,10 +42,10 @@ class User(UserMixin):
self.username = username
self.password_hash = password_hash
# Хранилище пользователей (в реальном приложении использовать базу данных)
users = {
'mario': User(1, 'mario', generate_password_hash('2htC9YlEMXAhNE'))
}
users = {"mario": User(1, "mario", generate_password_hash("2htC9YlEMXAhNE"))}
@login_manager.user_loader
def load_user(user_id):
@@ -37,11 +54,12 @@ def load_user(user_id):
return user
return None
app = Flask(__name__)
# Добавляем после создания app
login_manager.init_app(app)
app.config['SECRET_KEY'] = 'your-secret-key-here' # Замените на случайный ключ
app.config["SECRET_KEY"] = "your-secret-key-here" # Замените на случайный ключ
# Глобальные настройки
app_settings = {"items_limit": -1} # Ограничение количества обрабатываемых товаров
@@ -69,6 +87,7 @@ parsing_status = {
"processed_items": 0,
"error": None,
}
active_translations = set()
def load_categories():
@@ -94,8 +113,7 @@ def extract_category(url: str) -> str:
return category
return None
def start_parsing(category):
# def start_parsing(category):
"""Запуск парсинга категории"""
global parsing_status
@@ -129,6 +147,48 @@ def start_parsing(category):
parsing_status["is_running"] = False
def start_parsing_with_status(category_url: str):
"""Універсальний запуск парсингу з оновленням статусу"""
category = extract_category(category_url)
parsing_status.update(
{
"is_running": True,
"total_items": 0,
"processed_items": 0,
"error": None,
"current_category": category,
}
)
try:
session = create_retry_session()
driver = setup_selenium()
try:
products = fetch_products(category, session, driver, parsing_status)
if products:
output_file = os.path.join("output", f"{category}_products.json")
with open(output_file, "w", encoding="utf-8") as f:
json.dump(products, f, ensure_ascii=False, indent=2)
# ⏩ Після успішного збереження — запустити переклад
threading.Thread(
target=start_translation, args=(f"{category}_products.json",)
).start()
finally:
driver.quit()
except Exception as e:
parsing_status["error"] = str(e)
print(f"❌ Error during parsing {category}: {e}")
finally:
parsing_status["is_running"] = False
def get_file_info(filename, directory="output"):
"""Получение информации о файле"""
filepath = os.path.join(directory, filename)
@@ -139,11 +199,13 @@ def get_file_info(filename, directory="output"):
"size": f"{stat.st_size / 1024:.1f} KB",
}
def get_oldest_parsed_file():
"""Повертає найстаріший _products.json файл"""
folder = "output"
files = [
f for f in os.listdir(folder)
f
for f in os.listdir(folder)
if f.endswith("_products.json") and not f.endswith("_translated_products.json")
]
if not files:
@@ -154,26 +216,28 @@ def get_oldest_parsed_file():
return category
@app.route('/login', methods=['GET', 'POST'])
@app.route("/login", methods=["GET", "POST"])
def login():
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
if request.method == "POST":
username = request.form.get("username")
password = request.form.get("password")
user = users.get(username)
if user and check_password_hash(user.password_hash, password):
login_user(user)
return redirect(url_for('index'))
return render_template('login.html', error='Неверный логин или пароль')
return render_template('login.html')
return redirect(url_for("index"))
@app.route('/logout')
return render_template("login.html", error="Неверный логин или пароль")
return render_template("login.html")
@app.route("/logout")
@login_required
def logout():
logout_user()
return redirect(url_for('login'))
return redirect(url_for("login"))
@app.route("/")
@login_required
@@ -201,6 +265,11 @@ def index():
files = [f for f in os.listdir("output/yml") if f.endswith(".yml")]
yml_files = [get_file_info(f, "output/yml") for f in files]
feed_file_info = None
feed_path = os.path.join("output/yml", "feed.yml")
if os.path.exists(feed_path):
feed_file_info = get_file_info("feed.yml", "output/yml")
# Загружаем категории
categories = load_categories()
@@ -213,25 +282,60 @@ def index():
yml_files=yml_files,
categories=categories,
app_settings=app_settings,
feed_file_info=feed_file_info,
)
@app.route("/manual-refresh-all", methods=["POST"])
def manual_refresh_all():
threading.Thread(target=refresh_all_categories_daily).start()
return jsonify({"message": "Оновлення всіх категорій запущено вручну"})
@app.route("/auto-refresh", methods=["POST"])
def auto_refresh():
"""Запускає парсинг найстарішої збереженої категорії"""
if parsing_status["is_running"]:
return jsonify({"error": "Парсинг уже запущено"})
try:
if parsing_status["is_running"]:
return jsonify({"error": "Парсинг вже запущений"})
category = get_oldest_parsed_file()
if not category:
return jsonify({"error": "Немає жодної категорії для оновлення"})
categories = load_categories()
print(f"[AUTO REFRESH] Повторний парсинг для категорії: {category}")
valid_categories = [c for c in categories if "url" in c and c["url"]]
if not valid_categories:
return jsonify({"error": "Немає категорій з URL"})
thread = threading.Thread(target=start_parsing, args=(category,))
thread.start()
oldest = None
oldest_mtime = float("inf")
return jsonify({"success": True, "category": category})
for cat in valid_categories:
category_url = cat["url"]
category_key = extract_category(category_url)
filename = f"{category_key}_products.json"
path = os.path.join("output", filename)
# ❗ Якщо файл не існує — одразу обираємо цю категорію
if not os.path.exists(path):
oldest = cat
break
mtime = os.path.getmtime(path)
if mtime < oldest_mtime:
oldest_mtime = mtime
oldest = cat
if not oldest:
return jsonify({"error": "Не вдалося знайти категорію для оновлення"})
# category = extract_category(oldest["url"])
thread = threading.Thread(
target=start_parsing_with_status, args=(oldest["url"],)
)
thread.start()
return jsonify({"success": True, "category": oldest["name"]})
except Exception as e:
return jsonify({"error": str(e)})
@app.route("/parse", methods=["POST"])
@@ -250,7 +354,7 @@ def parse():
return jsonify({"error": "Парсинг уже запущен"})
# Запускаем парсинг в отдельном потоке
thread = threading.Thread(target=start_parsing, args=(category,))
thread = threading.Thread(target=start_parsing_with_status, args=(url,))
thread.start()
return jsonify({"status": "ok"})
@@ -258,8 +362,21 @@ def parse():
@app.route("/status")
def get_status():
"""Получение статуса парсинга"""
return jsonify(parsing_status)
# Повертаємо додатково обчислений прогрес
progress = 0
if parsing_status["total_items"]:
progress = round(
(parsing_status["processed_items"] / parsing_status["total_items"]) * 100
)
return jsonify(
{
**parsing_status,
"current_product": parsing_status["processed_items"],
"total_products": parsing_status["total_items"],
"progress": progress,
}
)
@app.route("/download/<path:filename>")
@@ -326,9 +443,16 @@ def get_translation_status():
def start_translation(filename: str):
"""Функция для запуска перевода в отдельнм потоке"""
global translation_status
print(f"[TRANSLATE] Починаємо переклад: {filename}")
if filename in active_translations:
print(f"[SKIP] Переклад вже йде: {filename}")
return
active_translations.add(filename)
translation_status["is_running"] = True
translation_status["processed_items"] = 0
translation_status["error"] = None
@@ -339,6 +463,17 @@ def start_translation(filename: str):
with open(os.path.join("output", filename), "r", encoding="utf-8") as f:
products = json.load(f)
# Витягуємо category_id з назви файлу
category_id = filename.split("_")[0]
categories = load_categories()
category = next((c for c in categories if str(c["id"]) == category_id), None)
# Якщо категорія знайдена — вставляємо ID в продукти
if category and "portal_id" in category:
for product in products:
product["portal_category_id"] = category["portal_id"]
product["local_category_id"] = category["id"]
# Ограничиваем количество товаров только если лимит больше 0
if app_settings["items_limit"] > 0:
products = products[: app_settings["items_limit"]]
@@ -368,6 +503,78 @@ def start_translation(filename: str):
print(f"Ошибка перевода: {e}")
finally:
translation_status["is_running"] = False
active_translations.discard(filename)
print(f"[DONE] Переклад завершено: {filename}")
def refresh_all_categories_daily():
"""Оновлює всі категорії один раз (але не частіше ніж раз на добу кожна)"""
if parsing_status["is_running"]:
print("[SKIP] Парсинг уже запущено. Пропускаємо автоматичне оновлення.")
return
print("[START] Щоденне оновлення категорій...")
categories = load_categories()
for category in categories:
category_name = category.get("name")
category_url = category.get("url")
if not category_url:
print(f"[SKIP] Категорія '{category_name}' не має URL")
continue
category_key = extract_category(category_url)
filename = f"{category_key}_products.json"
filepath = os.path.join("output", filename)
# ⏱ Перевіряємо, коли файл востаннє оновлювався
if os.path.exists(filepath):
modified_time = os.path.getmtime(filepath)
age_seconds = time.time() - modified_time
if age_seconds < 86400:
print(f"[SKIP] {category_name} — оновлено менше доби тому")
continue
print(f"[PARSE] {category_name}")
try:
parsing_status["is_running"] = True
start_parsing_with_status(category_url)
except Exception as e:
print(f"[ERROR] Не вдалося оновити {category_name}: {e}")
finally:
parsing_status["is_running"] = False
time.sleep(5)
print("[DONE] Автоматичне оновлення завершено")
def translate_all_parsed_once():
"""Одноразово запускає переклад для всіх ще не перекладених файлів"""
print("[START] Одноразовий переклад всіх категорій...")
parsed_folder = "output"
translated_folder = "output/translated"
for filename in os.listdir(parsed_folder):
if filename.endswith("_products.json") and not filename.endswith(
"_translated_products.json"
):
parsed_path = os.path.join(parsed_folder, filename)
translated_name = filename.replace(
"_products.json", "_translated_products.json"
)
translated_path = os.path.join(translated_folder, translated_name)
if os.path.exists(translated_path):
print(f"[SKIP] Вже перекладено: {filename}")
continue
print(f"[TRANSLATE] Запускаємо переклад: {filename}")
threading.Thread(target=start_translation, args=(filename,)).start()
time.sleep(2) # трохи затримки між потоками, щоб не навантажувати
print("[DONE] Запуск перекладу завершено.")
@app.route("/update-settings", methods=["POST"])
@@ -386,6 +593,75 @@ def update_settings():
return jsonify({"error": str(e)})
@app.route("/manual-translate-all", methods=["POST"])
@login_required
def manual_translate_all():
threading.Thread(target=translate_all_parsed_once).start()
return jsonify({"message": "Запущено переклад усіх категорій"})
@app.route("/generate-full-yml", methods=["POST"])
def generate_full_yml():
try:
translated_folder = "output/translated"
categories = load_categories()
all_products = []
for file in os.listdir(translated_folder):
if file.endswith("_translated_products.json"):
with open(
os.path.join(translated_folder, file), "r", encoding="utf-8"
) as f:
products = json.load(f)
if not products:
continue
# Перевірка на наявність ID
if not all(
"portal_category_id" in p and "local_category_id" in p
for p in products
):
print(f"[SKIP] {file} — відсутні ID")
continue
all_products.extend(products)
if not all_products:
return jsonify({"error": "Не знайдено товарів з валідними категоріями"})
generator = RobotVacuumYMLGenerator(
base_url=BASE_URL, categories_data=categories
)
# Додаємо тільки ті категорії, що реально використовуються
added_categories = set()
for product in all_products:
cid = str(product.get("local_category_id"))
if cid and cid not in added_categories:
match = next((c for c in categories if str(c["id"]) == cid), None)
if match:
generator.add_category(cid, match["name"])
added_categories.add(cid)
output_path = "output/yml/feed.yml"
os.makedirs("output/yml", exist_ok=True)
success = generator.generate_yml(all_products, output_path)
if success:
return jsonify({"success": True})
else:
return jsonify({"error": "Помилка при генерації повного YML"})
except Exception as e:
return jsonify({"error": str(e)})
@app.route("/yml/feed.yml")
def serve_feed():
return send_from_directory("output/yml", "feed.yml", mimetype="application/xml")
@app.route("/generate-yml", methods=["POST"])
def generate_yml():
"""Обработчик генерации YML файла"""
@@ -401,14 +677,20 @@ def generate_yml():
# Загружаем категории
categories = load_categories()
category = next((c for c in categories if str(c["id"]) == str(category_id)), None)
category = next(
(c for c in categories if str(c["id"]) == str(category_id)), None
)
if not category:
return jsonify({"error": "Категорія не знайдена"})
portal_category_id = category.get("portal_id")
if not portal_category_id:
return jsonify({"error": "Категорія не має portal_id (ідентифікатор категорії Prom.ua)"})
return jsonify(
{
"error": "Категорія не має portal_id (ідентифікатор категорії Prom.ua)"
}
)
os.makedirs("output/yml", exist_ok=True)
@@ -426,7 +708,10 @@ def generate_yml():
product["local_category_id"] = category["id"]
# Создаем генератор YML с указанием базового URL
generator = RobotVacuumYMLGenerator(base_url=BASE_URL)
categories = load_categories()
generator = RobotVacuumYMLGenerator(
base_url=BASE_URL, categories_data=categories
)
generator.add_category(str(category["id"]), category["name"])
# Генерируем имя выходного файла
@@ -453,23 +738,24 @@ def add_category():
data = request.json
categories = load_categories()
# Перевірка обов'язкових полів
if "id" not in data or "name" not in data:
return jsonify({"error": "Обов'язкові поля: id, name"})
# Перевірка наявності name
if "name" not in data:
return jsonify({"error": "Поле name є обов'язковим"})
# Перевірка унікальності ID
if any(str(c["id"]) == str(data["id"]) for c in categories):
return jsonify({"error": "Категорія з таким ID вже існує"})
# Генерація нового ID (максимальний + 1 або 1)
if categories:
new_id = max(int(c["id"]) for c in categories) + 1
else:
new_id = 1
# Додаємо категорію з optional portal_id
new_category = {
"id": data["id"],
"name": data["name"],
}
new_category = {"id": str(new_id), "name": data["name"]}
if "portal_id" in data:
new_category["portal_id"] = data["portal_id"]
if "url" in data:
new_category["url"] = data["url"]
categories.append(new_category)
save_categories(categories)
@@ -479,14 +765,15 @@ def add_category():
return jsonify({"error": str(e)})
@app.route("/delete-category", methods=["POST"])
def delete_category():
"""Удаление категории"""
"""Видалення категорії"""
try:
data = request.json
target_id = str(data["id"])
categories = load_categories()
categories = [c for c in categories if c["id"] != data["id"]]
categories = [c for c in categories if str(c["id"]) != target_id]
save_categories(categories)
return jsonify({"success": True})
@@ -602,12 +889,15 @@ def get_file_info(filename, directory):
if __name__ == "__main__":
# Создаем необходимые директории
for directory in ["output", "output/translated", "output/yml"]:
os.makedirs(directory, exist_ok=True)
# Создаем файл категорий, если его нет
if not os.path.exists(CATEGORIES_FILE):
save_categories([])
app.run(host='0.0.0.0', port=5000, debug=True)
# Запуск планувальника
scheduler = BackgroundScheduler()
scheduler.add_job(refresh_all_categories_daily, "interval", days=1)
scheduler.start()
app.run(host="0.0.0.0", port=5000, debug=True)