40-validation/find-semantic-duplicate-candidates.py

#!/usr/bin/env python3
import argparse,itertools,json,re
from pathlib import Path
def tokens(v):return set(re.findall(r"[a-zäöüß0-9]+",v.lower()))
def jac(a,b):return len(a&b)/len(a|b) if a|b else 0.0
def find(data,config):
 units=data.get("units",[]);w=config["weights"];candidates=[]
 for a,b in itertools.combinations(units,2):
  lexical=jac(tokens(a.get("label","")+" "+a.get("text","")),tokens(b.get("label","")+" "+b.get("text","")))
  structural=jac(set(a.get("sections",[])),set(b.get("sections",[])));references=jac(set(a.get("references",[])),set(b.get("references",[])))
  score=lexical*w["lexical"]+structural*w["structural"]+references*w["references"]
  if lexical>=config["minimumLexical"] and score>=config["threshold"]:
   candidates.append({"left":a.get("id"),"right":b.get("id"),"score":round(score,4),"signals":{"lexical":round(lexical,4),"structural":round(structural,4),"references":round(references,4)},"decision":"REVIEW_REQUIRED","automaticMerge":False})
 return {"schemaVersion":"1.0","workItem":"SE-0032","status":"REVIEW_REQUIRED" if candidates else "PASSED","units":len(units),"pairsCompared":len(units)*(len(units)-1)//2,"thresholds":{"score":config["threshold"],"minimumLexical":config["minimumLexical"],"weights":w},"candidates":candidates,"automaticMerges":0}
def main():
 p=argparse.ArgumentParser();p.add_argument("input",type=Path);p.add_argument("--config",type=Path,required=True);p.add_argument("--output",type=Path);a=p.parse_args();r=find(json.loads(a.input.read_text()),json.loads(a.config.read_text()));text=json.dumps(r,ensure_ascii=False,indent=2)+"\n"
 if a.output:a.output.parent.mkdir(parents=True,exist_ok=True);a.output.write_text(text)
 else:print(text,end="")
if __name__=="__main__":main()