Spaces:
Running
Running
| """Centralny, typowany kontekst firmy i projektu (Faza 1: ugruntowanie danych). | |
| Cel: jedno źródło prawdy przekazywane do KAŻDEGO agenta generującego, renderowane | |
| strukturalnie (nie jako surowy zrzut JSON obcinany do N znaków). Jawnie rozróżnia | |
| dane ZNANE od LUK, żeby generator wiedział, gdzie wolno pisać treść, a gdzie musi | |
| wstawić placeholder do weryfikacji. | |
| Moduł jest czysty (bez DB / sieci) — przyjmuje obiekt projektu (ORM) lub dict, | |
| dzięki czemu jest w pełni testowalny. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass, field | |
| from typing import Any | |
| def _as_list(value: Any) -> list[str]: | |
| if value is None: | |
| return [] | |
| if isinstance(value, str): | |
| return [value] if value.strip() else [] | |
| if isinstance(value, (list, tuple, set)): | |
| return [str(v).strip() for v in value if str(v).strip()] | |
| return [str(value)] | |
| def _clean(value: Any) -> str: | |
| if value is None: | |
| return "" | |
| text = str(value).strip() | |
| lowered = text.lower() | |
| if lowered in ("", "none", "null", "brak", "brak danych", "n/a", "nan"): | |
| return "" | |
| return text | |
| def _first(*values: Any) -> str: | |
| for value in values: | |
| cleaned = _clean(value) | |
| if cleaned: | |
| return cleaned | |
| return "" | |
| class CompanyContext: | |
| """Ugruntowany profil wnioskodawcy złożony z rejestrów + danych użytkownika.""" | |
| name: str = "" | |
| nip: str = "" | |
| regon: str = "" | |
| krs: str = "" | |
| address: str = "" | |
| voivodeship: str = "" | |
| legal_form: str = "" | |
| size: str = "" | |
| msp_status: str = "" | |
| msp_verified: bool = False | |
| pkd_codes: list[str] = field(default_factory=list) | |
| revenue: str = "" | |
| employment: str = "" | |
| de_minimis_total_eur: str = "" | |
| de_minimis_risk: str = "" | |
| sudop_configured: bool = False | |
| krs_shareholders: int = 0 | |
| krs_board: int = 0 | |
| web_intel_summary: str = "" | |
| gaps: list[str] = field(default_factory=list) | |
| risk_hints: list[str] = field(default_factory=list) | |
| def is_grounded(self) -> bool: | |
| """Czy mamy minimum danych, by pisać z konkretami, a nie ogólnikami.""" | |
| return bool(self.name and self.nip and self.pkd_codes) | |
| def known_facts(self) -> dict[str, str]: | |
| facts: dict[str, str] = {} | |
| if self.name: | |
| facts["Nazwa wnioskodawcy"] = self.name | |
| if self.nip: | |
| facts["NIP"] = self.nip | |
| if self.regon: | |
| facts["REGON"] = self.regon | |
| if self.krs: | |
| facts["KRS"] = self.krs | |
| if self.address: | |
| facts["Adres siedziby"] = self.address | |
| if self.voivodeship: | |
| facts["Województwo"] = self.voivodeship | |
| if self.legal_form: | |
| facts["Forma prawna"] = self.legal_form | |
| if self.size: | |
| facts["Wielkość firmy"] = self.size | |
| if self.msp_status: | |
| verified = " (zweryfikowany)" if self.msp_verified else " (deklarowany)" | |
| facts["Status MŚP"] = f"{self.msp_status}{verified}" | |
| if self.pkd_codes: | |
| facts["Kody PKD"] = ", ".join(self.pkd_codes[:8]) | |
| if self.revenue: | |
| facts["Przychód roczny"] = self.revenue | |
| if self.employment: | |
| facts["Zatrudnienie"] = self.employment | |
| if self.de_minimis_total_eur: | |
| facts["Pomoc de minimis (suma EUR)"] = self.de_minimis_total_eur | |
| if self.de_minimis_risk: | |
| facts["Ryzyko de minimis"] = self.de_minimis_risk | |
| if self.krs_shareholders or self.krs_board: | |
| facts["Struktura KRS"] = ( | |
| f"wspólnicy: {self.krs_shareholders}, zarząd: {self.krs_board}" | |
| ) | |
| return facts | |
| class ProjectContext: | |
| """Ujednolicony kontekst projektu wraz z profilem wnioskodawcy.""" | |
| project_id: str = "" | |
| title: str = "" | |
| description: str = "" | |
| program_name: str = "" | |
| program_type: str = "" | |
| estimated_value: str = "" | |
| company: CompanyContext = field(default_factory=CompanyContext) | |
| def to_facts_dict(self) -> dict[str, str]: | |
| """Serializowalna mapa zweryfikowanych faktów (label -> wartość). | |
| Używana do przeniesienia ugruntowanych danych wnioskodawcy przez stan | |
| LangGraph (JSON-serializowalny słownik, bez żywego dataclass), tak aby | |
| Auto-Fill Agent mógł deterministycznie uzupełniać placeholdery. | |
| """ | |
| return dict(self.company.known_facts()) | |
| def to_prompt_block(self, max_chars: int = 6000) -> str: | |
| """Strukturalny blok do wstrzyknięcia do promptu (nie surowy JSON).""" | |
| lines: list[str] = ["=== KONTEKST PROJEKTU (jedyne źródło prawdy) ==="] | |
| if self.title: | |
| lines.append(f"Tytuł projektu: {self.title}") | |
| program = _first(self.program_name, self.program_type) | |
| if program: | |
| lines.append(f"Program dotacyjny: {program}") | |
| if self.estimated_value: | |
| lines.append(f"Szacowana wartość: {self.estimated_value}") | |
| if self.description: | |
| lines.append(f"Opis / cel projektu: {self.description[:1500]}") | |
| facts = self.company.known_facts() | |
| if facts: | |
| lines.append("\n--- DANE WNIOSKODAWCY (ZWERYFIKOWANE — używaj wprost) ---") | |
| for label, value in facts.items(): | |
| lines.append(f"- {label}: {value}") | |
| if self.company.web_intel_summary: | |
| lines.append( | |
| f"\n--- WYWIAD (web) ---\n{self.company.web_intel_summary[:800]}" | |
| ) | |
| gaps = self.company.gaps | |
| if gaps: | |
| lines.append( | |
| "\n--- LUKI DANYCH (NIE wymyślaj — użyj [DO WERYFIKACJI: ...]) ---" | |
| ) | |
| for gap in gaps: | |
| lines.append(f"- {gap}") | |
| if self.company.risk_hints: | |
| lines.append("\n--- SYGNAŁY RYZYKA ---") | |
| for hint in self.company.risk_hints: | |
| lines.append(f"- {hint}") | |
| lines.append( | |
| "\nZASADA UGRUNTOWANIA: Dane wnioskodawcy powyżej są autorytatywne — " | |
| "nie pytaj o nie ponownie i nie zmieniaj ich. Dla pól z listy LUK wstaw " | |
| "wyraźny placeholder [DO WERYFIKACJI: opis]. Zakaz wymyślania kwot, dat, " | |
| "nazw i wskaźników, których nie ma w tym kontekście." | |
| ) | |
| block = "\n".join(lines) | |
| if len(block) > max_chars: | |
| block = block[:max_chars].rstrip() + "\n[... kontekst skrócony ...]" | |
| return block | |
| def _extract_company(external_context: dict[str, Any]) -> CompanyContext: | |
| ec = external_context or {} | |
| company_data = ec.get("company_data") | |
| if not isinstance(company_data, dict): | |
| company_data = {} | |
| financials = company_data.get("financials") | |
| if not isinstance(financials, dict): | |
| financials = {} | |
| sudop = company_data.get("sudop") | |
| if not isinstance(sudop, dict): | |
| sudop = ec.get("sudop") if isinstance(ec.get("sudop"), dict) else {} | |
| krs_graph = company_data.get("krs_graph") | |
| if not isinstance(krs_graph, dict): | |
| krs_graph = {} | |
| pkd = company_data.get("pkd_codes") or company_data.get("pkd") | |
| pkd_codes = [ | |
| code | |
| for code in _as_list(pkd) | |
| if code and "brak pkd" not in code.lower() | |
| ] | |
| revenue = _first( | |
| financials.get("revenue"), | |
| company_data.get("revenue"), | |
| ) | |
| if revenue in ("0", "0.0"): | |
| revenue = "" | |
| employment = _first( | |
| financials.get("employment"), | |
| company_data.get("employment"), | |
| ) | |
| if employment in ("0", "0.0"): | |
| employment = "" | |
| de_minimis_total = _first(sudop.get("de_minimis_total_eur")) | |
| if de_minimis_total in ("0", "0.0"): | |
| de_minimis_total = "" | |
| gaps = _as_list(ec.get("gaps") or ec.get("data_gaps")) | |
| risk_hints = _as_list(ec.get("risk_hints")) | |
| return CompanyContext( | |
| name=_first(company_data.get("name"), ec.get("company_name")), | |
| nip=_first(company_data.get("nip"), ec.get("nip")), | |
| regon=_first(company_data.get("regon")), | |
| krs=_first(company_data.get("krs")), | |
| address=_first(company_data.get("address")), | |
| voivodeship=_first( | |
| company_data.get("voivodeship"), company_data.get("region") | |
| ), | |
| legal_form=_first(company_data.get("legal_form")), | |
| size=_first(company_data.get("size")), | |
| msp_status=_first(company_data.get("msp_status")), | |
| msp_verified=bool(company_data.get("msp_verified")), | |
| pkd_codes=pkd_codes, | |
| revenue=revenue, | |
| employment=employment, | |
| de_minimis_total_eur=de_minimis_total, | |
| de_minimis_risk=_first(company_data.get("de_minimis_risk")), | |
| sudop_configured=bool(sudop.get("configured")), | |
| krs_shareholders=int(krs_graph.get("wspolnicy") or 0), | |
| krs_board=int(krs_graph.get("zarzad") or 0), | |
| web_intel_summary=_first(company_data.get("web_intel_summary")), | |
| gaps=gaps, | |
| risk_hints=risk_hints, | |
| ) | |
| def _profile_nip(profile: Any) -> str: | |
| if profile is None: | |
| return "" | |
| if isinstance(profile, dict): | |
| return _clean(profile.get("nip")) | |
| return _clean(getattr(profile, "nip", None)) | |
| def select_profile_for_project(company_nip: str, profiles: Any) -> Any: | |
| """Wybiera profil firmy pasujący do NIP projektu (WIELE profili per użytkownik). | |
| ``profiles`` może być pojedynczym profilem (dict/ORM) lub listą. Preferowany jest | |
| profil o NIP zgodnym z projektem; w razie braku dopasowania zwracany jest pierwszy | |
| dostępny profil (zachowanie zgodne wstecz). | |
| """ | |
| if profiles is None: | |
| return None | |
| if not isinstance(profiles, (list, tuple)): | |
| # Pojedynczy profil — dopasuj po NIP, jeśli podano. | |
| if company_nip and _profile_nip(profiles) and _profile_nip(profiles) != _clean(company_nip): | |
| return None | |
| return profiles | |
| profile_list = list(profiles) | |
| if not profile_list: | |
| return None | |
| target = _clean(company_nip) | |
| if target: | |
| for prof in profile_list: | |
| if _profile_nip(prof) == target: | |
| return prof | |
| return profile_list[0] | |
| def _merge_company_profile(company: CompanyContext, profile: Any) -> CompanyContext: | |
| """ | |
| Uzupełnia CompanyContext (zbudowany z external_context) danymi z twardego | |
| magazynu `company_profiles` (FAZA 2). external_context pozostaje źródłem | |
| pierwszorzędnym; profil wypełnia wyłącznie brakujące pola (backward-compatible). | |
| Gdy ``profile`` jest listą (wiele firm per użytkownik) — dobierany jest profil | |
| pasujący do NIP projektu (``select_profile_for_project``). | |
| """ | |
| if profile is None: | |
| return company | |
| if isinstance(profile, (list, tuple)): | |
| profile = select_profile_for_project(company.nip, profile) | |
| if profile is None: | |
| return company | |
| def _get(attr: str) -> Any: | |
| if isinstance(profile, dict): | |
| return profile.get(attr) | |
| return getattr(profile, attr, None) | |
| # Jeśli profil dotyczy innej firmy (inny NIP niż projekt) — nie mieszaj danych. | |
| prof_nip = _clean(_get("nip")) | |
| if company.nip and prof_nip and prof_nip != company.nip: | |
| return company | |
| # P4#21: ZWERYFIKOWANY profil (last_verified_at) jest ŹRÓDŁEM PRAWDY dla pól | |
| # rejestrowych — ma priorytet nad danymi z external_context (które mogą być | |
| # nieaktualne, bo edycja profilu nie propaguje wstecznie do starych projektów). | |
| # Profil niezweryfikowany uzupełnia jedynie brakujące pola (backward-compatible). | |
| verified = bool(_get("last_verified_at")) | |
| def _apply(field_value: str, profile_value: Any) -> str: | |
| pv = _clean(profile_value) | |
| if verified and pv: | |
| return pv | |
| return field_value or pv | |
| company.name = _apply(company.name, _get("name")) | |
| company.nip = _apply(company.nip, _get("nip")) | |
| company.regon = _apply(company.regon, _get("regon")) | |
| company.krs = _apply(company.krs, _get("krs")) | |
| company.voivodeship = _apply(company.voivodeship, _get("region")) | |
| company.size = _apply(company.size, _get("size")) | |
| profile_pkd = _as_list(_get("pkd_codes")) | |
| if profile_pkd and (verified or not company.pkd_codes): | |
| company.pkd_codes = profile_pkd | |
| if not company.employment: | |
| employees = _get("employees") | |
| if employees: | |
| company.employment = str(employees) | |
| if not company.revenue: | |
| turnover = _get("turnover") | |
| if turnover: | |
| company.revenue = str(turnover) | |
| # Znacznik zweryfikowania z rejestrów (jeśli profil był weryfikowany) | |
| if not company.msp_verified and verified: | |
| company.msp_verified = True | |
| return company | |
| def build_project_context(project: Any, company_profile: Any = None) -> ProjectContext: | |
| """Buduje ProjectContext z obiektu ORM Project lub ze zwykłego dict. | |
| Opcjonalny ``company_profile`` (ORM CompanyProfile lub dict) uzupełnia dane | |
| wnioskodawcy z twardego magazynu ``company_profiles`` (FAZA 2). | |
| """ | |
| if isinstance(project, dict): | |
| external_context = project.get("external_context") or {} | |
| title = project.get("title", "") | |
| description = project.get("description", "") | |
| program_name = project.get("program_name", "") | |
| program_type = project.get("program_type", "") | |
| estimated_value = project.get("estimated_value", "") | |
| project_id = project.get("id", "") | |
| else: | |
| external_context = getattr(project, "external_context", None) or {} | |
| title = getattr(project, "title", "") or "" | |
| description = getattr(project, "description", "") or "" | |
| program_name = getattr(project, "program_name", "") or "" | |
| program_type = getattr(project, "program_type", "") or "" | |
| estimated_value = getattr(project, "estimated_value", "") or "" | |
| project_id = getattr(project, "id", "") or "" | |
| company = _extract_company(external_context) | |
| company = _merge_company_profile(company, company_profile) | |
| return ProjectContext( | |
| project_id=str(project_id or ""), | |
| title=_clean(title), | |
| description=_clean(description), | |
| program_name=_clean(program_name), | |
| program_type=_clean(program_type), | |
| estimated_value=_clean(estimated_value), | |
| company=company, | |
| ) | |