✅ SAFE — Excel reads from local cache only. 0% lock risk.
| Endpoint | Method | Lock Risk |
|---|---|---|
| /api/status | GET | 0% |
| /api/employees | GET | 0% |
| /api/attendance | GET | 0% |
| /api/attendance/today | GET | 0% |
| /api/attendance/month | GET | 0% |
| /api/sync | POST | Protected ⚠️ |
let
Source = Json.Document(Web.Contents("http://localhost:5000/api/attendance")),
data = Source[data],
table = Table.FromRecords(data),
renamed = Table.RenameColumns(table, {
{"date", "Ngay"}, {"time", "Gio"},
{"employee_no", "Ma_NV"}, {"employee_name", "Ten_NV"},
{"status", "Trang_thai"}
})
in renamed'''
@app.route('/api/status')
def api_status():
stats = get_stats()
return jsonify({'server': 'running', 'cache': stats,
'sync': {'running': sync_status['running'], 'last_sync': sync_status['last_sync']},
'lock_safety': 'GET endpoints read from local cache only. 0% device contact.'})
@app.route('/api/employees')
def api_employees():
e = get_employees()
return jsonify({'data': e, 'count': len(e)})
@app.route('/api/attendance')
def api_attendance():
df = request.args.get('from', '')
dt = request.args.get('to', '')
emp = request.args.get('employee', '')
rows = get_attendance(date_from=df or None, date_to=dt or None, employee_no=emp or None)
return jsonify({'data': rows, 'count': len(rows), 'filters': {'from': df, 'to': dt, 'employee': emp}})
@app.route('/api/attendance/today')
def api_today():
today = datetime.now().strftime('%Y-%m-%d')
rows = get_attendance(date_from=today, date_to=today)
return jsonify({'data': rows, 'count': len(rows), 'date': today})
@app.route('/api/attendance/month')
def api_month():
y = request.args.get('year', datetime.now().year, type=int)
m = request.args.get('month', datetime.now().month, type=int)
df = f"{y}-{m:02d}-01"
nxt = datetime(y + (1 if m == 12 else 0), (m % 12) + 1, 1)
dt = (nxt - timedelta(days=1)).strftime('%Y-%m-%d')
rows = get_attendance(date_from=df, date_to=dt)
return jsonify({'data': rows, 'count': len(rows), 'year': y, 'month': m})
@app.route('/api/sync', methods=['POST'])
def api_sync():
if sync_status['running']:
return jsonify({'error': 'Sync already running'}), 429
if sync_status['last_sync'] and time.time() - sync_status['last_sync'] < 30:
return jsonify({'error': 'Wait 30s between syncs', 'lock_warning': 'Rate limited'}), 429
data = request.get_json() or {}
df = data.get('from', datetime.now().strftime('%Y-%m-%d'))
dt = data.get('to', datetime.now().strftime('%Y-%m-%d'))
def do_sync():
sync_status['running'] = True
try:
dev = HikvisionDevice()
def prog(d, c, t): sync_status['progress'] = f'{d}: {c} (total: {t})'
total, new, msg = dev.sync_date_range(df, dt, prog)
sync_status['result'] = {'total': total, 'new': new, 'message': msg,
'sessions': dev.session_count, 'posts': dev.total_posts}
except Exception as e:
sync_status['result'] = {'error': str(e)}
finally:
sync_status['running'] = False; sync_status['last_sync'] = time.time()
threading.Thread(target=do_sync, daemon=True).start()
log_action("API", f"Sync started: {df} → {dt}")
return jsonify({'status': 'started', 'range': {'from': df, 'to': dt}})
@app.route('/api/sync/status')
def api_sync_status():
return jsonify(sync_status)
@app.route('/api/sync/logs')
def api_sync_logs():
return jsonify({'data': get_sync_logs()})
@app.route('/api/test-connection')
def api_test():
dev = HikvisionDevice()
ok, msg = dev.test_connection()
return jsonify({'connected': ok, 'message': msg})
if __name__ == '__main__':
init_db()
log_action("SERVER", f"Starting on {API_HOST}:{API_PORT}")
print("=" * 50)
print(f" Attendance API: http://{API_HOST}:{API_PORT}")
print(f" Lock Risk: 0% for GET endpoints")
print("=" * 50)
app.run(host=API_HOST, port=API_PORT, debug=False)