-
Notifications
You must be signed in to change notification settings - Fork 0
/
validate.py
61 lines (52 loc) · 3.21 KB
/
validate.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
from embed import Embed
class Validate:
def __init__(self, data, node_type, private_key):
self.data = data
self.embed = Embed(data, node_type, private_key=private_key)
def validate_watermark(self, wm_id_dict, watermark_cover_field):
# Convert id_list to a set for fast membership checks
# id_set = set(str(key) for key in id_list) # Use set to avoid duplicates and improve lookup speed
# Iterate over each record in the list with its index (record number)
for index, record in enumerate(self.data):
attributes = [key for key, value in record.items() if isinstance(value, (int, float))]
# print(attributes)
# Ensure the record has a 'watermark_id'
if watermark_cover_field in record:
# print("cover in record")
watermark_id = int(record[watermark_cover_field])
if watermark_id in wm_id_dict.keys():
# print("key in wm secret")
_, hashed_secret_int = self.embed.watermark_pseudo_node(pseudo_node=record, watermark_identity=str(watermark_id),
watermark_id_field=watermark_cover_field, attributes=attributes, validate=True)
# print(f"Watermark id: {watermark_id}")
# print(attributes)
# print(record)
# print("BOTH HASHES")
# print(hashed_secret_int, wm_id_dict[watermark_id])
if hashed_secret_int == wm_id_dict[watermark_id]:
return True
return False
def validate_watermark_all(self, wm_id_dict, watermark_cover_field):
# Convert id_list to a set for fast membership checks
# id_set = set(str(key) for key in id_list) # Use set to avoid duplicates and improve lookup speed
pseudo_node_count = 0
# Iterate over each record in the list with its index (record number)
for index, record in enumerate(self.data):
attributes = sorted([key for key, value in record.items() if isinstance(value, (int, float))])
# print(attributes)
# Ensure the record has a 'watermark_id'
if watermark_cover_field in record:
# print("cover in record")
watermark_id = int(record[watermark_cover_field])
if watermark_id in wm_id_dict.keys():
# print("key in wm secret")
_, hashed_secret_int = self.embed.watermark_pseudo_node(pseudo_node=record, watermark_identity=str(watermark_id),
watermark_id_field=watermark_cover_field, attributes=attributes, validate=True)
# print(f"Watermark id: {watermark_id}")
# print(attributes)
# print(record)
# print("BOTH HASHES")
# print(hashed_secret_int, wm_id_dict[watermark_id])
if hashed_secret_int == wm_id_dict[watermark_id]:
pseudo_node_count += 1
return pseudo_node_count