| 1 |
"""
|
| 2 |
|
| 3 |
RelativitySearchCriteria
|
| 4 |
|
| 5 |
Created by:
|
| 6 |
Emanuel Borges
|
| 7 |
06.20.2025
|
| 8 |
|
| 9 |
This very simple program will take a Relativity saved search criteria report (for now from Lighthouse) and perform a sync between search folders.
|
| 10 |
|
| 11 |
"""
|
| 12 |
|
| 13 |
import os
|
| 14 |
from win32com.client import Dispatch
|
| 15 |
from dataclasses import dataclass, field, asdict
|
| 16 |
from typing import Optional, List, Dict
|
| 17 |
|
| 18 |
version = "0.1"
|
| 19 |
|
| 20 |
@dataclass
|
| 21 |
class RelativitySearchCriteria:
|
| 22 |
artifact_id: int
|
| 23 |
search_name: str
|
| 24 |
search_folder_path: str
|
| 25 |
related_items: Optional[str] = None
|
| 26 |
search_index: Optional[str] = None
|
| 27 |
index_search_text: Optional[str] = None
|
| 28 |
search_conditions: List[str] = field(default_factory=list)
|
| 29 |
#search_condition_connector: Optional[str] = None
|
| 30 |
|
| 31 |
|
| 32 |
|
| 33 |
|
| 34 |
class RelativitySearchManager:
|
| 35 |
def __init__(self):
|
| 36 |
self.entries: Dict[int, RelativitySearchCriteria] = {}
|
| 37 |
|
| 38 |
def add_entry(self, artifact_id: int, search_name: str, search_folder_path: str) -> None:
|
| 39 |
if artifact_id in self.entries:
|
| 40 |
raise ValueError(f"Entry with ARtifact ID {artifact_id} already exists.")
|
| 41 |
self.entries[artifact_id] = RelativitySearchCriteria(artifact_id = artifact_id, search_name = search_name, search_folder_path = search_folder_path)
|
| 42 |
|
| 43 |
|
| 44 |
def update_entry(self, artifact_id: int, related_items: Optional[str] = None, search_index: Optional[str] = None, index_search_text: Optional[str] = None):
|
| 45 |
if artifact_id not in self.entries:
|
| 46 |
raise KeyError(f"No entry with Artifact ID {artifact_id} found.")
|
| 47 |
entry = self.entries[artifact_id]
|
| 48 |
if related_items is not None:
|
| 49 |
entry.related_items = related_items
|
| 50 |
if search_index is not None:
|
| 51 |
entry.search_index = search_index
|
| 52 |
if index_search_text is not None:
|
| 53 |
entry.index_search_text = index_search_text
|
| 54 |
|
| 55 |
|
| 56 |
def add_condition(self, artifact_id: int, condition: str) -> None:
|
| 57 |
if artifact_id not in self.entries:
|
| 58 |
raise KeyError(f"No entry with Artifact ID {artifact_id} found.")
|
| 59 |
self.entries[artifact_id].search_conditions.append(condition)
|
| 60 |
|
| 61 |
|
| 62 |
def get_entry(self, artifact_id:int) -> RelativitySearchCriteria:
|
| 63 |
if artifact_id not in self.entries:
|
| 64 |
raise KeyError(f"No entry with Artifact ID {artifact_id} found.")
|
| 65 |
return self.entries[artifact_id]
|
| 66 |
|
| 67 |
def to_dict(self, artifact_id: int) -> dict:
|
| 68 |
return asdict(self.get_entry(artifact_id))
|
| 69 |
|
| 70 |
def all_entries(self) -> List[RelativitySearchCriteria]:
|
| 71 |
return list(self.entries.values())
|
| 72 |
|
| 73 |
def compare_entries(self, artifact_id_1: int, artifact_id_2: int, criteria_ignore_list: Optional[List[str]] = None) -> Dict[str, List[str]]:
|
| 74 |
if artifact_id_1 not in self.entries or artifact_id_2 not in self.entries:
|
| 75 |
raise KeyError("One or both artifact IDs not found.")
|
| 76 |
|
| 77 |
entry1 = self.entries[artifact_id_1]
|
| 78 |
entry2 = self.entries[artifact_id_2]
|
| 79 |
ignore_list = criteria_ignore_list if criteria_ignore_list else []
|
| 80 |
|
| 81 |
differences = {}
|
| 82 |
|
| 83 |
## Compare the simple fields
|
| 84 |
fields_to_compare = ['related_items', 'search_index', 'index_search_text']
|
| 85 |
for field_name in fields_to_compare:
|
| 86 |
val1 = getattr(entry1, field_name)
|
| 87 |
val2 = getattr(entry2, field_name)
|
| 88 |
if val1 != val2:
|
| 89 |
differences[field_name] = [val1, val2]
|
| 90 |
|
| 91 |
|
| 92 |
## Compare search_conditions
|
| 93 |
conds1 = entry1.search_conditions
|
| 94 |
conds2 = entry2.search_conditions
|
| 95 |
|
| 96 |
## Compare the lengths first
|
| 97 |
if len(conds1) != len(conds2):
|
| 98 |
differences['search_conditions_length'] = [len(conds1), len(conds2)]
|
| 99 |
|
| 100 |
## Compare the individual conditions
|
| 101 |
cond_diffs = []
|
| 102 |
for i, (c1,c2) in enumerate(zip(conds1, conds2)):
|
| 103 |
if c1 != c2:
|
| 104 |
if c1 not in ignore_list or c2 not in ignore_list:
|
| 105 |
cond_diffs.append(f"Index {i}: '{c1}' vs '{c2}'")
|
| 106 |
|
| 107 |
|
| 108 |
## Check for extra conditions beyond the zip
|
| 109 |
longer_list = conds1 if len(conds1) > len(conds2) else conds2
|
| 110 |
if len(conds1) != len(conds2):
|
| 111 |
for i in range(len(conds1), len(longer_list)):
|
| 112 |
val = longer_list[i]
|
| 113 |
if val not in ignore_list:
|
| 114 |
cond_diffs.append(f"Index {i}: '{val}' (only in one entry).")
|
| 115 |
|
| 116 |
|
| 117 |
if cond_diffs:
|
| 118 |
differences['search_conditions'] = cond_diffs
|
| 119 |
|
| 120 |
return differences
|
| 121 |
|
| 122 |
|
| 123 |
|
| 124 |
if __name__ == '__main__':
|
| 125 |
manager = RelativitySearchManager()
|
| 126 |
|
| 127 |
## Add a new entry
|
| 128 |
manager.add_entry(artifact_id=3359288, search_name = '04.00 Limited Priv "Request to counsel"', search_folder_path = r'GDC - AG\DOJCID_Priv Eval QC\04_Last Ditch + Liimited Priv Terms')
|
| 129 |
|
| 130 |
## Update with some optional fields
|
| 131 |
manager.update_entry(3359288, search_index = 'dtSearch', index_search_text = '("legal advice" AND NOT "seeking legal advice") OR "talk to legal" OR "talk with legal" OR "speak with legal" OR "speak to legal" OR "spoke with legal" OR "spoke to legal" OR "legal said" OR "legal has said" OR "legal told" OR "legal has told" OR "heard from legal" OR "legal advised" OR "legal has advised" OR "ask legal" OR "asked legal" OR "legal asked"')
|
| 132 |
|
| 133 |
## Add some conditions
|
| 134 |
manager.add_condition(3359288, ' [savedsearch] in "AG Source Population__UPDATE-NEXT PROD*-Not SEC/DJ" ')
|
| 135 |
manager.add_condition(3359288, 'and')
|
| 136 |
manager.add_condition(3359288, ' [1L Privilege] not in ( "Privileged Withhold","Partially Privileged (Redact)") ')
|
| 137 |
|
| 138 |
|
| 139 |
## Add another new entry
|
| 140 |
manager.add_entry(artifact_id=3359281, search_name = '04.00 Limited Priv "Request to counsel"', search_folder_path = r'GDC - DOJ - Go Gets\DOJCID_Priv Eval QC\04_Last Ditch + Limited Priv Terms')
|
| 141 |
|
| 142 |
## Add the second optional fields
|
| 143 |
manager.update_entry(3359281, search_index = 'dtSearch', index_search_text = '("legal advice" AND NOT "seeking legal advice") OR "talk to legal" OR "talk with legal" OR "speak with legal" OR "speak to legal" OR "spoke with legal" OR "spoke to legal" OR "legal said" OR "legal has said" OR "legal told" OR "legal has told" OR "heard from legal" OR "legal advised" OR "legal has advised" OR "ask legal" OR "asked legal" OR "legal asked"')
|
| 144 |
|
| 145 |
## Add some conditions to the second entry
|
| 146 |
manager.add_condition(3359281, ' [savedsearch] in "GoGets Source Population__UPDATE-NEXT PROD*" ')
|
| 147 |
manager.add_condition(3359281, 'and')
|
| 148 |
manager.add_condition(3359281, ' [1L Privilege] not in ( "Privileged Withhold","Partially Privileged (Redact)") ')
|
| 149 |
|
| 150 |
|
| 151 |
#print(manager.all_entries())
|
| 152 |
|
| 153 |
|
| 154 |
diff = manager.compare_entries(3359288, 3359281, criteria_ignore_list = [' [savedsearch] in "AG Source Population__UPDATE-NEXT PROD*-Not SEC/DJ" ',' [savedsearch] in "GoGets Source Population__UPDATE-NEXT PROD*" '])
|
| 155 |
if diff:
|
| 156 |
print("Differences found:")
|
| 157 |
for key, vals in diff.items():
|
| 158 |
print(f"{key}:")
|
| 159 |
for v in vals:
|
| 160 |
print(f" - {v}")
|
| 161 |
else:
|
| 162 |
print("No relevant differences.") |
