Untitled
unknown
plain_text
a month ago
23 kB
2
Indexable
from datetime import datetime, timedelta import os import csv from flask import render_template, flash, redirect, url_for, request, jsonify, abort, 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 import pandas as pd from app import app, db from app.models import User, Store, StoreStats, StoreAddition from app.forms import LoginForm, UploadForm, StoreForm, DateRangeForm from app.utils import allowed_file, parse_date, format_date @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, func.max(StoreStats.sales).label('max_sales') ).join(StoreStats).group_by(Store.id).order_by(desc('max_sales')).limit(5).all() except: pass 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()) @app.route('/login', methods=['GET', 'POST']) def login(): if current_user.is_authenticated: return redirect(url_for('index')) form = LoginForm() if form.validate_on_submit(): user = User.query.filter_by(username=form.username.data).first() if user is None or not user.check_password(form.password.data): flash('Λάθος όνομα χρήστη ή κωδικός πρόσβασης', 'danger') return redirect(url_for('login')) login_user(user, remember=form.remember_me.data) return redirect(url_for('index')) return render_template('login.html', title='Σύνδεση', form=form, now=datetime.now()) @app.route('/logout') def logout(): logout_user() 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') # Προετοιμασία του query query = Store.query # Εφαρμογή αναζήτησης αν υπάρχει if search: query = query.filter(Store.name.contains(search) | Store.url.contains(search)) # Εφαρμογή ταξινόμησης 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 με pagination stores = query.paginate(page=page, per_page=20) return render_template('stores.html', title='Καταστήματα', stores=stores, search=search, sort=sort, order=order, now=datetime.now()) @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, now=datetime.now()) @app.route('/store/add', methods=['GET', 'POST']) @login_required def add_store(): form = StoreForm() if form.validate_on_submit(): store = Store( name=form.name.data, url=form.url.data, open_date=form.open_date.data ) db.session.add(store) db.session.commit() flash(f'Το κατάστημα {store.name} προστέθηκε επιτυχώς!', 'success') return redirect(url_for('stores')) return render_template('store_form.html', title='Προσθήκη Καταστήματος', form=form, now=datetime.now()) @app.route('/store/edit/<int:id>', methods=['GET', 'POST']) @login_required def edit_store(id): store = Store.query.get_or_404(id) form = StoreForm(obj=store) if form.validate_on_submit(): store.name = form.name.data store.url = form.url.data store.open_date = form.open_date.data db.session.commit() flash(f'Το κατάστημα {store.name} ενημερώθηκε επιτυχώς!', 'success') return redirect(url_for('store_detail', id=store.id)) return render_template('store_form.html', title=f'Επεξεργασία Καταστήματος: {store.name}', form=form, edit=True, now=datetime.now()) @app.route('/store/delete/<int:id>', methods=['POST']) @login_required def delete_store(id): store = Store.query.get_or_404(id) # Διαγραφή των στατιστικών που σχετίζονται με το κατάστημα StoreStats.query.filter_by(store_id=id).delete() db.session.delete(store) db.session.commit() flash(f'Το κατάστημα {store.name} διαγράφηκε επιτυχώς!', 'success') return redirect(url_for('stores')) @app.route('/stats') @login_required def stats(): form = DateRangeForm(request.args) # Προεπιλεγμένες ημερομηνίες (τελευταίες 30 ημέρες) today = datetime.now().date() default_start = today - timedelta(days=30) default_end = today # Λήψη παραμέτρων αναζήτησης και φιλτραρίσματος start_date = parse_date(request.args.get('start_date'), default_start) end_date = parse_date(request.args.get('end_date'), default_end) store_id = request.args.get('store_id', type=int) sort = request.args.get('sort', 'date') order = request.args.get('order', 'desc') # Προετοιμασία του 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 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 με ασφάλεια try: stats_data = query.all() except Exception as e: app.logger.error(f"Error fetching stats: {str(e)}") stats_data = [] # Υπολογισμός συνολικών αθροισμάτων με ασφάλεια totals = { 'sales': sum(getattr(stat, 'sales', 0) or 0 for stat in stats_data), 'products': sum(getattr(stat, 'products', 0) or 0 for stat in stats_data) } # Λίστα όλων των καταστημάτων για το φίλτρο επιλογής try: all_stores = Store.query.order_by(Store.name).all() except Exception as e: app.logger.error(f"Error fetching stores: {str(e)}") all_stores = [] return render_template('stats.html', title='Στατιστικά', stats=stats_data, stores=all_stores, selected_store_id=store_id, start_date=format_date(start_date), end_date=format_date(end_date), sort=sort, order=order, totals=totals, form=form, now=datetime.now()) @app.route('/upload', methods=['GET', 'POST']) @login_required def upload_file(): form = UploadForm() if form.validate_on_submit(): file = form.file.data 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) # Επεξεργασία του CSV αρχείου try: process_csv_file(file_path) flash('Το αρχείο ανέβηκε και επεξεργάστηκε επιτυχώς!', 'success') except Exception as e: flash(f'Σφάλμα κατά την επεξεργασία του αρχείου: {str(e)}', 'danger') return redirect(url_for('stores')) return render_template('upload.html', title='Ανέβασμα Δεδομένων', form=form, now=datetime.now()) def process_csv_file(file_path): """Επεξεργάζεται το CSV αρχείο και εισάγει τα δεδομένα στη βάση""" today = datetime.now().date() with open(file_path, 'r', encoding='utf-8-sig') as csvfile: reader = csv.DictReader(csvfile) for row in reader: store_name = row.get('Store', '').strip() store_url = row.get('URL', '').strip() open_date = row.get('Open Date', '').strip() # Προαιρετικά πεδία sales = 0 products = 0 scrape_date = today # Έλεγχος για τα προαιρετικά πεδία if 'Sales' in row: try: sales = int(row.get('Sales', 0)) except: sales = 0 if 'Products' in row: try: products = int(row.get('Products', 0)) except: products = 0 if 'Scrape Date' in row: scrape_date_str = row.get('Scrape Date', None) if scrape_date_str: try: scrape_date = datetime.strptime(scrape_date_str, '%d-%m-%Y').date() except: scrape_date = today if not store_name or not store_url: continue # Αναζήτηση ή δημιουργία καταστήματος store = Store.query.filter_by(url=store_url).first() if not store: store = Store(name=store_name, url=store_url, open_date=open_date) 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) else: # Ενημέρωση υπαρχόντων στοιχείων store.name = store_name store.open_date = open_date # Έλεγχος αν χρειάζεται να προσθέσουμε στατιστικά (μόνο αν έχουμε sales ή products) if sales > 0 or products > 0: # Έλεγχος αν υπάρχουν ήδη στατιστικά για αυτή την ημερομηνία existing_stats = StoreStats.query.filter_by( store_id=store.id, scrape_date=scrape_date ).first() if existing_stats: # Ενημέρωση υπαρχόντων στατιστικών existing_stats.sales = sales existing_stats.products = products else: # Δημιουργία νέων στατιστικών stats = StoreStats( store_id=store.id, sales=sales, products=products, scrape_date=scrape_date ) db.session.add(stats) # Αποθήκευση όλων των αλλαγών db.session.commit() @app.route('/export_data') @login_required def export_data(): """Εξάγει τα δεδομένα σε CSV""" # Λήψη παραμέτρων start_date = parse_date(request.args.get('start_date')) end_date = parse_date(request.args.get('end_date')) store_id = request.args.get('store_id', type=int) # Προετοιμασία του query query = db.session.query( Store.name.label('Store'), Store.url.label('URL'), Store.open_date.label('Open Date'), StoreStats.sales.label('Sales'), StoreStats.products.label('Products'), StoreStats.scrape_date.label('Scrape Date') ).join(Store) # Εφαρμογή φίλτρων if start_date and end_date: query = query.filter(StoreStats.scrape_date.between(start_date, end_date)) if store_id: query = query.filter(Store.id == store_id) # Εκτέλεση του query data = query.all() # Μετατροπή σε DataFrame για εύκολη εξαγωγή σε CSV df = pd.DataFrame(data) # Δημιουργία προσωρινού CSV αρχείου timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') csv_file = os.path.join(app.config['UPLOAD_FOLDER'], f'jamjar_export_{timestamp}.csv') df.to_csv(csv_file, index=False, encoding='utf-8-sig') return send_file( csv_file, as_attachment=True, download_name=f'jamjar_export_{timestamp}.csv', mimetype='text/csv' ) @app.route('/run_scraper', methods=['POST']) @login_required def run_scraper(): """Εκτελεί το scraper χειροκίνητα""" try: from app.scraper import run_scraper result = run_scraper() flash(f'Το scraping ολοκληρώθηκε επιτυχώς! {result}', 'success') except Exception as e: flash(f'Σφάλμα κατά την εκτέλεση του scraper: {str(e)}', 'danger') return redirect(url_for('index')) @app.route('/store_additions') @login_required def store_additions(): # Προεπιλεγμένες ημερομηνίες (τελευταίες 30 ημέρες) today = datetime.now().date() default_start = today - timedelta(days=30) default_end = today # Λήψη παραμέτρων φιλτραρίσματος start_date = parse_date(request.args.get('start_date'), default_start) end_date = parse_date(request.args.get('end_date'), default_end) source = request.args.get('source', '') # Base query - συγκεντρωτικά στατιστικά ανά ημέρα stats_query = db.session.query( StoreAddition.addition_date, func.count(StoreAddition.id).label('count') ).filter( StoreAddition.addition_date.between(start_date, end_date) ) # Εφαρμογή φίλτρου πηγής if source: stats_query = stats_query.filter(StoreAddition.source == source) # Ομαδοποίηση ανά ημέρα και ταξινόμηση stats_query = stats_query.group_by(StoreAddition.addition_date).order_by(StoreAddition.addition_date.desc()) # Εκτέλεση του query daily_stats = stats_query.all() # Πρόσφατα προστεθέντα καταστήματα recent_stores_query = db.session.query( Store, StoreAddition.addition_date, StoreAddition.source ).join(StoreAddition).filter( StoreAddition.addition_date.between(start_date, end_date) ) # Εφαρμογή φίλτρου πηγής if source: recent_stores_query = recent_stores_query.filter(StoreAddition.source == source) # Ταξινόμηση και περιορισμός recent_stores = recent_stores_query.order_by(StoreAddition.addition_date.desc()).limit(20).all() # Συγκεντρωτικά στατιστικά total_additions = sum(stat.count for stat in daily_stats) # Στατιστικά ανά πηγή source_stats = db.session.query( StoreAddition.source, func.count(StoreAddition.id).label('count') ).filter( StoreAddition.addition_date.between(start_date, end_date) ).group_by(StoreAddition.source).all() # Δεδομένα για γράφημα chart_dates = [stat.addition_date.strftime('%d-%m-%Y') for stat in daily_stats] chart_counts = [stat.count for stat in daily_stats] return render_template('store_additions.html', title='Νέα Καταστήματα', daily_stats=daily_stats, recent_stores=recent_stores, total_additions=total_additions, source_stats=source_stats, start_date=format_date(start_date), end_date=format_date(end_date), selected_source=source, chart_dates=chart_dates, chart_counts=chart_counts, now=datetime.now()) @app.route('/export_additions') @login_required def export_additions(): """Εξάγει τα στατιστικά προσθηκών καταστημάτων σε CSV""" # Λήψη παραμέτρων start_date = parse_date(request.args.get('start_date')) end_date = parse_date(request.args.get('end_date')) source = request.args.get('source', '') # Προετοιμασία του query query = db.session.query( Store.name.label('Store'), Store.url.label('URL'), Store.open_date.label('Open Date'), StoreAddition.addition_date.label('Addition Date'), StoreAddition.source.label('Source') ).join(StoreAddition) # Εφαρμογή φίλτρων if start_date and end_date: query = query.filter(StoreAddition.addition_date.between(start_date, end_date)) if source: query = query.filter(StoreAddition.source == source) # Ταξινόμηση query = query.order_by(StoreAddition.addition_date.desc()) # Εκτέλεση του query data = query.all() # Μετατροπή σε DataFrame για εύκολη εξαγωγή σε CSV df = pd.DataFrame(data) # Δημιουργία προσωρινού CSV αρχείου timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') csv_file = os.path.join(app.config['UPLOAD_FOLDER'], f'jamjar_additions_{timestamp}.csv') df.to_csv(csv_file, index=False, encoding='utf-8-sig') return send_file( csv_file, as_attachment=True, download_name=f'jamjar_additions_{timestamp}.csv', mimetype='text/csv' )
Editor is loading...
Leave a Comment