"""DEPL Machine Log — motor ampere maintenance system.""" from __future__ import annotations import json import os import sqlite3 from datetime import date, datetime, timedelta from functools import wraps from pathlib import Path from flask import ( Flask, flash, g, jsonify, redirect, render_template, request, session, url_for, ) from werkzeug.middleware.proxy_fix import ProxyFix from werkzeug.security import check_password_hash, generate_password_hash BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" DB_PATH = DATA_DIR / "depl_machine_log.db" try: from dotenv import load_dotenv load_dotenv(BASE_DIR / ".env") except ImportError: pass def env_flag(name: str, default: bool = False) -> bool: val = os.environ.get(name) if val is None: return default return val.strip().lower() in {"1", "true", "yes", "on"} PRODUCTION = ( os.environ.get("FLASK_ENV") == "production" or env_flag("PRODUCTION") or bool(os.environ.get("PASSENGER_APP_ENV")) ) app = Flask(__name__) app.secret_key = os.environ.get("SECRET_KEY") or "depl-machine-log-local-2026" app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1) app.config.update( TEMPLATES_AUTO_RELOAD=not PRODUCTION, SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE="Lax", SESSION_COOKIE_SECURE=PRODUCTION, PREFERRED_URL_SCHEME="https" if PRODUCTION else "http", ) @app.template_filter("amp") def amp_filter(value): if value is None or value == "": return "" try: return f"{float(value):.2f}" except (TypeError, ValueError): return value I18N = { "en": { "app_name": "DEPL Machine Log", "tagline": "Motor ampere maintenance log", "login": "Sign in", "username": "Username", "password": "Password", "logout": "Sign out", "menu": "Menu", "dashboard": "Dashboard", "machines": "Machines", "daily_log": "Daily Log", "analysis": "Analysis", "companies": "Companies", "company": "Company", "date": "Date", "machine_no": "Machine No", "motor_hp": "Motor (HP)", "machine_on": "Machine On", "machine_off": "Machine Off", "empty_amper": "Empty Amper", "loading_amper": "Loading Amper", "overload_amper": "Overload Amper", "on_time": "On time", "off_time": "Off time", "save": "Save", "add": "Add", "edit": "Edit", "delete": "Delete", "cancel": "Cancel", "add_machine": "Add machine", "add_company": "Add company", "add_motor": "Add motor", "motor": "Motor", "location": "Location", "notes": "Notes", "remarks": "Remarks", "status": "Status", "alerts": "Alerts", "normal": "Normal", "watch": "Watch", "alert": "Alert", "no_alerts": "No machine problems detected from the latest readings.", "welcome": "Welcome", "today_logs": "Today's logs", "machines_count": "Machines", "open_alerts": "Open alerts", "recent_activity": "Latest ampere trend", "baseline": "Expected amper (baseline)", "expected_empty": "Expected empty A", "expected_loading": "Expected loading A", "expected_overload": "Expected overload A", "contact": "Contact person", "phone": "Phone", "address": "Address", "company_name": "Company name", "login_account": "Login account", "full_name": "Full name", "invalid_login": "Username or password is incorrect.", "saved": "Saved successfully.", "deleted": "Deleted.", "need_machine": "Add at least one machine before filling the daily log.", "print": "Print log sheet", "select_machine": "Select machine", "findings": "Findings", "no_data": "Not enough readings yet. Fill the daily log for a few days.", "no_readings": "No readings yet", "language": "Language", "admin": "DEPL Admin", "hp": "HP", "label": "Motor name", "confirm_delete": "Delete this record?", "fill_log": "Fill today's ampere log", "view_analysis": "View analysis", "signed_in_as": "Signed in as", "demo_hint": "DEPL admin: admin / DEPL@2026", "empty_hint": "No-load current", "loading_hint": "Normal load current", "overload_hint": "Overload current", "problem_summary": "Machine health", "all_companies": "All companies", "create_login": "This login is given to the company that bought the machine.", "company_profiles": "Company profiles", "open_profile": "Open profile", "viewing_all": "All company profiles", "viewing_profile": "Company profile", "this_profile": "This company", "profile_machines": "Machine profiles", "system_overview": "Whole system", "no_company_profiles": "No company profiles yet. Add a company to start.", "operator": "Machine operator", "operator_logins": "Operator logins", "add_operator": "Add operator login", "operator_help": "This login can open only the Daily Log for this company. Create it from the company profile.", "no_operators": "No operator logins yet.", "operator_only_log": "Operator access is Daily Log only.", "ask_admin_machines": "Ask the administrator to add machines for this company.", "motor_help": "A machine can have more than one motor (3HP, 5HP, …).", "log_help": "Enter Empty, Loading and Overload amper for Machine On and Machine Off. Log and chart always use the same added machine.", "trend_up": "Current is rising", "healthy": "Healthy", "needs_check": "Needs check", "critical": "Critical", "days": "days", "readings": "Readings", "last_reading": "Last reading", "no_machines": "No machines yet. Add each machine with a machine number.", "no_companies": "No customer companies yet.", "password_optional": "Leave blank to keep the current password.", "active": "Active", "save_log": "Save daily log", "pick_date": "Log date", "ampere": "Amper", "attached_machines": "Added machines", "machine_chart": "Ampere chart for this machine", "open_log": "Daily log + chart", "linked_help": "Only machines you add appear in Daily Log and Chart. The same machine is used for today and future dates.", "hp_auto_fill": "When you enter HP, standard 400V 3-phase Unload / Loading / Overload amper is filled. You can still change it.", "std_hint": "Std 400V 3φ", "legend_ok": "Normal", "legend_high": "High", "legend_too_high": "Too high", "legend_low": "Low", "legend_too_low": "Too low", "color_legend": "Amper vs standard for this HP", "ton_output": "Today's ton output", "target_tons": "Daily target (ton)", "run_hours": "Run hours", "got_output": "Got today's output", "got_yes": "Yes", "not_yet": "Not yet", "no_target": "No target", "today_tons": "Today tons", "today_hours": "Today run hours", "today_production": "Today's production", "hours_unit": "h", "tons_unit": "T", "on_off": "On – Off", "plant_tons_help": "Ton output and run hours are for all machines together, not one machine at a time.", "plant_target_hint": "Daily target for every machine of this company combined", "save_production": "Save tons", "select_company_first": "Select a company first.", "view_date": "Select date", "day_tons": "Tons that day", "day_hours": "Run hours that day", "day_production": "Production that day", "got_day_output": "Got that day's output", "day_logs": "Logs that day", "ton_output_day": "Ton output for this date", "back_today": "Today", "view_month": "Select month", "month_tons": "Month tons", "month_hours": "Month run hours", "month_target": "Month target (ton)", "month_production": "This month's production", "got_month_output": "Got this month's output", "days_recorded": "Days with tons", "this_month": "This month", "enter_day_prod": "Enter day's tons", "all_machines_chart": "All machines together", "this_machine_chart": "Selected machine", "monthly_loading": "On Loading Amper this month", "all_avg": "All machines average", "search_month": "Search", "download_pdf": "Download PDF", "downloading_pdf": "Preparing PDF…", "day_amp_compare": "Today vs next day amper", "prev_log_day": "Previous day", "next_log_day": "Next day", "same_amp": "Same", "amp_changed": "Not the same", "no_day_compare": "Need two log days to compare.", "vs_prev_day": "vs previous day", }, "si": { "app_name": "DEPL Machine Log", "tagline": "මෝටර් ඇම්පියර් නඩත්තු ලොග්", "login": "පිවිසෙන්න", "username": "පරිශීලක නාමය", "password": "මුරපදය", "logout": "ඉවත් වන්න", "menu": "මෙනුව", "dashboard": "මුල් පිටුව", "machines": "යන්ත්‍ර", "daily_log": "දෛනික ලොග්", "analysis": "විශ්ලේෂණය", "companies": "සමාගම්", "company": "සමාගම", "date": "දිනය", "machine_no": "යන්ත්‍ර අංකය", "motor_hp": "මෝටර් (HP)", "machine_on": "Machine On", "machine_off": "Machine Off", "empty_amper": "Empty Amper", "loading_amper": "Loading Amper", "overload_amper": "Overload Amper", "on_time": "On වේලාව", "off_time": "Off වේලාව", "save": "සුරකින්න", "add": "එකතු කරන්න", "edit": "සංස්කරණය", "delete": "මකන්න", "cancel": "අවලංගු", "add_machine": "යන්ත්‍රය එකතු කරන්න", "add_company": "සමාගම එකතු කරන්න", "add_motor": "මෝටර් එකතු කරන්න", "motor": "මෝටර්", "location": "ස්ථානය", "notes": "සටහන්", "remarks": "සටහන්", "status": "තත්ත්වය", "alerts": "අනතුරු ඇඟවීම්", "normal": "සාමාන්‍ය", "watch": "බලාගන්න", "alert": "අවවාදය", "no_alerts": "නවතම කියවීම් අනුව යන්ත්‍ර ගැටලුවක් නොපෙනේ.", "welcome": "සාදරයෙන් පිළිගනිමු", "today_logs": "අද ලොග්", "machines_count": "යන්ත්‍ර", "open_alerts": "අනතුරු ඇඟවීම්", "recent_activity": "නවතම ඇම්පියර් ප්‍රවණතාව", "baseline": "බලාපොරොත්තු ඇම්පියර්", "expected_empty": "Expected empty A", "expected_loading": "Expected loading A", "expected_overload": "Expected overload A", "contact": "සම්බන්ධකරු", "phone": "දුරකථනය", "address": "ලිපිනය", "company_name": "සමාගමේ නම", "login_account": "Login ගිණුම", "full_name": "නම", "invalid_login": "පරිශීලක නාමය හෝ මුරපදය වැරදියි.", "saved": "සාර්ථකව සුරකින ලදී.", "deleted": "මකන ලදී.", "need_machine": "දෛනික ලොග් පුරවන්නට පෙර යන්ත්‍රයක් එකතු කරන්න.", "print": "ලොග් පත්‍රය මුද්‍රණය", "select_machine": "යන්ත්‍රය තෝරන්න", "findings": "සොයාගැනීම්", "no_data": "තවම කියවීම් මදි. දින කිහිපයක දෛනික ලොග් පුරවන්න.", "no_readings": "කියවීම් නැත", "language": "භාෂාව", "admin": "DEPL Admin", "hp": "HP", "label": "මෝටර් නම", "confirm_delete": "මෙය මකන්නද?", "fill_log": "අද ඇම්පියර් ලොග් පුරවන්න", "view_analysis": "විශ්ලේෂණය බලන්න", "signed_in_as": "පිවිසී ඇත්තේ", "demo_hint": "DEPL admin: admin / DEPL@2026", "empty_hint": "හිස් බර (no-load) ධාරාව", "loading_hint": "සාමාන්‍ය බර ධාරාව", "overload_hint": "අධිබර ධාරාව", "problem_summary": "යන්ත්‍ර සෞඛ්‍යය", "all_companies": "සියලු සමාගම්", "create_login": "යන්ත්‍රය විකුණු සමාගමට මෙම login එක දෙන්න.", "company_profiles": "සමාගම් profiles", "open_profile": "Profile එක අරින්න", "viewing_all": "සියලු සමාගම් profiles", "viewing_profile": "සමාගමේ profile", "this_profile": "මෙම සමාගම", "profile_machines": "යන්ත්‍ර profiles", "system_overview": "මුළු පද්ධතිය", "no_company_profiles": "තවම සමාගම් profiles නැහැ. සමාගමක් එකතු කරන්න.", "operator": "යන්ත්‍ර operator", "operator_logins": "Operator logins", "add_operator": "Operator login එකතු කරන්න", "operator_help": "මෙම login එකට Daily Log පිටුව විතරක් තියෙනවා. Admin විතරක් company profile එකෙන් හදන්න පුළුවන්.", "no_operators": "තවම operator logins නැහැ.", "operator_only_log": "Operator ට Daily Log විතරක් තියෙනවා.", "ask_admin_machines": "මෙම සමාගමට යන්ත්‍ර එකතු කරන්න admin ගෙන් කියන්න.", "motor_help": "යන්ත්‍රයකට මෝටර් කිහිපයක් තිබිය හැක (3HP, 5HP, …).", "log_help": "එකතු කළ යන්ත්‍රයම Daily Log සහ Chart දෙකටම යයි. ඊළඟ දිනවලටත් එයම තියෙනවා.", "trend_up": "ධාරාව ඉහළ යයි", "healthy": "හොඳයි", "needs_check": "පරීක්ෂා කරන්න", "critical": "බරපතලයි", "days": "දින", "readings": "කියවීම්", "last_reading": "අවසන් කියවීම", "no_machines": "යන්ත්‍ර නැත. යන්ත්‍ර අංකයක් දී එකතු කරන්න.", "no_companies": "පාරිභෝගික සමාගම් නැත.", "password_optional": "මුරපදය වෙනස් නොකරන්නේ නම් හිස්ව තබන්න.", "active": "සක්‍රිය", "save_log": "දෛනික ලොග් සුරකින්න", "pick_date": "ලොග් දිනය", "ampere": "ඇම්පියර්", "attached_machines": "එකතු කළ යන්ත්‍ර", "machine_chart": "මෙම යන්ත්‍රයේ ඇම්පියර් ප්‍රස්ථාරය", "open_log": "දෛනික ලොග් + ප්‍රස්ථාරය", "linked_help": "Machines පිටුවෙන් එකතු කළ යන්ත්‍ර විතරක් Daily Log සහ Chart වලට යනවා.", "hp_auto_fill": "HP එක දාම 400V 3-phase standard Unload / Loading / Overload Amper auto-fill වෙනවා. ඕන නම් වෙනස් කරන්න පුළුවන්.", "std_hint": "Std 400V 3φ", "legend_ok": "සාමාන්‍ය", "legend_high": "වැඩියි", "legend_too_high": "ඉතා වැඩියි", "legend_low": "අඩුයි", "legend_too_low": "ඉතා අඩුයි", "color_legend": "මෙම HP එකේ standard එකට සාපේක්ෂව", "ton_output": "අද ton output", "target_tons": "දිනකට target (ton)", "run_hours": "Run hours", "got_output": "අද output එක ගත්තා", "got_yes": "ඔව්", "not_yet": "තවම නැහැ", "no_target": "Target නැහැ", "today_tons": "අද tons", "today_hours": "අද run hours", "today_production": "අද නිෂ්පාදනය", "hours_unit": "h", "tons_unit": "T", "on_off": "On – Off", "plant_tons_help": "Ton සහ run hours මෙම යන්ත්‍ර සියල්ලේ එකතුවෙන්. යන්ත්‍රයෙන් යන්ත්‍රයට දාන්න එපා.", "plant_target_hint": "මෙම සමාගමේ යන්ත්‍ර සියල්ලටම එක daily target එකක්", "save_production": "Tons සුරකින්න", "select_company_first": "පළමුව සමාගමක් තෝරන්න.", "view_date": "දිනය තෝරන්න", "day_tons": "ඒ දිනයේ tons", "day_hours": "ඒ දිනයේ run hours", "day_production": "ඒ දිනයේ නිෂ්පාදනය", "got_day_output": "ඒ දිනයේ output එක ගත්තා", "day_logs": "ඒ දිනයේ ලොග්", "ton_output_day": "මෙම දිනයේ ton output", "back_today": "අද", "view_month": "මාසය තෝරන්න", "month_tons": "මෙම මාසයේ tons", "month_hours": "මෙම මාසයේ run hours", "month_target": "මාසයේ target (ton)", "month_production": "මෙම මාසයේ නිෂ්පාදනය", "got_month_output": "මෙම මාසයේ output එක ගත්තා", "days_recorded": "Tons තියෙන දින", "this_month": "මේ මාසය", "enter_day_prod": "තෝරාගත් දිනයට tons සහ run hours දාන්න", "all_machines_chart": "යන්ත්‍ර සියල්ල එකතුව", "this_machine_chart": "තෝරාගත් යන්ත්‍රය", "monthly_loading": "මෙම මාසයේ On Loading Amper", "all_avg": "සියලු යන්ත්‍ර සාමාන්‍ය", "search_month": "සොයන්න", "download_pdf": "PDF බාගන්න", "downloading_pdf": "PDF සූදානම් වෙමින්…", "day_amp_compare": "අද සහ ඊළඟ දිනයේ Amper", "prev_log_day": "කලින් දිනය", "next_log_day": "ඊළඟ දිනය", "same_amp": "සමානයි", "amp_changed": "සමාන නැහැ", "no_day_compare": "සසඳන්නට දින දෙකක ලොග් ඕනෑ.", "vs_prev_day": "කලින් දිනයට සාපේක්ෂව", }, } FINDINGS = { "empty_high": { "en": "Empty (no-load) current is high. Check bearings, winding condition, or shaft alignment.", "si": "Empty Amper වැඩියි. බෙයරිං, වයරිං, හෝ shaft alignment පරීක්ෂා කරන්න.", }, "empty_low": { "en": "Empty current dropped suddenly. Check supply voltage, loose connections, or a phase issue.", "si": "Empty Amper හදිසියේ අඩු වී ඇත. වෝල්ටීයතාව, සම්බන්ධතා, හෝ phase ගැටලුවක් බලන්න.", }, "loading_high": { "en": "Loading current is high. Check mechanical load, belts, jams, blunt tools, or lubrication.", "si": "Loading Amper වැඩියි. යාන්ත්‍රික බර, බෙල්ට්, jam, මෙවලම් ගෙවීම හෝ තෙල් ගැලීම බලන්න.", }, "loading_low": { "en": "Loading current is unusually low. Confirm the machine is actually under load.", "si": "Loading Amper අසාමාන්‍ය ලෙස අඩුයි. යන්ත්‍රයට බර තියෙනවද බලන්න.", }, "overload_high": { "en": "Overload current is too high. Stop the machine and inspect before running again.", "si": "Overload Amper ඉතා වැඩියි. යන්ත්‍රය නවත්වා පරීක්ෂා කරන්න.", }, "on_off_spread": { "en": "Machine On and Off readings differ a lot. Recheck the measurement or look for a developing fault.", "si": "On සහ Off කියවීම් අතර වෙනස වැඩියි. මැනීම නැවත බලන්න හෝ ගැටලුවක් හැදෙනවද බලන්න.", }, "rising": { "en": "Ampere has been rising over recent days. Plan maintenance before a breakdown.", "si": "පසුගිය දිනවල ඇම්පියර් ඉහළ යයි. කැඩීමකට කලින් නඩත්තුවක් සැලසුම් කරන්න.", }, "day_mismatch": { "en": "Today's ampere is not the same as the previous day's. Check the machine.", "si": "අද Amper එක කලින් දිනයේ Amper එකට සමාන නැහැ. යන්ත්‍රය පරීක්ෂා කරන්න.", }, } def t(key: str) -> str: lang = session.get("lang", "en") return I18N.get(lang, I18N["en"]).get(key, I18N["en"].get(key, key)) def finding_text(code: str) -> str: lang = session.get("lang", "en") item = FINDINGS.get(code, {}) return item.get(lang) or item.get("en") or code def fmt_date(value: str | date | None) -> str: if not value: return "—" if isinstance(value, date): return value.strftime("%Y.%m.%d") try: return datetime.strptime(str(value)[:10], "%Y-%m-%d").strftime("%Y.%m.%d") except ValueError: return str(value) def parse_iso_date(value) -> str: raw = str(value or "").strip()[:10] try: return date.fromisoformat(raw).isoformat() except ValueError: return date.today().isoformat() def parse_float(value) -> float | None: if value is None: return None text = str(value).strip() if text == "": return None try: return round(float(text), 2) except ValueError: return None # Typical 3-phase 400/415V 50Hz induction motor full-load current (Loading Amper). # Sources: IEC/industrial 415V FLA charts and UL 508A 380-415V table. FLA_400V3 = { 0.25: 0.6, 0.5: 1.1, 0.75: 1.5, 1: 1.8, 1.5: 2.6, 2: 3.4, 3: 4.8, 4: 6.6, 5: 7.6, 5.5: 8.5, 7.5: 11.0, 10: 14.8, 12.5: 18.5, 15: 21.0, 20: 28.0, 25: 34.0, 30: 41.0, 40: 55.0, 50: 68.0, 60: 82.0, 75: 99.0, 100: 135.0, } def typical_loading(hp: float) -> float: keys = sorted(FLA_400V3) if hp in FLA_400V3: return FLA_400V3[hp] if hp <= keys[0]: return round(FLA_400V3[keys[0]] * hp / keys[0], 2) if hp >= keys[-1]: return round(FLA_400V3[keys[-1]] * hp / keys[-1], 2) for low, high in zip(keys, keys[1:]): if low <= hp <= high: ratio = (hp - low) / (high - low) return round(FLA_400V3[low] + ratio * (FLA_400V3[high] - FLA_400V3[low]), 2) return round(hp * 1.6, 2) def typical_amperes(hp: float | None) -> dict: if not hp or hp <= 0: return {"empty": None, "loading": None, "overload": None} loading = typical_loading(float(hp)) return { "empty": round(loading * 0.40, 2), "loading": round(loading, 2), "overload": round(loading * 1.25, 2), } def expected_for_motor(motor) -> dict: typ = typical_amperes(motor["hp"]) return { "empty": motor["expected_empty"] if motor["expected_empty"] is not None else typ["empty"], "loading": motor["expected_loading"] if motor["expected_loading"] is not None else typ["loading"], "overload": motor["expected_overload"] if motor["expected_overload"] is not None else typ["overload"], } def amp_delta_status(value, expected) -> str: if value is None or expected is None or expected == 0: return "unknown" pct = (float(value) - float(expected)) / float(expected) * 100 if pct >= 15: return "high-alert" if pct >= 8: return "high-watch" if pct <= -15: return "low-alert" if pct <= -8: return "low-watch" return "ok" def hours_from_times(on_time, off_time) -> float | None: if not on_time or not off_time: return None try: start = datetime.strptime(str(on_time)[:5], "%H:%M") end = datetime.strptime(str(off_time)[:5], "%H:%M") except ValueError: return None hours = (end - start).total_seconds() / 3600 if hours < 0: hours += 24 return round(hours, 2) def fmt_hours(value) -> str: if value is None or value == "": return "—" try: n = float(value) except (TypeError, ValueError): return "—" if n == int(n): return f"{int(n)} h" return f"{n:.1f} h" def fmt_tons(value) -> str: if value is None or value == "": return "—" try: n = float(value) except (TypeError, ValueError): return "—" if n == int(n): return f"{int(n)} T" return f"{n:.1f} T" def upsert_daily_production(db, company_id: int, log_date: str, ton_output, run_hours=None, on_time="", off_time=""): on_time = (on_time or "").strip() off_time = (off_time or "").strip() existing = db.execute( "SELECT id FROM daily_production WHERE company_id = ? AND log_date = ?", (company_id, log_date), ).fetchone() empty = ton_output is None and run_hours is None and not on_time and not off_time if empty: if existing: db.execute("DELETE FROM daily_production WHERE id = ?", (existing["id"],)) return if existing: db.execute( """ UPDATE daily_production SET ton_output = ?, run_hours = ?, on_time = ?, off_time = ? WHERE id = ? """, (ton_output, run_hours, on_time, off_time, existing["id"]), ) else: db.execute( """ INSERT INTO daily_production (company_id, log_date, ton_output, run_hours, on_time, off_time) VALUES (?, ?, ?, ?, ?, ?) """, (company_id, log_date, ton_output, run_hours, on_time, off_time), ) def plant_production(db, company_id, log_date: str): empty = {"tons": None, "hours": None, "on_time": "", "off_time": "", "target": None} if company_id: prod = db.execute( "SELECT * FROM daily_production WHERE company_id = ? AND log_date = ?", (company_id, log_date), ).fetchone() company = db.execute( "SELECT target_tons FROM companies WHERE id = ?", (company_id,), ).fetchone() if prod: empty["tons"] = prod["ton_output"] empty["hours"] = prod["run_hours"] empty["on_time"] = prod["on_time"] or "" empty["off_time"] = prod["off_time"] or "" empty["target"] = company["target_tons"] if company else None return empty prod = db.execute( """ SELECT SUM(ton_output) AS tons, SUM(run_hours) AS hours FROM daily_production WHERE log_date = ? """, (log_date,), ).fetchone() target_row = db.execute("SELECT SUM(target_tons) AS n FROM companies").fetchone() empty["tons"] = prod["tons"] if prod else None empty["hours"] = prod["hours"] if prod else None empty["target"] = target_row["n"] if target_row else None return empty def parse_year_month(value): raw = str(value or "").strip() try: year_s, month_s = raw.split("-")[:2] year, month = int(year_s), int(month_s) if 1 <= month <= 12 and year >= 2000: return year, month except (TypeError, ValueError): pass today = date.today() return today.year, today.month def month_last_day(year: int, month: int) -> date: if month == 12: return date(year, 12, 31) return date(year, month + 1, 1) - timedelta(days=1) def calendar_month_range(year: int, month: int): return date(year, month, 1), month_last_day(year, month) def month_day_range(year: int, month: int): start = date(year, month, 1) last = month_last_day(year, month) today = date.today() if (year, month) == (today.year, today.month): return start, min(last, today) return start, last def month_overview(db, company_id, year: int, month: int): start, end = month_day_range(year, month) target = None if company_id: company = db.execute( "SELECT target_tons FROM companies WHERE id = ?", (company_id,), ).fetchone() target = company["target_tons"] if company else None recs = db.execute( """ SELECT log_date, ton_output, run_hours, on_time, off_time FROM daily_production WHERE company_id = ? AND log_date >= ? AND log_date <= ? ORDER BY log_date """, (company_id, start.isoformat(), end.isoformat()), ).fetchall() else: recs = db.execute( """ SELECT log_date, SUM(ton_output) AS ton_output, SUM(run_hours) AS run_hours FROM daily_production WHERE log_date >= ? AND log_date <= ? GROUP BY log_date ORDER BY log_date """, (start.isoformat(), end.isoformat()), ).fetchall() summed = db.execute("SELECT SUM(target_tons) AS n FROM companies").fetchone() target = summed["n"] if summed else None by_date = {str(r["log_date"])[:10]: r for r in recs} days = [] tons_total = 0.0 hours_total = 0.0 has_tons = False has_hours = False recorded = 0 met = 0 cursor = start while cursor <= end: rec = by_date.get(cursor.isoformat()) tons = rec["ton_output"] if rec else None hours = rec["run_hours"] if rec else None on_time = "" off_time = "" if rec is not None: keys = rec.keys() if "on_time" in keys: on_time = rec["on_time"] or "" if "off_time" in keys: off_time = rec["off_time"] or "" if tons is not None: tons_total += float(tons) has_tons = True recorded += 1 if output_met(tons, target) == "got_output": met += 1 if hours is not None: hours_total += float(hours) has_hours = True days.append( { "log_date": cursor.isoformat(), "display": fmt_date(cursor), "tons": tons, "hours": hours, "on_time": on_time, "off_time": off_time, "met": output_met(tons, target), "is_today": cursor == date.today(), } ) cursor += timedelta(days=1) day_count = (end - start).days + 1 month_target = float(target) * day_count if target is not None else None return { "days": days, "tons": tons_total if has_tons else None, "hours": hours_total if has_hours else None, "daily_target": target, "month_target": month_target, "recorded": recorded, "met": met, "day_count": day_count, } def output_met(actual, target) -> str: if target is None: return "no_target" if actual is None: return "not_yet" return "got_output" if float(actual) + 1e-9 >= float(target) else "not_yet" def ensure_column(db, table: str, name: str, spec: str): cols = {row[1] for row in db.execute(f"PRAGMA table_info({table})")} if name not in cols: db.execute(f"ALTER TABLE {table} ADD COLUMN {name} {spec}") def migrate_user_roles(db): row = db.execute( "SELECT sql FROM sqlite_master WHERE type='table' AND name='users'" ).fetchone() sql = (row[0] if row else "") or "" if "operator" in sql: return db.execute("PRAGMA foreign_keys = OFF") db.executescript( """ CREATE TABLE users_mig ( id INTEGER PRIMARY KEY AUTOINCREMENT, company_id INTEGER, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, full_name TEXT DEFAULT '', role TEXT NOT NULL CHECK(role IN ('admin', 'company', 'operator')), active INTEGER NOT NULL DEFAULT 1, FOREIGN KEY (company_id) REFERENCES companies(id) ON DELETE CASCADE ); INSERT INTO users_mig (id, company_id, username, password_hash, full_name, role, active) SELECT id, company_id, username, password_hash, full_name, role, active FROM users; DROP TABLE users; ALTER TABLE users_mig RENAME TO users; """ ) db.execute("PRAGMA foreign_keys = ON") def fill_typical_if_missing(hp_val, empty, loading, overload): typ = typical_amperes(hp_val) return ( empty if empty is not None else typ["empty"], loading if loading is not None else typ["loading"], overload if overload is not None else typ["overload"], ) def get_db() -> sqlite3.Connection: if "db" not in g: DATA_DIR.mkdir(exist_ok=True) conn = sqlite3.connect(DB_PATH, timeout=30) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") conn.execute("PRAGMA journal_mode=WAL") g.db = conn return g.db @app.teardown_appcontext def close_db(_exc): db = g.pop("db", None) if db is not None: db.close() def init_db(): DATA_DIR.mkdir(exist_ok=True) db = sqlite3.connect(DB_PATH, timeout=30) db.execute("PRAGMA journal_mode=WAL") db.executescript( """ CREATE TABLE IF NOT EXISTS companies ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, contact_person TEXT DEFAULT '', phone TEXT DEFAULT '', address TEXT DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, company_id INTEGER, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, full_name TEXT DEFAULT '', role TEXT NOT NULL CHECK(role IN ('admin', 'company', 'operator')), active INTEGER NOT NULL DEFAULT 1, FOREIGN KEY (company_id) REFERENCES companies(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS machines ( id INTEGER PRIMARY KEY AUTOINCREMENT, company_id INTEGER NOT NULL, machine_number TEXT NOT NULL, location TEXT DEFAULT '', notes TEXT DEFAULT '', target_tons REAL, created_at TEXT NOT NULL, UNIQUE(company_id, machine_number), FOREIGN KEY (company_id) REFERENCES companies(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS motors ( id INTEGER PRIMARY KEY AUTOINCREMENT, machine_id INTEGER NOT NULL, hp REAL NOT NULL, label TEXT DEFAULT '', expected_empty REAL, expected_loading REAL, expected_overload REAL, FOREIGN KEY (machine_id) REFERENCES machines(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS daily_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, machine_id INTEGER NOT NULL, log_date TEXT NOT NULL, on_time TEXT DEFAULT '', off_time TEXT DEFAULT '', remarks TEXT DEFAULT '', ton_output REAL, run_hours REAL, created_at TEXT NOT NULL, UNIQUE(machine_id, log_date), FOREIGN KEY (machine_id) REFERENCES machines(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS readings ( id INTEGER PRIMARY KEY AUTOINCREMENT, log_id INTEGER NOT NULL, motor_id INTEGER NOT NULL, on_empty REAL, on_loading REAL, on_overload REAL, off_empty REAL, off_loading REAL, off_overload REAL, UNIQUE(log_id, motor_id), FOREIGN KEY (log_id) REFERENCES daily_logs(id) ON DELETE CASCADE, FOREIGN KEY (motor_id) REFERENCES motors(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS daily_production ( id INTEGER PRIMARY KEY AUTOINCREMENT, company_id INTEGER NOT NULL, log_date TEXT NOT NULL, ton_output REAL, run_hours REAL, on_time TEXT DEFAULT '', off_time TEXT DEFAULT '', UNIQUE(company_id, log_date), FOREIGN KEY (company_id) REFERENCES companies(id) ON DELETE CASCADE ); """ ) db.row_factory = sqlite3.Row ensure_column(db, "machines", "target_tons", "REAL") ensure_column(db, "companies", "target_tons", "REAL") ensure_column(db, "daily_logs", "ton_output", "REAL") ensure_column(db, "daily_logs", "run_hours", "REAL") db.execute( """ CREATE TABLE IF NOT EXISTS daily_production ( id INTEGER PRIMARY KEY AUTOINCREMENT, company_id INTEGER NOT NULL, log_date TEXT NOT NULL, ton_output REAL, run_hours REAL, on_time TEXT DEFAULT '', off_time TEXT DEFAULT '', UNIQUE(company_id, log_date), FOREIGN KEY (company_id) REFERENCES companies(id) ON DELETE CASCADE ) """ ) ensure_column(db, "daily_production", "run_hours", "REAL") ensure_column(db, "daily_production", "on_time", "TEXT") ensure_column(db, "daily_production", "off_time", "TEXT") migrate_user_roles(db) admin = db.execute("SELECT id FROM users WHERE username = 'admin'").fetchone() if not admin: db.execute( """ INSERT INTO users (company_id, username, password_hash, full_name, role, active) VALUES (NULL, 'admin', ?, 'DEPL Administrator', 'admin', 1) """, (generate_password_hash("DEPL@2026"),), ) db.execute( """ INSERT INTO companies (name, contact_person, phone, address, created_at) VALUES (?, ?, ?, ?, ?) """, ( "Demo Packing (Pvt) Ltd", "Nimal Perera", "077 123 4567", "Kandy Road, Kurunegala", datetime.now().isoformat(timespec="seconds"), ), ) company_id = db.execute("SELECT last_insert_rowid()").fetchone()[0] db.execute( """ INSERT INTO users (company_id, username, password_hash, full_name, role, active) VALUES (?, 'demo', ?, 'Demo Company User', 'company', 1) """, (company_id, generate_password_hash("demo123")), ) now = datetime.now().isoformat(timespec="seconds") db.execute( """ INSERT INTO machines (company_id, machine_number, location, notes, target_tons, created_at) VALUES (?, 'DP-01', 'Line 1', 'Main packing machine', 12, ?) """, (company_id, now), ) m1 = db.execute("SELECT last_insert_rowid()").fetchone()[0] db.execute( """ INSERT INTO motors (machine_id, hp, label, expected_empty, expected_loading, expected_overload) VALUES (?, 3, 'Main drive', 2.1, 4.8, 6.2) """, (m1,), ) db.execute( """ INSERT INTO machines (company_id, machine_number, location, notes, target_tons, created_at) VALUES (?, 'DP-02', 'Line 2', 'Sealer — sample rising current', 10, ?) """, (company_id, now), ) m2 = db.execute("SELECT last_insert_rowid()").fetchone()[0] db.execute( """ INSERT INTO motors (machine_id, hp, label, expected_empty, expected_loading, expected_overload) VALUES (?, 5, 'Sealer motor', 3.4, 7.6, 9.8) """, (m2,), ) db.commit() db.execute( """ UPDATE readings SET on_empty = ROUND(on_empty, 2), on_loading = ROUND(on_loading, 2), on_overload = ROUND(on_overload, 2), off_empty = ROUND(off_empty, 2), off_loading = ROUND(off_loading, 2), off_overload = ROUND(off_overload, 2) """ ) db.commit() db.execute("UPDATE machines SET target_tons = 12 WHERE machine_number = 'DP-01' AND target_tons IS NULL") db.execute("UPDATE machines SET target_tons = 10 WHERE machine_number = 'DP-02' AND target_tons IS NULL") for row in db.execute("SELECT id, machine_id, on_time, off_time, run_hours FROM daily_logs"): if row["run_hours"] is None: hours = hours_from_times(row["on_time"], row["off_time"]) if hours is not None: db.execute("UPDATE daily_logs SET run_hours = ? WHERE id = ?", (hours, row["id"])) for company in db.execute("SELECT id, target_tons FROM companies"): if company["target_tons"] is None: summed = db.execute( "SELECT SUM(target_tons) AS n FROM machines WHERE company_id = ?", (company["id"],), ).fetchone()["n"] if summed is not None: db.execute( "UPDATE companies SET target_tons = ? WHERE id = ?", (summed, company["id"]), ) for row in db.execute( """ SELECT m.company_id, d.log_date, SUM(d.ton_output) AS tons FROM daily_logs d JOIN machines m ON m.id = d.machine_id WHERE d.ton_output IS NOT NULL GROUP BY m.company_id, d.log_date """ ): exists = db.execute( "SELECT id FROM daily_production WHERE company_id = ? AND log_date = ?", (row["company_id"], row["log_date"]), ).fetchone() if not exists: db.execute( "INSERT INTO daily_production (company_id, log_date, ton_output) VALUES (?, ?, ?)", (row["company_id"], row["log_date"], row["tons"]), ) for row in db.execute( """ SELECT m.company_id, d.log_date, MAX(d.run_hours) AS hours FROM daily_logs d JOIN machines m ON m.id = d.machine_id GROUP BY m.company_id, d.log_date """ ): times = db.execute( """ SELECT d.on_time, d.off_time, d.run_hours FROM daily_logs d JOIN machines m ON m.id = d.machine_id WHERE m.company_id = ? AND d.log_date = ? ORDER BY d.run_hours DESC LIMIT 1 """, (row["company_id"], row["log_date"]), ).fetchone() on_time = times["on_time"] if times else "" off_time = times["off_time"] if times else "" hours = row["hours"] prod = db.execute( "SELECT id, run_hours FROM daily_production WHERE company_id = ? AND log_date = ?", (row["company_id"], row["log_date"]), ).fetchone() if prod: if prod["run_hours"] is None and hours is not None: db.execute( """ UPDATE daily_production SET run_hours = ?, on_time = ?, off_time = ? WHERE id = ? """, (hours, on_time or "", off_time or "", prod["id"]), ) elif hours is not None: db.execute( """ INSERT INTO daily_production (company_id, log_date, run_hours, on_time, off_time) VALUES (?, ?, ?, ?, ?) """, (row["company_id"], row["log_date"], hours, on_time or "", off_time or ""), ) db.commit() db.close() def current_user(): uid = session.get("user_id") if not uid: return None return get_db().execute( """ SELECT u.*, c.name AS company_name FROM users u LEFT JOIN companies c ON c.id = u.company_id WHERE u.id = ? AND u.active = 1 """, (uid,), ).fetchone() def login_required(view): @wraps(view) def wrapped(*args, **kwargs): if not current_user(): return redirect(url_for("login")) return view(*args, **kwargs) return wrapped def admin_required(view): @wraps(view) def wrapped(*args, **kwargs): user = current_user() if not user: return redirect(url_for("login")) if user["role"] != "admin": flash("Admin access only.", "error") return redirect(url_for("dashboard")) return view(*args, **kwargs) return wrapped OPERATOR_ENDPOINTS = {"logs_page", "logout", "set_lang"} def home_for(user): if user and user["role"] == "operator": return url_for("logs_page") return url_for("dashboard") @app.before_request def limit_operator_access(): if not request.endpoint or request.endpoint == "static": return None user = current_user() if not user or user["role"] != "operator": return None if request.endpoint not in OPERATOR_ENDPOINTS: flash(t("operator_only_log"), "error") return redirect(url_for("logs_page")) return None def company_scope_id(user) -> int | None: if not user: return None if user["role"] != "admin": return user["company_id"] if "company_id" in request.args: raw = request.args.get("company_id") if not raw: session.pop("company_id", None) return None try: cid = int(raw) except (TypeError, ValueError): return session.get("company_id") row = get_db().execute("SELECT id FROM companies WHERE id = ?", (cid,)).fetchone() if not row: session.pop("company_id", None) return None session["company_id"] = cid return cid cid = session.get("company_id") if not cid: return None row = get_db().execute("SELECT id FROM companies WHERE id = ?", (cid,)).fetchone() if not row: session.pop("company_id", None) return None return cid def visible_companies(user): db = get_db() if user["role"] == "admin": return db.execute("SELECT * FROM companies ORDER BY name").fetchall() return db.execute("SELECT * FROM companies WHERE id = ?", (user["company_id"],)).fetchall() def company_by_id(company_id: int | None): if not company_id: return None return get_db().execute("SELECT * FROM companies WHERE id = ?", (company_id,)).fetchone() def company_profile_summaries(user): db = get_db() rank = {"unknown": 0, "ok": 1, "watch": 2, "alert": 3} out = [] for company in visible_companies(user): machines = machines_for(user, company["id"]) worst = "unknown" alert_n = watch_n = ok_n = 0 for machine in machines: health = machine_health(machine) if rank.get(health["status"], 0) > rank.get(worst, 0): worst = health["status"] if health["status"] == "alert": alert_n += 1 elif health["status"] == "watch": watch_n += 1 elif health["status"] == "ok": ok_n += 1 login = db.execute( "SELECT username FROM users WHERE company_id = ? AND role = 'company'", (company["id"],), ).fetchone() out.append( { "company": company, "machine_count": len(machines), "status": worst if machines else "unknown", "alert_count": alert_n, "watch_count": watch_n, "ok_count": ok_n, "username": login["username"] if login else "", "operator_count": db.execute( "SELECT COUNT(*) AS n FROM users WHERE company_id = ? AND role = 'operator'", (company["id"],), ).fetchone()["n"], } ) return out def machines_for(user, company_id: int | None = None): db = get_db() if user["role"] == "admin": if company_id: return db.execute( """ SELECT m.*, c.name AS company_name FROM machines m JOIN companies c ON c.id = m.company_id WHERE m.company_id = ? ORDER BY m.machine_number """, (company_id,), ).fetchall() return db.execute( """ SELECT m.*, c.name AS company_name FROM machines m JOIN companies c ON c.id = m.company_id ORDER BY c.name, m.machine_number """ ).fetchall() return db.execute( """ SELECT m.*, c.name AS company_name FROM machines m JOIN companies c ON c.id = m.company_id WHERE m.company_id = ? ORDER BY m.machine_number """, (user["company_id"],), ).fetchall() def motors_for_machine(machine_id: int): return get_db().execute( "SELECT * FROM motors WHERE machine_id = ? ORDER BY hp, id", (machine_id,), ).fetchall() def can_access_machine(user, machine_id: int) -> bool: row = get_db().execute("SELECT * FROM machines WHERE id = ?", (machine_id,)).fetchone() if not row: return False if user["role"] == "admin": return True return row["company_id"] == user["company_id"] def pct_change(current: float | None, baseline: float | None) -> float | None: if current is None or baseline is None or baseline == 0: return None return round((current - baseline) / baseline * 100, 1) def classify(pct: float | None) -> str: if pct is None: return "unknown" if pct >= 15 or pct <= -15: return "alert" if pct >= 8 or pct <= -8: return "watch" return "ok" AMP_DAY_FIELDS = ("on_empty", "on_loading", "on_overload", "off_empty") def amps_equal(left, right) -> bool: if left is None or right is None: return False return abs(float(left) - float(right)) < 0.01 def day_field_status(prev, curr) -> str: if prev is None or curr is None: return "unknown" if amps_equal(prev, curr): return "ok" pct = pct_change(curr, prev) status = classify(pct) if status == "ok": return "watch" return status def reading_before(motor_id: int, before_date) -> dict | None: row = get_db().execute( """ SELECT r.*, d.log_date FROM readings r JOIN daily_logs d ON d.id = r.log_id WHERE r.motor_id = ? AND d.log_date < ? ORDER BY d.log_date DESC LIMIT 1 """, (motor_id, str(before_date)[:10]), ).fetchone() return dict(row) if row else None def compare_day_amps(prev, curr) -> dict: fields = {} changed = False worst = "unknown" rank = {"unknown": 0, "ok": 1, "watch": 2, "alert": 3} for field in AMP_DAY_FIELDS: left = prev[field] if prev else None right = curr[field] if curr else None status = day_field_status(left, right) if left is not None and right is not None and not amps_equal(left, right): changed = True fields[field] = {"prev": left, "curr": right, "status": status} if rank[status] > rank[worst]: worst = status if changed and worst in ("unknown", "ok"): worst = "watch" if not changed and worst == "unknown": worst = "ok" return { "changed": changed, "status": worst if changed else "ok", "fields": fields, "prev_date": prev["log_date"] if prev else None, "curr_date": curr["log_date"] if curr else None, } def machine_day_amp_compare(machine) -> dict: motors = motors_for_machine(machine["id"]) rows = [] worst = "ok" rank = {"unknown": 0, "ok": 1, "watch": 2, "alert": 3} for motor in motors: latest = latest_reading(motor["id"]) if not latest: continue curr = dict(latest) prev = reading_before(motor["id"], curr["log_date"]) compared = compare_day_amps(prev, curr) compared["motor"] = motor rows.append(compared) if compared["changed"] and rank[compared["status"]] > rank[worst]: worst = compared["status"] return { "status": worst, "changed": any(row["changed"] for row in rows), "rows": rows, } def history_values(motor_id: int, field: str, limit: int = 14) -> list[float]: rows = get_db().execute( f""" SELECT r.{field} AS val FROM readings r JOIN daily_logs d ON d.id = r.log_id WHERE r.motor_id = ? AND r.{field} IS NOT NULL ORDER BY d.log_date DESC LIMIT ? """, (motor_id, limit), ).fetchall() return [row["val"] for row in rows] def baseline_for(motor, field: str, reading_field: str) -> float | None: expected = motor[field] if expected is not None: return expected typ = expected_for_motor(motor) mapping = { "expected_empty": "empty", "expected_loading": "loading", "expected_overload": "overload", } if field in mapping and typ.get(mapping[field]) is not None: return typ[mapping[field]] hist = history_values(motor["id"], reading_field, 10) if len(hist) >= 3: older = hist[1:] return round(sum(older) / len(older), 2) return None def analyze_motor(motor, reading) -> dict: issues = [] worst = "ok" pairs = [ ("expected_empty", "on_empty", "empty_high", "empty_low"), ("expected_loading", "on_loading", "loading_high", "loading_low"), ("expected_overload", "on_overload", "overload_high", None), ] details = {} if not reading: return {"status": "unknown", "issues": [], "details": {}} if not isinstance(reading, dict): reading = dict(reading) for expected_field, value_field, high_code, low_code in pairs: current = reading[value_field] base = baseline_for(motor, expected_field, value_field) pct = pct_change(current, base) status = classify(pct) details[value_field] = { "value": current, "baseline": base, "pct": pct, "status": status, } if status == "alert": worst = "alert" if pct is not None and pct > 0: issues.append(high_code) elif low_code: issues.append(low_code) elif status == "watch" and worst != "alert": worst = "watch" if pct is not None and pct > 0: issues.append(high_code) if reading["on_loading"] is not None and reading["off_loading"] is not None: spread = abs(reading["on_loading"] - reading["off_loading"]) if reading["on_loading"] and spread / reading["on_loading"] >= 0.12: issues.append("on_off_spread") if worst == "ok": worst = "watch" loading_hist = list(reversed(history_values(motor["id"], "on_loading", 6))) if len(loading_hist) >= 4: first = sum(loading_hist[:2]) / 2 last = sum(loading_hist[-2:]) / 2 if first and (last - first) / first >= 0.08: issues.append("rising") if worst == "ok": worst = "watch" prev = reading_before(motor["id"], reading["log_date"]) if reading.get("log_date") else None if prev: day = compare_day_amps(prev, reading) if day["changed"]: issues.append("day_mismatch") rank = {"unknown": 0, "ok": 1, "watch": 2, "alert": 3} if rank[day["status"]] > rank[worst]: worst = day["status"] # unique issues preserving order seen = [] for code in issues: if code not in seen: seen.append(code) return {"status": worst, "issues": seen, "details": details} def latest_reading(motor_id: int): return get_db().execute( """ SELECT r.*, d.log_date, d.on_time, d.off_time, d.remarks, d.machine_id FROM readings r JOIN daily_logs d ON d.id = r.log_id WHERE r.motor_id = ? ORDER BY d.log_date DESC LIMIT 1 """, (motor_id,), ).fetchone() def machine_health(machine) -> dict: motors = motors_for_machine(machine["id"]) worst = "unknown" issues = [] motor_results = [] rank = {"unknown": 0, "ok": 1, "watch": 2, "alert": 3} for motor in motors: reading = latest_reading(motor["id"]) result = analyze_motor(motor, reading) motor_results.append({"motor": motor, "reading": reading, "result": result}) if rank[result["status"]] > rank[worst]: worst = result["status"] issues.extend(result["issues"]) if worst == "unknown" and motors: worst = "unknown" return {"status": worst, "issues": issues, "motors": motor_results} def machine_from_list(machine_list, selected_id: int | None): if not machine_list: return None if selected_id: for machine in machine_list: if machine["id"] == selected_id: return machine return machine_list[0] def load_series_for_machine(machine): health = machine_health(machine) series = [] db = get_db() for item in health["motors"]: motor = item["motor"] rows = db.execute( """ SELECT d.log_date, r.on_empty, r.on_loading, r.on_overload, r.off_empty, r.off_loading, r.off_overload FROM readings r JOIN daily_logs d ON d.id = r.log_id WHERE r.motor_id = ? ORDER BY d.log_date """, (motor["id"],), ).fetchall() series.append( { "motor": motor, "result": item["result"], "reading": item["reading"], "points": [dict(r) for r in rows], } ) return health, series @app.context_processor def inject_globals(): user = current_user() scope_id = company_scope_id(user) if user else None return { "t": t, "fmt_date": fmt_date, "fmt_hours": fmt_hours, "fmt_tons": fmt_tons, "lang": session.get("lang", "en"), "user": user, "today": date.today().isoformat(), "today_display": fmt_date(date.today()), "scope_company_id": scope_id, "active_company": company_by_id(scope_id) if user else None, "viewing_all": bool(user and user["role"] == "admin" and not scope_id), "nav_companies": visible_companies(user) if user else [], } @app.route("/lang/") def set_lang(code: str): session["lang"] = "si" if code == "si" else "en" return redirect(request.referrer or url_for("dashboard")) @app.route("/profile") @login_required def set_profile(): user = current_user() if user["role"] != "admin": return redirect(url_for("dashboard")) raw = request.args.get("company_id") if raw: try: cid = int(raw) except (TypeError, ValueError): cid = None row = get_db().execute("SELECT id FROM companies WHERE id = ?", (cid,)).fetchone() if cid else None if row: session["company_id"] = cid else: session.pop("company_id", None) else: session.pop("company_id", None) nxt = request.args.get("next") or url_for("dashboard") if not str(nxt).startswith("/"): nxt = url_for("dashboard") return redirect(nxt) @app.route("/login", methods=["GET", "POST"]) def login(): if current_user(): return redirect(home_for(current_user())) if request.method == "POST": username = (request.form.get("username") or "").strip() password = request.form.get("password") or "" row = get_db().execute( "SELECT * FROM users WHERE username = ? AND active = 1", (username,), ).fetchone() if row and check_password_hash(row["password_hash"], password): lang = session.get("lang", "en") session.clear() session["user_id"] = row["id"] session["lang"] = lang return redirect(home_for(row)) flash(t("invalid_login"), "error") return render_template("login.html") @app.route("/logout") def logout(): session.clear() return redirect(url_for("login")) @app.route("/") @login_required def dashboard(): user = current_user() company_id = company_scope_id(user) db = get_db() today = date.today() year, month = parse_year_month(request.values.get("month") or today.strftime("%Y-%m")) is_this_month = (year, month) == (today.year, today.month) machines = machines_for(user, company_id) cards = [] alerts = [] ok_count = watch_count = alert_count = 0 production = [] day_amp_rows = [] changed_count = 0 for machine in machines: health = machine_health(machine) day_amp = machine_day_amp_compare(machine) production.append({"machine": machine, "health": health}) cards.append({"machine": machine, "health": health, "day_amp": day_amp}) if day_amp["rows"]: day_amp_rows.append({"machine": machine, "day_amp": day_amp}) if day_amp["changed"]: changed_count += 1 if health["status"] == "ok": ok_count += 1 elif health["status"] == "watch": watch_count += 1 elif health["status"] == "alert": alert_count += 1 if health["status"] in ("watch", "alert"): alerts.append({"machine": machine, "health": health}) today_count = 0 if machines: placeholders = ",".join("?" * len(machines)) ids = [m["id"] for m in machines] today_count = db.execute( f""" SELECT COUNT(*) AS n FROM daily_logs WHERE log_date = ? AND machine_id IN ({placeholders}) """, [today.isoformat(), *ids], ).fetchone()["n"] overview = month_overview(db, company_id, year, month) overall_met = output_met(overview["tons"], overview["month_target"]) return render_template( "dashboard.html", cards=cards, alerts=alerts, production=production, ok_count=ok_count, watch_count=watch_count, alert_count=alert_count, today_count=today_count, month_days=overview["days"], month_tons=overview["tons"], month_hours=overview["hours"], month_target=overview["month_target"], daily_target=overview["daily_target"], days_recorded=overview["recorded"], days_met=overview["met"], overall_met=overall_met, machine_count=len(machines), companies=visible_companies(user), company_id=company_id, finding_text=finding_text, month_value=f"{year:04d}-{month:02d}", is_this_month=is_this_month, view_display=f"{year}.{month:02d}", day_amp_rows=day_amp_rows, changed_count=changed_count, profile_summaries=company_profile_summaries(user) if user["role"] == "admin" else [], ) @app.route("/machines", methods=["GET", "POST"]) @login_required def machines_page(): user = current_user() db = get_db() if request.method == "POST": company_id = company_scope_id(user) or user["company_id"] if user["role"] == "admin": company_id = int(request.form["company_id"]) machine_number = (request.form.get("machine_number") or "").strip() location = (request.form.get("location") or "").strip() notes = (request.form.get("notes") or "").strip() machine_id = request.form.get("machine_id") if not machine_number: flash("Machine number is required.", "error") return redirect(url_for("machines_page")) try: if machine_id: mid = int(machine_id) if not can_access_machine(user, mid): flash("Not allowed.", "error") return redirect(url_for("machines_page")) db.execute( """ UPDATE machines SET machine_number = ?, location = ?, notes = ? WHERE id = ? """, (machine_number, location, notes, mid), ) target_id = mid else: db.execute( """ INSERT INTO machines (company_id, machine_number, location, notes, created_at) VALUES (?, ?, ?, ?, ?) """, ( company_id, machine_number, location, notes, datetime.now().isoformat(timespec="seconds"), ), ) target_id = db.execute("SELECT last_insert_rowid()").fetchone()[0] motor_ids = request.form.getlist("motor_id") hps = request.form.getlist("motor_hp") labels = request.form.getlist("motor_label") empties = request.form.getlist("motor_empty") loadings = request.form.getlist("motor_loading") overloads = request.form.getlist("motor_overload") kept_ids = [] for motor_id, hp, label, empty, loading, overload in zip( motor_ids, hps, labels, empties, loadings, overloads ): hp_val = parse_float(hp) if hp_val is None: continue motor_label = (label or "").strip() or f"{hp_val:g}HP Motor" empty_v, loading_v, overload_v = fill_typical_if_missing( hp_val, parse_float(empty), parse_float(loading), parse_float(overload), ) values = ( hp_val, motor_label, empty_v, loading_v, overload_v, ) if str(motor_id).isdigit(): db.execute( """ UPDATE motors SET hp = ?, label = ?, expected_empty = ?, expected_loading = ?, expected_overload = ? WHERE id = ? AND machine_id = ? """, (*values, int(motor_id), target_id), ) kept_ids.append(int(motor_id)) else: db.execute( """ INSERT INTO motors (machine_id, hp, label, expected_empty, expected_loading, expected_overload) VALUES (?, ?, ?, ?, ?, ?) """, (target_id, *values), ) kept_ids.append(db.execute("SELECT last_insert_rowid()").fetchone()[0]) if not kept_ids: typ = typical_amperes(3) db.execute( """ INSERT INTO motors (machine_id, hp, label, expected_empty, expected_loading, expected_overload) VALUES (?, 3, '3HP Motor', ?, ?, ?) """, (target_id, typ["empty"], typ["loading"], typ["overload"]), ) else: existing = db.execute( "SELECT id FROM motors WHERE machine_id = ?", (target_id,), ).fetchall() for row in existing: if row["id"] not in kept_ids: db.execute("DELETE FROM motors WHERE id = ?", (row["id"],)) db.commit() flash(t("saved"), "ok") except sqlite3.IntegrityError: db.rollback() flash("That machine number already exists for this company.", "error") return redirect(url_for("machines_page", company_id=company_id if user["role"] == "admin" else None)) if not machine_id: return redirect( url_for( "logs_page", machine_id=target_id, company_id=company_id if user["role"] == "admin" else None, ) ) return redirect(url_for("machines_page", company_id=company_id if user["role"] == "admin" else None)) company_id = company_scope_id(user) rows = [] for machine in machines_for(user, company_id): rows.append({"machine": machine, "motors": motors_for_machine(machine["id"]), "health": machine_health(machine)}) return render_template( "machines.html", rows=rows, companies=visible_companies(user), company_id=company_id, finding_text=finding_text, fla_table_json=json.dumps(FLA_400V3), ) @app.route("/machines//json") @login_required def machine_json(machine_id: int): user = current_user() if not can_access_machine(user, machine_id): return jsonify({"error": "not allowed"}), 403 machine = get_db().execute("SELECT * FROM machines WHERE id = ?", (machine_id,)).fetchone() motors = [dict(m) for m in motors_for_machine(machine_id)] return jsonify({"machine": dict(machine), "motors": motors}) @app.route("/api/typical-amperes") @login_required def typical_amperes_api(): hp = parse_float(request.args.get("hp")) data = typical_amperes(hp) data["hp"] = hp data["supply"] = "400V 3-phase" return jsonify(data) @app.route("/machines//delete", methods=["POST"]) @login_required def delete_machine(machine_id: int): user = current_user() if not can_access_machine(user, machine_id): flash("Not allowed.", "error") return redirect(url_for("machines_page")) get_db().execute("DELETE FROM machines WHERE id = ?", (machine_id,)) get_db().commit() flash(t("deleted"), "ok") return redirect(url_for("machines_page")) @app.route("/logs", methods=["GET", "POST"]) @login_required def logs_page(): user = current_user() db = get_db() log_date = parse_iso_date(request.values.get("log_date") or date.today().isoformat()) company_id = company_scope_id(user) machine_list = machines_for(user, company_id) selected = machine_from_list(machine_list, request.values.get("machine_id", type=int)) if request.method == "POST" and request.form.get("intent") == "production": if user["role"] == "operator": flash(t("operator_only_log"), "error") return redirect(url_for("logs_page", log_date=log_date, machine_id=request.form.get("machine_id"))) cid = company_id if user["role"] == "admin": cid = request.form.get("company_id", type=int) or company_id if not cid: flash(t("select_company_first"), "error") return redirect(url_for("logs_page", log_date=log_date, machine_id=request.form.get("machine_id"))) target_tons = parse_float(request.form.get("target_tons")) ton_output = parse_float(request.form.get("ton_output")) on_time = (request.form.get("on_time") or "").strip() off_time = (request.form.get("off_time") or "").strip() run_hours = hours_from_times(on_time, off_time) db.execute("UPDATE companies SET target_tons = ? WHERE id = ?", (target_tons, cid)) upsert_daily_production(db, cid, log_date, ton_output, run_hours, on_time, off_time) db.commit() flash(t("saved"), "ok") return redirect( url_for( "logs_page", log_date=log_date, machine_id=request.form.get("machine_id"), company_id=cid if user["role"] == "admin" else None, ) ) if request.method == "POST": selected = machine_from_list(machine_list, request.form.get("machine_id", type=int)) if not selected: flash(t("need_machine"), "error") return redirect(url_for("logs_page", log_date=log_date, company_id=company_id)) mid = selected["id"] remarks = (request.form.get(f"remarks_{mid}") or "").strip() motors = motors_for_machine(mid) has_any = False values = {} for motor in motors: packed = {} for field in ("on_empty", "on_loading", "on_overload", "off_empty", "off_loading", "off_overload"): packed[field] = parse_float(request.form.get(f"{field}_{motor['id']}")) if packed[field] is not None: has_any = True values[motor["id"]] = packed existing = db.execute( "SELECT id FROM daily_logs WHERE machine_id = ? AND log_date = ?", (mid, log_date), ).fetchone() if not has_any and not remarks: if existing: db.execute("DELETE FROM daily_logs WHERE id = ?", (existing["id"],)) else: if existing: log_id = existing["id"] db.execute( "UPDATE daily_logs SET remarks = ? WHERE id = ?", (remarks, log_id), ) else: db.execute( """ INSERT INTO daily_logs (machine_id, log_date, remarks, created_at) VALUES (?, ?, ?, ?) """, ( mid, log_date, remarks, datetime.now().isoformat(timespec="seconds"), ), ) log_id = db.execute("SELECT last_insert_rowid()").fetchone()[0] for motor in motors: packed = values[motor["id"]] db.execute( "DELETE FROM readings WHERE log_id = ? AND motor_id = ?", (log_id, motor["id"]), ) db.execute( """ INSERT INTO readings (log_id, motor_id, on_empty, on_loading, on_overload, off_empty, off_loading, off_overload) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( log_id, motor["id"], packed["on_empty"], packed["on_loading"], packed["on_overload"], packed["off_empty"], packed["off_loading"], packed["off_overload"], ), ) db.commit() flash(t("saved"), "ok") return redirect( url_for( "logs_page", log_date=log_date, machine_id=mid, company_id=company_id, ) ) sheet = [] health = None series = [] log_day_changed = False log_prev_date = None if selected: log = db.execute( "SELECT * FROM daily_logs WHERE machine_id = ? AND log_date = ?", (selected["id"], log_date), ).fetchone() motor_rows = [] for motor in motors_for_machine(selected["id"]): reading = None if log: reading = db.execute( "SELECT * FROM readings WHERE log_id = ? AND motor_id = ?", (log["id"], motor["id"]), ).fetchone() exp = expected_for_motor(motor) cells = {} for field, key in ( ("on_empty", "empty"), ("on_loading", "loading"), ("on_overload", "overload"), ("off_empty", "empty"), ("off_loading", "loading"), ("off_overload", "overload"), ): current = reading[field] if reading else None cells[field] = amp_delta_status(current, exp[key]) prev = reading_before(motor["id"], log_date) curr_for_day = dict(reading) if reading else None if curr_for_day is not None: curr_for_day["log_date"] = log_date day_cmp = compare_day_amps(prev, curr_for_day) motor_rows.append( { "motor": motor, "reading": reading, "prev_reading": prev, "day_cmp": day_cmp, "analysis": analyze_motor( motor, {**dict(reading), "log_date": log_date} if reading else None, ), "expected": exp, "cells": cells, } ) sheet.append({"machine": selected, "log": log, "motors": motor_rows}) health, series = load_series_for_machine(selected) log_day_changed = any(m["day_cmp"]["changed"] for m in motor_rows) log_prev_date = next((m["day_cmp"]["prev_date"] for m in motor_rows if m["day_cmp"]["prev_date"]), None) plant = plant_production(db, company_id, log_date) return render_template( "logs.html", machines=machine_list, selected=selected, sheet=sheet, log_date=log_date, companies=visible_companies(user), company_id=company_id, health=health, series=series, series_json=json.dumps(series_for_chart(series), default=str), finding_text=finding_text, switch_endpoint="logs_page", prod_date=log_date, is_today=log_date == date.today().isoformat(), today_tons=plant["tons"], today_hours=plant["hours"], today_target=plant["target"], plant_on=plant["on_time"], plant_off=plant["off_time"], log_day_changed=log_day_changed, log_prev_date=log_prev_date, ) @app.route("/analysis") @login_required def analysis_page(): user = current_user() company_id = company_scope_id(user) machine_list = machines_for(user, company_id) selected = machine_from_list(machine_list, request.args.get("machine_id", type=int)) today = date.today() year, month = parse_year_month(request.args.get("month") or today.strftime("%Y-%m")) is_this_month = (year, month) == (today.year, today.month) health = None series = [] if selected: health, series = load_series_for_machine(selected) series = filter_series_month(series, year, month) combined = combined_loading_chart( machine_list, year, month, selected_id=selected["id"] if selected else None, ) machine_code = selected["machine_number"] if selected else "all" safe_code = "".join(ch if ch.isalnum() or ch in "-_" else "-" for ch in str(machine_code)) pdf_meta = { "title": "DEPL Machine Log — Monthly analysis", "machine": selected["machine_number"] if selected else "", "company": selected["company_name"] if selected else "", "location": selected["location"] if selected else "", "month": f"{year:04d}.{month:02d}", "filename": f"DEPL-{safe_code}-{year:04d}-{month:02d}.pdf", "cols": ["Date", "On Empty Amper", "On Loading Amper", "On Overload Amper", "Off Empty Amper"], } return render_template( "analysis.html", machines=machine_list, selected=selected, health=health, series=series, series_json=json.dumps(series_for_chart(series), default=str), combined_json=json.dumps(combined, default=str), combined_has_data=any(v is not None for ds in combined["datasets"] for v in ds["data"]), companies=visible_companies(user), company_id=company_id, finding_text=finding_text, switch_endpoint="analysis_page", month_value=f"{year:04d}-{month:02d}", is_this_month=is_this_month, view_display=f"{year}.{month:02d}", pdf_meta=pdf_meta, pdf_meta_json=json.dumps(pdf_meta), ) def series_for_chart(series): out = [] for item in series: motor = item["motor"] out.append( { "id": motor["id"], "label": f"{motor['hp']:g}HP {motor['label']}".strip(), "expected": { "empty": motor["expected_empty"], "loading": motor["expected_loading"], "overload": motor["expected_overload"], }, "dates": [p["log_date"] for p in item["points"]], "on_empty": [None if p["on_empty"] is None else round(p["on_empty"], 2) for p in item["points"]], "on_loading": [None if p["on_loading"] is None else round(p["on_loading"], 2) for p in item["points"]], "on_overload": [None if p["on_overload"] is None else round(p["on_overload"], 2) for p in item["points"]], "off_empty": [None if p["off_empty"] is None else round(p["off_empty"], 2) for p in item["points"]], "off_loading": [None if p["off_loading"] is None else round(p["off_loading"], 2) for p in item["points"]], "off_overload": [None if p["off_overload"] is None else round(p["off_overload"], 2) for p in item["points"]], } ) return out def filter_series_month(series, year: int, month: int): start, end = calendar_month_range(year, month) start_s, end_s = start.isoformat(), end.isoformat() out = [] for item in series: points = [ p for p in item["points"] if start_s <= str(p["log_date"])[:10] <= end_s ] out.append({**item, "points": points}) return out def avg_or_none(values): nums = [float(v) for v in values if v is not None] if not nums: return None return round(sum(nums) / len(nums), 2) def combined_loading_chart(machines, year: int, month: int, selected_id: int | None = None): start, end = calendar_month_range(year, month) dates = [] cursor = start while cursor <= end: dates.append(cursor.isoformat()) cursor += timedelta(days=1) colors = ["#1f4e79", "#9b2c2c", "#2f6f4e", "#c48a12", "#5b4b8a", "#0e7490", "#9a3412"] datasets = [] by_day_all = {day: [] for day in dates} for i, machine in enumerate(machines): _health, series = load_series_for_machine(machine) by_date = {day: [] for day in dates} for item in series: for point in item["points"]: day = str(point["log_date"])[:10] if day in by_date and point["on_loading"] is not None: by_date[day].append(point["on_loading"]) values = [avg_or_none(by_date[day]) for day in dates] for day, val in zip(dates, values): if val is not None: by_day_all[day].append(val) is_selected = selected_id is not None and machine["id"] == selected_id datasets.append( { "label": machine["machine_number"], "data": values, "borderColor": colors[i % len(colors)], "borderWidth": 3 if is_selected else 2, } ) if len(datasets) > 1: datasets.append( { "label": t("all_avg"), "data": [avg_or_none(by_day_all[day]) for day in dates], "borderColor": "#17324d", "borderDash": [6, 4], "borderWidth": 2, } ) keep_idx = [i for i, day in enumerate(dates) if by_day_all[day]] if keep_idx: dates = [dates[i] for i in keep_idx] for ds in datasets: ds["data"] = [ds["data"][i] for i in keep_idx] else: dates = [] for ds in datasets: ds["data"] = [] return {"dates": dates, "datasets": datasets} @app.route("/companies", methods=["GET", "POST"]) @admin_required def companies_page(): db = get_db() if request.method == "POST": name = (request.form.get("name") or "").strip() contact = (request.form.get("contact_person") or "").strip() phone = (request.form.get("phone") or "").strip() address = (request.form.get("address") or "").strip() target_tons = parse_float(request.form.get("target_tons")) username = (request.form.get("username") or "").strip() password = request.form.get("password") or "" full_name = (request.form.get("full_name") or "").strip() company_id = request.form.get("company_id") if not name: flash("Company name is required.", "error") return redirect(url_for("companies_page")) try: if company_id: cid = int(company_id) db.execute( """ UPDATE companies SET name = ?, contact_person = ?, phone = ?, address = ?, target_tons = ? WHERE id = ? """, (name, contact, phone, address, target_tons, cid), ) user_row = db.execute( "SELECT * FROM users WHERE company_id = ? AND role = 'company'", (cid,), ).fetchone() if user_row: if username: db.execute("UPDATE users SET username = ?, full_name = ? WHERE id = ?", (username, full_name, user_row["id"])) if password: db.execute( "UPDATE users SET password_hash = ? WHERE id = ?", (generate_password_hash(password), user_row["id"]), ) elif username and password: db.execute( """ INSERT INTO users (company_id, username, password_hash, full_name, role, active) VALUES (?, ?, ?, ?, 'company', 1) """, (cid, username, generate_password_hash(password), full_name), ) else: if not username or not password: flash("Username and password are required for a new company login.", "error") return redirect(url_for("companies_page")) db.execute( """ INSERT INTO companies (name, contact_person, phone, address, target_tons, created_at) VALUES (?, ?, ?, ?, ?, ?) """, (name, contact, phone, address, target_tons, datetime.now().isoformat(timespec="seconds")), ) cid = db.execute("SELECT last_insert_rowid()").fetchone()[0] db.execute( """ INSERT INTO users (company_id, username, password_hash, full_name, role, active) VALUES (?, ?, ?, ?, 'company', 1) """, (cid, username, generate_password_hash(password), full_name or name), ) db.commit() flash(t("saved"), "ok") if not company_id: session["company_id"] = cid return redirect(url_for("dashboard")) except sqlite3.IntegrityError: db.rollback() flash("That username is already taken.", "error") return redirect(url_for("companies_page")) rows = db.execute( """ SELECT c.*, u.username, u.full_name, (SELECT COUNT(*) FROM machines m WHERE m.company_id = c.id) AS machine_count FROM companies c LEFT JOIN users u ON u.company_id = c.id AND u.role = 'company' ORDER BY c.name """ ).fetchall() return render_template( "companies.html", rows=rows, profiles=company_profile_summaries(current_user()), ) @app.route("/operators", methods=["GET", "POST"]) @admin_required def operators_page(): user = current_user() company_id = company_scope_id(user) company = company_by_id(company_id) if not company: flash(t("select_company_first"), "error") return redirect(url_for("companies_page")) db = get_db() if request.method == "POST": full_name = (request.form.get("full_name") or "").strip() username = (request.form.get("username") or "").strip() password = request.form.get("password") or "" operator_id = request.form.get("operator_id") if operator_id: oid = int(operator_id) row = db.execute( "SELECT * FROM users WHERE id = ? AND company_id = ? AND role = 'operator'", (oid, company_id), ).fetchone() if not row: flash("Not allowed.", "error") return redirect(url_for("operators_page")) if username: db.execute( "UPDATE users SET username = ?, full_name = ? WHERE id = ?", (username, full_name, oid), ) else: db.execute("UPDATE users SET full_name = ? WHERE id = ?", (full_name, oid)) if password: db.execute( "UPDATE users SET password_hash = ? WHERE id = ?", (generate_password_hash(password), oid), ) else: if not username or not password: flash("Username and password are required.", "error") return redirect(url_for("operators_page")) db.execute( """ INSERT INTO users (company_id, username, password_hash, full_name, role, active) VALUES (?, ?, ?, ?, 'operator', 1) """, (company_id, username, generate_password_hash(password), full_name or username), ) try: db.commit() flash(t("saved"), "ok") except sqlite3.IntegrityError: db.rollback() flash("That username is already taken.", "error") return redirect(url_for("operators_page")) operators = db.execute( """ SELECT * FROM users WHERE company_id = ? AND role = 'operator' ORDER BY username """, (company_id,), ).fetchall() return render_template( "operators.html", company=company, operators=operators, ) @app.route("/operators//delete", methods=["POST"]) @admin_required def delete_operator(operator_id: int): company_id = company_scope_id(current_user()) if not company_id: flash(t("select_company_first"), "error") return redirect(url_for("companies_page")) db = get_db() db.execute( "DELETE FROM users WHERE id = ? AND company_id = ? AND role = 'operator'", (operator_id, company_id), ) db.commit() flash(t("deleted"), "ok") return redirect(url_for("operators_page")) @app.route("/companies//json") @admin_required def company_json(company_id: int): company = get_db().execute("SELECT * FROM companies WHERE id = ?", (company_id,)).fetchone() user_row = get_db().execute( "SELECT username, full_name FROM users WHERE company_id = ? AND role = 'company'", (company_id,), ).fetchone() if not company: return jsonify({"error": "missing"}), 404 payload = dict(company) payload["username"] = user_row["username"] if user_row else "" payload["full_name"] = user_row["full_name"] if user_row else "" return jsonify(payload) @app.route("/companies//delete", methods=["POST"]) @admin_required def delete_company(company_id: int): get_db().execute("DELETE FROM companies WHERE id = ?", (company_id,)) get_db().commit() if session.get("company_id") == company_id: session.pop("company_id", None) flash(t("deleted"), "ok") return redirect(url_for("companies_page")) init_db() if __name__ == "__main__": init_db() port = int(os.environ.get("PORT", "5050")) print("DEPL Machine Log -> http://127.0.0.1:%s" % port) if not PRODUCTION: print("Admin login: admin / DEPL@2026") print("Demo company: demo / demo123") app.run(host="127.0.0.1" if PRODUCTION else "0.0.0.0", port=port, debug=not PRODUCTION)