PasteShr

New Paste Browse

"""
Bridge SC105 / ZKTeco -> Laravel attendance API.

Mode input (ATTENDANCE_INPUT_MODE):
  - zk   : baca log absensi lewat LAN (pyzk, port 4370) — untuk SC105MF + kabel LAN
  - mock : uji tanpa mesin
  - serial : reader serial (COM port), bukan SC105 standar
"""

import contextlib
import io
import json
import os
import random
import sys
import time
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

import requests

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
LOG_FILE = os.getenv("BRIDGE_LOG_FILE", os.path.join(BASE_DIR, "bridge.log"))
_STATE_FILE_PATH = None
_MEMORY_STATE = {}

def load_simple_env(filepath):
    if not os.path.exists(filepath):
        return
    with open(filepath, "r", encoding="utf-8") as f:
        for raw in f:
            line = raw.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, value = line.split("=", 1)
            key = key.strip()
            value = value.strip().strip('"').strip("'")
            if key and key not in os.environ:
                os.environ[key] = value

load_simple_env(os.path.join(BASE_DIR, ".env.bridge"))

API_URL = os.getenv("ATTENDANCE_API_URL", "https://absencakra.online/api/attendance/device/logs")
DEVICE_TOKEN = os.getenv("ATTENDANCE_DEVICE_TOKEN", "")
DEVICE_CODE = os.getenv("ATTENDANCE_DEVICE_CODE", "SC105MF-01")
DEVICE_NAME = os.getenv("ATTENDANCE_DEVICE_NAME", "ZKTeco SC105MF LAN Bridge")
POLL_SECONDS = int(os.getenv("ATTENDANCE_POLL_SECONDS", "10"))
MOCK_MODE = os.getenv("ATTENDANCE_MOCK_MODE", "false").lower() == "true"
MOCK_CARD_UID = os.getenv("ATTENDANCE_MOCK_CARD_UID", "A1B2C3D4")
INPUT_MODE = os.getenv("ATTENDANCE_INPUT_MODE", "zk").lower()
SERIAL_PORT = os.getenv("ATTENDANCE_SERIAL_PORT", "COM3")
SERIAL_BAUDRATE = int(os.getenv("ATTENDANCE_SERIAL_BAUDRATE", "9600"))
SERIAL_TIMEOUT = float(os.getenv("ATTENDANCE_SERIAL_TIMEOUT", "0.5"))
ATTENDANCE_TIMEZONE = os.getenv("ATTENDANCE_TIMEZONE", "Asia/Jakarta")
ZK_DEVICE_IP = os.getenv("ATTENDANCE_DEVICE_IP", "192.168.1.201")
ZK_DEVICE_PORT = int(os.getenv("ATTENDANCE_DEVICE_PORT", "4370"))
ZK_DEVICE_PASSWORD = int(os.getenv("ATTENDANCE_DEVICE_PASSWORD", "0"))
ZK_TIMEOUT = int(os.getenv("ATTENDANCE_DEVICE_TIMEOUT", "10"))
# username = kirim PIN mesin (user_id) -> cocokkan users.username di Laravel
# card = kirim nomor kartu RFID dari mesin
UID_MODE = os.getenv("ATTENDANCE_UID_MODE", "pin").lower()

_SERIAL_CONN = None
_SCAN_TZ = None
_ZK_CONN = None
_PIN_CARD_CACHE = None

def log(level, message):
    line = f"[{datetime.now().isoformat()}] [{level}] {message}"
    print(line, flush=True)
    try:
        with open(LOG_FILE, "a", encoding="utf-8") as handle:
            handle.write(line + "\n")
    except Exception:
        pass

@contextlib.contextmanager
def silence_noisy_stdout():
    """pyzk kadang menulis progress dengan \\r — di redirect file jadi 1 huruf saja."""
    old_stdout = sys.stdout
    sys.stdout = io.StringIO()
    try:
        yield
    finally:
        sys.stdout = old_stdout

def resolve_scan_timezone():
    global _SCAN_TZ
    if _SCAN_TZ is not None:
        return _SCAN_TZ

try:
        _SCAN_TZ = ZoneInfo(ATTENDANCE_TIMEZONE)
        return _SCAN_TZ
    except Exception:
        if ATTENDANCE_TIMEZONE.lower() in ("asia/jakarta", "jakarta", "utc+7", "gmt+7", "wib"):
            _SCAN_TZ = timezone(timedelta(hours=7))
            log("WARN", "Timezone 'Asia/Jakarta' tidak tersedia di sistem, fallback ke UTC+7.")
            return _SCAN_TZ

_SCAN_TZ = datetime.now().astimezone().tzinfo
        log("WARN", f"Timezone '{ATTENDANCE_TIMEZONE}' tidak dikenali, fallback ke timezone lokal sistem.")
        return _SCAN_TZ

def _can_write_directory(directory):
    try:
        os.makedirs(directory, exist_ok=True)
        test_file = os.path.join(directory, ".bridge_write_test")
        with open(test_file, "w", encoding="utf-8") as handle:
            handle.write("ok")
        os.remove(test_file)
        return True
    except OSError:
        return False

def resolve_state_file():
    global _STATE_FILE_PATH
    if _STATE_FILE_PATH is not None:
        return _STATE_FILE_PATH

candidates = []
    custom = os.getenv("BRIDGE_STATE_FILE", "").strip()
    if custom:
        candidates.append(custom)

candidates.extend(
        [
            os.path.join(BASE_DIR, "bridge_state.json"),
            os.path.join(BASE_DIR, ".bridge_state.json"),
        ]
    )

local_app = os.environ.get("LOCALAPPDATA") or os.environ.get("APPDATA")
    if local_app:
        candidates.append(os.path.join(local_app, "AbsenCakraBridge", "bridge_state.json"))

for path in candidates:
        directory = os.path.dirname(os.path.abspath(path))
        if _can_write_directory(directory):
            _STATE_FILE_PATH = path
            log("INFO", f"State file: {path}")
            return path

_STATE_FILE_PATH = ""
    log(
        "WARN",
        "Tidak bisa menulis state file. Pakai mode memori saja "
        "(set BRIDGE_STATE_FILE=C:\\bridge-absencakra\\bridge_state.json di .env.bridge).",
    )
    return ""

def load_state():
    global _MEMORY_STATE

path = resolve_state_file()
    if not path:
        return dict(_MEMORY_STATE)

for candidate in (path, os.path.join(BASE_DIR, "bridge_state.json"), os.path.join(BASE_DIR, ".bridge_state.json")):
        if not candidate or not os.path.exists(candidate):
            continue
        try:
            with open(candidate, "r", encoding="utf-8") as handle:
                data = json.load(handle)
                if isinstance(data, dict):
                    _MEMORY_STATE = data
                    return data
        except Exception:
            continue

return dict(_MEMORY_STATE)

def save_state(state):
    global _MEMORY_STATE
    _MEMORY_STATE = dict(state)

path = resolve_state_file()
    if not path:
        return

directory = os.path.dirname(os.path.abspath(path))
    os.makedirs(directory, exist_ok=True)
    tmp_path = path + ".tmp"
    with open(tmp_path, "w", encoding="utf-8") as handle:
        json.dump(state, handle)
    os.replace(tmp_path, path)

def format_scan_time(dt):
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=resolve_scan_timezone())
    return dt.astimezone(resolve_scan_timezone()).isoformat()

def get_serial_conn():
    global _SERIAL_CONN
    if _SERIAL_CONN is not None:
        return _SERIAL_CONN

try:
        import serial  # type: ignore
    except Exception as exc:
        raise RuntimeError(f"pyserial belum terpasang: {exc}")

_SERIAL_CONN = serial.Serial(
        port=SERIAL_PORT,
        baudrate=SERIAL_BAUDRATE,
        timeout=SERIAL_TIMEOUT,
    )
    log("INFO", f"Serial connected: port={SERIAL_PORT}, baudrate={SERIAL_BAUDRATE}")
    return _SERIAL_CONN

def get_zk_conn():
    global _ZK_CONN
    if _ZK_CONN is not None:
        return _ZK_CONN

try:
        from zk import ZK  # type: ignore
    except Exception as exc:
        raise RuntimeError(f"pyzk belum terpasang. Jalankan: pip install pyzk ({exc})")

zk = ZK(ZK_DEVICE_IP, port=ZK_DEVICE_PORT, timeout=ZK_TIMEOUT, password=ZK_DEVICE_PASSWORD)
    with silence_noisy_stdout():
        _ZK_CONN = zk.connect()
    log("INFO", f"ZK connected: {ZK_DEVICE_IP}:{ZK_DEVICE_PORT}")
    try:
        log("INFO", f"Firmware: {_ZK_CONN.get_firmware_version()}")
    except Exception:
        pass
    return _ZK_CONN

def reset_zk_conn():
    global _ZK_CONN
    if _ZK_CONN is not None:
        try:
            _ZK_CONN.disconnect()
        except Exception:
            pass
    _ZK_CONN = None

def normalize_uid(value):
    if value is None:
        return None
    text = str(value).strip()
    if not text or text == "0":
        return None
    if UID_MODE in ("username", "pin"):
        return text
    return text.upper()

def load_pin_card_cache(conn):
    """Map PIN mesin (user_id) -> nomor kartu RFID di mesin."""
    global _PIN_CARD_CACHE
    if _PIN_CARD_CACHE is not None:
        return _PIN_CARD_CACHE

_PIN_CARD_CACHE = {}
    try:
        for user in conn.get_users():
            pin = normalize_uid(getattr(user, "user_id", None))
            if not pin:
                continue
            card = normalize_uid(getattr(user, "card", None))
            if UID_MODE == "card":
                _PIN_CARD_CACHE[pin] = card if card else pin
            else:
                _PIN_CARD_CACHE[pin] = pin
        log("INFO", f"Cache user mesin: {len(_PIN_CARD_CACHE)} PIN terdaftar")
    except Exception as exc:
        log("WARN", f"Gagal memuat daftar user dari mesin: {exc}")
        _PIN_CARD_CACHE = {}

return _PIN_CARD_CACHE

def resolve_card_uid(attendance, conn):
    pin = normalize_uid(getattr(attendance, "user_id", None))
    if not pin:
        return None

if UID_MODE in ("username", "pin"):
        return pin

cache = load_pin_card_cache(conn)
    card_uid = cache.get(pin, pin)
    if card_uid != pin:
        log("INFO", f"PIN {pin} -> kartu {card_uid}")
    return card_uid

def read_from_zk():
    global _PIN_CARD_CACHE
    state = load_state()
    state_key = f"last_uid_{ZK_DEVICE_IP}"

try:
        conn = get_zk_conn()
        attendances = conn.get_attendance() or []
    except Exception as exc:
        reset_zk_conn()
        raise RuntimeError(f"Gagal baca mesin {ZK_DEVICE_IP}: {exc}") from exc

if not attendances:
        return []

max_uid = max(getattr(a, "uid", 0) or 0 for a in attendances)

if state_key not in state:
        state[state_key] = max_uid
        save_state(state)
        log("INFO", f"Baseline: abaikan log lama. Mulai dari tap baru (uid > {max_uid}).")
        return []

last_uid = int(state[state_key])
    logs = []
    highest = last_uid

for att in sorted(attendances, key=lambda a: getattr(a, "uid", 0) or 0):
        att_uid = getattr(att, "uid", 0) or 0
        if att_uid <= last_uid:
            continue

pin = normalize_uid(getattr(att, "user_id", None))
        card_uid = resolve_card_uid(att, conn)
        if not card_uid:
            log("WARN", f"Lewati uid={att_uid}: card/user_id kosong")
            highest = max(highest, att_uid)
            continue

timestamp = getattr(att, "timestamp", None)
        if not timestamp:
            timestamp = datetime.now(resolve_scan_timezone())

logs.append(
            {
                "card_uid": card_uid,
                "scan_time": format_scan_time(timestamp),
                "event_type": "tap",
                "meta": {
                    "mode": "zk",
                    "device_ip": ZK_DEVICE_IP,
                    "device_pin": pin,
                    "attendance_uid": att_uid,
                    "status": getattr(att, "status", None),
                    "punch": getattr(att, "punch", None),
                },
            }
        )
        highest = max(highest, att_uid)

if highest > last_uid:
        state[state_key] = highest
        save_state(state)

return logs

def read_from_device():
    scan_time = datetime.now(resolve_scan_timezone()).isoformat()

if INPUT_MODE in ("zk", "lan", "tcp"):
        return read_from_zk()

if INPUT_MODE == "serial":
        conn = get_serial_conn()
        raw = conn.readline().decode("utf-8", errors="ignore").strip()
        if raw:
            return [
                {
                    "card_uid": normalize_uid(raw) or raw,
                    "scan_time": scan_time,
                    "event_type": "tap",
                    "meta": {"mode": "serial", "port": SERIAL_PORT},
                }
            ]
        return []

if MOCK_MODE or INPUT_MODE == "mock":
        if random.random() < 0.25:
            return [
                {
                    "card_uid": MOCK_CARD_UID,
                    "scan_time": scan_time,
                    "event_type": "tap",
                    "meta": {"mode": "mock"},
                }
            ]
    return []

def push_logs(logs):
    headers = {
        "Content-Type": "application/json",
        "X-Device-Token": DEVICE_TOKEN,
    }
    payload = {
        "device_code": DEVICE_CODE,
        "device_name": DEVICE_NAME,
        "logs": logs,
    }
    response = requests.post(API_URL, headers=headers, data=json.dumps(payload), timeout=20)
    if response.status_code >= 400:
        raise RuntimeError(
            f"HTTP {response.status_code} {response.reason} | body={response.text[:500]}"
        )
    return response.json()

def main():
    if not DEVICE_TOKEN:
        raise RuntimeError("ATTENDANCE_DEVICE_TOKEN belum diisi.")
    if "domain-anda.com" in API_URL.lower() or "your-domain.com" in API_URL.lower():
        raise RuntimeError(
            "ATTENDANCE_API_URL masih placeholder. "
            "Set di .env.bridge — production: https://absencakra.online/api/attendance/device/logs"
        )

log("INFO", f"Bridge start: device={DEVICE_CODE}, input_mode={INPUT_MODE}, uid_mode={UID_MODE}")
    log("INFO", f"Target API: {API_URL}")
    if INPUT_MODE in ("zk", "lan", "tcp"):
        log("INFO", f"ZK LAN: {ZK_DEVICE_IP}:{ZK_DEVICE_PORT}")
    elif INPUT_MODE == "serial":
        log("INFO", f"Serial: port={SERIAL_PORT}, baudrate={SERIAL_BAUDRATE}")

while True:
        try:
            logs = read_from_device()
            if logs:
                result = push_logs(logs)
                log("PUSH", f"count={len(logs)} card_uid={logs[0].get('card_uid')} result={result}")
            else:
                log("WAIT", "Tidak ada tap baru pada siklus ini")
        except Exception as exc:
            log("ERROR", str(exc))
        time.sleep(POLL_SECONDS)

if __name__ == "__main__":
    try:
        log("INFO", f"Python {sys.version.split()[0]} | log={LOG_FILE}")
        main()
    except Exception as exc:
        log("FATAL", str(exc))
        raise

Paste Settings

Paste Title : [Optional]

Paste Folder : [Optional]

Syntax Highlighting : [Optional]

Paste Expiration : [Optional]

Paste Status : [Optional]

Password : [Optional]

Description: [Optional]

Tags: [Optional]

Encrypt Paste (?)

You are currently not logged in, this means you can not edit or delete anything you paste. Sign Up or Login

Recent Pastes

Untitled
Plaintext | 8 | 2 hours ago
Untitled
Plaintext | 9 | 2 hours ago
Untitled
Plaintext | 223 | 4 hours ago
Untitled
Plaintext | 16 | 4 hours ago
Untitled
Plaintext | 7 | 4 hours ago
Untitled
Plaintext | 25 | 5 hours ago
Untitled
Plaintext | 61 | 6 hours ago

Paste Settings

Site Languages