""" World-class regulation-grounded advisor (pure evaluation path). Checks application content against a structured advisor brief without requiring a live LLM — suitable for unit tests and as a hard gate before soft-pass export. """ from __future__ import annotations import re from dataclasses import asdict, dataclass, field from typing import Any, Dict, List, Optional, Sequence @dataclass class AdvisorFinding: code: str severity: str # critical | major | minor | info message: str category: str # program_alignment | missing_required | quality | grounding section: str = "" blocking: bool = False def to_dict(self) -> Dict[str, Any]: return asdict(self) @dataclass class AdvisorReport: passed: bool score: int grounding_mode: str regulation_grounded_pass: bool findings: List[AdvisorFinding] = field(default_factory=list) blockers: List[str] = field(default_factory=list) covered_sections: List[str] = field(default_factory=list) missing_sections: List[str] = field(default_factory=list) missing_attachments_mentioned: List[str] = field(default_factory=list) attention_addressed: List[str] = field(default_factory=list) attention_open: List[str] = field(default_factory=list) brief_usable: bool = False summary: str = "" def to_dict(self) -> Dict[str, Any]: d = asdict(self) d["findings"] = [f if isinstance(f, dict) else f.to_dict() for f in self.findings] return d def _norm(s: str) -> str: return re.sub(r"\s+", " ", (s or "").lower().strip()) # Generic tokens that must NOT count as rule hits (corporate fluff matches these) _RULE_STOPWORDS = frozenset( { "wnioskodawca", "beneficjent", "projekt", "projekty", "musi", "muszą", "powinien", "powinna", "posiadać", "posiada", "status", "oraz", "przez", "który", "która", "które", "zgodnie", "zawierać", "zawiera", "następujące", "elementy", "wymagane", "wymagany", "opis", "treść", "sekcja", "program", "naboru", "regulamin", "sprawdź", "potwierdź", "udokumentuj", "przygotuj", "zweryfikuj", "uwzględnij", "punkt", "uwagi", "minimum", "wynosi", "kosztów", "koszty", "koszt", "zasada", "spełniać", "spełnia", } ) # Domain signals that, if present in brief, must appear in the application _CORE_SIGNAL_GROUPS: List[tuple[str, tuple[str, ...]]] = [ ("mśp", ("mśp", "msp", "mikroprzedsiębior", "małe przedsiębior", "średnie przedsiębior", "mikro firma")), ("dnsh", ("dnsh", "do no significant harm", "significant harm", "wpływ na środowisko", "wpływ środowisk")), ("wkład_własny", ("wkład własny", "wklad wlasny", "finansowanie własne", "finansowanie wlasne")), ("de_minimis", ("de minimis", "pomoc publiczna", "pomocy publicznej")), ("trl", ("trl", "gotowości technologicz", "gotowosci technologicz")), ("niekwalifikowalne", ("niekwalifikow", "koszty niekwalifikowalne")), ] def _blob_from_sections(sections: Optional[Dict[str, str]], document_text: str = "") -> str: parts: List[str] = [] if document_text: parts.append(document_text) if sections: for title, body in sections.items(): parts.append(f"## {title}\n{body or ''}") return "\n".join(parts) def _section_present(required: str, sections: Dict[str, str], blob: str) -> bool: """True if required section has meaningful content under a matching title.""" nr = _norm(required) if not nr: return True # Direct title match with substantial body only (no free-text weak match) for title, body in (sections or {}).items(): nt = _norm(title) if nr in nt or nt in nr or difflib_ratio(nr, nt) >= 0.55: if body and len(body.strip()) >= 80 and "[UZUPEŁNIĆ" not in body: return True return False def difflib_ratio(a: str, b: str) -> float: import difflib return difflib.SequenceMatcher(None, a, b).ratio() def _distinctive_terms_from_rule(rule: str) -> List[str]: """ Extract distinctive multi-token phrases / domain terms from a rule. Drops stopwords so 'Projekt musi…' alone never counts as alignment. """ n = _norm(rule) terms: List[str] = [] # Prefer known domain multi-word / acronyms first for _name, variants in _CORE_SIGNAL_GROUPS: for v in variants: if v in n: terms.append(v) # Multi-word chunks of 2–3 content words words = re.findall(r"[a-ząćęłńóśźż0-9%]{3,}", n) content = [w for w in words if w not in _RULE_STOPWORDS and len(w) >= 4] for i in range(len(content) - 1): bigram = f"{content[i]} {content[i + 1]}" if bigram not in terms: terms.append(bigram) # Long single tokens (≥7) that aren't stopwords — acronyms like mśp already handled for w in content: if len(w) >= 7 and w not in terms and w not in _RULE_STOPWORDS: terms.append(w) return terms[:12] def rule_is_addressed(rule: str, blob: str) -> bool: """True only when distinctive signal(s) from the rule appear in application text.""" b = _norm(blob) if not b or not rule: return False terms = _distinctive_terms_from_rule(rule) if not terms: # No distinctive content in rule → cannot claim hit from fluff return False # Need at least one multi-word term OR two distinct single-domain hits multi = [t for t in terms if " " in t or any(t in g[1] for g in _CORE_SIGNAL_GROUPS)] hits = [t for t in terms if t in b] if not hits: return False if any(t in b for t in multi): return True # Single long distinctive tokens: require ≥2 different hits for generic rules single_hits = [t for t in hits if " " not in t and len(t) >= 7] return len(set(single_hits)) >= 2 or (len(single_hits) >= 1 and any(t in b for t in multi)) def core_signals_required_by_brief(brief: Dict[str, Any]) -> List[str]: """Which core domain signals appear in brief (rules + attention + eligibility).""" blob = " ".join( [ " ".join(str(x) for x in (brief.get("key_rules") or [])), " ".join(str(x) for x in (brief.get("attention_points") or [])), " ".join(str(x) for x in (brief.get("eligibility_signals") or [])), " ".join(str(x) for x in (brief.get("funding_limits") or [])), ] ) n = _norm(blob) required: List[str] = [] for name, variants in _CORE_SIGNAL_GROUPS: if any(v in n for v in variants): required.append(name) return required def core_signal_present(name: str, blob: str) -> bool: b = _norm(blob) for gname, variants in _CORE_SIGNAL_GROUPS: if gname == name: return any(v in b for v in variants) return False def _attention_is_critical(point: str) -> bool: p = _norm(point) critical_markers = ( "dnsh", "środowisk", "srodowisk", "mśp", "msp", "wkład", "własn", "wlasn", "de minimis", "pomoc publiczn", "trl", "niekwalifikow", ) return any(m in p for m in critical_markers) def _attention_addressed(point: str, blob: str) -> bool: p = _norm(point) b = _norm(blob) keywords: List[str] = [] if "dnsh" in p or "środowisk" in p or "srodowisk" in p: keywords = ["dnsh", "do no significant harm", "wpływ na środowisko", "wpływ środowisk", "środowisk", "srodowisk"] elif "mśp" in p or "msp" in p: keywords = ["mśp", "msp", "mikroprzedsiębior", "małe przedsiębior", "średnie przedsiębior"] elif "wkład" in p or "własn" in p or "wlasn" in p: keywords = ["wkład własny", "wklad wlasny", "finansowanie własne", "finansowanie wlasne"] elif "de minimis" in p or "pomoc publiczn" in p: keywords = ["de minimis", "pomoc publiczna", "pomocy publicznej"] elif "trl" in p: keywords = ["trl", "gotowości technologicz", "gotowosci technologicz"] elif "załącznik" in p or "zalacznik" in p: keywords = ["załącznik", "zalacznik", "oświadczenie", "oswiadczenie"] elif "niekwalifikow" in p: keywords = ["niekwalifikow", "koszty niekwalifikowalne"] else: # Require multi-token distinctive match, not lone stopwords keywords = _distinctive_terms_from_rule(point)[:4] if not keywords: return False return any(k in b for k in keywords) def evaluate_application( *, document_text: str = "", sections: Optional[Dict[str, str]] = None, brief: Optional[Dict[str, Any]] = None, grounding_mode: str = "regulation", min_score: int = 70, min_section_chars: int = 80, ) -> AdvisorReport: """ Evaluate application against regulation-derived brief. structure_only / blocked / blind modes never yield regulation_grounded_pass=True. """ brief = brief if isinstance(brief, dict) else {} sections = sections if isinstance(sections, dict) else {} mode = (grounding_mode or "regulation").lower().strip() blob = _blob_from_sections(sections, document_text) findings: List[AdvisorFinding] = [] score = 100 # --- Grounding hard rules --- if mode in ("structure_only", "blocked", "blind"): findings.append( AdvisorFinding( code="GROUNDING_NOT_REGULATION", severity="critical", message=( f"Tryb {mode}: ocena nie może zakończyć się regulation-grounded pass. " "Brak ugruntowania w regulaminie naboru." ), category="grounding", blocking=True, ) ) score -= 40 usable = bool(brief.get("usable")) if "usable" in brief else ( bool(brief.get("key_rules") or brief.get("required_sections") or brief.get("required_attachments")) ) if mode == "regulation" and not usable and not ( brief.get("key_rules") or brief.get("required_sections") ): findings.append( AdvisorFinding( code="BRIEF_EMPTY", severity="critical", message="Brief doradcy pusty — brak reguł/sekcji z regulaminu. Nie można ugruntować oceny.", category="grounding", blocking=True, ) ) score -= 35 # --- Instrument mismatch (Eurogranty vs SMART modules etc.) --- try: from core.projects.instrument_profile import ( detect_instrument_mismatch, resolve_program_type, ) prog_type = str( brief.get("program_type") or (brief.get("instrument_program_type") if isinstance(brief, dict) else "") or "" ) # Allow caller to pass via document_text meta later; also scan section titles mismatch = detect_instrument_mismatch( program_type=prog_type or resolve_program_type(program_name=str(brief.get("name") or "")), document_text=blob, section_titles=list(sections.keys()), ) if mismatch.get("mismatch"): for msg in mismatch.get("findings") or []: findings.append( AdvisorFinding( code="INSTRUMENT_MISMATCH", severity="critical", message=msg, category="program_alignment", blocking=bool(mismatch.get("blocking")), ) ) score -= int(mismatch.get("score_penalty") or 0) except Exception: pass # --- Required sections (program alignment + missing elements) --- required_sections = list(brief.get("required_sections") or []) covered: List[str] = [] missing: List[str] = [] for req in required_sections: if _section_present(req, sections, blob): covered.append(req) else: missing.append(req) findings.append( AdvisorFinding( code="MISSING_REQUIRED_SECTION", severity="critical", message=f"Brak wymaganej sekcji/treści: {req}", category="missing_required", section=req, blocking=True, ) ) score -= 12 # --- Key rules: distinctive multi-token / domain coverage (no fluff hits) --- rules = list(brief.get("key_rules") or []) rules_hit = 0 rules_checked = rules[:12] for rule in rules_checked: if rule_is_addressed(rule, blob): rules_hit += 1 if rules_checked and mode == "regulation": rule_ratio = rules_hit / max(len(rules_checked), 1) if rule_ratio < 0.5: findings.append( AdvisorFinding( code="WEAK_RULE_ALIGNMENT", severity="critical" if rule_ratio < 0.35 else "major", message=( f"Słabe dopasowanie do reguł regulaminu " f"({rules_hit}/{len(rules_checked)} reguł z distinctive signals w treści)." ), category="program_alignment", blocking=rule_ratio < 0.5, ) ) score -= 25 if rule_ratio < 0.35 else 12 elif rule_ratio >= 0.7: score = min(100, score + 5) # --- Core domain signals required by brief must appear in application --- if mode == "regulation": for sig in core_signals_required_by_brief(brief): if not core_signal_present(sig, blob): findings.append( AdvisorFinding( code="MISSING_CORE_SIGNAL", severity="critical", message=f"Brak kluczowego sygnału regulaminu w treści wniosku: {sig}", category="program_alignment", blocking=True, ) ) score -= 15 # --- Attachments mentioned when brief requires them --- missing_att: List[str] = [] for att in list(brief.get("required_attachments") or [])[:10]: na = _norm(att)[:40] if na and na[:12] not in _norm(blob): # Also check generic "załącznik" coverage if "załącznik" not in _norm(blob) and "zalacznik" not in _norm(blob): missing_att.append(att) findings.append( AdvisorFinding( code="ATTACHMENT_NOT_ADDRESSED", severity="major", message=f"Brak odniesienia do wymaganego załącznika: {att}", category="missing_required", blocking=False, ) ) score -= 4 # --- Attention points (critical ones block regulation pass) --- attention = list(brief.get("attention_points") or []) addressed: List[str] = [] open_pts: List[str] = [] for pt in attention: if _attention_addressed(pt, blob): addressed.append(pt) else: open_pts.append(pt) critical = _attention_is_critical(pt) findings.append( AdvisorFinding( code="ATTENTION_OPEN", severity="critical" if critical else "minor", message=f"Punkt uwagi regulaminu niezaadresowany: {pt}", category="quality", blocking=critical, ) ) score -= 12 if critical else 3 # --- Critical quality / readiness (empty/short sections) --- short_sections = 0 for title, body in sections.items(): body = body or "" if len(body.strip()) < min_section_chars or "[UZUPEŁNIĆ" in body: short_sections += 1 findings.append( AdvisorFinding( code="SECTION_TOO_THIN", severity="major", message=f"Sekcja zbyt krótka lub niekompletna: {title}", category="quality", section=title, blocking=len(body.strip()) < 20, ) ) score -= 6 if not sections and len(blob.strip()) < 120: findings.append( AdvisorFinding( code="DOCUMENT_EMPTY", severity="critical", message="Dokument wniosku pusty lub zbyt krótki.", category="quality", blocking=True, ) ) score -= 40 score = max(0, min(100, score)) blockers = [f.message for f in findings if f.blocking] critical = [f for f in findings if f.severity == "critical"] # Explicit: never soft-pass structure_only / blind / blocked as regulation-grounded if mode in ("structure_only", "blocked", "blind"): regulation_grounded_pass = False else: regulation_grounded_pass = ( mode == "regulation" and usable and score >= min_score and not blockers and len(critical) == 0 ) if mode == "regulation": passed = regulation_grounded_pass elif mode == "structure_only": # Structural readiness only — never regulation_grounded_pass passed = ( len(blob.strip()) >= 200 and short_sections == 0 and score >= max(40, min_score - 25) and not any(f.code == "DOCUMENT_EMPTY" for f in findings) ) else: passed = False summary_bits = [ f"score={score}", f"mode={mode}", f"missing_sections={len(missing)}", f"blockers={len(blockers)}", f"regulation_grounded_pass={regulation_grounded_pass}", ] return AdvisorReport( passed=passed, score=score, grounding_mode=mode, regulation_grounded_pass=regulation_grounded_pass, findings=findings, blockers=blockers, covered_sections=covered, missing_sections=missing, missing_attachments_mentioned=missing_att, attention_addressed=addressed, attention_open=open_pts, brief_usable=usable, summary="; ".join(summary_bits), ) def advisor_findings_to_rewrite_targets( report: AdvisorReport | Dict[str, Any], plan_titles: Sequence[str], ) -> Dict[str, List[str]]: """Map advisor findings onto quality_loop section targets.""" from core.generation.quality_loop import section_title_match if isinstance(report, AdvisorReport): findings = report.findings missing = report.missing_sections else: findings = report.get("findings") or [] missing = report.get("missing_sections") or [] targets: Dict[str, List[str]] = {} for req in missing: matched = section_title_match(str(req), plan_titles) note = f"[world_class_advisor] Uzupełnij wymaganą treść regulaminu: {req}" if matched: targets.setdefault(matched, []).append(note) elif plan_titles: targets.setdefault(list(plan_titles)[0], []).append(note) for f in findings: if isinstance(f, AdvisorFinding): code, msg, section, sev = f.code, f.message, f.section, f.severity elif isinstance(f, dict): code = f.get("code") or "FINDING" msg = f.get("message") or "" section = f.get("section") or "" sev = f.get("severity") or "major" else: continue if code in ("GROUNDING_NOT_REGULATION", "BRIEF_EMPTY"): # global note on all titles (limited later by pick) for t in plan_titles: targets.setdefault(t, []).append(f"[world_class_advisor|{sev}] {msg}") break matched = section_title_match(section, plan_titles) if section else None line = f"[world_class_advisor|{sev}|{code}] {msg}" if matched: targets.setdefault(matched, []).append(line) elif plan_titles and sev == "critical": # Global alignment/core-signal issues: attach to budget/alignment-like titles if any, # never blindly rewrite the first healthy section (e.g. full Wstęp). if code in ("WEAK_RULE_ALIGNMENT", "MISSING_CORE_SIGNAL"): domain_keys = ("budżet", "budzet", "finans", "koszt", "opis", "innowacj", "dopasow") hit_any = False for t in plan_titles: nt = _norm(t) if any(k in nt for k in domain_keys): targets.setdefault(t, []).append(line) hit_any = True if not hit_any: # last resort: last plan title (often budget/closing), not first intro targets.setdefault(list(plan_titles)[-1], []).append(line) else: targets.setdefault(list(plan_titles)[0], []).append(line) return targets def evaluate_from_generator_state(state: Dict[str, Any]) -> AdvisorReport: """Convenience: build brief + sections from generator/external_context state.""" ext = state.get("external_context") if isinstance(state.get("external_context"), dict) else {} generated = state.get("generated_sections") if isinstance(state.get("generated_sections"), dict) else {} mode = str(ext.get("grounding_mode") or state.get("grounding_mode") or "regulation").lower() brief = dict(ext.get("advisor_brief") or {}) if isinstance(ext.get("advisor_brief"), dict) else {} if not brief: brief = { "key_rules": list(ext.get("regulation_key_rules") or ext.get("key_rules") or []), "required_sections": list(ext.get("required_sections") or []), "required_attachments": list(ext.get("required_attachments") or []), "attention_points": list(ext.get("attention_points") or []), } brief["usable"] = bool( brief["key_rules"] or brief["required_sections"] or brief["required_attachments"] ) # Instrument family for mismatch detection try: from core.projects.instrument_profile import resolve_program_type brief["program_type"] = resolve_program_type( program_type=str(ext.get("instrument_program_type") or ext.get("program_type") or ""), program_name=str(ext.get("program_name") or ext.get("grant_name") or ""), grant_id=str(ext.get("grant_id") or ""), ) brief["name"] = str(ext.get("program_name") or ext.get("grant_name") or "") except Exception: pass return evaluate_application( sections=generated, document_text=state.get("full_document") or "", brief=brief, grounding_mode=mode, )