Untitled
unknown
plain_text
25 days ago
38 kB
2
Indexable
import os import csv from datetime import datetime, timedelta from flask import render_template, redirect, url_for, flash, request, jsonify, send_file from flask_login import login_user, logout_user, login_required, current_user from werkzeug.utils import secure_filename from sqlalchemy import func, desc from app import app, db from app.database_setup import User, Store, StoreStats, StoreAddition, StoreCategory, StoreGroup, Notification from app.jj_scraper import run_scraper from app.scripts.new_stores_checker import run_checker # Helper functions def allowed_file(filename): """Ελέγχει αν το αρχείο έχει επιτρεπόμενη κατάληξη""" ALLOWED_EXTENSIONS = {'csv'} return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS @app.route('/login', methods=['GET', 'POST']) def login(): """Σελίδα σύνδεσης χρήστη""" if current_user.is_authenticated: next_page = url_for('dashboard') if request.method == 'POST': username = request.form.get('username') password = request.form.get('password') remember = 'remember' in request.form user = User.query.filter_by(username=username).first() if user is None or not user.check_password(password): flash('Λάθος όνομα χρήστη ή κωδικός πρόσβασης', 'danger') return redirect(url_for('login')) login_user(user, remember=remember) next_page = request.args.get('next') if not next_page or next_page.startswith('/'): next_page = url_for('dashboard') return redirect(next_page) return render_template('login.html', title='Σύνδεση') @app.route('/logout') @login_required def logout(): """Αποσύνδεση χρήστη""" logout_user() flash('Έχετε αποσυνδεθεί επιτυχώς!', 'success') return redirect(url_for('login')) @app.route('/stores') @login_required def stores(): """Λίστα καταστημάτων με φίλτρα και αναζήτηση""" page = request.args.get('page', 1, type=int) search = request.args.get('search', '') sort = request.args.get('sort', 'name') order = request.args.get('order', 'asc') lovel_filter = request.args.get('lovel', '') # Προετοιμασία του query query = Store.query # Εφαρμογή αναζήτησης if search: query = query.filter(Store.name.ilike(f'%{search}%') | Store.url.ilike(f'%{search}%')) # Εφαρμογή φίλτρου Lovel if lovel_filter == 'yes': query = query.filter_by(is_lovel_store=True) elif lovel_filter == 'no': query = query.filter_by(is_lovel_store=False) # Εφαρμογή ταξινόμησης if sort == 'name': query = query.order_by(Store.name.asc() if order == 'asc' else Store.name.desc()) elif sort == 'open_date': query = query.order_by(Store.open_date.asc() if order == 'asc' else Store.open_date.desc()) elif sort == 'created_at': query = query.order_by(Store.created_at.asc() if order == 'asc' else Store.created_at.desc()) # Εκτέλεση του query με σελιδοποίηση stores = query.paginate(page=page, per_page=20) # Λήψη όλων των ομάδων Lovel lovel_groups = db.session.query(Store.lovel_group).filter(Store.lovel_group.isnot(None)).distinct().all() lovel_groups = [group[0] for group in lovel_groups if group[0]] return render_template('stores.html', title='Καταστήματα', stores=stores, search=search, sort=sort, order=order, lovel_filter=lovel_filter, lovel_groups=lovel_groups) @app.route('/store/<int:id>') @login_required def store_detail(id): """Προβολή λεπτομερειών καταστήματος""" store = Store.query.get_or_404(id) # Λήψη των στατιστικών του καταστήματος stats = StoreStats.query.filter_by(store_id=id).order_by(StoreStats.scrape_date.desc()).all() # Προετοιμασία δεδομένων για το γράφημα dates = [stat.scrape_date.strftime('%d-%m-%Y') for stat in stats] sales = [stat.sales for stat in stats] products = [stat.products for stat in stats] return render_template('store_detail.html', title=f'Κατάστημα: {store.name}', store=store, stats=stats, chart_dates=dates, chart_sales=sales, chart_products=products) @app.route('/store/add', methods=['GET', 'POST']) @login_required def add_store(): """Προσθήκη νέου καταστήματος""" if request.method == 'POST': name = request.form.get('name') url = request.form.get('url') open_date = request.form.get('open_date') is_lovel_store = 'is_lovel_store' in request.form lovel_group = request.form.get('lovel_group') if is_lovel_store else None # Έλεγχοι εγκυρότητας if not name or not url: flash('Το όνομα και το URL είναι υποχρεωτικά πεδία!', 'danger') return redirect(url_for('add_store')) if Store.query.filter_by(url=url).first(): flash('Υπάρχει ήδη κατάστημα με αυτό το URL!', 'danger') return redirect(url_for('add_store')) # Δημιουργία νέου καταστήματος store = Store( name=name, url=url, open_date=open_date, is_lovel_store=is_lovel_store, lovel_group=lovel_group ) db.session.add(store) db.session.flush() # Για να λάβουμε το ID # Καταγραφή της προσθήκης store_addition = StoreAddition( store_id=store.id, addition_date=datetime.now().date(), source='manual' ) db.session.add(store_addition) db.session.commit() flash(f'Το κατάστημα {name} προστέθηκε επιτυχώς!', 'success') return redirect(url_for('stores')) # Λήψη των υπαρχουσών ομάδων Lovel για το dropdown lovel_groups = db.session.query(Store.lovel_group).filter(Store.lovel_group.isnot(None)).distinct().all() lovel_groups = [group[0] for group in lovel_groups if group[0]] return render_template('store_form.html', title='Προσθήκη Καταστήματος', lovel_groups=lovel_groups) @app.route('/store/edit/<int:id>', methods=['GET', 'POST']) @login_required def edit_store(id): """Επεξεργασία καταστήματος""" store = Store.query.get_or_404(id) if request.method == 'POST': name = request.form.get('name') url = request.form.get('url') open_date = request.form.get('open_date') is_lovel_store = 'is_lovel_store' in request.form lovel_group = request.form.get('lovel_group') if is_lovel_store else None is_active = 'is_active' in request.form # Έλεγχοι εγκυρότητας if not name or not url: flash('Το όνομα και το URL είναι υποχρεωτικά πεδία!', 'danger') return redirect(url_for('edit_store', id=id)) # Έλεγχος αν το URL ανήκει σε άλλο κατάστημα existing_store = Store.query.filter_by(url=url).first() if existing_store and existing_store.id != id: flash('Υπάρχει ήδη κατάστημα με αυτό το URL!', 'danger') return redirect(url_for('edit_store', id=id)) # Ενημέρωση καταστήματος store.name = name store.url = url store.open_date = open_date store.is_lovel_store = is_lovel_store store.lovel_group = lovel_group store.is_active = is_active db.session.commit() flash(f'Το κατάστημα {name} ενημερώθηκε επιτυχώς!', 'success') return redirect(url_for('store_detail', id=id)) # Λήψη των υπαρχουσών ομάδων Lovel για το dropdown lovel_groups = db.session.query(Store.lovel_group).filter(Store.lovel_group.isnot(None)).distinct().all() lovel_groups = [group[0] for group in lovel_groups if group[0]] return render_template('store_form.html', title='Επεξεργασία Καταστήματος', store=store, lovel_groups=lovel_groups, edit=True) @app.route('/store/delete/<int:id>', methods=['POST']) @login_required def delete_store(id): """Διαγραφή καταστήματος""" store = Store.query.get_or_404(id) store_name = store.name db.session.delete(store) db.session.commit() flash(f'Το κατάστημα {store_name} διαγράφηκε επιτυχώς!', 'success') return redirect(url_for('stores')) @app.route('/stores/bulk-edit', methods=['POST']) @login_required def bulk_edit_stores(): """Μαζική επεξεργασία καταστημάτων""" # Λήψη λίστας ID καταστημάτων store_ids = request.form.getlist('store_ids') # Λήψη τιμών από τη φόρμα is_lovel_store = 'is_lovel_store' in request.form lovel_group = request.form.get('lovel_group', '').strip() if is_lovel_store else None is_active = 'is_active' in request.form if not store_ids: flash('Δεν επιλέχθηκαν καταστήματα', 'warning') return redirect(url_for('stores')) # Ενημέρωση των επιλεγμένων καταστημάτων update_values = {} if 'is_lovel_store' in request.form: update_values['is_lovel_store'] = is_lovel_store update_values['lovel_group'] = lovel_group if 'is_active' in request.form: update_values['is_active'] = is_active if update_values: updated_count = Store.query.filter(Store.id.in_(store_ids)).update( update_values, synchronize_session=False ) db.session.commit() flash(f'Ενημερώθηκαν {updated_count} καταστήματα', 'success') else: flash('Δεν επιλέχθηκαν αλλαγές για εφαρμογή', 'warning') return redirect(url_for('stores')) @app.route('/statistics') @login_required def statistics(): """Προβολή στατιστικών""" # Παράμετροι φιλτραρίσματος start_date_str = request.args.get('start_date') end_date_str = request.args.get('end_date') store_id = request.args.get('store_id', type=int) sort = request.args.get('sort', 'date') order = request.args.get('order', 'desc') lovel_only = 'lovel_only' in request.args # Προεπιλεγμένες ημερομηνίες (τελευταίες 7 ημέρες) today = datetime.now().date() default_start = today - timedelta(days=7) default_end = today # Μετατροπή ημερομηνιών try: start_date = datetime.strptime(start_date_str, '%Y-%m-%d').date() if start_date_str else default_start end_date = datetime.strptime(end_date_str, '%Y-%m-%d').date() if end_date_str else default_end except ValueError: start_date = default_start end_date = default_end # Προετοιμασία του query για τα στατιστικά query = db.session.query( StoreStats.id, Store.name.label('store_name'), Store.id.label('store_id'), StoreStats.sales, StoreStats.products, StoreStats.scrape_date ).join(Store) # Εφαρμογή φίλτρων query = query.filter(StoreStats.scrape_date.between(start_date, end_date)) if store_id: query = query.filter(Store.id == store_id) if lovel_only: query = query.filter(Store.is_lovel_store == True) # Εφαρμογή ταξινόμησης if sort == 'store': query = query.order_by(Store.name.asc() if order == 'asc' else Store.name.desc()) elif sort == 'sales': query = query.order_by(StoreStats.sales.asc() if order == 'asc' else StoreStats.sales.desc()) elif sort == 'products': query = query.order_by(StoreStats.products.asc() if order == 'asc' else StoreStats.products.desc()) else: # date query = query.order_by(StoreStats.scrape_date.asc() if order == 'asc' else StoreStats.scrape_date.desc()) # Εκτέλεση του query stats_data = query.all() # Υπολογισμός συνολικών στοιχείων totals = { 'sales': sum(row.sales for row in stats_data), 'products': sum(row.products for row in stats_data) } # Λίστα καταστημάτων για το dropdown stores = Store.query.order_by(Store.name).all() return render_template('statistics.html', title='Στατιστικά', stats=stats_data, stores=stores, selected_store_id=store_id, start_date=start_date, end_date=end_date, sort=sort, order=order, lovel_only=lovel_only, totals=totals) @app.route('/lovel-report') @login_required def lovel_stores(): """Έκθεση για τα καταστήματα Lovel""" # Λίστα Lovel καταστημάτων lovel_stores = Store.query.filter_by(is_lovel_store=True).order_by(Store.name).all() # Ομαδοποίηση ανά Lovel group lovel_groups = {} for store in lovel_stores: group = store.lovel_group or "Άλλα" if group not in lovel_groups: lovel_groups[group] = [] lovel_groups[group].append(store) # Συγκεντρωτικά στατιστικά today = datetime.now().date() # Συνολικά στατιστικά Lovel καταστημάτων lovel_stats = db.session.query( func.sum(StoreStats.sales).label('total_sales'), func.sum(StoreStats.products).label('total_products') ).join( Store, StoreStats.store_id == Store.id ).filter( Store.is_lovel_store == True, StoreStats.scrape_date == today ).first() lovel_sales = lovel_stats.total_sales if lovel_stats and lovel_stats.total_sales else 0 lovel_products = lovel_stats.total_products if lovel_stats and lovel_stats.total_products else 0 # Στατιστικά ανά ομάδα group_stats = {} for group, stores in lovel_groups.items(): store_ids = [store.id for store in stores] group_data = db.session.query( func.sum(StoreStats.sales).label('total_sales'), func.sum(StoreStats.products).label('total_products') ).filter( StoreStats.store_id.in_(store_ids), StoreStats.scrape_date == today ).first() group_stats[group] = { 'stores': len(stores), 'sales': group_data.total_sales if group_data and group_data.total_sales else 0, 'products': group_data.total_products if group_data and group_data.total_products else 0 } return render_template('lovel_report.html', title='Έκθεση Lovel', lovel_stores=lovel_stores, lovel_groups=lovel_groups, lovel_sales=lovel_sales, lovel_products=lovel_products, group_stats=group_stats) @app.route('/new-stores') @login_required def new_stores(): """Προβολή νέων καταστημάτων""" # Παράμετροι days = request.args.get('days', 30, type=int) # Υπολογισμός ημερομηνίας έναρξης today = datetime.now().date() start_date = today - timedelta(days=days) # Εύρεση νέων καταστημάτων new_stores_data = db.session.query( Store, StoreAddition ).join( StoreAddition, Store.id == StoreAddition.store_id ).filter( StoreAddition.addition_date >= start_date ).order_by( StoreAddition.addition_date.desc() ).all() # Ομαδοποίηση ανά ημέρα grouped_by_date = {} for store, addition in new_stores_data: date_str = addition.addition_date.strftime('%Y-%m-%d') if date_str not in grouped_by_date: grouped_by_date[date_str] = [] grouped_by_date[date_str].append((store, addition)) # Στατιστικά νέων καταστημάτων total_new_stores = len(new_stores_data) stores_by_source = {} for _, addition in new_stores_data: source = addition.source if source not in stores_by_source: stores_by_source[source] = 0 stores_by_source[source] += 1 return render_template('new_stores.html', title='Νέα Καταστήματα', grouped_by_date=grouped_by_date, total_new_stores=total_new_stores, stores_by_source=stores_by_source, days=days) @app.route('/upload', methods=['GET', 'POST']) @login_required def upload_file(): """Ανέβασμα αρχείου CSV""" if request.method == 'POST': # Έλεγχος αν υπάρχει αρχείο στο request if 'file' not in request.files: flash('Δεν υπάρχει αρχείο', 'danger') return redirect(request.url) file = request.files['file'] # Έλεγχος αν δεν επιλέχθηκε αρχείο if file.filename == '': flash('Δεν επιλέχθηκε αρχείο', 'danger') return redirect(request.url) # Έλεγχος αν είναι επιτρεπόμενο αρχείο if file and allowed_file(file.filename): filename = secure_filename(file.filename) file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename) file.save(file_path) try: # Επεξεργασία αρχείου added, updated = process_csv_file(file_path) flash(f'Το αρχείο ανέβηκε και επεξεργάστηκε επιτυχώς! Προστέθηκαν {added} και ενημερώθηκαν {updated} καταστήματα.', 'success') except Exception as e: flash(f'Σφάλμα κατά την επεξεργασία του αρχείου: {str(e)}', 'danger') return redirect(url_for('stores')) flash('Μη επιτρεπόμενο αρχείο. Επιτρέπονται μόνο αρχεία CSV.', 'danger') return redirect(request.url) return render_template('upload.html', title='Ανέβασμα Αρχείου') def process_csv_file(file_path): """Επεξεργασία αρχείου CSV""" added_count = 0 updated_count = 0 with open(file_path, 'r', encoding='utf-8-sig') as csvfile: reader = csv.DictReader(csvfile) for row in reader: name = row.get('Store', '').strip() url = row.get('URL', '').strip() open_date = row.get('Open Date', '').strip() # Προαιρετικά πεδία sales = int(row.get('Sales', 0)) if row.get('Sales', '').strip() else 0 products = int(row.get('Products', 0)) if row.get('Products', '').strip() else 0 is_lovel_store = row.get('Lovel Store', '').strip().lower() in ['ναι', 'yes', 'true', '1'] lovel_group = row.get('Lovel Group', '').strip() if is_lovel_store else None if not name or not url: continue # Έλεγχος αν υπάρχει ήδη το κατάστημα store = Store.query.filter_by(url=url).first() if store: # Ενημέρωση υπάρχοντος καταστήματος store.name = name store.open_date = open_date store.is_lovel_store = is_lovel_store store.lovel_group = lovel_group updated_count += 1 else: # Δημιουργία νέου καταστήματος store = Store( name=name, url=url, open_date=open_date, is_lovel_store=is_lovel_store, lovel_group=lovel_group ) db.session.add(store) db.session.flush() # Για να πάρουμε το ID # Καταγραφή της προσθήκης store_addition = StoreAddition( store_id=store.id, addition_date=datetime.now().date(), source='import' ) db.session.add(store_addition) added_count += 1 # Αν έχουμε πληροφορίες για πωλήσεις και προϊόντα, τα αποθηκεύουμε if sales > 0 or products > 0: # Έλεγχος αν υπάρχουν ήδη στατιστικά για σήμερα today = datetime.now().date() existing_stats = StoreStats.query.filter_by( store_id=store.id, scrape_date=today ).first() if existing_stats: # Ενημέρωση υπαρχόντων στατιστικών existing_stats.sales = sales existing_stats.products = products existing_stats.last_update = datetime.now() else: # Δημιουργία νέων στατιστικών stats = StoreStats( store_id=store.id, sales=sales, products=products, scrape_date=today, last_update=datetime.now() ) db.session.add(stats) # Αποθήκευση όλων των αλλαγών db.session.commit() return added_count, updated_count @app.route('/run-scraper', methods=['POST']) @login_required def run_scraper(): """Εκτέλεση του scraper πωλήσεων/προϊόντων χειροκίνητα""" try: from app.jj_scraper import run_scraper as execute_scraper result = execute_scraper() flash(f'Το scraping ολοκληρώθηκε επιτυχώς! {result}', 'success') except Exception as e: flash(f'Σφάλμα κατά την εκτέλεση του scraper: {str(e)}', 'danger') return redirect(url_for('dashboard')) @app.route('/run-store-checker', methods=['POST']) @login_required def run_store_checker(): """Εκτέλεση του ελέγχου νέων καταστημάτων χειροκίνητα""" try: from app.scripts.new_stores_checker import run_checker result = run_checker() flash(f'Ο έλεγχος νέων καταστημάτων ολοκληρώθηκε επιτυχώς! {result}', 'success') except Exception as e: flash(f'Σφάλμα κατά την εκτέλεση του ελέγχου νέων καταστημάτων: {str(e)}', 'danger') return redirect(url_for('dashboard')) @app.route('/') @app.route('/index') @login_required def index(): # Στατιστικά για το dashboard today = datetime.now().date() yesterday = today - timedelta(days=1) # Σύνολο καταστημάτων total_stores = Store.query.count() # Σιγουρευτείτε ότι έχετε αντικείμενα για τα στατιστικά today_stats = { 'total_sales': 0, 'total_products': 0 } yesterday_stats = { 'total_sales': 0, 'total_products': 0 } # Στατιστικά τελευταίων ημερών today_result = db.session.query( func.sum(StoreStats.sales).label('total_sales'), func.sum(StoreStats.products).label('total_products') ).filter(StoreStats.scrape_date == today).first() yesterday_result = db.session.query( func.sum(StoreStats.sales).label('total_sales'), func.sum(StoreStats.products).label('total_products') ).filter(StoreStats.scrape_date == yesterday).first() # Ενημέρωση στατιστικών αν έχουμε δεδομένα if today_result and today_result.total_sales is not None: today_stats['total_sales'] = today_result.total_sales if today_result and today_result.total_products is not None: today_stats['total_products'] = today_result.total_products if yesterday_result and yesterday_result.total_sales is not None: yesterday_stats['total_sales'] = yesterday_result.total_sales if yesterday_result and yesterday_result.total_products is not None: yesterday_stats['total_products'] = yesterday_result.total_products # Πρόσφατα προστεθέντα καταστήματα recent_stores = Store.query.order_by(Store.created_at.desc()).limit(5).all() # Καταστήματα με τις περισσότερες πωλήσεις top_selling_stores = [] try: top_selling_stores = db.session.query( Store.name, Store.id, func.max(StoreStats.sales).label('max_sales') ).join(StoreStats).group_by(Store.id, Store.name).order_by(desc('max_sales')).limit(5).all() except Exception as e: app.logger.error(f"Σφάλμα κατά την ανάκτηση των κορυφαίων καταστημάτων: {str(e)}") # Νέα καταστήματα που προστέθηκαν πρόσφατα new_stores_count = StoreAddition.query.filter( StoreAddition.addition_date >= today - timedelta(days=7) ).count() # Πληροφορίες για τα τελευταία scraping last_scraping = StoreStats.query.order_by(StoreStats.created_at.desc()).first() last_scraping_time = last_scraping.created_at if last_scraping else None last_store_check = StoreAddition.query.order_by(StoreAddition.addition_date.desc()).first() last_store_check_time = last_store_check.addition_date if last_store_check else None # Υπολογισμός επόμενων προγραμματισμένων εκτελέσεων next_scraping_time = datetime.now().replace(hour=23, minute=1, second=0) if next_scraping_time < datetime.now(): next_scraping_time += timedelta(days=1) next_stores_check_time = datetime.now().replace(hour=22, minute=45, second=0) if next_stores_check_time < datetime.now(): next_stores_check_time += timedelta(days=1) return render_template('index.html', title='Dashboard', total_stores=total_stores, today_stats=today_stats, yesterday_stats=yesterday_stats, recent_stores=recent_stores, top_selling_stores=top_selling_stores, now=datetime.now(), new_stores_count=new_stores_count, last_scraping_time=last_scraping_time, last_store_check_time=last_store_check_time, next_scraping_time=next_scraping_time, next_stores_check_time=next_stores_check_time) import os import sys import subprocess from datetime import datetime, timedelta from flask import render_template, flash, redirect, url_for, request, jsonify from app import app, db from app.database_setup import Store, StoreStats, StoreAddition @app.route('/') @app.route('/index') @app.route('/dashboard') @login_required def index(): today = datetime.now().date() yesterday = today - timedelta(days=1) # Βασικά στατιστικά total_stores = Store.query.count() # Υπολογισμός Λοβελ καταστημάτων lovel_stores = Store.query.filter_by(is_lovel_store=True).count() # Σημερινά στατιστικά today_stats = db.session.query( func.sum(StoreStats.sales).label('total_sales'), func.sum(StoreStats.products).label('total_products') ).filter(StoreStats.scrape_date == today).first() today_sales = today_stats.total_sales if today_stats and today_stats.total_sales else 0 today_products = today_stats.total_products if today_stats and today_stats.total_products else 0 # Πρόσφατα νέα καταστήματα recent_new_stores = db.session.query(Store, StoreAddition)\ .join(StoreAddition, Store.id == StoreAddition.store_id)\ .order_by(StoreAddition.addition_date.desc())\ .limit(5).all() # Μετατροπή σε λίστα από καταστήματα με επιπλέον πληροφορίες recent_new_stores_data = [] for store, addition in recent_new_stores: store_data = { 'id': store.id, 'name': store.name, 'url': store.url, 'open_date': store.open_date, 'is_lovel_store': store.is_lovel_store, 'added_date': addition.addition_date.strftime('%d/%m/%Y') } recent_new_stores_data.append(store_data) # Top καταστήματα με βάση τις πωλήσεις top_stores = db.session.query( Store, func.max(StoreStats.sales).label('max_sales'), func.max(StoreStats.products).label('max_products') ).join(StoreStats).group_by(Store.id).order_by(db.desc('max_sales')).limit(5).all() top_stores_data = [] for store, sales, products in top_stores: store_data = { 'id': store.id, 'name': store.name, 'is_lovel_store': store.is_lovel_store, 'sales': sales, 'products': products } top_stores_data.append(store_data) # Προγραμματισμένες εργασίες now = datetime.now() # Επόμενη εκτέλεση ελέγχου νέων καταστημάτων (22:45) next_new_stores_check = now.replace(hour=22, minute=45, second=0) if next_new_stores_check < now: next_new_stores_check += timedelta(days=1) # Επόμενη εκτέλεση scraping πωλήσεων/προϊόντων (23:01) next_sales_scrape = now.replace(hour=23, minute=1, second=0) if next_sales_scrape < now: next_sales_scrape += timedelta(days=1) # Πρόσφατη δραστηριότητα (παράδειγμα) activity_log = [ { 'icon': 'sync-alt', 'level': 'primary', 'message': 'Ολοκληρώθηκε το scraping πωλήσεων/προϊόντων', 'timestamp': (now - timedelta(hours=2)).strftime('%d/%m/%Y %H:%M') }, { 'icon': 'plus-circle', 'level': 'success', 'message': 'Προστέθηκαν 3 νέα καταστήματα', 'timestamp': (now - timedelta(days=1)).strftime('%d/%m/%Y %H:%M') } ] # Για AJAX αιτήματα if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return jsonify({ 'success': True, 'data': { 'total_stores': total_stores, 'lovel_stores': lovel_stores, 'today_sales': today_sales, 'today_products': today_products } }) return render_template('dashboard.html', title='Dashboard', total_stores=total_stores, lovel_stores=lovel_stores, today_sales=today_sales, today_products=today_products, recent_new_stores=recent_new_stores_data, top_stores=top_stores_data, next_new_stores_check=next_new_stores_check.strftime('%d/%m/%Y %H:%M'), next_sales_scrape=next_sales_scrape.strftime('%d/%m/%Y %H:%M'), activity_log=activity_log) @app.route('/run_scraper', methods=['POST']) @login_required def run_scraper(): """Εκτελεί το scraper χειροκίνητα""" try: # Διαδρομή προς το jj_scraper.py scraper_path = os.path.join(app.root_path, 'jj_scraper.py') # Εκτέλεση του scraper ως υποδιεργασία result = subprocess.run( [sys.executable, scraper_path], capture_output=True, text=True, timeout=300 # 5 λεπτά timeout ) if result.returncode == 0: flash(f'Το scraping ολοκληρώθηκε επιτυχώς!', 'success') app.logger.info(f'Επιτυχής εκτέλεση του scraper. Output: {result.stdout}') else: flash(f'Προέκυψε σφάλμα κατά την εκτέλεση του scraper: {result.stderr}', 'danger') app.logger.error(f'Σφάλμα κατά την εκτέλεση του scraper: {result.stderr}') except subprocess.TimeoutExpired: flash('Το scraping διήρκεσε πολύ και τερματίστηκε. Δοκιμάστε ξανά αργότερα.', 'warning') app.logger.warning('Timeout κατά την εκτέλεση του scraper') except Exception as e: flash(f'Σφάλμα κατά την εκτέλεση του scraper: {str(e)}', 'danger') app.logger.error(f'Εξαίρεση κατά την εκτέλεση του scraper: {str(e)}') return redirect(url_for('index')) @app.route('/run_stores_check', methods=['POST']) @login_required def run_stores_check(): """Εκτελεί τον έλεγχο για νέα καταστήματα χειροκίνητα""" try: # Διαδρομή προς το new_stores_checker.py checker_path = os.path.join(app.root_path, 'scripts', 'new_stores_checker.py') # Εκτέλεση του checker ως υποδιεργασία result = subprocess.run( [sys.executable, checker_path], capture_output=True, text=True, timeout=300 # 5 λεπτά timeout ) if result.returncode == 0: flash(f'Ο έλεγχος για νέα καταστήματα ολοκληρώθηκε επιτυχώς!', 'success') app.logger.info(f'Επιτυχής εκτέλεση του new_stores_checker. Output: {result.stdout}') else: flash(f'Προέκυψε σφάλμα κατά τον έλεγχο για νέα καταστήματα: {result.stderr}', 'danger') app.logger.error(f'Σφάλμα κατά την εκτέλεση του new_stores_checker: {result.stderr}') except subprocess.TimeoutExpired: flash('Ο έλεγχος διήρκεσε πολύ και τερματίστηκε. Δοκιμάστε ξανά αργότερα.', 'warning') app.logger.warning('Timeout κατά την εκτέλεση του new_stores_checker') except Exception as e: flash(f'Σφάλμα κατά τον έλεγχο για νέα καταστήματα: {str(e)}', 'danger') app.logger.error(f'Εξαίρεση κατά την εκτέλεση του new_stores_checker: {str(e)}') return redirect(url_for('index'))
Editor is loading...
Leave a Comment