ป้องกันทุจริตด้วย Perceptual Hashing

1from PIL import Image
2
3def dhash(image: Image.Image, hash_size: int = 8) -> int:
4    # ย่อเป็น (hash_size+1) x hash_size แล้วแปลงเป็นเทา
5    gray = image.convert("L").resize(
6        (hash_size + 1, hash_size), Image.Resampling.LANCZOS
7    )
8    pixels = list(gray.getdata())
9    width = hash_size + 1
10
11    bits = 0
12    for row in range(hash_size):
13        for col in range(hash_size):
14            left = pixels[row * width + col]
15            right = pixels[row * width + col + 1]
16            bits = (bits << 1) | int(left > right)
17    return bits  # 64-bit integer
18
19def dhash_hex(image: Image.Image) -> str:
20    return format(dhash(image), "016x")

1import numpy as np
2from PIL import Image
3from scipy.fftpack import dct
4
5def phash(image: Image.Image, hash_size: int = 8, highfreq_factor: int = 4) -> int:
6    img_size = hash_size * highfreq_factor  # 32x32
7    gray = image.convert("L").resize(
8        (img_size, img_size), Image.Resampling.LANCZOS
9    )
10    pixels = np.asarray(gray, dtype=np.float32)
11
12    # DCT 2 มิติ แล้วเก็บมุมบนซ้าย (ความถี่ต่ำ)
13    coeffs = dct(dct(pixels, axis=0, norm="ortho"), axis=1, norm="ortho")
14    low = coeffs[:hash_size, :hash_size]
15
16    # ใช้ median ของบล็อก (ข้าม DC term ที่ [0,0] เพื่อตัด bias ความสว่างรวม)
17    med = np.median(low[1:].flatten())
18
19    bits = 0
20    for value in low.flatten():
21        bits = (bits << 1) | int(value > med)
22    return bits

1def hamming(a: int, b: int) -> int:
2    return (a ^ b).bit_count()  # Python 3.10+
3
4DUP_THRESHOLD = 8  # <= ถือว่าเป็นสลิปเดียวกัน
5
6def is_duplicate(hash_a: int, hash_b: int) -> bool:
7    return hamming(hash_a, hash_b) <= DUP_THRESHOLD

1import io
2from PIL import Image
3
4class DuplicateSlipError(Exception):
5    def __init__(self, existing_id: str, distance: int):
6        self.existing_id = existing_id
7        self.distance = distance
8        super().__init__(f"ซ้ำกับสลิป {existing_id} (distance={distance})")
9
10async def verify_slip(raw_bytes: bytes, repo) -> int:
11    """คืนค่า hash ถ้าผ่าน, โยน error ถ้าซ้ำ - เรียกก่อนเครดิตเงินเสมอ"""
12    image = Image.open(io.BytesIO(raw_bytes))
13    h = phash(image)
14
15    # ดึงเฉพาะ candidate ที่ใกล้ (ดูหัวข้อ BK-tree) แล้วเทียบ
16    for existing_id, existing_hash in await repo.candidates(h):
17        dist = hamming(h, existing_hash)
18        if dist <= DUP_THRESHOLD:
19            raise DuplicateSlipError(existing_id, dist)
20    return h

1class BKTree:
2    def __init__(self, distance_fn):
3        self.distance = distance_fn
4        self.root = None  # (value, payload, {dist: child})
5
6    def add(self, value: int, payload):
7        if self.root is None:
8            self.root = (value, payload, {})
9            return
10        node = self.root
11        while True:
12            d = self.distance(value, node[0])
13            child = node[2].get(d)
14            if child is None:
15                node[2][d] = (value, payload, {})
16                return
17            node = child
18
19    def query(self, value: int, max_dist: int):
20        if self.root is None:
21            return []
22        results, stack = [], [self.root]
23        while stack:
24            v, payload, children = stack.pop()
25            d = self.distance(value, v)
26            if d <= max_dist:
27                results.append((payload, d))
28            # triangle inequality: เหลือเฉพาะกิ่งที่ระยะอยู่ในช่วงที่เป็นไปได้
29            for edge, child in children.items():
30                if d - max_dist <= edge <= d + max_dist:
31                    stack.append(child)
32        return results