"""
Hikvision DS-K1A8503F - Core Module (v2 - with logging & rules)
================================================================
RULES ENFORCED:
R1: Lock prevention (18 POST/session, 9s cooldown)
R2: Security (no hardcoded passwords)
R3: Data integrity (UNIQUE keys, no delete)
R4: Full logging & audit trail
R5: Performance timing
"""
import requests
from requests.auth import HTTPDigestAuth
import time, re, os, json, sqlite3
from datetime import datetime, timedelta
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
from rules.project_rules import (
MAX_POSTS_PER_SESSION, SESSION_COOLDOWN_SECONDS,
MAX_CONSECUTIVE_ERRORS, LOCK_CHECK_BEFORE_SYNC,
PERFORMANCE_WARN_SYNC_DAY_MS, PERFORMANCE_WARN_QUERY_MS
)
from logger import log_action, log_audit, log_performance, Timer, timed
BASE_DIR = os.path.dirname(__file__)
CONFIG_PATH = os.path.join(BASE_DIR, 'config.json')
DB_PATH = os.path.join(BASE_DIR, 'data', 'attendance.db')
DEFAULT_CONFIG = {
"device_ip": "192.168.1.15",
"username": "admin",
"password": "digi111111",
"timezone": "+08:00",
"port": 80,
}
def load_config():
if os.path.exists(CONFIG_PATH):
with open(CONFIG_PATH, 'r', encoding='utf-8') as f:
cfg = {**DEFAULT_CONFIG, **json.load(f)}
log_action("CONFIG", f"Loaded from {CONFIG_PATH} (IP: {cfg['device_ip']})")
return cfg
log_action("CONFIG", "Using default config", "warning")
return DEFAULT_CONFIG.copy()
def save_config(cfg):
with open(CONFIG_PATH, 'w', encoding='utf-8') as f:
json.dump(cfg, f, indent=2, ensure_ascii=False)
# R2.4: Don't log password
safe = {k: ('***' if k == 'password' else v) for k, v in cfg.items()}
log_action("CONFIG", f"Saved: {safe}")
def init_db():
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
# R3.5: WAL mode -> Power Query / API server có thể đọc song song khi sync ghi
c.execute('PRAGMA journal_mode=WAL')
c.execute('PRAGMA synchronous=NORMAL')
# R3.1: UNIQUE key on raw_time + employee_no
c.execute('''CREATE TABLE IF NOT EXISTS employees (
employee_no TEXT PRIMARY KEY, name TEXT
)''')
c.execute('''CREATE TABLE IF NOT EXISTS attendance (
id INTEGER PRIMARY KEY AUTOINCREMENT,
date TEXT, time TEXT, employee_no TEXT, employee_name TEXT,
status TEXT, card_no TEXT, raw_time TEXT,
card_reader_no TEXT,
event_type INTEGER,
sub_event_type INTEGER,
current_verify_mode TEXT,
UNIQUE(raw_time, employee_no)
)''')
# Migration cho DB cũ: thêm cột nếu thiếu (idempotent)
c.execute("PRAGMA table_info(attendance)")
cols = {r[1] for r in c.fetchall()}
for col, typ in [('card_reader_no','TEXT'),('event_type','INTEGER'),
('sub_event_type','INTEGER'),('current_verify_mode','TEXT')]:
if col not in cols:
c.execute(f'ALTER TABLE attendance ADD COLUMN {col} {typ}')
c.execute('''CREATE TABLE IF NOT EXISTS sync_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sync_time TEXT, records_added INTEGER,
date_from TEXT, date_to TEXT,
status TEXT, message TEXT,
duration_ms REAL, sessions_used INTEGER, total_posts INTEGER
)''')
# R3.4: Indexes
c.execute('CREATE INDEX IF NOT EXISTS idx_att_date ON attendance(date)')
c.execute('CREATE INDEX IF NOT EXISTS idx_att_emp ON attendance(employee_no)')
conn.commit()
conn.close()
log_action("DB", f"Initialized: {DB_PATH}")
class HikvisionDevice:
"""
R1: Lock-safe device connector.
Enforces 18 POST/session, 9s cooldown, full audit trail.
"""
def __init__(self, config=None):
self.config = config or load_config()
self.base_url = f"http://{self.config['device_ip']}:{self.config.get('port', 80)}"
self.tz = self.config.get('timezone', '+08:00')
self.session = None
self.post_count = 0
self.session_count = 0
self.total_posts = 0
self.last_error = None
self.is_locked = False
log_action("DEVICE", f"Created connector: {self.base_url}")
def _new_session(self):
"""R1.1-R1.6: Create new session with full safety checks."""
if self.session:
self.session.close()
log_action("SESSION", f"Closed session #{self.session_count} after {self.post_count} POSTs. Cooling down {SESSION_COOLDOWN_SECONDS}s...")
time.sleep(SESSION_COOLDOWN_SECONDS) # R1.2
self.session = requests.Session()
self.session.auth = HTTPDigestAuth(self.config['username'], self.config['password'])
self.session.verify = False
self.post_count = 0
self.session_count += 1
# R1.6: Log session creation
log_action("SESSION", f"New session #{self.session_count}")
try:
resp = self.session.get(f"{self.base_url}/ISAPI/System/deviceInfo", timeout=15)
except Exception as e:
self.last_error = f"Connection failed: {e}"
log_action("SESSION", f"FAIL: {self.last_error}", "error")
return False
if resp.status_code == 401:
if 'lock' in str(resp.text):
m = re.search(r'(\d+)', resp.text)
wait = int(m.group(1)) if m else 0
self.is_locked = True
self.last_error = f"LOCKED {wait}s ({wait//60}m)"
# R1.3: Log lock event
log_audit("R1.3", f"Device LOCKED for {wait}s. Will NOT create new sessions.")
log_action("DEVICE", f"LOCKED: {wait}s remaining", "error")
return False
self.last_error = "Authentication failed"
log_audit("R2", "Auth failed - check credentials")
return False
self.is_locked = False
log_action("SESSION", f"Session #{self.session_count} authenticated OK")
return resp.status_code == 200
def _safe_post(self, path, body):
"""R1.1: POST with automatic session rotation at limit."""
if not self.session or self.post_count >= MAX_POSTS_PER_SESSION:
# R1.1: Log rotation reason
if self.post_count >= MAX_POSTS_PER_SESSION:
log_action("SESSION", f"Rotating: {self.post_count} >= {MAX_POSTS_PER_SESSION} limit")
if not self._new_session():
return None
try:
resp = self.session.post(f"{self.base_url}{path}", json=body, timeout=60)
self.post_count += 1
self.total_posts += 1
# R1.5: If 401, stop immediately
if resp.status_code == 401:
log_audit("R1.5", f"Got 401 at POST #{self.post_count}. Stopping immediately.")
log_action("DEVICE", "401 received - stopping to prevent lock", "error")
return resp
return resp
except Exception as e:
self.last_error = str(e)
log_action("DEVICE", f"POST error: {e}", "error")
return None
def test_connection(self):
"""Test device connectivity (1 GET only)."""
with Timer("test_connection"):
try:
s = requests.Session()
s.auth = HTTPDigestAuth(self.config['username'], self.config['password'])
s.verify = False
resp = s.get(f"{self.base_url}/ISAPI/System/deviceInfo", timeout=10)
s.close()
if resp.status_code == 200:
log_action("TEST", "Connection OK")
return True, "Connected OK!"
elif resp.status_code == 401:
if 'lock' in str(resp.text):
m = re.search(r'(\d+)', resp.text)
wait = int(m.group(1)) if m else 0
log_action("TEST", f"Device locked: {wait}s", "warning")
return False, f"LOCKED ({wait}s remaining)"
return False, "Wrong username/password"
return False, f"HTTP {resp.status_code}"
except Exception as e:
log_action("TEST", f"Error: {e}", "error")
return False, f"Connection error: {e}"
def fetch_employees(self):
"""Fetch employee list."""
log_action("FETCH", "Fetching employee list")
if not self._new_session():
return None
users = []
pos = 0
while True:
resp = self._safe_post("/ISAPI/AccessControl/UserInfo/Search?format=json",
{"UserInfoSearchCond": {"searchID": "u", "searchResultPosition": pos, "maxResults": 30}})
if not resp or resp.status_code != 200:
break
d = resp.json().get("UserInfoSearch", {})
ul = d.get("UserInfo", [])
if not ul:
break
for u in ul:
users.append({'employee_no': u.get('employeeNo',''), 'name': u.get('name','')})
pos += len(ul)
if pos >= d.get("totalMatches", 0):
break
log_action("FETCH", f"Got {len(users)} employees")
return users
def fetch_day(self, date_str):
"""Fetch attendance for one day.
FIX BUG: device DS-K1A8503F có hard limit ~10 records/response dù request
maxResults=200. Phải paginate qua searchResultPosition để lấy hết.
Sử dụng responseStatusStrg='MORE' và totalMatches để biết khi nào dừng.
"""
with Timer(f"fetch_day({date_str})", PERFORMANCE_WARN_SYNC_DAY_MS) as t:
records = []
status_map = {'checkIn':'Vào','checkOut':'Ra','breakOut':'Ra nghỉ',
'breakIn':'Vào nghỉ','overtimeIn':'Vào TC','overtimeOut':'Ra TC'}
pos = 0
page = 0
MAX_PAGES = 100 # an toàn — 100 pages × 10 = 1000 records/ngày
while page < MAX_PAGES:
body = {
"AcsEventCond": {
"searchID": f"d{date_str.replace('-','')}p{page}",
"searchResultPosition": pos,
"maxResults": 200,
"major": 0, "minor": 0,
"startTime": f"{date_str}T00:00:00{self.tz}",
"endTime": f"{date_str}T23:59:59{self.tz}",
}
}
resp = self._safe_post("/ISAPI/AccessControl/AcsEvent?format=json", body)
if not resp or resp.status_code != 200:
log_action("FETCH", f"{date_str} page {page} pos {pos}: HTTP fail", "warning")
return None if page == 0 else records
ae = resp.json().get("AcsEvent", {})
info_list = ae.get("InfoList", [])
total_matches = ae.get("totalMatches", 0)
status_str = ae.get("responseStatusStrg", "")
if not info_list:
break
for e in info_list:
raw_time = e.get('time', '')
date_part = raw_time.split('T')[0] if 'T' in raw_time else date_str
time_part = ''
if 'T' in raw_time:
tp = raw_time.split('T')[1]
time_part = tp.split('+')[0] if '+' in tp else tp
raw_status = e.get('attendanceStatus', '')
records.append({
'date': date_part, 'time': time_part,
'employee_no': e.get('employeeNoString', e.get('employeeNo', '')),
'employee_name': e.get('name', ''),
'status': status_map.get(raw_status, raw_status),
'card_no': e.get('cardNo', ''),
'raw_time': raw_time,
'card_reader_no': str(e.get('cardReaderNo', '') or ''),
'event_type': e.get('major'),
'sub_event_type': e.get('minor'),
'current_verify_mode': e.get('currentVerifyMode', ''),
})
pos += len(info_list)
page += 1
# Dieu kien dung: het MORE hoac da fetch du totalMatches
if status_str != "MORE":
break
if total_matches and pos >= total_matches:
break
log_action("FETCH", f"{date_str}: {len(records)} records ({page} pages)")
return records
def sync_date_range(self, date_from, date_to, callback=None):
"""
Sync attendance data with full logging & timing.
R1.4: Check lock before sync.
R3.2: INSERT OR IGNORE (no delete).
R4.1-4.3: Full log trail.
"""
sync_start = time.perf_counter()
log_action("SYNC", f"Starting sync: {date_from} → {date_to}")
# R1.4: Check lock status first
if LOCK_CHECK_BEFORE_SYNC:
ok, msg = self.test_connection()
if not ok:
log_audit("R1.4", f"Pre-sync check failed: {msg}")
return 0, 0, f"Pre-sync check failed: {msg}"
log_action("SYNC", "Pre-sync lock check: OK")
init_db()
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
current = datetime.strptime(date_from, '%Y-%m-%d')
end = datetime.strptime(date_to, '%Y-%m-%d')
total_records = 0
total_new = 0
errors = 0
day_timings = []
while current <= end:
date_str = current.strftime('%Y-%m-%d')
day_start = time.perf_counter()
records = self.fetch_day(date_str)
if records is None:
errors += 1
log_action("SYNC", f"{date_str}: ERROR (#{errors})", "warning")
if errors > MAX_CONSECUTIVE_ERRORS:
msg = f"Stopped: {errors} consecutive errors at {date_str}"
log_audit("R1.5", msg)
self._log_sync(conn, total_new, date_from, date_to, 'error', msg,
time.perf_counter() - sync_start)
conn.close()
return total_records, total_new, msg
current += timedelta(days=1)
continue
errors = 0
total_records += len(records)
# R3.2: INSERT OR IGNORE
for r in records:
try:
c.execute('''INSERT OR IGNORE INTO attendance
(date, time, employee_no, employee_name, status, card_no, raw_time,
card_reader_no, event_type, sub_event_type, current_verify_mode)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''',
(r['date'], r['time'], r['employee_no'], r['employee_name'],
r['status'], r['card_no'], r['raw_time'],
r.get('card_reader_no',''), r.get('event_type'),
r.get('sub_event_type'), r.get('current_verify_mode','')))
if c.rowcount > 0:
total_new += 1
except Exception:
pass
conn.commit()
day_ms = (time.perf_counter() - day_start) * 1000
day_timings.append(day_ms)
if callback:
callback(date_str, len(records), total_records)
current += timedelta(days=1)
time.sleep(0.05)
# Update employees
employees = self.fetch_employees()
if employees:
for emp in employees:
c.execute('INSERT OR REPLACE INTO employees VALUES (?, ?)', (emp['employee_no'], emp['name']))
conn.commit()
duration_ms = (time.perf_counter() - sync_start) * 1000
avg_day_ms = sum(day_timings) / len(day_timings) if day_timings else 0
msg = f"OK: {total_records} fetched, {total_new} new, {len(day_timings)} days"
log_action("SYNC", f"Complete: {msg} in {duration_ms:.0f}ms (avg {avg_day_ms:.0f}ms/day)")
log_performance("sync_total", duration_ms)
log_performance("sync_avg_per_day", avg_day_ms, PERFORMANCE_WARN_SYNC_DAY_MS)
self._log_sync(conn, total_new, date_from, date_to, 'ok', msg, duration_ms)
conn.close()
if self.session:
self.session.close()
self.session = None
return total_records, total_new, msg
def _log_sync(self, conn, records, date_from, date_to, status, message, duration_ms=0):
# R4.1: Log every sync
conn.cursor().execute(
'''INSERT INTO sync_log (sync_time, records_added, date_from, date_to,
status, message, duration_ms, sessions_used, total_posts)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)''',
(datetime.now().isoformat(), records, date_from, date_to,
status, message, duration_ms, self.session_count, self.total_posts))
conn.commit()
# === Database Query Functions ===
@timed("get_attendance", PERFORMANCE_WARN_QUERY_MS)
def get_attendance(date_from=None, date_to=None, employee_no=None):
"""R5.2: Must complete in <1s. R3.7: JOIN employees để fill tên NV nếu rỗng."""
init_db()
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
c = conn.cursor()
query = '''SELECT a.id, a.date, a.time, a.employee_no,
COALESCE(NULLIF(a.employee_name,''), e.name, '') AS employee_name,
a.status, a.card_no, a.raw_time,
a.card_reader_no, a.event_type, a.sub_event_type, a.current_verify_mode
FROM attendance a
LEFT JOIN employees e ON a.employee_no = e.employee_no
WHERE 1=1'''
params = []
if date_from: query += " AND a.date >= ?"; params.append(date_from)
if date_to: query += " AND a.date <= ?"; params.append(date_to)
if employee_no: query += " AND a.employee_no = ?"; params.append(employee_no)
query += " ORDER BY a.date, a.time"
c.execute(query, params)
rows = [dict(r) for r in c.fetchall()]
conn.close()
return rows
@timed("get_employees", PERFORMANCE_WARN_QUERY_MS)
def get_employees():
init_db()
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
c = conn.cursor()
c.execute("SELECT * FROM employees ORDER BY CAST(employee_no AS INTEGER)")
rows = [dict(r) for r in c.fetchall()]
conn.close()
return rows
def get_sync_logs():
init_db()
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
c = conn.cursor()
c.execute("SELECT * FROM sync_log ORDER BY id DESC LIMIT 20")
rows = [dict(r) for r in c.fetchall()]
conn.close()
return rows
@timed("get_stats", PERFORMANCE_WARN_QUERY_MS)
def get_stats():
init_db()
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
c.execute("SELECT COUNT(*) FROM attendance")
total = c.fetchone()[0]
c.execute("SELECT COUNT(DISTINCT employee_no) FROM attendance")
emp_count = c.fetchone()[0]
c.execute("SELECT MIN(date), MAX(date) FROM attendance")
row = c.fetchone()
date_from, date_to = row[0], row[1]
c.execute("SELECT COUNT(DISTINCT date) FROM attendance")
days = c.fetchone()[0]
c.execute("SELECT date, COUNT(*) FROM attendance WHERE date = date('now', 'localtime') GROUP BY date")
today_row = c.fetchone()
today_count = today_row[1] if today_row else 0
conn.close()
return {'total_records': total, 'employee_count': emp_count,
'date_from': date_from or '', 'date_to': date_to or '',
'total_days': days, 'today_count': today_count}