from __future__ import annotations import hashlib import re import uuid from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from typing import Any, Iterable APP_TITLE = "Dr Drastic: Unified Evidence Engine" APP_VERSION = "2.0.0" @dataclass class SourceDocument: doc_id: str name: str doc_type: str extracted_text: str = "" role: str = "unknown" extraction_status: str = "loaded" @dataclass class FactEvent: event_id: str date: str sort_date: str actor: str fact: str source_doc: str source_role: str tags: list[str] = field(default_factory=list) confidence: float = 0.7 @dataclass class Contradiction: contradiction_id: str kind: str left_doc: str right_doc: str issue: str left_excerpt: str right_excerpt: str severity: str confidence: float @dataclass class OmissionFlag: doc: str category: str reason: str confidence: float @dataclass class EvidenceRow: item_id: str source_doc: str source_role: str date: str actor: str fact: str tags: list[str] legal_significance: str contradiction_link: str = "" omission_flag: str = "" confidence: float = 0.7 next_action: str = "Verify against the source document" @dataclass class AnalysisResult: release_id: str owner: str = "" documents: list[SourceDocument] = field(default_factory=list) events: list[FactEvent] = field(default_factory=list) contradictions: list[Contradiction] = field(default_factory=list) omissions: list[OmissionFlag] = field(default_factory=list) evidence_matrix: list[EvidenceRow] = field(default_factory=list) routes: list[str] = field(default_factory=list) score: dict[str, Any] = field(default_factory=dict) warnings: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "AnalysisResult": data = data or {} return cls( release_id=data.get("release_id") or make_release_id(), owner=data.get("owner", ""), documents=[SourceDocument(**x) for x in data.get("documents", [])], events=[FactEvent(**x) for x in data.get("events", [])], contradictions=[Contradiction(**x) for x in data.get("contradictions", [])], omissions=[OmissionFlag(**x) for x in data.get("omissions", [])], evidence_matrix=[EvidenceRow(**x) for x in data.get("evidence_matrix", [])], routes=list(data.get("routes", [])), score=dict(data.get("score", {})), warnings=list(data.get("warnings", [])), ) PARTY_DICTIONARY = { "claimant": ["dwayne", "galloway", "claimant", "complainant", "applicant"], "dbl-max": ["dbl-max", "dbl max", "db l max", "seller", "vendor"], "halifax": ["halifax", "bank of scotland", "lloyds", "lloyds banking group"], "eversheds": ["eversheds", "eversheds sutherland"], "fos": ["financial ombudsman", "ombudsman", "adjudicator", "final decision"], "ebay": ["ebay", "e-bay", "marketplace"], } ROLE_HINTS = { "claimant": PARTY_DICTIONARY["claimant"], "respondent_bank": PARTY_DICTIONARY["halifax"], "respondent_solicitor": PARTY_DICTIONARY["eversheds"] + ["solicitor", "legal team"], "adjudicator": PARTY_DICTIONARY["fos"], "third_party_seller": PARTY_DICTIONARY["dbl-max"], "third_party_platform": PARTY_DICTIONARY["ebay"], } TAG_RULES = { "vulnerability": [ "vulnerab", "disability", "reasonable adjustment", "protected characteristic", "hardship", "medical", "pip", "accessibility", ], "admission": ["admit", "accepted", "confirmed", "acknowledged", "admission"], "regulatory": ["fca", "sra", "ehrc", "ico", "ombudsman", "fos", "gdpr"], "account_interference": [ "account closed", "closure", "blocked", "standing order", "froze", "frozen", "freeze", "clawback", ], "evidence_failure": [ "ignored", "misrecorded", "not raised", "failed to", "missing", "omitted", "suppressed", "misrepresent", ], "loss_or_harm": [ "distress", "loss", "eviction", "malnutrition", "injury", "health decline", "business loss", "commercial loss", ], "consumer_rights": [ "refund", "return", "reject", "defect", "faulty", "not as described", "brake", "consumer rights", ], } LEGAL_ROUTES = { "Equality Act / vulnerability": ( ["disability", "reasonable adjustment", "equality act", "pip", "vulnerab"], "Review Equality Act 2010 service-provider duties and evidence of knowledge, " "disadvantage, and proposed adjustments.", ), "Consumer rights": ( ["consumer rights", "right to reject", "refund", "defect", "faulty", "not as described"], "Review the Consumer Rights Act 2015 issues, remedy dates, trader identity, " "product evidence, and any platform protections.", ), "Banking complaint / FCA / FOS": ( ["fca", "chargeback", "account", "bank", "standing order", "consumer duty", "fos"], "Build a dated complaint trail and check the applicable FCA DISP/FOS route, " "including limitation and final-response dates.", ), "Data protection": ( ["subject access", "sar", "gdpr", "personal data", "data protection", "ico"], "Map each data request, response deadline, missing category, and ICO escalation.", ), "Professional conduct": ( ["solicitor", "eversheds", "sra", "professional conduct"], "Separate litigation disagreement from evidence of a potential professional-" "conduct issue and verify the applicable SRA rule.", ), "Fraud / misrepresentation": ( ["fraud", "false representation", "misrepresent", "deceiv"], "Identify the exact representation, speaker, date, falsity, reliance, and loss. " "Do not label conduct criminal without evidence supporting each element.", ), "Loss and causation": ( ["distress", "loss", "eviction", "arrears", "injury", "commercial"], "Create a loss schedule with documents, dates, causation, mitigation, and a " "clearly separated estimate for each head of loss.", ), } DATE_TOKEN = re.compile( r"\b(" r"\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|" r"\d{4}-\d{2}-\d{2}|" r"\d{1,2}\s+(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|" r"Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:t(?:ember)?)?|Oct(?:ober)?|" r"Nov(?:ember)?|Dec(?:ember)?)\s+\d{2,4}|" r"(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|" r"Jul(?:y)?|Aug(?:ust)?|Sep(?:t(?:ember)?)?|Oct(?:ober)?|" r"Nov(?:ember)?|Dec(?:ember)?)\s+\d{4}" r")\b", re.IGNORECASE, ) def make_release_id() -> str: stamp = datetime.now(timezone.utc).strftime("%Y%m%d") return f"DDU-{stamp}-{uuid.uuid4().hex[:8].upper()}" def normalise_whitespace(text: str) -> str: text = (text or "").replace("\r\n", "\n").replace("\r", "\n") text = re.sub(r"[ \t]+", " ", text) return re.sub(r"\n{3,}", "\n\n", text).strip() def stable_doc_id(name: str, text: str = "") -> str: seed = f"{name.lower()}:{len(text)}:{text[:200]}".encode("utf-8", errors="replace") return "DOC-" + hashlib.sha256(seed).hexdigest()[:8].upper() def guess_role(doc_name: str, text: str) -> str: haystack = f"{doc_name}\n{text[:6000]}".lower() scores = { role: sum(haystack.count(hint) for hint in hints) for role, hints in ROLE_HINTS.items() } role, score = max(scores.items(), key=lambda item: item[1]) return role if score else "unknown" def resolve_actor(text: str) -> str: lower = text.lower() for party, aliases in PARTY_DICTIONARY.items(): if any(re.search(rf"\b{re.escape(alias)}\b", lower) for alias in aliases): return party sender = re.search(r"\b(?:from|by)\s*:\s*([^,;\n]{2,80})", text, re.IGNORECASE) return sender.group(1).strip() if sender else "unknown" def normalise_date(value: str) -> tuple[str, str]: raw = value.strip() formats = ( "%d/%m/%Y", "%d-%m-%Y", "%Y-%m-%d", "%d/%m/%y", "%d-%m-%y", "%d %b %Y", "%d %B %Y", "%d %b %y", "%d %B %y", "%b %Y", "%B %Y", ) for fmt in formats: try: parsed = datetime.strptime(raw, fmt) display = parsed.strftime("%d %B %Y") if "%d" in fmt else parsed.strftime("%B %Y") return display, parsed.strftime("%Y-%m-%d") except ValueError: continue return raw, "" def tag_text(text: str) -> list[str]: lower = text.lower() tags = [tag for tag, keywords in TAG_RULES.items() if any(k in lower for k in keywords)] return tags or ["general"] def _sentences(text: str) -> Iterable[str]: clean = normalise_whitespace(text) for paragraph in clean.splitlines(): for sentence in re.split(r"(?<=[.!?])\s+", paragraph): sentence = sentence.strip(" -\t") if len(sentence) >= 24: yield sentence def _event_candidate(line: str) -> tuple[str, str] | None: line = line.strip() if len(line) < 8: return None match = DATE_TOKEN.search(line) if not match: return None date = match.group(1) before = line[:match.start()].strip(" :-|>") after = line[match.end():].strip(" :-|>") fact = after or before return (date, fact) if len(fact) >= 8 else None def extract_events(documents: list[SourceDocument], chronology: str = "") -> list[FactEvent]: raw_events: list[tuple[str, str, str, str]] = [] if chronology.strip(): for line in chronology.splitlines(): candidate = _event_candidate(line) if candidate: raw_events.append((candidate[0], candidate[1], "CHRONOLOGY", "claimant")) for doc in documents: candidates = list(doc.extracted_text.splitlines()) + list(_sentences(doc.extracted_text)) for text in candidates[:800]: candidate = _event_candidate(text) if candidate: raw_events.append((candidate[0], candidate[1], doc.doc_id, doc.role)) seen: set[str] = set() events: list[FactEvent] = [] for date_raw, fact, source_doc, source_role in raw_events: key = re.sub(r"\W+", " ", fact.lower()).strip()[:180] if key in seen: continue seen.add(key) display_date, sort_date = normalise_date(date_raw) events.append( FactEvent( event_id="", date=display_date, sort_date=sort_date, actor=resolve_actor(fact), fact=re.sub(r"\s+", " ", fact).strip()[:1000], source_doc=source_doc, source_role=source_role, tags=tag_text(fact), confidence=0.82 if source_doc == "CHRONOLOGY" else 0.72, ) ) events.sort(key=lambda event: (event.sort_date or "9999-99-99", event.source_doc, event.fact)) for index, event in enumerate(events, 1): event.event_id = f"EVT-{index:03d}" return events def _excerpt(text: str, needle: str, radius: int = 160) -> str: index = text.lower().find(needle.lower()) if index < 0: return "" value = text[max(0, index - radius): index + len(needle) + radius] return re.sub(r"\s+", " ", value).strip() def find_contradictions(documents: list[SourceDocument]) -> list[Contradiction]: terms = { "warranty": ["warranty"], "refund or return": ["refund", "return", "reject"], "product defect": ["defect", "faulty", "brake"], "account access": ["account closed", "frozen", "blocked"], "reasonable adjustment": ["reasonable adjustment", "disability"], "evidence handling": ["missing evidence", "ignored evidence", "omitted"], } contradictions: list[Contradiction] = [] seen: set[tuple[str, str, str]] = set() for left_index, left in enumerate(documents): for right in documents[left_index + 1:]: for topic, needles in terms.items(): left_needle = next((n for n in needles if n in left.extracted_text.lower()), "") right_needle = next((n for n in needles if n in right.extracted_text.lower()), "") if not left_needle or not right_needle: continue left_excerpt = _excerpt(left.extracted_text, left_needle) right_excerpt = _excerpt(right.extracted_text, right_needle) left_words = set(re.findall(r"\w+", left_excerpt.lower())) right_words = set(re.findall(r"\w+", right_excerpt.lower())) overlap = len(left_words & right_words) / max(1, len(left_words | right_words)) negation_mismatch = any( marker in left_excerpt.lower() and marker not in right_excerpt.lower() or marker in right_excerpt.lower() and marker not in left_excerpt.lower() for marker in (" not ", " never ", " no ", " deny", "refus") ) if overlap > 0.82 and not negation_mismatch: continue key = (left.doc_id, right.doc_id, topic) if key in seen: continue seen.add(key) contradictions.append( Contradiction( contradiction_id=f"CONTR-{len(contradictions) + 1:03d}", kind=f"Potentially conflicting accounts: {topic}", left_doc=left.doc_id, right_doc=right.doc_id, issue="The documents discuss the same topic differently. Human review is required.", left_excerpt=left_excerpt[:500], right_excerpt=right_excerpt[:500], severity="high" if negation_mismatch else "medium", confidence=0.76 if negation_mismatch else 0.58, ) ) return contradictions def find_omissions(documents: list[SourceDocument]) -> list[OmissionFlag]: checks = { "admission or acknowledgement": ["admitted", "confirmed", "acknowledged"], "disability or vulnerability": ["disability", "pip", "reasonable adjustment", "vulnerab"], "hardship": ["hardship", "eviction", "arrears", "financial difficulty"], "product safety or compliance": ["illegal", "not compliant", "unsafe", "defect", "brake"], "burden or standard of proof": ["burden", "onus", "standard of proof"], "data protection": ["subject access", "sar", "gdpr", "personal data"], } all_text = " ".join(doc.extracted_text.lower() for doc in documents) omissions: list[OmissionFlag] = [] decision_roles = {"adjudicator", "respondent_bank", "respondent_solicitor"} for category, keywords in checks.items(): if not any(keyword in all_text for keyword in keywords): continue for doc in documents: if doc.role not in decision_roles: continue if not any(keyword in doc.extracted_text.lower() for keyword in keywords): omissions.append( OmissionFlag( doc=doc.doc_id, category=category, reason=f"Material about {category} appears elsewhere but was not detected in this document.", confidence=0.68, ) ) return omissions def legal_significance(tags: list[str]) -> str: mapping = { "vulnerability": "Potential Equality Act / vulnerability issue; verify duty, knowledge, and evidence.", "consumer_rights": "Potential Consumer Rights Act issue; verify trader, timing, defect, and remedy.", "account_interference": "Potential banking complaint issue; verify account terms, notice, and FCA/FOS rules.", "evidence_failure": "Potential evidence-handling issue; preserve originals and compare the decision trail.", "regulatory": "Potential regulatory route; verify jurisdiction, deadline, and the current rule text.", "admission": "Potential admission or acknowledgement; retain the complete document and context.", "loss_or_harm": "Potential loss item; document amount, date, causation, and mitigation.", } return " ".join(mapping[tag] for tag in tags if tag in mapping) or "Review for relevance and corroboration." def build_routes(documents: list[SourceDocument]) -> list[str]: routes: list[str] = [] for title, (keywords, recommendation) in LEGAL_ROUTES.items(): hits: list[str] = [] for doc in documents: for sentence in _sentences(doc.extracted_text): if any(keyword in sentence.lower() for keyword in keywords): hits.append(f"[{doc.doc_id}] {sentence[:260]}") break if len(hits) == 3: break if hits: routes.append(f"{title}\n{recommendation}\n" + "\n".join(hits)) return routes or ["No specific route was detected. Review the source documents manually."] def score_case(events: list[FactEvent], contradictions: list[Contradiction], omissions: list[OmissionFlag]) -> dict[str, Any]: tag_counts = {tag: 0 for tag in TAG_RULES} corroborated_sources = set() for event in events: corroborated_sources.add(event.source_doc) for tag in event.tags: if tag in tag_counts: tag_counts[tag] += 1 evidence_points = min(35, len(events) * 2) source_points = min(25, len(corroborated_sources) * 5) issue_points = min(20, sum(1 for value in tag_counts.values() if value)) review_points = min(20, len(contradictions) * 2 + len(omissions)) completeness = min(100, evidence_points + source_points + issue_points + review_points) return { "completeness_score": completeness, "event_count": len(events), "source_count": len(corroborated_sources), "tag_counts": tag_counts, "contradiction_count": len(contradictions), "omission_count": len(omissions), "note": ( "This is an evidence-pack completeness heuristic, not a prediction of legal " "success or damages." ), } def analyze(owner: str, documents: list[SourceDocument], chronology: str = "") -> AnalysisResult: documents = [doc for doc in documents if doc.extracted_text.strip()] events = extract_events(documents, chronology) contradictions = find_contradictions(documents) omissions = find_omissions(documents) matrix: list[EvidenceRow] = [] for index, event in enumerate(events, 1): contradiction = next( (item.contradiction_id for item in contradictions if event.source_doc in (item.left_doc, item.right_doc)), "", ) omission = next((item.category for item in omissions if item.doc == event.source_doc), "") matrix.append( EvidenceRow( item_id=f"EVD-{index:03d}", source_doc=event.source_doc, source_role=event.source_role, date=event.date, actor=event.actor, fact=event.fact, tags=event.tags, legal_significance=legal_significance(event.tags), contradiction_link=contradiction, omission_flag=omission, confidence=event.confidence, ) ) warnings = [ "Automated findings are leads for human verification, not factual or legal conclusions.", "Check current law, limitation dates, jurisdiction, and source context before relying on the report.", ] if not events: warnings.append("No dated events were detected. Add a line such as '20 Jan 2025: event description'.") result = AnalysisResult( release_id=make_release_id(), owner=owner.strip(), documents=documents, events=events, contradictions=contradictions, omissions=omissions, evidence_matrix=matrix, routes=build_routes(documents), warnings=warnings, ) result.score = score_case(events, contradictions, omissions) return result def build_text_report(result: AnalysisResult) -> str: score = result.score lines = [ "=" * 78, APP_TITLE.upper(), f"Version: {APP_VERSION}", f"Reference: {result.release_id}", f"Generated: {datetime.now(timezone.utc).strftime('%d %B %Y, %H:%M UTC')}", f"Owner / claimant: {result.owner or 'Not specified'}", "=" * 78, "", "EXECUTIVE DASHBOARD", "-" * 78, f"Documents: {len(result.documents)}", f"Events: {len(result.events)}", f"Potential contradictions: {len(result.contradictions)}", f"Potential omissions: {len(result.omissions)}", f"Evidence-pack completeness: {score.get('completeness_score', 0)}/100", score.get("note", ""), "", "SOURCE DOCUMENTS", "-" * 78, ] lines.extend( f"{doc.doc_id} | {doc.name} | {doc.doc_type} | role={doc.role} | status={doc.extraction_status}" for doc in result.documents ) lines += ["", "CHRONOLOGY", "-" * 78] for event in result.events: lines += [ f"{event.event_id} | {event.date or 'Undated'} | {event.actor} | {event.source_doc}", event.fact, f"Tags: {', '.join(event.tags)} | Confidence: {event.confidence:.0%}", "", ] lines += ["POTENTIAL CONTRADICTIONS", "-" * 78] if not result.contradictions: lines.append("None detected.") for item in result.contradictions: lines += [ f"{item.contradiction_id} | {item.severity.upper()} | {item.kind}", f"{item.left_doc}: {item.left_excerpt}", f"{item.right_doc}: {item.right_excerpt}", f"Review note: {item.issue}", "", ] lines += ["POTENTIAL OMISSIONS", "-" * 78] if not result.omissions: lines.append("None detected.") for item in result.omissions: lines.append(f"{item.doc} | {item.category} | {item.reason}") lines += ["", "LEGAL / REGULATORY ROUTES", "-" * 78] for route in result.routes: lines += [route, ""] lines += ["EVIDENCE MATRIX", "-" * 78] for row in result.evidence_matrix: lines += [ f"{row.item_id} | {row.date or 'Undated'} | {row.actor} | {row.source_doc}", f"Fact: {row.fact}", f"Significance: {row.legal_significance}", f"Links: contradiction={row.contradiction_link or 'none'}; omission={row.omission_flag or 'none'}", f"Next action: {row.next_action}", "", ] lines += ["WARNINGS", "-" * 78] lines.extend(f"- {warning}" for warning in result.warnings) return "\n".join(lines).strip() + "\n" def format_panels(result: AnalysisResult) -> dict[str, str]: events = "\n\n".join( f"{event.event_id} | {event.date or 'Undated'} | {event.actor} | {event.source_doc}\n" f"{event.fact}\nTags: {', '.join(event.tags)}" for event in result.events ) or "No dated events detected." contradictions = "\n\n".join( f"{item.contradiction_id} [{item.severity.upper()}] {item.kind}\n" f"{item.left_doc}: {item.left_excerpt}\n{item.right_doc}: {item.right_excerpt}\n" f"{item.issue}" for item in result.contradictions ) or "No potential contradictions detected." omissions = "\n".join( f"{item.doc} | {item.category} | {item.reason}" for item in result.omissions ) or "No potential omissions detected." routes = "\n\n".join(result.routes) matrix = "\n\n".join( f"{row.item_id} | {row.date or 'Undated'} | {row.actor} | {row.source_doc}\n" f"Fact: {row.fact}\nSignificance: {row.legal_significance}\n" f"Next: {row.next_action}" for row in result.evidence_matrix ) or "No evidence rows generated." dashboard = ( f"Reference: {result.release_id}\n" f"Documents: {len(result.documents)} | Events: {len(result.events)} | " f"Contradictions: {len(result.contradictions)} | Omissions: {len(result.omissions)}\n" f"Evidence-pack completeness: {result.score.get('completeness_score', 0)}/100\n" f"{result.score.get('note', '')}" ) return { "dashboard": dashboard, "events": events, "contradictions": contradictions, "omissions": omissions, "routes": routes, "matrix": matrix, "report": build_text_report(result), }