""" Data Audit Script ================= Kiểm tra DỮ LIỆU (khác audit.py vốn kiểm CODE). Bao phủ: - Gap ngày (Mon-Sat; CN bỏ qua) - Records thiếu employee_no / employee_name - Orphan employee_no - NV không có attendance - status='undefined' (lỗi mapping) - Ngày Mon-Sat có < N records (suspicious) - sync_log gần đây có lỗi không Chạy: python audit_data.py """ import os, sys, io, sqlite3, json from datetime import datetime, timedelta # Force UTF-8 stdout (Windows console) try: sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') except Exception: pass BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DB_PATH = os.path.join(BASE_DIR, 'data', 'attendance.db') LOG_DIR = os.path.join(BASE_DIR, 'logs') os.makedirs(LOG_DIR, exist_ok=True) # Holidays VN (mở rộng theo nhu cầu) — gap đúng các ngày này KHÔNG báo lỗi VN_HOLIDAYS = { '2025-09-01', '2025-09-02', '2026-01-01', '2026-02-16', '2026-02-17', '2026-02-18', '2026-02-19', '2026-02-20', '2026-02-21', '2026-04-30', '2026-05-01', '2026-05-02', '2026-09-01', '2026-09-02', } SUSPICIOUS_DAILY_MIN = 3 # ngày Mon-Sat có < 3 scans → suspicious FINDINGS = [] # (severity, code, message) def add(severity, code, msg): FINDINGS.append((severity, code, msg)) icon = {'INFO': 'i', 'WARN': '!', 'ERROR': 'X', 'OK': '+'}[severity] print(f' [{icon}] {code}: {msg}') def run(): print('=' * 70) print(f'DATA AUDIT - {datetime.now().isoformat(timespec="seconds")}') print(f'DB: {DB_PATH}') print('=' * 70) if not os.path.exists(DB_PATH): add('ERROR', 'DB.0', f'Database not found: {DB_PATH}') return _finish() conn = sqlite3.connect(DB_PATH) c = conn.cursor() # Overview c.execute('SELECT COUNT(*) FROM attendance'); total = c.fetchone()[0] c.execute('SELECT COUNT(DISTINCT employee_no) FROM attendance'); ed = c.fetchone()[0] c.execute('SELECT COUNT(*) FROM employees'); etbl = c.fetchone()[0] c.execute('SELECT MIN(date), MAX(date), COUNT(DISTINCT date) FROM attendance') dmin, dmax, ddist = c.fetchone() print(f'\nRange: {dmin} -> {dmax} ({ddist} days, {total} records, {ed} distinct emp, employees table={etbl})') print('\n[1] Gap days (Mon-Sat, excluding holidays)') c.execute('SELECT DISTINCT date FROM attendance ORDER BY date') have = set(r[0] for r in c.fetchall()) if dmin and dmax: d0 = datetime.strptime(dmin, '%Y-%m-%d') d1 = datetime.strptime(dmax, '%Y-%m-%d') bad = [] cur = d0 while cur <= d1: s = cur.strftime('%Y-%m-%d') if cur.weekday() != 6 and s not in have and s not in VN_HOLIDAYS: bad.append(s) cur += timedelta(days=1) if not bad: add('OK', 'GAP.1', 'No unexplained gaps (Mon-Sat).') else: add('WARN', 'GAP.1', f'{len(bad)} gap days Mon-Sat not in holiday list') for s in bad[:30]: print(f' - {s}') if len(bad) > 30: print(f' ... and {len(bad)-30} more') print('\n[2] Missing fields') c.execute("SELECT COUNT(*) FROM attendance WHERE employee_no='' OR employee_no IS NULL") n = c.fetchone()[0] if n: add('WARN', 'MISS.NO', f'{n} records missing employee_no (suspicious scans / unknown person)') else: add('OK', 'MISS.NO', '0 records missing employee_no') c.execute("SELECT COUNT(*) FROM attendance WHERE employee_name='' OR employee_name IS NULL") n = c.fetchone()[0] if n: add('WARN', 'MISS.NAME', f'{n}/{total} records missing employee_name ' '-> use JOIN with employees when exporting (R3.7)') else: add('OK', 'MISS.NAME', '0 records missing employee_name') c.execute("SELECT COUNT(*) FROM attendance WHERE status='' OR status IS NULL") n = c.fetchone()[0] if n: add('WARN', 'MISS.ST', f'{n} records missing status') else: add('OK', 'MISS.ST', '0 records missing status') c.execute("SELECT COUNT(*) FROM attendance WHERE status='undefined'") n = c.fetchone()[0] if n: add('ERROR', 'STATUS.U', f'{n}/{total} records have status="undefined" ' '(status mapping broken or CSV import)') else: add('OK', 'STATUS.U', 'No status="undefined"') print('\n[3] Foreign-key / Orphans') c.execute('''SELECT a.employee_no, COUNT(*) FROM attendance a LEFT JOIN employees e ON a.employee_no = e.employee_no WHERE e.employee_no IS NULL AND a.employee_no != '' GROUP BY a.employee_no''') orphans = c.fetchall() if orphans: add('WARN', 'ORPH.1', f'{len(orphans)} orphan employee_no in attendance not in employees') for eno, cnt in orphans: print(f' - {eno!r}: {cnt} records') else: add('OK', 'ORPH.1', 'All employee_no resolve in employees table') c.execute('''SELECT e.employee_no, e.name FROM employees e LEFT JOIN (SELECT DISTINCT employee_no FROM attendance) a USING(employee_no) WHERE a.employee_no IS NULL''') no_attend = c.fetchall() if no_attend: add('INFO', 'NV.NOAT', f'{len(no_attend)}/{etbl} employees never appear in attendance ' '(may have left, new, or device cannot recognise)') print('\n[4] Duplicates (raw_time + employee_no)') c.execute('''SELECT raw_time, employee_no, COUNT(*) FROM attendance GROUP BY raw_time, employee_no HAVING COUNT(*) > 1''') dups = c.fetchall() if dups: add('ERROR', 'DUP.1', f'{len(dups)} duplicate keys (violates R3.1 UNIQUE)') else: add('OK', 'DUP.1', 'No duplicates (UNIQUE intact)') print('\n[5] Days with very low activity (Mon-Sat)') c.execute(f'''SELECT date, COUNT(*) FROM attendance GROUP BY date HAVING COUNT(*) < {SUSPICIOUS_DAILY_MIN} ORDER BY date''') low_all = c.fetchall() low = [(d, n) for d, n in low_all if datetime.strptime(d, '%Y-%m-%d').weekday() != 6 and d not in VN_HOLIDAYS] if low: add('WARN', 'LOW.1', f'{len(low)} working days with < {SUSPICIOUS_DAILY_MIN} scans') for d, n in low[:10]: print(f' - {d}: {n}') else: add('OK', 'LOW.1', f'No suspicious low-activity days') print('\n[6] Recent sync_log') c.execute('SELECT id, sync_time, status, message FROM sync_log ORDER BY id DESC LIMIT 5') syncs = c.fetchall() err = [s for s in syncs if s[2] != 'ok'] if err: add('ERROR', 'SYNC.1', f'{len(err)}/5 recent syncs failed') for s in err: print(f' - #{s[0]} {s[1]}: {s[2]} - {s[3]}') else: add('OK', 'SYNC.1', 'Last 5 syncs all OK') print('\n[7] DB hygiene') c.execute('PRAGMA journal_mode') jm = c.fetchone()[0] if jm != 'wal': add('WARN', 'DB.WAL', f'journal_mode={jm} (R3.5 prefers WAL for concurrent reads)') else: add('OK', 'DB.WAL', 'WAL mode enabled') c.execute("PRAGMA index_list('attendance')") idx = c.fetchall() have_idx = {r[1] for r in idx} needed = {'idx_att_date', 'idx_att_emp'} if needed - have_idx: add('WARN', 'DB.IDX', f'missing indexes: {needed - have_idx}') else: add('OK', 'DB.IDX', 'Required indexes present') conn.close() return _finish() def _finish(): print('\n' + '=' * 70) counts = {} for s, *_ in FINDINGS: counts[s] = counts.get(s, 0) + 1 line = ' '.join(f'{k}={v}' for k, v in sorted(counts.items())) print(f'SUMMARY: {line}') print('=' * 70) rpt = os.path.join(LOG_DIR, f'data_audit_{datetime.now().strftime("%Y%m%d_%H%M%S")}.txt') with open(rpt, 'w', encoding='utf-8') as f: f.write(f'DATA AUDIT - {datetime.now().isoformat()}\n') f.write('=' * 70 + '\n') for sev, code, msg in FINDINGS: f.write(f'[{sev}] {code}: {msg}\n') f.write('\n' + line + '\n') print(f'\nReport: {rpt}') # Exit non-zero if any ERROR return 0 if not any(s == 'ERROR' for s, *_ in FINDINGS) else 1 if __name__ == '__main__': sys.exit(run())