dr_drastic_core (13).py

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),
    }