Spaces:
Sleeping
Sleeping
| """ | |
| Pre-generation data completeness gate. | |
| Before full autopilot: ready_to_generate | blocked_missing_fields | | |
| blocked_dossier | structure_only. | |
| """ | |
| from __future__ import annotations | |
| from typing import Any, Dict, List, Optional, Tuple | |
| STATUS_READY = "ready_to_generate" | |
| STATUS_MISSING = "blocked_missing_fields" | |
| STATUS_DOSSIER = "blocked_dossier" | |
| STATUS_STRUCTURE = "structure_only" | |
| STATUS_BLOCKED_CONSENT = "blocked_consent" | |
| def _get_nested(d: Dict[str, Any], *keys: str) -> Any: | |
| cur: Any = d | |
| for k in keys: | |
| if not isinstance(cur, dict): | |
| return None | |
| cur = cur.get(k) | |
| return cur | |
| def _has_value(v: Any) -> bool: | |
| if v is None: | |
| return False | |
| if isinstance(v, str): | |
| s = v.strip() | |
| if not s or s.startswith("[DO WERYFIKACJI") or s.lower() in ("brak", "n/a", "tbd"): | |
| return False | |
| return True | |
| if isinstance(v, (list, dict)): | |
| return len(v) > 0 | |
| return True | |
| def extract_company_field_map(external_context: Optional[Dict[str, Any]]) -> Dict[str, Any]: | |
| ext = dict(external_context or {}) | |
| company = ext.get("company_data") if isinstance(ext.get("company_data"), dict) else {} | |
| return { | |
| "nip": company.get("nip") or ext.get("nip"), | |
| "company_name": company.get("name") or company.get("company_name") or ext.get("company_name"), | |
| "pkd_codes": company.get("pkd") | |
| or company.get("pkd_codes") | |
| or company.get("pkd_list") | |
| or ext.get("pkd_codes"), | |
| "msp_status": company.get("msp_status") | |
| or company.get("sme_status") | |
| or ext.get("msp_status") | |
| or _get_nested(ext, "msp_analysis", "status"), | |
| "de_minimis": company.get("de_minimis") | |
| or ext.get("de_minimis") | |
| or company.get("de_minimis_eur"), | |
| "closed_financial_year": company.get("closed_financial_year") | |
| or ext.get("closed_financial_year") | |
| or company.get("financial_year_closed"), | |
| "revenue_3y": company.get("revenue_3y") | |
| or company.get("revenues") | |
| or ext.get("revenue_3y"), | |
| "employment_fte": company.get("employment_fte") | |
| or company.get("fte") | |
| or company.get("employees") | |
| or ext.get("employment_fte"), | |
| "address": company.get("address") or company.get("adres"), | |
| "regon": company.get("regon") or ext.get("regon"), | |
| } | |
| def extract_project_field_map( | |
| external_context: Optional[Dict[str, Any]], | |
| *, | |
| description: str = "", | |
| title: str = "", | |
| ) -> Dict[str, Any]: | |
| ext = dict(external_context or {}) | |
| proj = ext.get("project_facts") if isinstance(ext.get("project_facts"), dict) else {} | |
| return { | |
| "project_description": description or proj.get("description") or ext.get("project_description"), | |
| "innovation_description": proj.get("innovation") or ext.get("innovation_description"), | |
| "eurogrant_programme": proj.get("eurogrant_programme") or ext.get("eurogrant_programme"), | |
| "eurogrant_call": proj.get("eurogrant_call") or ext.get("eurogrant_call"), | |
| "eurogrant_role": proj.get("eurogrant_role") or ext.get("eurogrant_role"), | |
| "specialist_analysis_yes_no": proj.get("specialist_analysis_yes_no") | |
| or ext.get("specialist_analysis_yes_no"), | |
| "lump_sum_amount_pln": proj.get("lump_sum_amount_pln") or ext.get("lump_sum_amount_pln"), | |
| "title": title or ext.get("title"), | |
| } | |
| _FIELD_LABELS_PL = { | |
| "nip": "NIP firmy", | |
| "company_name": "Nazwa firmy", | |
| "pkd_codes": "Kody PKD (lista)", | |
| "msp_status": "Status MŚP", | |
| "de_minimis": "Pomoc de minimis (wykorzystana)", | |
| "closed_financial_year": "Potwierdzenie zamkniętego roku obrotowego (≥12 mies.)", | |
| "revenue_3y": "Przychody za ostatnie 3 lata", | |
| "employment_fte": "Zatrudnienie FTE", | |
| "project_description": "Opis projektu", | |
| "innovation_description": "Opis innowacji / TRL", | |
| "eurogrant_programme": "Program UE (target Eurograntu)", | |
| "eurogrant_call": "Konkurs / call / Work Programme", | |
| "eurogrant_role": "Rola (samodzielny / koordynator / członek)", | |
| "specialist_analysis_yes_no": "Czy analizy specjalistyczne (tak/nie + uzasadnienie)", | |
| "lump_sum_amount_pln": "Kwota ryczałtowa (PLN)", | |
| } | |
| def evaluate_data_completeness( | |
| *, | |
| external_context: Optional[Dict[str, Any]] = None, | |
| instrument_schema: Optional[Dict[str, Any]] = None, | |
| description: str = "", | |
| title: str = "", | |
| grounding_mode: str = "", | |
| dossier_level: str = "", | |
| ) -> Dict[str, Any]: | |
| """ | |
| Returns readiness status + missing field descriptors for HITL UI. | |
| """ | |
| ext = dict(external_context or {}) | |
| schema = instrument_schema or ext.get("instrument_schema") or {} | |
| mode = (grounding_mode or ext.get("grounding_mode") or "").lower() | |
| level = ( | |
| dossier_level | |
| or (ext.get("program_dossier") or {}).get("readiness", {}).get("level") | |
| or ext.get("dossier_readiness") | |
| or "" | |
| ) | |
| level = str(level).lower() | |
| if mode == "structure_only": | |
| # Structure-only consent allows *structural* generation without a regulation | |
| # pack — but Full Autopilot must still pass the eligibility spine (MŚP / | |
| # de minimis). Early-return used to skip apply_eligibility_to_completeness. | |
| result = { | |
| "status": STATUS_STRUCTURE, | |
| "ready_to_generate": True, # structural only — allowed with consent | |
| "full_autopilot_allowed": True, # may be cleared by spine below | |
| "missing_fields": [], | |
| "warnings": [ | |
| "Tryb structuralny: generacja bez ugruntowania w regulaminie — nie składać bez weryfikacji." | |
| ], | |
| "instrument_family": schema.get("family"), | |
| "dossier_level": level or "blind", | |
| } | |
| try: | |
| from core.eligibility.spine import ( | |
| apply_eligibility_to_completeness, | |
| build_eligibility_report, | |
| ) | |
| company = extract_company_field_map(ext) | |
| report = build_eligibility_report( | |
| nip=str(company.get("nip") or ""), | |
| company_data=ext.get("company_data") | |
| if isinstance(ext.get("company_data"), dict) | |
| else {}, | |
| external_context=ext, | |
| ) | |
| result = apply_eligibility_to_completeness(result, report) | |
| # Preserve structure_only generation consent; spine still owns autopilot. | |
| result["ready_to_generate"] = True | |
| if result.get("status") != STATUS_STRUCTURE: | |
| result["status"] = STATUS_STRUCTURE | |
| warns = list(result.get("warnings") or []) | |
| structural_warn = ( | |
| "Tryb structuralny: generacja bez ugruntowania w regulaminie — nie składać bez weryfikacji." | |
| ) | |
| if structural_warn not in warns: | |
| warns.insert(0, structural_warn) | |
| result["warnings"] = warns | |
| except Exception: | |
| pass | |
| return result | |
| if mode == "blocked" or (level == "blind" and mode != "structure_only"): | |
| # Dossier blind without consent | |
| from core.projects.generation_consent import ( | |
| resolve_grounding_mode, | |
| GROUNDING_STRUCTURE, | |
| GROUNDING_REGULATION, | |
| ) | |
| gm = resolve_grounding_mode(ext) | |
| if gm == GROUNDING_STRUCTURE: | |
| return evaluate_data_completeness( | |
| external_context={**ext, "grounding_mode": "structure_only"}, | |
| instrument_schema=schema, | |
| description=description, | |
| title=title, | |
| grounding_mode="structure_only", | |
| dossier_level=level, | |
| ) | |
| if gm != GROUNDING_REGULATION and level == "blind": | |
| return { | |
| "status": STATUS_DOSSIER, | |
| "ready_to_generate": False, | |
| "full_autopilot_allowed": False, | |
| "missing_fields": [ | |
| { | |
| "field": "regulation_pack", | |
| "label": "Regulamin / paczka dokumentów naboru", | |
| "category": "dossier", | |
| "hint": "Dołącz URL/PDF regulaminu albo jawnie zaakceptuj tryb strukturalny.", | |
| } | |
| ], | |
| "warnings": ["Dossier blind — brak reguł naboru w systemie."], | |
| "instrument_family": schema.get("family"), | |
| "dossier_level": level or "blind", | |
| } | |
| company = extract_company_field_map(ext) | |
| project = extract_project_field_map(ext, description=description, title=title) | |
| required_company = list(schema.get("required_company_fields") or ["nip", "company_name"]) | |
| required_project = list(schema.get("required_project_fields") or ["project_description"]) | |
| missing: List[Dict[str, str]] = [] | |
| for f in required_company: | |
| if not _has_value(company.get(f)): | |
| missing.append( | |
| { | |
| "field": f, | |
| "label": _FIELD_LABELS_PL.get(f, f), | |
| "category": "company", | |
| "hint": "Uzupełnij w profilu firmy lub wgraj dokumenty (CEIDG, sprawozdania).", | |
| } | |
| ) | |
| for f in required_project: | |
| if not _has_value(project.get(f)): | |
| missing.append( | |
| { | |
| "field": f, | |
| "label": _FIELD_LABELS_PL.get(f, f), | |
| "category": "project", | |
| "hint": "Uzupełnij w opisie projektu / faktach projektu (project_facts).", | |
| } | |
| ) | |
| # Minimum always: NIP + name for full autopilot quality | |
| if not _has_value(company.get("nip")): | |
| if not any(m["field"] == "nip" for m in missing): | |
| missing.append( | |
| { | |
| "field": "nip", | |
| "label": _FIELD_LABELS_PL["nip"], | |
| "category": "company", | |
| "hint": "Wymagane do full autopilot.", | |
| } | |
| ) | |
| if missing: | |
| result = { | |
| "status": STATUS_MISSING, | |
| "ready_to_generate": False, | |
| "full_autopilot_allowed": False, | |
| "missing_fields": missing, | |
| "warnings": [], | |
| "instrument_family": schema.get("family"), | |
| "dossier_level": level or "unknown", | |
| "present_company": {k: _has_value(v) for k, v in company.items()}, | |
| "present_project": {k: _has_value(v) for k, v in project.items()}, | |
| } | |
| else: | |
| result = { | |
| "status": STATUS_READY, | |
| "ready_to_generate": True, | |
| "full_autopilot_allowed": True, | |
| "missing_fields": [], | |
| "warnings": [], | |
| "instrument_family": schema.get("family"), | |
| "dossier_level": level or "unknown", | |
| "present_company": {k: _has_value(v) for k, v in company.items()}, | |
| "present_project": {k: _has_value(v) for k, v in project.items()}, | |
| } | |
| # Eligibility spine 2026 (MŚP + de minimis) — product gate when flag on | |
| try: | |
| from core.eligibility.spine import ( | |
| apply_eligibility_to_completeness, | |
| build_eligibility_report, | |
| ) | |
| report = build_eligibility_report( | |
| nip=str(company.get("nip") or ""), | |
| company_data=ext.get("company_data") if isinstance(ext.get("company_data"), dict) else {}, | |
| external_context=ext, | |
| ) | |
| result = apply_eligibility_to_completeness(result, report) | |
| except Exception: | |
| pass | |
| return result | |
| def apply_project_facts( | |
| external_context: Optional[Dict[str, Any]], | |
| facts: Dict[str, Any], | |
| ) -> Dict[str, Any]: | |
| """Merge user-provided facts into external_context.project_facts + top-level mirrors.""" | |
| ext = dict(external_context or {}) | |
| pf = dict(ext.get("project_facts") or {}) if isinstance(ext.get("project_facts"), dict) else {} | |
| for k, v in (facts or {}).items(): | |
| if v is None: | |
| continue | |
| pf[k] = v | |
| # mirror common fields to top-level for extractors | |
| if k in ( | |
| "eurogrant_programme", | |
| "eurogrant_call", | |
| "eurogrant_role", | |
| "specialist_analysis_yes_no", | |
| "lump_sum_amount_pln", | |
| ): | |
| ext[k] = v | |
| ext["project_facts"] = pf | |
| # optional company patch | |
| company_patch = facts.get("company") if isinstance(facts.get("company"), dict) else None | |
| if company_patch: | |
| cd = dict(ext.get("company_data") or {}) if isinstance(ext.get("company_data"), dict) else {} | |
| cd.update({k: v for k, v in company_patch.items() if v is not None}) | |
| ext["company_data"] = cd | |
| return ext | |