wehub-resource-sync
79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
import os
|
|
import json
|
|
|
|
"""
|
|
NegationAPIPairManager Interface
|
|
"""
|
|
class NegationAPIPairManager:
|
|
def __init__(self):
|
|
return None
|
|
|
|
def get_success(self, forward_call):
|
|
# return a single backward call that have previously been reported successful
|
|
# this backward call will be used to perform the reversion if requested
|
|
raise NotImplementedError
|
|
|
|
def get_failure(self, forward_call):
|
|
# return a list of backward calls that have previously been reported unsuccessful
|
|
raise NotImplementedError
|
|
|
|
def insert_log(self, forward_call, backward_call, result):
|
|
# insert a new entry or log to the negation manager
|
|
raise NotImplementedError
|
|
|
|
"""
|
|
Simple NegationManager with JSON based key-value pair storage
|
|
"""
|
|
class NaiveNegationAPIPairManager(NegationAPIPairManager):
|
|
def __init__(self, reverse_log_json_path):
|
|
self.reverse_log_path = os.path.join(reverse_log_json_path, "negation_log.json")
|
|
|
|
def get_success(self, forward_call):
|
|
logs = {}
|
|
try:
|
|
with open(self.reverse_log_path, "r") as reverse_log:
|
|
logs = json.load(reverse_log)
|
|
if forward_call in logs:
|
|
if "true" in logs[forward_call]:
|
|
return logs[forward_call]["true"][0]
|
|
|
|
except Exception as e:
|
|
pass
|
|
|
|
return None
|
|
|
|
def get_failure(self, forward_call):
|
|
logs = {}
|
|
try:
|
|
with open(self.reverse_log_path, "r") as reverse_log:
|
|
logs = json.load(reverse_log)
|
|
if forward_call in logs:
|
|
if "false" in logs[forward_call]:
|
|
return logs[forward_call]["false"]
|
|
|
|
except Exception as e:
|
|
pass
|
|
|
|
return None
|
|
|
|
def insert_log(self, forward_call, backward_call, result):
|
|
logs = {}
|
|
try:
|
|
with open(self.reverse_log_path, "r") as reverse_log:
|
|
logs = json.load(reverse_log)
|
|
|
|
except Exception as e:
|
|
pass
|
|
|
|
with open(self.reverse_log_path, "w") as reverse_log:
|
|
if forward_call not in logs:
|
|
logs[forward_call] = {}
|
|
|
|
result = str(result).lower()
|
|
if result in logs[forward_call]:
|
|
logs[forward_call][result].append(backward_call)
|
|
else:
|
|
logs[forward_call][result] = [backward_call]
|
|
|
|
json.dump(logs, reverse_log)
|