from flask import Flask, render_template, request, redirect, url_for, jsonify, Response import sqlite3, csv, io, os, zipfile, re, math from datetime import datetime, timedelta, date as date_cls app = Flask(__name__) DATABASE = os.environ.get('DATABASE_PATH', '/data/pricing.db') def get_db(): conn = sqlite3.connect(DATABASE) conn.row_factory = sqlite3.Row return conn def init_db(): os.makedirs(os.path.dirname(DATABASE), exist_ok=True) conn = get_db() conn.executescript(''' CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT); CREATE TABLE IF NOT EXISTS clients ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, email TEXT DEFAULT '', design_multiplier REAL DEFAULT 1.80, notes TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS materials ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, brand TEXT DEFAULT '', type TEXT DEFAULT 'PLA', color TEXT DEFAULT '', price_per_kg REAL NOT NULL DEFAULT 25.0, stock_g REAL DEFAULT 0, low_stock_threshold_g REAL DEFAULT 200, notes TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS material_price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, material_id INTEGER REFERENCES materials(id) ON DELETE CASCADE, price_per_kg REAL NOT NULL, recorded_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS machine_profiles ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, printer_price REAL DEFAULT 400.0, printer_lifespan_hours REAL DEFAULT 5000, nozzle_price REAL DEFAULT 10.0, nozzle_lifespan_hours REAL DEFAULT 250, plate_price REAL DEFAULT 25.0, plate_lifespan_hours REAL DEFAULT 1000, printer_power_kw REAL DEFAULT 0.35, electricity_price_kwh REAL DEFAULT 0.18, notes TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS handling_profiles ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, handling_rate_per_hour REAL DEFAULT 40.0, handling_minutes_per_plate REAL DEFAULT 5, notes TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS material_profiles ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, material_margin_pct REAL DEFAULT 20.0, notes TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS pricing_profiles ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, design_multiplier REAL DEFAULT 1.80, gross_margin_pct REAL DEFAULT 27.5, notes TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS jobs ( id INTEGER PRIMARY KEY AUTOINCREMENT, client_id INTEGER REFERENCES clients(id) ON DELETE SET NULL, material_id INTEGER REFERENCES materials(id) ON DELETE SET NULL, machine_profile_id INTEGER REFERENCES machine_profiles(id) ON DELETE SET NULL, handling_profile_id INTEGER REFERENCES handling_profiles(id) ON DELETE SET NULL, material_profile_id INTEGER REFERENCES material_profiles(id) ON DELETE SET NULL, pricing_profile_id INTEGER REFERENCES pricing_profiles(id) ON DELETE SET NULL, source_file TEXT DEFAULT '', name TEXT NOT NULL, description TEXT DEFAULT '', weight_g REAL NOT NULL, print_time_s INTEGER NOT NULL, design_multiplier REAL NOT NULL, gross_margin_pct REAL NOT NULL DEFAULT 27.5, discount_pct REAL DEFAULT 0, pieces_per_plate INTEGER DEFAULT 1, order_qty INTEGER DEFAULT 1, material_cost REAL, material_margin REAL, design_cost REAL, handling_cost REAL, wear_cost REAL, electricity_cost REAL, cout_fixe REAL, total_marge REAL, subtotal REAL, margin_amount REAL, cotisations_amount REAL, vfl_amount REAL, tva_amount REAL, other_taxes_amount REAL, tax_amount REAL, price_per_piece REAL, final_price REAL, notes TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS printers ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, is_active INTEGER DEFAULT 1, notes TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS working_schedule ( day_of_week INTEGER PRIMARY KEY, mode TEXT DEFAULT 'onsite', tt_window_start TEXT DEFAULT '10:00', tt_window_end TEXT DEFAULT '22:00', onsite_slot_1 TEXT DEFAULT '07:00', onsite_slot_2 TEXT DEFAULT '18:30' ); CREATE TABLE IF NOT EXISTS calendar_blocks ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, type TEXT DEFAULT 'off', printer_id INTEGER, notes TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS print_slots ( id INTEGER PRIMARY KEY AUTOINCREMENT, job_id INTEGER REFERENCES jobs(id) ON DELETE CASCADE, printer_id INTEGER REFERENCES printers(id) ON DELETE SET NULL, planned_start TEXT NOT NULL, planned_end TEXT NOT NULL, status TEXT DEFAULT 'planned', failure_note TEXT DEFAULT '', created_at TEXT DEFAULT (datetime('now')) ); ''') defaults = [ ('filament_price_per_kg','25.0'),('electricity_price_kwh','0.18'), ('printer_price','400.0'),('printer_lifespan_hours','5000'), ('nozzle_price','10.0'),('nozzle_lifespan_hours','250'), ('plate_price','25.0'),('plate_lifespan_hours','1000'), ('handling_rate_per_hour','40.0'),('handling_minutes_per_plate','5'), ('material_margin_pct','20.0'), ('cotisations_rate_pct','12.3'), ('vfl_rate_pct','1.0'), ('tva_rate_pct','20.0'), ('other_taxes_rate_pct','1.0'), ('printer_power_kw','0.35'), ('cooldown_minutes','15'), ('failure_buffer_pct','10'), ('max_simultaneous_printers','5'), ] conn.executemany('INSERT OR IGNORE INTO settings VALUES (?,?)', defaults) # Default printers: 6 A1, #1-#5 active, #6 inactive for i in range(1, 7): conn.execute('INSERT OR IGNORE INTO printers(id,name,is_active) VALUES(?,?,?)', (i, f'A1 #{i}', 1 if i <= 5 else 0)) # Default working schedule: Mon/Wed/Fri = TT, others = onsite tt_days = {0, 2, 4} for dow in range(7): mode = 'tt' if dow in tt_days else 'onsite' conn.execute('''INSERT OR IGNORE INTO working_schedule (day_of_week,mode,tt_window_start,tt_window_end,onsite_slot_1,onsite_slot_2) VALUES(?,?,?,?,?,?)''', (dow, mode, '10:00', '22:00', '07:00', '18:30')) # Migrations migrations = [ 'ALTER TABLE jobs ADD COLUMN machine_profile_id INTEGER', 'ALTER TABLE jobs ADD COLUMN handling_profile_id INTEGER', 'ALTER TABLE jobs ADD COLUMN material_profile_id INTEGER', 'ALTER TABLE jobs ADD COLUMN pricing_profile_id INTEGER', 'ALTER TABLE jobs ADD COLUMN cout_fixe REAL', 'ALTER TABLE jobs ADD COLUMN total_marge REAL', 'ALTER TABLE jobs ADD COLUMN cotisations_amount REAL', 'ALTER TABLE jobs ADD COLUMN vfl_amount REAL', 'ALTER TABLE jobs ADD COLUMN tva_amount REAL', 'ALTER TABLE jobs ADD COLUMN other_taxes_amount REAL', 'ALTER TABLE jobs ADD COLUMN pieces_per_plate INTEGER DEFAULT 1', 'ALTER TABLE jobs ADD COLUMN order_qty INTEGER DEFAULT 1', 'ALTER TABLE jobs ADD COLUMN price_per_piece REAL', 'ALTER TABLE clients ADD COLUMN default_machine_profile_id INTEGER', 'ALTER TABLE clients ADD COLUMN default_handling_profile_id INTEGER', 'ALTER TABLE clients ADD COLUMN default_material_profile_id INTEGER', 'ALTER TABLE clients ADD COLUMN default_pricing_profile_id INTEGER', ] for sql in migrations: try: conn.execute(sql) except Exception: pass conn.commit(); conn.close() def get_settings(): conn = get_db() rows = conn.execute('SELECT key,value FROM settings').fetchall() conn.close() return {r['key']: float(r['value']) for r in rows} def parse_3mf(file_stream): result = {'weight_g': None, 'hours': None, 'minutes': None, 'filename': None} try: with zipfile.ZipFile(file_stream) as z: for name in z.namelist(): if not any(name.endswith(e) for e in ['.xml','.model','.config','.json','.txt']): continue try: raw = z.read(name).decode('utf-8', errors='ignore') except Exception: continue if result['weight_g'] is None: for pat in [ r'filament[_\s]*weight[^>]*?["\s](\d+[\.,]\d+)', r'(\d+[\.,]\d+)\s*g\b(?!\s*[/])', r'"weight"\s*[:=]\s*"?(\d+[\.,]\d+)', r'used\s*\[g\][^"]*"\s*value\s*=\s*"(\d+[\.,]\d+)', ]: m = re.search(pat, raw, re.IGNORECASE) if m: try: result['weight_g'] = float(m.group(1).replace(',','.')) break except ValueError: pass if result['hours'] is None: for pat in [ r'(\d+)h\s*(\d+)m', r'(\d+)\s*hours?\s*(\d+)\s*min', r'printing.time[^"]*"\s*value\s*=\s*"(\d+)h\s*(\d+)', ]: m = re.search(pat, raw, re.IGNORECASE) if m: try: h, mn = int(m.group(1)), int(m.group(2)) if 0 <= h <= 500 and 0 <= mn <= 59: result['hours'] = h result['minutes'] = mn break except (ValueError, IndexError): pass except Exception: pass return result def calc(weight_g, print_time_s, design_mult, gross_margin_pct, discount_pct, price_per_kg_override=None, s=None, pieces_per_plate=1): if s is None: s = get_settings() if pieces_per_plate < 1: pieces_per_plate = 1 # weight_g et print_time_s = valeurs PLATEAU complet (Bambu Studio) # Tous les coûts sont ramenés à la pièce en divisant par pieces_per_plate price_kg = price_per_kg_override if price_per_kg_override is not None else s['filament_price_per_kg'] weight_per_piece = weight_g / pieces_per_plate mat = (price_kg / 1000) * weight_per_piece mat_margin = mat * (s['material_margin_pct'] / 100) design = mat * design_mult # Manutention, usure et élec divisées par pièces/plateau handling = (s['handling_rate_per_hour'] / 60) * s['handling_minutes_per_plate'] / pieces_per_plate printer_w = (s['printer_price'] / (s['printer_lifespan_hours'] * 3600)) * print_time_s / pieces_per_plate nozzle_w = (s['nozzle_price'] / (s['nozzle_lifespan_hours'] * 3600)) * print_time_s / pieces_per_plate plate_w = (s['plate_price'] / (s['plate_lifespan_hours'] * 3600)) * print_time_s / pieces_per_plate wear = printer_w + nozzle_w + plate_w elec = s['printer_power_kw'] * (print_time_s / 3600) * s['electricity_price_kwh'] / pieces_per_plate cout_fixe = mat + mat_margin + wear + elec sous_total = cout_fixe + handling + design margin = sous_total * (gross_margin_pct / 100) total_marge= handling + design + margin prix_ht = cout_fixe + total_marge cot_rate = s.get('cotisations_rate_pct', 12.3) / 100 vfl_rate = s.get('vfl_rate_pct', 0.0) / 100 tva_rate = s.get('tva_rate_pct', 0.0) / 100 other_taxes_rate = s.get('other_taxes_rate_pct', 1.0) / 100 total_charges = cot_rate + vfl_rate + other_taxes_rate if total_charges < 1.0: prix_ht_net = prix_ht / (1 - total_charges) else: prix_ht_net = prix_ht cot_amount = prix_ht_net * cot_rate vfl_amount = prix_ht_net * vfl_rate other_taxes_amount = prix_ht_net * other_taxes_rate tva_amount = prix_ht_net * tva_rate prix_ttc = prix_ht_net + tva_amount final = prix_ttc * (1 - discount_pct / 100) marge_pct_on_ht = (total_marge / prix_ht_net * 100) if prix_ht_net > 0 else 0 return { 'material_cost': round(mat, 4), 'material_margin': round(mat_margin, 4), 'wear_cost': round(wear, 4), 'printer_wear': round(printer_w, 4), 'nozzle_wear': round(nozzle_w, 4), 'plate_wear': round(plate_w, 4), 'electricity_cost': round(elec, 4), 'kwh_used': round(s['printer_power_kw'] * (print_time_s / 3600), 4), 'cout_fixe': round(cout_fixe, 4), 'handling_cost': round(handling, 4), 'design_cost': round(design, 4), 'subtotal': round(sous_total, 4), 'margin_amount': round(margin, 4), 'total_marge': round(total_marge, 4), 'marge_pct_on_ht': round(marge_pct_on_ht, 1), 'prix_ht': round(prix_ht, 4), 'cotisations_amount': round(cot_amount, 4), 'vfl_amount': round(vfl_amount, 4), 'other_taxes_amount': round(other_taxes_amount, 4), 'tva_amount': round(tva_amount, 4), 'prix_ht_net': round(prix_ht_net, 4), 'prix_ttc': round(prix_ttc, 4), 'tax_amount': round(cot_amount + vfl_amount + other_taxes_amount, 4), 'final_price': round(final, 2), '_cot_pct': round(cot_rate * 100, 2), '_vfl_pct': round(vfl_rate * 100, 2), '_tva_pct': round(tva_rate * 100, 2), '_other_taxes_pct': round(other_taxes_rate * 100, 2), '_material_margin_pct': s['material_margin_pct'], '_handling_rate': s['handling_rate_per_hour'], '_handling_minutes': s['handling_minutes_per_plate'], '_elec_price': s['electricity_price_kwh'], '_pieces_per_plate': pieces_per_plate, } # ── Planning helpers ────────────────────────────────────────────────────────── def get_schedule_dict(): conn = get_db() rows = conn.execute('SELECT * FROM working_schedule ORDER BY day_of_week').fetchall() conn.close() return {r['day_of_week']: dict(r) for r in rows} def get_blocks_dict(from_date=None, to_date=None): conn = get_db() if from_date and to_date: rows = conn.execute( 'SELECT * FROM calendar_blocks WHERE date BETWEEN ? AND ? ORDER BY date', (str(from_date), str(to_date))).fetchall() else: rows = conn.execute('SELECT * FROM calendar_blocks ORDER BY date').fetchall() conn.close() return [dict(r) for r in rows] def get_day_launches(check_date, schedule, blocks, cycle_min): """Returns number of possible launches on a given date (per machine).""" date_str = str(check_date) dow = check_date.weekday() block = next((b for b in blocks if b['date'] == date_str and b['printer_id'] is None), None) mode = block['type'] if block else schedule[dow]['mode'] if mode == 'off': return 0, mode elif mode == 'tt': sched = schedule[dow] h0, m0 = [int(x) for x in sched['tt_window_start'].split(':')] h1, m1 = [int(x) for x in sched['tt_window_end'].split(':')] window_min = (h1 * 60 + m1) - (h0 * 60 + m0) launches = max(1, int(window_min / cycle_min)) if cycle_min > 0 else 1 return launches, mode elif mode == 'onsite': return 2, mode return 0, mode def calc_capacity(print_time_s, pieces_per_plate, days_ahead=30): """Calcule la capacité de production sur N jours à venir.""" s = get_settings() cooldown_min = s.get('cooldown_minutes', 15) max_printers = int(s.get('max_simultaneous_printers', 5)) failure_buf = s.get('failure_buffer_pct', 10) / 100 cycle_min = print_time_s / 60 + cooldown_min if cycle_min <= 0: cycle_min = 1 conn = get_db() active_count = conn.execute('SELECT COUNT(*) FROM printers WHERE is_active=1').fetchone()[0] conn.close() effective = min(active_count, max_printers) schedule = get_schedule_dict() today = date_cls.today() blocks = get_blocks_dict(today, today + timedelta(days=days_ahead)) pieces_per_day_first = None week_total = 0 month_total = 0 for offset in range(days_ahead): d = today + timedelta(days=offset) launches, _ = get_day_launches(d, schedule, blocks, cycle_min) pieces = launches * pieces_per_plate * effective if pieces_per_day_first is None and pieces > 0: pieces_per_day_first = pieces if offset < 7: week_total += pieces month_total += pieces # Nb plateaux pour 1 commande (calculé côté appelant) return { 'pieces_per_day': pieces_per_day_first or 0, 'pieces_per_week': week_total, 'pieces_per_month': month_total, 'effective_printers': effective, 'cycle_minutes': round(cycle_min, 1), 'failure_buffer_pct': round(failure_buf * 100, 1), } def find_next_slot(print_time_s): """Trouve le prochain créneau libre sur une imprimante disponible.""" s = get_settings() cooldown_min = int(s.get('cooldown_minutes', 15)) max_printers = int(s.get('max_simultaneous_printers', 5)) total_sec = print_time_s + cooldown_min * 60 conn = get_db() printers = conn.execute( 'SELECT * FROM printers WHERE is_active=1 ORDER BY id LIMIT ?', (max_printers,) ).fetchall() existing = conn.execute( "SELECT printer_id, planned_start, planned_end FROM print_slots WHERE status NOT IN ('cancelled','failed')" ).fetchall() conn.close() schedule = get_schedule_dict() today = date_cls.today() blocks = get_blocks_dict(today, today + timedelta(days=60)) # Build busy intervals per printer busy = {} for sl in existing: pid = sl['printer_id'] if pid not in busy: busy[pid] = [] busy[pid].append((sl['planned_start'], sl['planned_end'])) for offset in range(60): d = today + timedelta(days=offset) launches, mode = get_day_launches(d, schedule, blocks, (print_time_s / 60) + cooldown_min) if launches == 0: continue date_str = str(d) sched = schedule[d.weekday()] # Build candidate launch times candidates = [] if mode == 'tt': h0, m0 = [int(x) for x in sched['tt_window_start'].split(':')] candidates.append(datetime(d.year, d.month, d.day, h0, m0)) # Add slots every cycle within window h1, m1 = [int(x) for x in sched['tt_window_end'].split(':')] window_end = datetime(d.year, d.month, d.day, h1, m1) t = candidates[0] while True: t = t + timedelta(seconds=total_sec) if t >= window_end: break candidates.append(t) else: h0, m0 = [int(x) for x in sched['onsite_slot_1'].split(':')] h1, m1 = [int(x) for x in sched['onsite_slot_2'].split(':')] candidates.append(datetime(d.year, d.month, d.day, h0, m0)) candidates.append(datetime(d.year, d.month, d.day, h1, m1)) now = datetime.now() for start_dt in candidates: if start_dt <= now: # ignorer les créneaux déjà passés continue end_dt = start_dt + timedelta(seconds=total_sec) for printer in printers: pid = printer['id'] printer_busy = busy.get(pid, []) conflict = any( not (end_dt.isoformat() <= bs or start_dt.isoformat() >= be) for bs, be in printer_busy ) if not conflict: return { 'printer_id': pid, 'printer_name': printer['name'], 'planned_start': start_dt.strftime('%Y-%m-%dT%H:%M'), 'planned_end': end_dt.strftime('%Y-%m-%dT%H:%M'), } return None def fmt_time(s): return f"{s//3600}h{(s%3600)//60:02d}m" app.jinja_env.filters['fmt_time'] = fmt_time # ─── Dashboard ──────────────────────────────────────────────────────────────── @app.route('/') def dashboard(): conn = get_db() recent = conn.execute(''' SELECT j.*, c.name as client_name, m.name as material_name FROM jobs j LEFT JOIN clients c ON j.client_id=c.id LEFT JOIN materials m ON j.material_id=m.id ORDER BY j.created_at DESC LIMIT 10''').fetchall() stats = conn.execute(''' SELECT COUNT(*) as total_jobs, ROUND(SUM(final_price),2) as total_revenue, ROUND(AVG(gross_margin_pct),1) as avg_margin, ROUND(SUM(CASE WHEN created_at>=date('now','-30 days') THEN final_price ELSE 0 END),2) as month_revenue FROM jobs''').fetchone() low_stock = conn.execute( 'SELECT * FROM materials WHERE stock_g<=low_stock_threshold_g ORDER BY stock_g ASC').fetchall() # Restock alerts planned_consumption = conn.execute(''' SELECT j.material_id, SUM(j.weight_g * j.order_qty) as total_g FROM print_slots ps JOIN jobs j ON ps.job_id=j.id WHERE ps.status='planned' AND j.material_id IS NOT NULL GROUP BY j.material_id''').fetchall() conn.close() return render_template('dashboard.html', jobs=recent, stats=stats, low_stock=low_stock, planned_consumption=planned_consumption) # ─── Clients ────────────────────────────────────────────────────────────────── @app.route('/clients') def clients(): conn = get_db() clients = conn.execute(''' SELECT c.*, COUNT(j.id) as job_count, ROUND(COALESCE(SUM(j.final_price),0),2) as total_revenue FROM clients c LEFT JOIN jobs j ON j.client_id=c.id GROUP BY c.id ORDER BY c.name''').fetchall() conn.close() return render_template('clients.html', clients=clients) def _get_all_profiles(conn): return { 'machine_profiles': conn.execute('SELECT id,name FROM machine_profiles ORDER BY name').fetchall(), 'handling_profiles': conn.execute('SELECT id,name FROM handling_profiles ORDER BY name').fetchall(), 'material_profiles': conn.execute('SELECT id,name FROM material_profiles ORDER BY name').fetchall(), 'pricing_profiles': conn.execute('SELECT id,name FROM pricing_profiles ORDER BY name').fetchall(), } @app.route('/clients/new', methods=['GET','POST']) def new_client(): conn = get_db() if request.method == 'POST': def _int(k): return int(request.form[k]) if request.form.get(k) else None conn.execute('''INSERT INTO clients (name,email,notes,default_machine_profile_id,default_handling_profile_id, default_material_profile_id,default_pricing_profile_id) VALUES(?,?,?,?,?,?,?)''', (request.form['name'], request.form.get('email',''), request.form.get('notes',''), _int('default_machine_profile_id'), _int('default_handling_profile_id'), _int('default_material_profile_id'), _int('default_pricing_profile_id'))) conn.commit(); conn.close() return redirect(url_for('clients')) profiles = _get_all_profiles(conn) conn.close() return render_template('client_form.html', client=None, **profiles) @app.route('/clients//edit', methods=['GET','POST']) def edit_client(id): conn = get_db() client = conn.execute('SELECT * FROM clients WHERE id=?',(id,)).fetchone() if request.method == 'POST': def _int(k): return int(request.form[k]) if request.form.get(k) else None conn.execute('''UPDATE clients SET name=?,email=?,notes=?, default_machine_profile_id=?,default_handling_profile_id=?, default_material_profile_id=?,default_pricing_profile_id=? WHERE id=?''', (request.form['name'], request.form.get('email',''), request.form.get('notes',''), _int('default_machine_profile_id'), _int('default_handling_profile_id'), _int('default_material_profile_id'), _int('default_pricing_profile_id'), id)) conn.commit(); conn.close() return redirect(url_for('clients')) profiles = _get_all_profiles(conn) conn.close() return render_template('client_form.html', client=client, **profiles) @app.route('/api/client//defaults') def api_client_defaults(id): conn = get_db() c = conn.execute('SELECT default_machine_profile_id, default_handling_profile_id, \ default_material_profile_id, default_pricing_profile_id FROM clients WHERE id=?',(id,)).fetchone() conn.close() if not c: return jsonify({}) return jsonify({ 'machine_profile_id': c['default_machine_profile_id'], 'handling_profile_id': c['default_handling_profile_id'], 'material_profile_id': c['default_material_profile_id'], 'pricing_profile_id': c['default_pricing_profile_id'], }) @app.route('/clients//delete', methods=['POST']) def delete_client(id): conn = get_db() conn.execute('DELETE FROM clients WHERE id=?',(id,)) conn.commit(); conn.close() return redirect(url_for('clients')) # ─── Matières ───────────────────────────────────────────────────────────────── @app.route('/materials') def materials(): conn = get_db() mats = conn.execute(''' SELECT m.*, COUNT(j.id) as job_count, ROUND(COALESCE(SUM(j.weight_g),0),1) as total_used_g FROM materials m LEFT JOIN jobs j ON j.material_id=m.id GROUP BY m.id ORDER BY m.type, m.name''').fetchall() conn.close() return render_template('materials.html', materials=mats) @app.route('/materials/new', methods=['GET','POST']) def new_material(): if request.method == 'POST': price = float(request.form['price_per_kg']) conn = get_db() conn.execute('''INSERT INTO materials(name,brand,type,color,price_per_kg,stock_g,low_stock_threshold_g,notes) VALUES(?,?,?,?,?,?,?,?)''', (request.form['name'], request.form.get('brand',''), request.form.get('type','PLA'), request.form.get('color',''), price, float(request.form.get('stock_g',0)), float(request.form.get('low_stock_threshold_g',200)), request.form.get('notes',''))) mat_id = conn.execute('SELECT last_insert_rowid()').fetchone()[0] conn.execute('INSERT INTO material_price_history(material_id,price_per_kg) VALUES(?,?)', (mat_id, price)) conn.commit(); conn.close() return redirect(url_for('materials')) return render_template('material_form.html', material=None, history=None) @app.route('/materials//edit', methods=['GET','POST']) def edit_material(id): conn = get_db() mat = conn.execute('SELECT * FROM materials WHERE id=?',(id,)).fetchone() history = conn.execute( 'SELECT * FROM material_price_history WHERE material_id=? ORDER BY recorded_at DESC LIMIT 10', (id,)).fetchall() if request.method == 'POST': new_price = float(request.form['price_per_kg']) old_price = mat['price_per_kg'] conn.execute('''UPDATE materials SET name=?,brand=?,type=?,color=?,price_per_kg=?, stock_g=?,low_stock_threshold_g=?,notes=? WHERE id=?''', (request.form['name'], request.form.get('brand',''), request.form.get('type','PLA'), request.form.get('color',''), new_price, float(request.form.get('stock_g', mat['stock_g'])), float(request.form.get('low_stock_threshold_g',200)), request.form.get('notes',''), id)) if abs(new_price - old_price) > 0.001: conn.execute('INSERT INTO material_price_history(material_id,price_per_kg) VALUES(?,?)', (id, new_price)) conn.commit(); conn.close() return redirect(url_for('materials')) conn.close() return render_template('material_form.html', material=mat, history=history) @app.route('/materials//restock', methods=['POST']) def restock_material(id): conn = get_db() add_g = float(request.form.get('add_g', 0)) conn.execute('UPDATE materials SET stock_g=stock_g+? WHERE id=?', (add_g, id)) new_price = request.form.get('new_price_per_kg') if new_price: new_price = float(new_price) mat = conn.execute('SELECT price_per_kg FROM materials WHERE id=?',(id,)).fetchone() conn.execute('UPDATE materials SET price_per_kg=? WHERE id=?', (new_price, id)) if abs(new_price - mat['price_per_kg']) > 0.001: conn.execute('INSERT INTO material_price_history(material_id,price_per_kg) VALUES(?,?)', (id, new_price)) conn.commit(); conn.close() return redirect(url_for('materials')) @app.route('/materials//delete', methods=['POST']) def delete_material(id): conn = get_db() conn.execute('DELETE FROM materials WHERE id=?',(id,)) conn.commit(); conn.close() return redirect(url_for('materials')) # ─── Profils / Templates ────────────────────────────────────────────────────── @app.route('/profiles') def profiles(): conn = get_db() machine = conn.execute('SELECT * FROM machine_profiles ORDER BY name').fetchall() handling = conn.execute('SELECT * FROM handling_profiles ORDER BY name').fetchall() material = conn.execute('SELECT * FROM material_profiles ORDER BY name').fetchall() pricing = conn.execute('SELECT * FROM pricing_profiles ORDER BY name').fetchall() conn.close() return render_template('profiles.html', machine_profiles=machine, handling_profiles=handling, material_profiles=material, pricing_profiles=pricing) @app.route('/profiles/machine/new', methods=['GET','POST']) def new_machine_profile(): if request.method == 'POST': conn = get_db() conn.execute('''INSERT INTO machine_profiles (name,printer_price,printer_lifespan_hours,nozzle_price,nozzle_lifespan_hours, plate_price,plate_lifespan_hours,printer_power_kw,electricity_price_kwh,notes) VALUES(?,?,?,?,?,?,?,?,?,?)''', ( request.form['name'], float(request.form['printer_price']), float(request.form['printer_lifespan_hours']), float(request.form['nozzle_price']), float(request.form['nozzle_lifespan_hours']), float(request.form['plate_price']), float(request.form['plate_lifespan_hours']), float(request.form['printer_power_kw']), float(request.form['electricity_price_kwh']), request.form.get('notes','') )) conn.commit(); conn.close() return redirect(url_for('profiles') + '#machine') return render_template('machine_profile_form.html', profile=None, settings=get_settings()) @app.route('/profiles/machine//edit', methods=['GET','POST']) def edit_machine_profile(id): conn = get_db() profile = conn.execute('SELECT * FROM machine_profiles WHERE id=?',(id,)).fetchone() if request.method == 'POST': conn.execute('''UPDATE machine_profiles SET name=?,printer_price=?,printer_lifespan_hours=?,nozzle_price=?,nozzle_lifespan_hours=?, plate_price=?,plate_lifespan_hours=?,printer_power_kw=?,electricity_price_kwh=?,notes=? WHERE id=?''', ( request.form['name'], float(request.form['printer_price']), float(request.form['printer_lifespan_hours']), float(request.form['nozzle_price']), float(request.form['nozzle_lifespan_hours']), float(request.form['plate_price']), float(request.form['plate_lifespan_hours']), float(request.form['printer_power_kw']), float(request.form['electricity_price_kwh']), request.form.get('notes',''), id )) conn.commit(); conn.close() return redirect(url_for('profiles') + '#machine') conn.close() return render_template('machine_profile_form.html', profile=profile, settings=get_settings()) @app.route('/profiles/machine//delete', methods=['POST']) def delete_machine_profile(id): conn = get_db() conn.execute('DELETE FROM machine_profiles WHERE id=?',(id,)) conn.commit(); conn.close() return redirect(url_for('profiles') + '#machine') @app.route('/profiles/handling/new', methods=['GET','POST']) def new_handling_profile(): if request.method == 'POST': conn = get_db() conn.execute('INSERT INTO handling_profiles(name,handling_rate_per_hour,handling_minutes_per_plate,notes) VALUES(?,?,?,?)', ( request.form['name'], float(request.form['handling_rate_per_hour']), float(request.form['handling_minutes_per_plate']), request.form.get('notes','') )) conn.commit(); conn.close() return redirect(url_for('profiles') + '#handling') return render_template('handling_profile_form.html', profile=None, settings=get_settings()) @app.route('/profiles/handling//edit', methods=['GET','POST']) def edit_handling_profile(id): conn = get_db() profile = conn.execute('SELECT * FROM handling_profiles WHERE id=?',(id,)).fetchone() if request.method == 'POST': conn.execute('UPDATE handling_profiles SET name=?,handling_rate_per_hour=?,handling_minutes_per_plate=?,notes=? WHERE id=?', ( request.form['name'], float(request.form['handling_rate_per_hour']), float(request.form['handling_minutes_per_plate']), request.form.get('notes',''), id )) conn.commit(); conn.close() return redirect(url_for('profiles') + '#handling') conn.close() return render_template('handling_profile_form.html', profile=profile, settings=get_settings()) @app.route('/profiles/handling//delete', methods=['POST']) def delete_handling_profile(id): conn = get_db() conn.execute('DELETE FROM handling_profiles WHERE id=?',(id,)) conn.commit(); conn.close() return redirect(url_for('profiles') + '#handling') @app.route('/profiles/material/new', methods=['GET','POST']) def new_material_profile(): if request.method == 'POST': conn = get_db() conn.execute('INSERT INTO material_profiles(name,material_margin_pct,notes) VALUES(?,?,?)', ( request.form['name'], float(request.form['material_margin_pct']), request.form.get('notes','') )) conn.commit(); conn.close() return redirect(url_for('profiles') + '#material') return render_template('material_profile_form.html', profile=None, settings=get_settings()) @app.route('/profiles/material//edit', methods=['GET','POST']) def edit_material_profile(id): conn = get_db() profile = conn.execute('SELECT * FROM material_profiles WHERE id=?',(id,)).fetchone() if request.method == 'POST': conn.execute('UPDATE material_profiles SET name=?,material_margin_pct=?,notes=? WHERE id=?', ( request.form['name'], float(request.form['material_margin_pct']), request.form.get('notes',''), id )) conn.commit(); conn.close() return redirect(url_for('profiles') + '#material') conn.close() return render_template('material_profile_form.html', profile=profile, settings=get_settings()) @app.route('/profiles/material//delete', methods=['POST']) def delete_material_profile(id): conn = get_db() conn.execute('DELETE FROM material_profiles WHERE id=?',(id,)) conn.commit(); conn.close() return redirect(url_for('profiles') + '#material') @app.route('/profiles/pricing/new', methods=['GET','POST']) def new_pricing_profile(): if request.method == 'POST': conn = get_db() conn.execute('INSERT INTO pricing_profiles(name,design_multiplier,gross_margin_pct,notes) VALUES(?,?,?,?)', ( request.form['name'], float(request.form['design_multiplier']), float(request.form['gross_margin_pct']), request.form.get('notes','') )) conn.commit(); conn.close() return redirect(url_for('profiles') + '#pricing') return render_template('pricing_profile_form.html', profile=None, settings=get_settings()) @app.route('/profiles/pricing//edit', methods=['GET','POST']) def edit_pricing_profile(id): conn = get_db() profile = conn.execute('SELECT * FROM pricing_profiles WHERE id=?',(id,)).fetchone() if request.method == 'POST': conn.execute('UPDATE pricing_profiles SET name=?,design_multiplier=?,gross_margin_pct=?,notes=? WHERE id=?', ( request.form['name'], float(request.form['design_multiplier']), float(request.form['gross_margin_pct']), request.form.get('notes',''), id )) conn.commit(); conn.close() return redirect(url_for('profiles') + '#pricing') conn.close() return render_template('pricing_profile_form.html', profile=profile, settings=get_settings()) @app.route('/profiles/pricing//delete', methods=['POST']) def delete_pricing_profile(id): conn = get_db() conn.execute('DELETE FROM pricing_profiles WHERE id=?',(id,)) conn.commit(); conn.close() return redirect(url_for('profiles') + '#pricing') # ─── Commandes ──────────────────────────────────────────────────────────────── @app.route('/jobs') def jobs(): conn = get_db() jobs = conn.execute(''' SELECT j.*, c.name as client_name, m.name as material_name, ps.status as slot_status, ps.planned_start FROM jobs j LEFT JOIN clients c ON j.client_id=c.id LEFT JOIN materials m ON j.material_id=m.id LEFT JOIN print_slots ps ON ps.job_id=j.id AND ps.id=( SELECT id FROM print_slots WHERE job_id=j.id ORDER BY planned_start LIMIT 1) ORDER BY j.created_at DESC''').fetchall() conn.close() return render_template('jobs.html', jobs=jobs) @app.route('/jobs/new', methods=['GET','POST']) def new_job(): conn = get_db() clients = conn.execute('SELECT * FROM clients ORDER BY name').fetchall() mats = conn.execute('SELECT * FROM materials ORDER BY type,name').fetchall() machine_profiles = conn.execute('SELECT * FROM machine_profiles ORDER BY name').fetchall() handling_profiles = conn.execute('SELECT * FROM handling_profiles ORDER BY name').fetchall() material_profiles = conn.execute('SELECT * FROM material_profiles ORDER BY name').fetchall() pricing_profiles = conn.execute('SELECT * FROM pricing_profiles ORDER BY name').fetchall() settings = get_settings() if request.method == 'POST': weight_g = float(request.form['weight_g']) hours = int(request.form.get('hours',0)) minutes = int(request.form.get('minutes',0)) print_time_s = hours*3600 + minutes*60 design_mult = float(request.form['design_multiplier']) gross_margin = float(request.form['gross_margin_pct']) discount = float(request.form.get('discount_pct',0)) client_id = request.form.get('client_id') or None material_id = request.form.get('material_id') or None source_file = request.form.get('source_file','').strip() machine_pid = request.form.get('machine_profile_id') or None handling_pid = request.form.get('handling_profile_id') or None material_pid = request.form.get('material_profile_id') or None pricing_pid = request.form.get('pricing_profile_id') or None pieces_per_plate = max(1, int(request.form.get('pieces_per_plate', 1))) order_qty = max(1, int(request.form.get('order_qty', 1))) s = dict(settings) if machine_pid: mp = conn.execute('SELECT * FROM machine_profiles WHERE id=?',(machine_pid,)).fetchone() if mp: s['printer_price'] = mp['printer_price'] s['printer_lifespan_hours'] = mp['printer_lifespan_hours'] s['nozzle_price'] = mp['nozzle_price'] s['nozzle_lifespan_hours'] = mp['nozzle_lifespan_hours'] s['plate_price'] = mp['plate_price'] s['plate_lifespan_hours'] = mp['plate_lifespan_hours'] s['printer_power_kw'] = mp['printer_power_kw'] s['electricity_price_kwh'] = mp['electricity_price_kwh'] if handling_pid: hp = conn.execute('SELECT * FROM handling_profiles WHERE id=?',(handling_pid,)).fetchone() if hp: s['handling_rate_per_hour'] = hp['handling_rate_per_hour'] s['handling_minutes_per_plate'] = hp['handling_minutes_per_plate'] if material_pid: marp = conn.execute('SELECT * FROM material_profiles WHERE id=?',(material_pid,)).fetchone() if marp: s['material_margin_pct'] = marp['material_margin_pct'] price_kg = None if material_id: mat = conn.execute('SELECT price_per_kg FROM materials WHERE id=?',(material_id,)).fetchone() if mat: price_kg = mat['price_per_kg'] r = calc(weight_g, print_time_s, design_mult, gross_margin, discount, price_kg, s, pieces_per_plate) conn.execute('''INSERT INTO jobs(client_id,material_id,machine_profile_id,handling_profile_id, material_profile_id,pricing_profile_id,source_file,name,description, weight_g,print_time_s,design_multiplier,gross_margin_pct,discount_pct, pieces_per_plate,order_qty, material_cost,material_margin,design_cost,handling_cost,wear_cost,electricity_cost, cout_fixe,total_marge,subtotal,margin_amount, cotisations_amount,vfl_amount,tva_amount,other_taxes_amount,tax_amount, price_per_piece,final_price,notes) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)''', (client_id,material_id,machine_pid,handling_pid,material_pid,pricing_pid, source_file,request.form['name'],request.form.get('description',''), weight_g,print_time_s,design_mult,gross_margin,discount, pieces_per_plate,order_qty, r['material_cost'],r['material_margin'],r['design_cost'],r['handling_cost'], r['wear_cost'],r['electricity_cost'], r['cout_fixe'],r['total_marge'],r['subtotal'],r['margin_amount'], r['cotisations_amount'],r['vfl_amount'],r['tva_amount'],r['other_taxes_amount'], r['tax_amount'],r['final_price'],round(r['final_price'],2), request.form.get('notes',''))) if material_id: conn.execute('UPDATE materials SET stock_g=MAX(0,stock_g-?) WHERE id=?', (weight_g, material_id)) conn.commit(); conn.close() return redirect(url_for('jobs')) conn.close() return render_template('new_job.html', clients=clients, materials=mats, settings=settings, machine_profiles=machine_profiles, handling_profiles=handling_profiles, material_profiles=material_profiles, pricing_profiles=pricing_profiles) @app.route('/jobs/') def job_detail(id): conn = get_db() job = conn.execute(''' SELECT j.*, c.name as client_name, m.name as material_name, m.color as material_color, m.type as material_type, mp.name as machine_profile_name, hp.name as handling_profile_name, marp.name as material_profile_name, pp.name as pricing_profile_name FROM jobs j LEFT JOIN clients c ON j.client_id=c.id LEFT JOIN materials m ON j.material_id=m.id LEFT JOIN machine_profiles mp ON j.machine_profile_id=mp.id LEFT JOIN handling_profiles hp ON j.handling_profile_id=hp.id LEFT JOIN material_profiles marp ON j.material_profile_id=marp.id LEFT JOIN pricing_profiles pp ON j.pricing_profile_id=pp.id WHERE j.id=?''', (id,)).fetchone() slots = conn.execute(''' SELECT ps.*, p.name as printer_name FROM print_slots ps LEFT JOIN printers p ON ps.printer_id=p.id WHERE ps.job_id=? ORDER BY ps.planned_start''', (id,)).fetchall() conn.close() if not job: return redirect(url_for('jobs')) return render_template('job_detail.html', job=job, slots=slots) @app.route('/jobs//delete', methods=['POST']) def delete_job(id): conn = get_db() conn.execute('DELETE FROM jobs WHERE id=?',(id,)) conn.commit(); conn.close() return redirect(url_for('jobs')) # ─── Planning ───────────────────────────────────────────────────────────────── @app.route('/planning') def planning(): conn = get_db() printers = [dict(r) for r in conn.execute('SELECT * FROM printers ORDER BY id').fetchall()] unscheduled = conn.execute(''' SELECT j.*, c.name as client_name FROM jobs j LEFT JOIN clients c ON j.client_id=c.id WHERE j.id NOT IN (SELECT DISTINCT job_id FROM print_slots WHERE status NOT IN ('cancelled','failed')) ORDER BY j.created_at DESC LIMIT 20''').fetchall() blocks = conn.execute( "SELECT * FROM calendar_blocks WHERE date >= date('now') ORDER BY date LIMIT 60" ).fetchall() schedule = conn.execute('SELECT * FROM working_schedule ORDER BY day_of_week').fetchall() conn.close() return render_template('planning.html', printers=printers, unscheduled=unscheduled, blocks=blocks, schedule=schedule) @app.route('/printers', methods=['GET','POST']) def printers_page(): conn = get_db() if request.method == 'POST': name = request.form['name'] active = 1 if request.form.get('is_active') else 0 notes = request.form.get('notes','') conn.execute('INSERT INTO printers(name,is_active,notes) VALUES(?,?,?)', (name,active,notes)) conn.commit() printers = conn.execute('SELECT * FROM printers ORDER BY id').fetchall() conn.close() return render_template('printers.html', printers=printers) @app.route('/printers//toggle', methods=['POST']) def toggle_printer(id): conn = get_db() conn.execute('UPDATE printers SET is_active=1-is_active WHERE id=?', (id,)) conn.commit(); conn.close() return redirect(url_for('printers_page')) @app.route('/printers//delete', methods=['POST']) def delete_printer(id): conn = get_db() conn.execute('DELETE FROM printers WHERE id=?', (id,)) conn.commit(); conn.close() return redirect(url_for('printers_page')) @app.route('/working-schedule', methods=['GET','POST']) def working_schedule(): if request.method == 'POST': conn = get_db() for dow in range(7): mode = request.form.get(f'mode_{dow}', 'off') tt_start = request.form.get(f'tt_start_{dow}', '10:00') tt_end = request.form.get(f'tt_end_{dow}', '22:00') onsite_slot1 = request.form.get(f'onsite_slot1_{dow}', '07:00') onsite_slot2 = request.form.get(f'onsite_slot2_{dow}', '18:30') conn.execute('''UPDATE working_schedule SET mode=?,tt_window_start=?,tt_window_end=?, onsite_slot_1=?,onsite_slot_2=? WHERE day_of_week=?''', (mode, tt_start, tt_end, onsite_slot1, onsite_slot2, dow)) conn.commit(); conn.close() return redirect(url_for('working_schedule')) conn = get_db() schedule = conn.execute('SELECT * FROM working_schedule ORDER BY day_of_week').fetchall() conn.close() return render_template('working_schedule.html', schedule=schedule) @app.route('/calendar-blocks', methods=['GET','POST']) def calendar_blocks(): conn = get_db() if request.method == 'POST': conn.execute('INSERT INTO calendar_blocks(date,type,printer_id,notes) VALUES(?,?,?,?)', ( request.form['date'], request.form.get('type','off'), request.form.get('printer_id') or None, request.form.get('notes','') )) conn.commit() blocks = conn.execute("SELECT cb.*, p.name as printer_name FROM calendar_blocks cb LEFT JOIN printers p ON cb.printer_id=p.id ORDER BY cb.date DESC LIMIT 50").fetchall() printers = conn.execute('SELECT * FROM printers ORDER BY id').fetchall() conn.close() return render_template('calendar_blocks.html', blocks=blocks, printers=printers) @app.route('/calendar-blocks//delete', methods=['POST']) def delete_calendar_block(id): conn = get_db() conn.execute('DELETE FROM calendar_blocks WHERE id=?', (id,)) conn.commit(); conn.close() return redirect(url_for('calendar_blocks')) # ─── API ────────────────────────────────────────────────────────────────────── @app.route('/api/calculate', methods=['POST']) def api_calculate(): d = request.json s = dict(get_settings()) conn = get_db() if d.get('machine_profile_id'): mp = conn.execute('SELECT * FROM machine_profiles WHERE id=?',(d['machine_profile_id'],)).fetchone() if mp: s['printer_price'] = mp['printer_price'] s['printer_lifespan_hours'] = mp['printer_lifespan_hours'] s['nozzle_price'] = mp['nozzle_price'] s['nozzle_lifespan_hours'] = mp['nozzle_lifespan_hours'] s['plate_price'] = mp['plate_price'] s['plate_lifespan_hours'] = mp['plate_lifespan_hours'] s['printer_power_kw'] = mp['printer_power_kw'] s['electricity_price_kwh'] = mp['electricity_price_kwh'] if d.get('handling_profile_id'): hp = conn.execute('SELECT * FROM handling_profiles WHERE id=?',(d['handling_profile_id'],)).fetchone() if hp: s['handling_rate_per_hour'] = hp['handling_rate_per_hour'] s['handling_minutes_per_plate'] = hp['handling_minutes_per_plate'] if d.get('material_profile_id'): marp = conn.execute('SELECT * FROM material_profiles WHERE id=?',(d['material_profile_id'],)).fetchone() if marp: s['material_margin_pct'] = marp['material_margin_pct'] price_kg = None if d.get('material_id'): mat = conn.execute('SELECT price_per_kg FROM materials WHERE id=?',(d['material_id'],)).fetchone() if mat: price_kg = mat['price_per_kg'] conn.close() pieces_per_plate = max(1, int(d.get('pieces_per_plate', 1))) order_qty = max(1, int(d.get('order_qty', 1))) print_time_s = int(d.get('hours',0))*3600 + int(d.get('minutes',0))*60 result = calc( weight_g=float(d['weight_g']), print_time_s=print_time_s, design_mult=float(d.get('design_multiplier',1.80)), gross_margin_pct=float(d.get('gross_margin_pct',27.5)), discount_pct=float(d.get('discount_pct',0)), price_per_kg_override=price_kg, s=s, pieces_per_plate=pieces_per_plate) # Multi-piece extras nb_plateaux = math.ceil(order_qty / pieces_per_plate) price_order = round(result['final_price'] * order_qty, 2) last_plate_qty = order_qty % pieces_per_plate plate_full = last_plate_qty == 0 cooldown_min = s.get('cooldown_minutes', 15) total_print_min = (print_time_s / 60 + cooldown_min) * nb_plateaux - cooldown_min capacity = calc_capacity(print_time_s, pieces_per_plate) result['order_qty'] = order_qty result['nb_plateaux'] = nb_plateaux result['price_order'] = price_order result['plate_full'] = plate_full result['last_plate_qty'] = last_plate_qty result['total_print_min']= round(total_print_min, 0) result['capacity'] = capacity return jsonify(result) @app.route('/api/clients/') def api_client(id): conn = get_db() c = conn.execute('SELECT * FROM clients WHERE id=?',(id,)).fetchone() conn.close() if not c: return jsonify({}), 404 return jsonify({'design_multiplier': c['design_multiplier']}) @app.route('/api/parse-3mf', methods=['POST']) def api_parse_3mf(): if 'file' not in request.files: return jsonify({'error': 'Aucun fichier'}), 400 f = request.files['file'] if not f.filename.lower().endswith('.3mf'): return jsonify({'error': 'Format .3mf requis'}), 400 result = parse_3mf(f.stream) result['filename'] = os.path.splitext(f.filename)[0] return jsonify(result) @app.route('/api/slots') def api_slots(): conn = get_db() slots = conn.execute(''' SELECT ps.id, ps.job_id, ps.printer_id, ps.planned_start, ps.planned_end, ps.status, ps.failure_note, j.name as job_name, j.print_time_s, j.final_price, j.pieces_per_plate, j.order_qty, j.weight_g, p.name as printer_name, c.name as client_name FROM print_slots ps JOIN jobs j ON ps.job_id=j.id JOIN printers p ON ps.printer_id=p.id LEFT JOIN clients c ON j.client_id=c.id WHERE ps.planned_start >= datetime('now', '-7 days') ORDER BY ps.planned_start''').fetchall() conn.close() return jsonify([dict(s) for s in slots]) @app.route('/api/slots/new', methods=['POST']) def api_new_slot(): d = request.json job_id = int(d['job_id']) printer_id = int(d['printer_id']) start_str = d['planned_start'] # ISO string conn = get_db() job = conn.execute('SELECT print_time_s FROM jobs WHERE id=?', (job_id,)).fetchone() s = get_settings() if not job: conn.close() return jsonify({'error': 'Job introuvable'}), 404 cooldown_s = int(s.get('cooldown_minutes', 15)) * 60 start_dt = datetime.fromisoformat(start_str) end_dt = start_dt + timedelta(seconds=job['print_time_s'] + cooldown_s) conn.execute('INSERT INTO print_slots(job_id,printer_id,planned_start,planned_end,status) VALUES(?,?,?,?,?)', (job_id, printer_id, start_dt.isoformat(), end_dt.isoformat(), 'planned')) conn.commit(); conn.close() return jsonify({'ok': True}) @app.route('/api/slots//move', methods=['POST']) def api_move_slot(id): d = request.json start_str = d['planned_start'] conn = get_db() slot = conn.execute('SELECT ps.*, j.print_time_s FROM print_slots ps JOIN jobs j ON ps.job_id=j.id WHERE ps.id=?', (id,)).fetchone() if not slot: conn.close() return jsonify({'error': 'Slot introuvable'}), 404 s = get_settings() cooldown_s = int(s.get('cooldown_minutes', 15)) * 60 start_dt = datetime.fromisoformat(start_str) end_dt = start_dt + timedelta(seconds=slot['print_time_s'] + cooldown_s) conn.execute('UPDATE print_slots SET planned_start=?,planned_end=? WHERE id=?', (start_dt.isoformat(), end_dt.isoformat(), id)) conn.commit(); conn.close() return jsonify({'ok': True}) @app.route('/api/slots//status', methods=['POST']) def api_slot_status(id): d = request.json conn = get_db() conn.execute('UPDATE print_slots SET status=?,failure_note=? WHERE id=?', (d.get('status','planned'), d.get('note',''), id)) conn.commit(); conn.close() return jsonify({'ok': True}) @app.route('/api/slots//delete', methods=['POST']) def api_delete_slot(id): conn = get_db() conn.execute('DELETE FROM print_slots WHERE id=?', (id,)) conn.commit(); conn.close() return jsonify({'ok': True}) @app.route('/api/suggest-slot', methods=['POST']) def api_suggest_slot(): d = request.json print_time_s = int(d.get('print_time_s', 0)) result = find_next_slot(print_time_s) if result: return jsonify(result) return jsonify({'error': 'Aucun créneau disponible dans les 60 prochains jours'}), 404 @app.route('/api/capacity', methods=['POST']) def api_capacity(): d = request.json print_time_s = int(d.get('print_time_s', 0)) pieces_per_plate = max(1, int(d.get('pieces_per_plate', 1))) result = calc_capacity(print_time_s, pieces_per_plate) return jsonify(result) @app.route('/api/availability') def api_availability(): """Returns unavailable periods for the next 4 weeks as JSON for vis.js background items.""" schedule = get_schedule_dict() today = date_cls.today() blocks = get_blocks_dict(today, today + timedelta(days=28)) result = [] for offset in range(28): d = today + timedelta(days=offset) date_str = str(d) block = next((b for b in blocks if b['date'] == date_str and b['printer_id'] is None), None) mode = block['type'] if block else schedule[d.weekday()]['mode'] if mode == 'off': result.append({ 'start': date_str + 'T00:00:00', 'end': date_str + 'T23:59:59', 'type': 'off' }) elif mode == 'tt': sched = schedule[d.weekday()] # Unavailable before tt_window_start result.append({'start': date_str+'T00:00:00', 'end': date_str+'T'+sched['tt_window_start']+':00', 'type':'unavailable'}) # Unavailable after tt_window_end result.append({'start': date_str+'T'+sched['tt_window_end']+':00', 'end': date_str+'T23:59:59', 'type':'unavailable'}) return jsonify(result) # ─── iCal Export ────────────────────────────────────────────────────────────── @app.route('/calendar.ics') def calendar_ics(): conn = get_db() slots = conn.execute(''' SELECT ps.*, j.name as job_name, j.print_time_s, j.order_qty, j.pieces_per_plate, j.weight_g, p.name as printer_name, c.name as client_name FROM print_slots ps JOIN jobs j ON ps.job_id=j.id JOIN printers p ON ps.printer_id=p.id LEFT JOIN clients c ON j.client_id=c.id WHERE ps.status NOT IN ('cancelled') ORDER BY ps.planned_start''').fetchall() conn.close() lines = [ 'BEGIN:VCALENDAR', 'VERSION:2.0', 'PRODID:-//3D Pricing//Planning//FR', 'CALSCALE:GREGORIAN', 'METHOD:PUBLISH', 'X-WR-CALNAME:Planning Impression 3D', 'X-WR-TIMEZONE:Europe/Paris', ] for s in slots: start = s['planned_start'].replace('-','').replace(':','').replace(' ','T') if '.' in start: start = start.split('.')[0] end = s['planned_end'].replace('-','').replace(':','').replace(' ','T') if '.' in end: end = end.split('.')[0] if 'T' not in start: start += 'T000000' if 'T' not in end: end += 'T000000' client_part = f" — {s['client_name']}" if s['client_name'] else '' lines += [ 'BEGIN:VEVENT', f"DTSTART:{start}", f"DTEND:{end}", f"SUMMARY:{s['job_name']}{client_part} [{s['printer_name']}]", f"DESCRIPTION:Qty: {s['order_qty']} pcs | {s['pieces_per_plate']} pcs/plateau | Statut: {s['status']}", f"UID:slot-{s['id']}@3d-pricing", f"STATUS:{'CONFIRMED' if s['status']=='done' else 'TENTATIVE'}", 'END:VEVENT', ] lines.append('END:VCALENDAR') ical = '\r\n'.join(lines) return Response(ical, mimetype='text/calendar', headers={'Content-Disposition': 'attachment; filename=planning_3d.ics'}) # ─── Paramètres ─────────────────────────────────────────────────────────────── @app.route('/settings', methods=['GET','POST']) def settings(): if request.method == 'POST': conn = get_db() for key in request.form: conn.execute('INSERT OR REPLACE INTO settings VALUES (?,?)', (key, request.form[key])) conn.commit(); conn.close() return redirect(url_for('settings')) return render_template('settings.html', settings=get_settings()) # ─── Statistiques ───────────────────────────────────────────────────────────── @app.route('/stats') def stats(): conn = get_db() by_client = conn.execute(''' SELECT COALESCE(c.name,'Sans client') as name, COUNT(j.id) as jobs, ROUND(SUM(j.final_price),2) as revenue, ROUND(AVG(j.gross_margin_pct),1) as avg_margin FROM jobs j LEFT JOIN clients c ON j.client_id=c.id GROUP BY j.client_id ORDER BY revenue DESC''').fetchall() by_month = conn.execute(''' SELECT strftime('%Y-%m',created_at) as month, COUNT(*) as jobs, ROUND(SUM(final_price),2) as revenue FROM jobs GROUP BY month ORDER BY month DESC LIMIT 12''').fetchall() by_material = conn.execute(''' SELECT COALESCE(m.name,'Non defini') as name, m.type, COUNT(j.id) as jobs, ROUND(SUM(j.weight_g),1) as total_g, ROUND(SUM(j.final_price),2) as revenue FROM jobs j LEFT JOIN materials m ON j.material_id=m.id GROUP BY j.material_id ORDER BY total_g DESC''').fetchall() conn.close() return render_template('stats.html', by_client=by_client, by_month=by_month, by_material=by_material) # ─── Export CSV ─────────────────────────────────────────────────────────────── @app.route('/export/csv') def export_csv(): conn = get_db() jobs = conn.execute(''' SELECT j.created_at, COALESCE(c.name,'') as client, COALESCE(m.name,'') as material, m.type as material_type, j.source_file, j.name, j.weight_g, j.print_time_s, j.pieces_per_plate, j.order_qty, j.design_multiplier, j.gross_margin_pct, j.discount_pct, j.cout_fixe, j.handling_cost, j.design_cost, j.total_marge, j.margin_amount, j.cotisations_amount, j.vfl_amount, j.other_taxes_amount, j.tva_amount, j.price_per_piece, j.final_price FROM jobs j LEFT JOIN clients c ON j.client_id=c.id LEFT JOIN materials m ON j.material_id=m.id ORDER BY j.created_at DESC''').fetchall() conn.close() out = io.StringIO() w = csv.writer(out, delimiter=';') w.writerow(['Date','Client','Matiere','Type','Fichier source','Nom', 'Poids(g)','Temps(s)','Pièces/plateau','Qté commande', 'Coeff Design','Marge(%)','Remise(%)', 'Cout fixe','Manutention','Design','Total marge', 'Marge brute','Cotisations','VFL','Autres taxes','TVA', 'Prix/pièce','Prix Final']) for j in jobs: w.writerow(list(j)) return Response(out.getvalue(), mimetype='text/csv', headers={'Content-Disposition': 'attachment; filename=tarifs_3d.csv'}) if __name__ == '__main__': init_db() app.run(host='0.0.0.0', port=5000, debug=False)