from flask import Flask, render_template_string, request, jsonify, session, redirect, url_for from flask_cors import CORS import sqlite3 import os from datetime import datetime, timedelta from functools import wraps app = Flask(__name__) # HuggingFace: SECRET_KEY env var မှ ဖတ်ပါ (fallback ပါ) app.secret_key = os.environ.get('SECRET_KEY', 'pos-secret-key-change-in-production') CORS(app) # HuggingFace Spaces: /data သည် persistent volume (container restart ကြည့်လည် ဒေတာ မပျောက်) # Local run: current directory ထဲမှာ pos.db ဖန်တီးမည် DATA_DIR = '/data' if os.path.exists('/data') and os.access('/data', os.W_OK) else '.' DB_PATH = os.path.join(DATA_DIR, 'pos.db') # ─── Inline HTML ───────────────────────────────────────────────────────────── LOGIN_HTML = r""" POS Login

Demo Accounts

👑 Admin
admin123
💰 Cashier
cashier123
""" CASHIER_HTML = r""" Cashier - POS
🛒

Cashier POS

...

🛒 Cart 0
🛍️Barcode Scan လုပ်ပြီး ကုန်ပစ္စည်း ထည့်ပါ
စုစုပေါင်းဈေး 0 ကျပ်
အကြွေ 0 ကျပ်
""" ADMIN_HTML = r""" Admin - POS Dashboard
Dashboard 🛒
💰
ယနေ့ ရောင်းရငွေ
-
-
📈
ဤသတ်ပတ် ရောင်းရငွေ
-
-
📦
ကုန်မျိုး
-
products
Top Seller
-
-
📊 ၇ ရက်အတွင်း ရောင်းချမှု
Loading...

🏆 ဤသတ်ပတ် အရောင်းကောင်းဆုံး

Loading...
""" # ─── Database Init ──────────────────────────────────────────────────────────── def get_db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): conn = get_db() c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, password TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'cashier' )''') c.execute('''CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, barcode TEXT UNIQUE NOT NULL, name TEXT NOT NULL, price REAL NOT NULL, stock INTEGER DEFAULT 0, created_at TEXT DEFAULT (datetime('now','localtime')) )''') c.execute('''CREATE TABLE IF NOT EXISTS sales ( id INTEGER PRIMARY KEY AUTOINCREMENT, total_amount REAL NOT NULL, cash_received REAL NOT NULL, change_amount REAL NOT NULL, cashier_id INTEGER, created_at TEXT DEFAULT (datetime('now','localtime')) )''') c.execute('''CREATE TABLE IF NOT EXISTS sale_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, sale_id INTEGER NOT NULL, product_id INTEGER NOT NULL, quantity INTEGER NOT NULL, price_at_time REAL NOT NULL, FOREIGN KEY (sale_id) REFERENCES sales(id), FOREIGN KEY (product_id) REFERENCES products(id) )''') # Default admin c.execute("SELECT id FROM users WHERE username='admin'") if not c.fetchone(): c.execute("INSERT INTO users (username, password, role) VALUES (?, ?, ?)", ('admin', 'admin123', 'admin')) # Default cashier c.execute("SELECT id FROM users WHERE username='cashier'") if not c.fetchone(): c.execute("INSERT INTO users (username, password, role) VALUES (?, ?, ?)", ('cashier', 'cashier123', 'cashier')) # Sample products c.execute("SELECT COUNT(*) FROM products") if c.fetchone()[0] == 0: sample_products = [ ('8850006111019', 'မာမာနူဒယ်လ် (ကြက်သား)', 500, 50), ('8852987122207', 'ကိုကာကိုလာ 330ml', 800, 100), ('8850006110302', 'ပေပ်စီ 500ml', 700, 80), ('8888888888881', 'ကြက်ဥ (တစ်လုံး)', 250, 200), ('8888888888882', 'ဆပ်ပြာတုံး (Lux)', 1200, 30), ('8888888888883', 'ဆန် ၁ပဲသား', 2500, 100), ('8888888888884', 'ကြက်သား ၁ပဲသား', 8000, 20), ('8888888888885', 'ငါးပိ (Shrimp Paste)', 1500, 40), ('8888888888886', 'သကြား ၁ပဲသား', 3000, 60), ('8888888888887', 'ကြော်ဆီ (Cooking Oil) 1L', 5500, 25), ] c.executemany("INSERT INTO products (barcode, name, price, stock) VALUES (?,?,?,?)", sample_products) conn.commit() conn.close() # ─── Auth Decorators ────────────────────────────────────────────────────────── def login_required(f): @wraps(f) def decorated(*args, **kwargs): if 'user_id' not in session: return jsonify({'error': 'Unauthorized'}), 401 return f(*args, **kwargs) return decorated def admin_required(f): @wraps(f) def decorated(*args, **kwargs): if 'user_id' not in session: return jsonify({'error': 'Unauthorized'}), 401 if session.get('role') != 'admin': return jsonify({'error': 'Admin only'}), 403 return f(*args, **kwargs) return decorated # ─── Page Routes ───────────────────────────────────────────────────────────── @app.route('/') def index(): if 'user_id' in session: if session.get('role') == 'admin': return redirect('/admin') return redirect('/cashier') return redirect('/login') @app.route('/login') def login_page(): return render_template_string(LOGIN_HTML) @app.route('/cashier') def cashier_page(): if 'user_id' not in session: return redirect('/login') return render_template_string(CASHIER_HTML) @app.route('/admin') def admin_page(): if 'user_id' not in session or session.get('role') != 'admin': return redirect('/login') return render_template_string(ADMIN_HTML) # ─── Auth API ───────────────────────────────────────────────────────────────── @app.route('/api/login', methods=['POST']) def login(): data = request.get_json() conn = get_db() user = conn.execute("SELECT * FROM users WHERE username=? AND password=?", (data['username'], data['password'])).fetchone() conn.close() if user: session['user_id'] = user['id'] session['username'] = user['username'] session['role'] = user['role'] return jsonify({'role': user['role'], 'username': user['username']}) return jsonify({'error': 'အသုံးပြုသူအမည် သို့မဟုတ် စကားဝှက် မမှန်ပါ'}), 401 @app.route('/api/logout', methods=['POST']) def logout(): session.clear() return jsonify({'ok': True}) @app.route('/api/me') def me(): if 'user_id' in session: return jsonify({'username': session['username'], 'role': session['role']}) return jsonify({'error': 'Not logged in'}), 401 # ─── Products API ───────────────────────────────────────────────────────────── @app.route('/api/products', methods=['GET']) @login_required def get_products(): conn = get_db() products = conn.execute("SELECT * FROM products ORDER BY name").fetchall() conn.close() return jsonify([dict(p) for p in products]) @app.route('/api/products/scan/', methods=['GET']) @login_required def scan_product(barcode): conn = get_db() product = conn.execute("SELECT * FROM products WHERE barcode=?", (barcode,)).fetchone() conn.close() if product: return jsonify(dict(product)) return jsonify({'error': 'product_not_found'}), 404 @app.route('/api/products', methods=['POST']) @admin_required def add_product(): data = request.get_json() try: conn = get_db() conn.execute("INSERT INTO products (barcode, name, price, stock) VALUES (?,?,?,?)", (data['barcode'], data['name'], float(data['price']), int(data.get('stock', 0)))) conn.commit() conn.close() return jsonify({'ok': True}) except sqlite3.IntegrityError: return jsonify({'error': 'Barcode ထပ်နေပါသည်'}), 400 @app.route('/api/products/', methods=['PUT']) @admin_required def update_product(pid): data = request.get_json() conn = get_db() conn.execute("UPDATE products SET barcode=?, name=?, price=?, stock=? WHERE id=?", (data['barcode'], data['name'], float(data['price']), int(data.get('stock', 0)), pid)) conn.commit() conn.close() return jsonify({'ok': True}) @app.route('/api/products/', methods=['DELETE']) @admin_required def delete_product(pid): conn = get_db() conn.execute("DELETE FROM products WHERE id=?", (pid,)) conn.commit() conn.close() return jsonify({'ok': True}) # ─── Sales API ──────────────────────────────────────────────────────────────── @app.route('/api/sales', methods=['POST']) @login_required def create_sale(): data = request.get_json() items = data.get('items', []) if not items: return jsonify({'error': 'Cart မှာ ကုန်ပစ္စည်း မပါပါ'}), 400 total = sum(i['price'] * i['quantity'] for i in items) cash = float(data.get('cash_received', total)) change = cash - total conn = get_db() cur = conn.execute( "INSERT INTO sales (total_amount, cash_received, change_amount, cashier_id) VALUES (?,?,?,?)", (total, cash, change, session['user_id']) ) sale_id = cur.lastrowid for item in items: conn.execute( "INSERT INTO sale_items (sale_id, product_id, quantity, price_at_time) VALUES (?,?,?,?)", (sale_id, item['product_id'], item['quantity'], item['price']) ) conn.commit() sale_row = conn.execute("SELECT * FROM sales WHERE id=?", (sale_id,)).fetchone() sale_items_rows = conn.execute(""" SELECT si.*, p.name FROM sale_items si JOIN products p ON si.product_id = p.id WHERE si.sale_id=? """, (sale_id,)).fetchall() conn.close() return jsonify({ 'sale_id': sale_id, 'total': total, 'cash': cash, 'change': change, 'created_at': dict(sale_row)['created_at'], 'items': [dict(r) for r in sale_items_rows] }) @app.route('/api/sales', methods=['GET']) @admin_required def get_sales(): period = request.args.get('period', 'today') conn = get_db() if period == 'today': where = "date(s.created_at) = date('now','localtime')" elif period == 'week': where = "s.created_at >= datetime('now','localtime','-7 days')" elif period == 'month': where = "s.created_at >= datetime('now','localtime','-30 days')" else: where = "1=1" sales = conn.execute(f""" SELECT s.*, u.username as cashier_name, COUNT(si.id) as item_count FROM sales s LEFT JOIN users u ON s.cashier_id = u.id LEFT JOIN sale_items si ON s.id = si.sale_id WHERE {where} GROUP BY s.id ORDER BY s.created_at DESC LIMIT 100 """).fetchall() summary = conn.execute(f""" SELECT COUNT(*) as total_sales, COALESCE(SUM(total_amount), 0) as total_revenue FROM sales s WHERE {where} """).fetchone() conn.close() return jsonify({ 'sales': [dict(s) for s in sales], 'summary': dict(summary) }) @app.route('/api/sales//items', methods=['GET']) @admin_required def get_sale_items(sid): conn = get_db() items = conn.execute(""" SELECT si.*, p.name, p.barcode FROM sale_items si JOIN products p ON si.product_id = p.id WHERE si.sale_id=? """, (sid,)).fetchall() conn.close() return jsonify([dict(i) for i in items]) @app.route('/api/dashboard', methods=['GET']) @admin_required def dashboard(): conn = get_db() today_sales = conn.execute(""" SELECT COALESCE(SUM(total_amount),0) as rev, COUNT(*) as cnt FROM sales WHERE date(created_at)=date('now','localtime') """).fetchone() week_sales = conn.execute(""" SELECT COALESCE(SUM(total_amount),0) as rev, COUNT(*) as cnt FROM sales WHERE created_at >= datetime('now','localtime','-7 days') """).fetchone() product_count = conn.execute("SELECT COUNT(*) as cnt FROM products").fetchone() top_products = conn.execute(""" SELECT p.name, SUM(si.quantity) as qty, SUM(si.quantity * si.price_at_time) as revenue FROM sale_items si JOIN products p ON si.product_id = p.id WHERE si.sale_id IN (SELECT id FROM sales WHERE created_at >= datetime('now','localtime','-7 days')) GROUP BY p.id ORDER BY qty DESC LIMIT 5 """).fetchall() daily_chart = conn.execute(""" SELECT date(created_at) as day, SUM(total_amount) as rev, COUNT(*) as cnt FROM sales WHERE created_at >= datetime('now','localtime','-7 days') GROUP BY day ORDER BY day """).fetchall() conn.close() return jsonify({ 'today': dict(today_sales), 'week': dict(week_sales), 'product_count': product_count['cnt'], 'top_products': [dict(p) for p in top_products], 'daily_chart': [dict(d) for d in daily_chart] }) if __name__ == '__main__': init_db() # HuggingFace Spaces requires port 7860 port = int(os.environ.get('PORT', 7860)) print(f"✅ POS App started at http://localhost:{port}") print(f" DB Path: {DB_PATH}") print(" Admin: admin / admin123") print(" Cashier: cashier / cashier123") app.run(host='0.0.0.0', port=port, debug=False)